X-Git-Url: http://git.sameswireless.fr/l2tpns.git/blobdiff_plain/ca4801db5521d0dc9f7c7f854f62f27c55420558..e020fbba4d413aada69619e5048f0ae4132611b5:/cluster.c?ds=inline diff --git a/cluster.c b/cluster.c index c529fea..41dc10f 100644 --- a/cluster.c +++ b/cluster.c @@ -1,6 +1,6 @@ // L2TPNS Clustering Stuff -char const *cvs_id_cluster = "$Id: cluster.c,v 1.39 2005/05/26 12:17:30 bodea Exp $"; +char const *cvs_id_cluster = "$Id: cluster.c,v 1.44 2005/06/28 14:48:19 bodea Exp $"; #include #include @@ -38,7 +38,7 @@ char const *cvs_id_cluster = "$Id: cluster.c,v 1.39 2005/05/26 12:17:30 bodea Ex */ // Module variables. -int cluster_sockfd = 0; // The filedescriptor for the cluster communications port. +extern int cluster_sockfd; // The filedescriptor for the cluster communications port. in_addr_t my_address = 0; // The network address of my ethernet port. static int walk_session_number = 0; // The next session to send when doing the slow table walk. @@ -270,13 +270,8 @@ static int peer_send_message(in_addr_t peer, int type, int more, char *data, int return peer_send_data(peer, buf, (p-buf) ); } -// -// Forward a state changing packet to the master. -// -// The master just processes the payload as if it had -// received it off the tun device. -// -int master_forward_packet(char *data, int size, in_addr_t addr, int port) +// send a packet to the master +static int _forward_packet(char *data, int size, in_addr_t addr, int port, int type) { char buf[65536]; // Vast overkill. char *p = buf; @@ -287,13 +282,30 @@ int master_forward_packet(char *data, int size, in_addr_t addr, int port) LOG(4, 0, 0, "Forwarding packet from %s to master (size %d)\n", fmtaddr(addr, 0), size); STAT(c_forwarded); - add_type(&p, C_FORWARD, addr, (char *) &port, sizeof(port)); + add_type(&p, type, addr, (char *) &port, sizeof(port)); // ick. should be uint16_t memcpy(p, data, size); p += size; return peer_send_data(config->cluster_master_address, buf, (p - buf)); } +// +// Forward a state changing packet to the master. +// +// The master just processes the payload as if it had +// received it off the tun device. +// +int master_forward_packet(char *data, int size, in_addr_t addr, int port) +{ + return _forward_packet(data, size, addr, port, C_FORWARD); +} + +// Forward a DAE RADIUS packet to the master. +int master_forward_dae_packet(char *data, int size, in_addr_t addr, int port) +{ + return _forward_packet(data, size, addr, port, C_FORWARD_DAE); +} + // // Forward a throttled packet to the master for handling. // @@ -402,10 +414,16 @@ void master_update_counts(void) if (config->cluster_iam_master) // Only happens on the slaves. return; - if (!config->cluster_master_address) // If we don't have a master, skip it for a while. + if (!config->cluster_master_address) // If we don't have a master, skip it for a while. return; - i = MAX_B_RECS * 5; // Examine max 2000 sessions; + // C_BYTES format changed in 2.1.0 (cluster version 5) + // during upgrade from previous versions, hang onto our counters + // for a bit until the new master comes up + if (config->cluster_last_hb_ver < 5) + return; + + i = MAX_B_RECS * 5; // Examine max 3000 sessions; if (config->cluster_highest_sessionid > i) i = config->cluster_highest_sessionid; @@ -416,17 +434,20 @@ void master_update_counts(void) walk_session_number = 1; if (!sess_local[walk_session_number].cin && !sess_local[walk_session_number].cout) - continue; // Unused. Skip it. + continue; // Unchanged. Skip it. b[c].sid = walk_session_number; - b[c].in = sess_local[walk_session_number].cin; - b[c].out = sess_local[walk_session_number].cout; - - if (++c > MAX_B_RECS) // Send a max of 400 elements in a packet. - break; + b[c].pin = sess_local[walk_session_number].pin; + b[c].pout = sess_local[walk_session_number].pout; + b[c].cin = sess_local[walk_session_number].cin; + b[c].cout = sess_local[walk_session_number].cout; // Reset counters. + sess_local[walk_session_number].pin = sess_local[walk_session_number].pout = 0; sess_local[walk_session_number].cin = sess_local[walk_session_number].cout = 0; + + if (++c > MAX_B_RECS) // Send a max of 600 elements in a packet. + break; } if (!c) // Didn't find any that changes. Get out of here! @@ -524,9 +545,14 @@ void cluster_check_master(void) return; // Everything's ok! config->cluster_last_hb = TIME + 1; // Just the one election thanks. + config->cluster_master_address = 0; LOG(0, 0, 0, "Master timed out! Holding election...\n"); + // In the process of shutting down, can't be master + if (main_quit) + return; + for (i = have_peers = 0; i < num_peers; i++) { if ((peers[i].timestamp + config->cluster_hb_timeout) < t) @@ -555,7 +581,6 @@ void cluster_check_master(void) // to become a master!!! config->cluster_iam_master = 1; - config->cluster_master_address = 0; LOG(0, 0, 0, "I am declaring myself the master!\n"); @@ -610,13 +635,17 @@ void cluster_check_master(void) // Reset die relative to our uptime rather than the old master's if (session[i].die) session[i].die = TIME; - // Accumulate un-sent byte counters. - session[i].cin += sess_local[i].cin; - session[i].cout += sess_local[i].cout; - session[i].total_cin += sess_local[i].cin; - session[i].total_cout += sess_local[i].cout; + // Accumulate un-sent byte/packet counters. + increment_counter(&session[i].cin, &session[i].cin_wrap, sess_local[i].cin); + increment_counter(&session[i].cout, &session[i].cout_wrap, sess_local[i].cout); + session[i].cin_delta += sess_local[i].cin; + session[i].cout_delta += sess_local[i].cout; + + session[i].pin += sess_local[i].pin; + session[i].pout += sess_local[i].pout; sess_local[i].cin = sess_local[i].cout = 0; + sess_local[i].pin = sess_local[i].pout = 0; sess_local[i].radius = 0; // Reset authentication as the radius blocks aren't up to date. @@ -1005,7 +1034,6 @@ static int cluster_add_peer(in_addr_t peer, time_t basetime, pingt *pp, int size config->cluster_master_address = 0; config->cluster_last_hb = 0; // Force an election. cluster_check_master(); - return 0; } if (i >= num_peers) @@ -1052,6 +1080,16 @@ static int cluster_set_master(in_addr_t peer, in_addr_t master) fmtaddr(master, 1)); config->cluster_master_address = master; + if (master) + { + // catchup with new master + peer_send_message(master, C_LASTSEEN, config->cluster_seq_number, NULL, 0); + + // delay next election + config->cluster_last_hb = TIME; + } + + // run election (or reset "probed" if master was set) cluster_check_master(); return 0; } @@ -1078,13 +1116,16 @@ static int cluster_handle_bytes(char *data, int size) return -1; /* Abort processing */ } - session[b->sid].total_cin += b->in; - session[b->sid].total_cout += b->out; + session[b->sid].pin += b->pin; + session[b->sid].pout += b->pout; + + increment_counter(&session[b->sid].cin, &session[b->sid].cin_wrap, b->cin); + increment_counter(&session[b->sid].cout, &session[b->sid].cout_wrap, b->cout); - session[b->sid].cin += b->in; - session[b->sid].cout += b->out; + session[b->sid].cin_delta += b->cin; + session[b->sid].cout_delta += b->cout; - if (b->in) + if (b->cin) session[b->sid].last_packet = time_now; // Reset idle timer! size -= sizeof(*b); @@ -1223,12 +1264,12 @@ static uint8_t *convert_session(struct oldsession *old) new.nr = old->nr; new.ns = old->ns; new.magic = old->magic; - new.cin = old->cin; - new.cout = old->cout; new.pin = old->pin; new.pout = old->pout; - new.total_cin = old->total_cin; - new.total_cout = old->total_cout; + new.cin = old->total_cin; + new.cout = old->total_cout; + new.cin_delta = old->cin; + new.cout_delta = old->cout; new.throttle_in = old->throttle_in; new.throttle_out = old->throttle_out; new.filter_in = old->filter_in; @@ -1348,8 +1389,7 @@ static int cluster_process_heartbeat(uint8_t *data, int size, int more, uint8_t // Note that after a clean failover, the cluster_master_address // is cleared, so this doesn't run. // - if (config->cluster_master_address && addr != config->cluster_master_address - && (config->cluster_last_hb + config->cluster_hb_timeout - 11) > TIME) { + if (config->cluster_master_address && addr != config->cluster_master_address) { LOG(0, 0, 0, "Ignoring stray heartbeat from %s, current master %s has not yet timed out (last heartbeat %.1f seconds ago).\n", fmtaddr(addr, 0), fmtaddr(config->cluster_master_address, 1), 0.1 * (TIME - config->cluster_last_hb)); @@ -1360,6 +1400,7 @@ static int cluster_process_heartbeat(uint8_t *data, int size, int more, uint8_t config->cluster_seq_number = h->seq; config->cluster_last_hb = TIME; // Reset to ensure that we don't become master!! + config->cluster_last_hb_ver = hb_ver; // remember what cluster version the master is using if (config->cluster_seq_number != h->seq) { // Out of sequence heartbeat! static int lastseen_seq = 0; @@ -1556,7 +1597,8 @@ int processcluster(char *data, int size, in_addr_t addr) p += sizeof(uint32_t); s -= sizeof(uint32_t); - switch (type) { + switch (type) + { case C_PING: // Update the peers table. return cluster_add_peer(addr, more, (pingt *) p, s); @@ -1566,24 +1608,36 @@ int processcluster(char *data, int size, in_addr_t addr) case C_LASTSEEN: // Catch up a slave (slave missed a packet). return cluster_catchup_slave(more, addr); - case C_FORWARD: { // Forwarded control packet. pass off to processudp. - struct sockaddr_in a; - a.sin_addr.s_addr = more; - - a.sin_port = *(int *) p; - s -= sizeof(int); - p += sizeof(int); + case C_FORWARD: // Forwarded control packet. pass off to processudp. + case C_FORWARD_DAE: // Forwarded DAE packet. pass off to processdae. + if (!config->cluster_iam_master) + { + LOG(0, 0, 0, "I'm not the master, but I got a C_FORWARD_%s from %s?\n", + type == C_FORWARD_DAE ? "_DAE" : "", fmtaddr(addr, 0)); - if (!config->cluster_iam_master) { // huh? - LOG(0, 0, 0, "I'm not the master, but I got a C_FORWARD from %s?\n", fmtaddr(addr, 0)); return -1; } + else + { + struct sockaddr_in a; + a.sin_addr.s_addr = more; + + a.sin_port = *(int *) p; + s -= sizeof(int); + p += sizeof(int); + + LOG(4, 0, 0, "Got a forwarded %spacket... (%s:%d)\n", + type == C_FORWARD_DAE ? "DAE " : "", fmtaddr(more, 0), a.sin_port); + + STAT(recv_forward); + if (type == C_FORWARD_DAE) + processdae(p, s, &a, sizeof(a)); + else + processudp(p, s, &a); + + return 0; + } - LOG(4, 0, 0, "Got a forwarded packet... (%s:%d)\n", fmtaddr(more, 0), a.sin_port); - STAT(recv_forward); - processudp(p, s, &a); - return 0; - } case C_THROTTLE: { // Receive a forwarded packet from a slave. if (!config->cluster_iam_master) { LOG(0, 0, 0, "I'm not the master, but I got a C_THROTTLE from %s?\n", fmtaddr(addr, 0));