Fix the inverted "delete/add" of the routes, in cluster mode.
[l2tpns.git] / l2tpns.c
index d1c1337..ed85a91 100644 (file)
--- a/l2tpns.c
+++ b/l2tpns.c
@@ -102,6 +102,7 @@ uint32_t eth_tx = 0;
 
 static uint32_t ip_pool_size = 1;      // Size of the pool of addresses used for dynamic address allocation.
 time_t time_now = 0;                   // Current time in seconds since epoch.
+uint64_t time_now_ms = 0;              // Current time in milliseconds since epoch.
 static char time_now_string[64] = {0}; // Current time as a string.
 static int time_changed = 0;           // time_now changed
 char main_quit = 0;                    // True if we're in the process of exiting.
@@ -161,6 +162,8 @@ config_descriptt config_values[] = {
        CONFIG("hostname", hostname, STRING),
        CONFIG("nexthop_address", nexthop_address, IPv4),
        CONFIG("nexthop6_address", nexthop6_address, IPv6),
+       CONFIG("echo_timeout", echo_timeout, INT),
+       CONFIG("idle_echo_timeout", idle_echo_timeout, INT),
        { NULL, 0, 0, 0 },
 };
 
@@ -240,6 +243,10 @@ static clockt now(double *f)
            time_now = t.tv_sec;
            time_changed++;
        }
+
+       // Time in milliseconds
+       time_now_ms = (t.tv_sec * 1000) + (t.tv_usec/1000);
+
        return (t.tv_sec - basetime) * 10 + t.tv_usec / 100000 + 1;
 }
 
@@ -1323,7 +1330,8 @@ static void update_session_out_stat(sessionidt s, sessiont *sp, int len)
 
 // process outgoing (to tunnel) IP
 //
-static void processipout(uint8_t *buf, int len)
+// (i.e. this routine writes to data[-8]).
+void processipout(uint8_t *buf, int len)
 {
        sessionidt s;
        sessiont *sp;
@@ -1462,6 +1470,14 @@ static void processipout(uint8_t *buf, int len)
 
        if(session[s].bundle != 0 && bundle[session[s].bundle].num_of_links > 1)
        {
+
+               if (!config->cluster_iam_master)
+               {
+                       // The MPPP packets must be managed by the Master.
+                       master_forward_mppp_packet(s, data, size);
+                       return;
+               }
+
                // Add on L2TP header
                sessionidt members[MAXBUNDLESES];
                bundleidt bid = session[s].bundle;
@@ -1483,7 +1499,7 @@ static void processipout(uint8_t *buf, int len)
 
                if (nb_opened < 1)
                {
-                       LOG(2, s, t, "MPPP: PROCESSIPOUT ERROR, no session opened in bundle:%d\n", bid);
+                       LOG(3, s, t, "MPPP: PROCESSIPOUT ERROR, no session opened in bundle:%d\n", bid);
                        return;
                }
 
@@ -1492,58 +1508,70 @@ static void processipout(uint8_t *buf, int len)
                s = members[b->current_ses];
                t = session[s].tunnel;
                sp = &session[s];
+               LOG(4, s, t, "MPPP: (1)Session number becomes: %d\n", s);
 
-               if ((num_of_links > 1) && (len > MINFRAGLEN))
+               if (num_of_links > 1)
                {
-                       LOG(4, s, t, "MPPP: (1)Session number becomes: %d\n", s);
-
-                       //for rotate traffic among the member links
-                       uint32_t divisor = num_of_links;
-                       if (divisor > 2)
-                               divisor--;
-
-                       // Partition the packet to "num_of_links" fragments
-                       uint32_t fraglen = len / divisor;
-                       uint32_t last_fraglen = fraglen + len % divisor;
-                       uint32_t remain = len;
-
-                       // send the first packet
-                       uint8_t *p = makeppp(fragbuf, sizeof(fragbuf), buf, fraglen, s, t, PPPIP, 0, bid, MP_BEGIN);
-                       if (!p) return;
-                       tunnelsend(fragbuf, fraglen + (p-fragbuf), t); // send it...
+                       if(len > MINFRAGLEN)
+                       {
+                               //for rotate traffic among the member links
+                               uint32_t divisor = num_of_links;
+                               if (divisor > 2)
+                                       divisor = divisor/2 + (divisor & 1);
+
+                               // Partition the packet to "num_of_links" fragments
+                               uint32_t fraglen = len / divisor;
+                               uint32_t last_fraglen = fraglen + len % divisor;
+                               uint32_t remain = len;
+
+                               // send the first packet
+                               uint8_t *p = makeppp(fragbuf, sizeof(fragbuf), buf, fraglen, s, t, PPPIP, 0, bid, MP_BEGIN);
+                               if (!p) return;
+                               tunnelsend(fragbuf, fraglen + (p-fragbuf), t); // send it...
 
-                       // statistics
-                       update_session_out_stat(s, sp, fraglen);
+                               // statistics
+                               update_session_out_stat(s, sp, fraglen);
 
-                       remain -= fraglen;
-                       while (remain > last_fraglen)
-                       { 
+                               remain -= fraglen;
+                               while (remain > last_fraglen)
+                               {
+                                       b->current_ses = (b->current_ses + 1) % num_of_links;
+                                       s = members[b->current_ses];
+                                       t = session[s].tunnel;
+                                       sp = &session[s];
+                                       LOG(4, s, t, "MPPP: (2)Session number becomes: %d\n", s);
+                                       p = makeppp(fragbuf, sizeof(fragbuf), buf+(len - remain), fraglen, s, t, PPPIP, 0, bid, 0);
+                                       if (!p) return;
+                                       tunnelsend(fragbuf, fraglen + (p-fragbuf), t); // send it...
+                                       update_session_out_stat(s, sp, fraglen);
+                                       remain -= fraglen;
+                               }
+                               // send the last fragment
                                b->current_ses = (b->current_ses + 1) % num_of_links;
                                s = members[b->current_ses];
                                t = session[s].tunnel;
                                sp = &session[s];
                                LOG(4, s, t, "MPPP: (2)Session number becomes: %d\n", s);
-                               p = makeppp(fragbuf, sizeof(fragbuf), buf+(len - remain), fraglen, s, t, PPPIP, 0, bid, 0);
+                               p = makeppp(fragbuf, sizeof(fragbuf), buf+(len - remain), remain, s, t, PPPIP, 0, bid, MP_END);
                                if (!p) return;
-                               tunnelsend(fragbuf, fraglen + (p-fragbuf), t); // send it...
-                               update_session_out_stat(s, sp, fraglen);
-                               remain -= fraglen;
+                               tunnelsend(fragbuf, remain + (p-fragbuf), t); // send it...
+                               update_session_out_stat(s, sp, remain);
+                               if (remain != last_fraglen)
+                                       LOG(3, s, t, "PROCESSIPOUT ERROR REMAIN != LAST_FRAGLEN, %d != %d\n", remain, last_fraglen);
+                       }
+                       else
+                       {
+                               // Send it as one frame
+                               uint8_t *p = makeppp(fragbuf, sizeof(fragbuf), buf, len, s, t, PPPIP, 0, bid, MP_BOTH_BITS);
+                               if (!p) return;
+                               tunnelsend(fragbuf, len + (p-fragbuf), t); // send it...
+                               LOG(4, s, t, "MPPP: packet sent as one frame\n");
+                               update_session_out_stat(s, sp, len);
                        }
-                       // send the last fragment
-                       b->current_ses = (b->current_ses + 1) % num_of_links;
-                       s = members[b->current_ses];
-                       t = session[s].tunnel;
-                       sp = &session[s];
-                       LOG(4, s, t, "MPPP: (2)Session number becomes: %d\n", s);
-                       p = makeppp(fragbuf, sizeof(fragbuf), buf+(len - remain), remain, s, t, PPPIP, 0, bid, MP_END);
-                       if (!p) return;
-                       tunnelsend(fragbuf, remain + (p-fragbuf), t); // send it...
-                       update_session_out_stat(s, sp, remain);
-                       if (remain != last_fraglen)
-                               LOG(3, s, t, "PROCESSIPOUT ERROR REMAIN != LAST_FRAGLEN, %d != %d\n", remain, last_fraglen);
                }
-               else {
-                       // Send it as one frame
+               else
+               {
+                       // Send it as one frame (NO MPPP Frame)
                        uint8_t *p = makeppp(fragbuf, sizeof(fragbuf), buf, len, s, t, PPPIP, 0, 0, 0);
                        if (!p) return;
                        tunnelsend(fragbuf, len + (p-fragbuf), t); // send it...
@@ -2016,11 +2044,11 @@ void sessionshutdown(sessionidt s, char const *reason, int cdn_result, int cdn_e
                                {
                                        uint8_t ml;
                                        for(ml = 0; ml<bundle[b].num_of_links; ml++)
-                                                       if(bundle[b].members[ml] == s)
-                                                       {
-                                                                       mem_num = ml;
-                                                                       break;
-                                                       }
+                                       if(bundle[b].members[ml] == s)
+                                       {
+                                                       mem_num = ml;
+                                                       break;
+                                       }
                                        bundle[b].members[mem_num] = bundle[b].members[bundle[b].num_of_links];
                                        LOG(3, s, session[s].tunnel, "MPPP: Adjusted member links array\n");
 
@@ -2596,7 +2624,7 @@ void processudp(uint8_t *buf, int len, struct sockaddr_in *addr)
                                case 0:     // message type
                                        message = ntohs(*(uint16_t *) b);
                                        mandatory = flags & 0x80;
-                                       LOG(4, s, t, "   Message type = %u (%s)\n", *b, l2tp_code(message));
+                                       LOG(4, s, t, "   Message type = %u (%s)\n", message, l2tp_code(message));
                                        break;
                                case 1:     // result code
                                        {
@@ -3191,6 +3219,7 @@ void processudp(uint8_t *buf, int len, struct sockaddr_in *addr)
 }
 
 // read and process packet on tun
+// (i.e. this routine writes to buf[-8]).
 static void processtun(uint8_t * buf, int len)
 {
        LOG_HEX(5, "Receive TUN Data", buf, len);
@@ -3464,8 +3493,8 @@ static void regular_cleanups(double period)
                        }
                }
 
-               // Drop sessions who have not responded within IDLE_TIMEOUT seconds
-               if (session[s].last_packet && (time_now - session[s].last_packet >= IDLE_TIMEOUT))
+               // Drop sessions who have not responded within IDLE_ECHO_TIMEOUT seconds
+               if (session[s].last_packet && (time_now - session[s].last_packet >= config->idle_echo_timeout))
                {
                        sessionshutdown(s, "No response to LCP ECHO requests.", CDN_ADMIN_DISC, TERM_LOST_SERVICE);
                        STAT(session_timeout);
@@ -3474,7 +3503,7 @@ static void regular_cleanups(double period)
                }
 
                // No data in ECHO_TIMEOUT seconds, send LCP ECHO
-               if (session[s].ppp.phase >= Establish && (time_now - session[s].last_packet >= ECHO_TIMEOUT) &&
+               if (session[s].ppp.phase >= Establish && (time_now - session[s].last_packet >= config->echo_timeout) &&
                        (time_now - sess_local[s].last_echo >= ECHO_TIMEOUT))
                {
                        uint8_t b[MAXETHER];
@@ -3755,6 +3784,8 @@ static void mainloop(void)
 {
        int i;
        uint8_t buf[65536];
+       uint8_t *p = buf + 8; // for the hearder of the forwarded MPPP packet (see C_MPPP_FORWARD)
+       int size_bufp = sizeof(buf) - 8;
        clockt next_cluster_ping = 0;   // send initial ping immediately
        struct epoll_event events[BASE_FDS + RADIUS_FDS + EXTRA_FDS];
        int maxevent = sizeof(events)/sizeof(*events);
@@ -3999,9 +4030,9 @@ static void mainloop(void)
                                // incoming IP
                                if (tun_ready)
                                {
-                                       if ((s = read(tunfd, buf, sizeof(buf))) > 0)
+                                       if ((s = read(tunfd, p, size_bufp)) > 0)
                                        {
-                                               processtun(buf, s);
+                                               processtun(p, s);
                                                tun_pkts++;
                                        }
                                        else
@@ -4250,6 +4281,9 @@ static void initdata(int optdebug, char *optconfig)
        config->ppp_max_failure = 5;
        config->kill_timedout_sessions = 1;
        strcpy(config->random_device, RANDOMDEVICE);
+       // Set default value echo_timeout and idle_echo_timeout
+       config->echo_timeout = ECHO_TIMEOUT;
+       config->idle_echo_timeout = IDLE_ECHO_TIMEOUT;
 
        log_stream = stderr;
 
@@ -5193,18 +5227,18 @@ int sessionsetup(sessionidt s, tunnelidt t)
        LOG(3, s, t, "Doing session setup for session\n");
 
        // Join a bundle if the MRRU option is accepted
-        if(session[s].mrru > 0 && session[s].bundle == 0)
-        {
-                LOG(3, s, t, "This session can be part of multilink bundle\n");
-                if (join_bundle(s) > 0)
-                       cluster_send_bundle(session[s].bundle);
+       if(session[s].mrru > 0 && session[s].bundle == 0)
+       {
+               LOG(3, s, t, "This session can be part of multilink bundle\n");
+               if (join_bundle(s) > 0)
+                       cluster_send_bundle(session[s].bundle);
                else
                {
                        LOG(0, s, t, "MPPP: Mismaching mssf option with other sessions in bundle\n");
                        sessionshutdown(s, "Mismaching mssf option.", CDN_NONE, TERM_SERVICE_UNAVAILABLE);
                        return 0;
                }
-        }
+       }
 
        if (!session[s].ip)
        {
@@ -5219,7 +5253,6 @@ int sessionsetup(sessionidt s, tunnelidt t)
                        fmtaddr(htonl(session[s].ip), 0));
        }
 
-
        // Make sure this is right
        session[s].tunnel = t;
 
@@ -5232,13 +5265,14 @@ int sessionsetup(sessionidt s, tunnelidt t)
                for (i = 1; i <= config->cluster_highest_sessionid; i++)
                {
                        if (i == s) continue;
-                       if (!session[s].opened) continue;
+                       if (!session[s].opened) break;
                        // Allow duplicate sessions for multilink ones of the same bundle.
-                        if (session[s].bundle && session[i].bundle && session[s].bundle == session[i].bundle)
-                                continue;
+                       if (session[s].bundle && session[i].bundle && session[s].bundle == session[i].bundle) continue;
+
                        if (ip == session[i].ip)
                        {
                                sessionkill(i, "Duplicate IP address");
+                               cluster_listinvert_session(s, i);
                                continue;
                        }
 
@@ -5246,16 +5280,16 @@ int sessionsetup(sessionidt s, tunnelidt t)
                        if (session[s].walled_garden || session[i].walled_garden) continue;
                        // Guest change
                        int found = 0;
-                        int gu;
-                        for (gu = 0; gu < guest_accounts_num; gu++)
-                        {
-                                if (!strcasecmp(user, guest_users[gu]))
-                                {
-                                        found = 1;
-                                        break;
-                                }
-                        }
-                        if (found) continue;
+                       int gu;
+                       for (gu = 0; gu < guest_accounts_num; gu++)
+                       {
+                               if (!strcasecmp(user, guest_users[gu]))
+                               {
+                                       found = 1;
+                                       break;
+                               }
+                       }
+                       if (found) continue;
 
                        // Drop the new session in case of duplicate sessionss, not the old one.
                        if (!strcasecmp(user, session[i].user))
@@ -5266,7 +5300,7 @@ int sessionsetup(sessionidt s, tunnelidt t)
        // no need to set a route for the same IP address of the bundle
        if (!session[s].bundle || (bundle[session[s].bundle].num_of_links == 1))
        {
-               int routed = 0;
+               int routed = 0;
 
                // Add the route for this session.
                for (r = 0; r < MAXROUTE && session[s].route[r].ip; r++)