static int walk_session_number = 0; // The next session to send when doing the slow table walk.
static int walk_bundle_number = 0; // The next bundle to send when doing the slow table walk.
static int walk_tunnel_number = 0; // The next tunnel to send when doing the slow table walk.
+static int walk_groupe_number = 0; // The next groupe to send when doing the slow table walk.
int forked = 0; // Sanity check: CLI must not diddle with heartbeat table
#define MAX_HEART_SIZE (8192) // Maximum size of heartbeat packet. Must be less than max IP packet size :)
config->cluster_undefined_sessions = MAXSESSION-1;
config->cluster_undefined_bundles = MAXBUNDLE-1;
config->cluster_undefined_tunnels = MAXTUNNEL-1;
+ config->cluster_undefined_groupes = MAXGROUPE-1;
if (!config->cluster_address)
return 0;
if (config->cluster_iam_uptodate)
return;
- if (config->cluster_undefined_sessions || config->cluster_undefined_tunnels || config->cluster_undefined_bundles)
+ if (config->cluster_undefined_sessions || config->cluster_undefined_tunnels ||
+ config->cluster_undefined_bundles || config->cluster_undefined_groupes)
return;
config->cluster_iam_uptodate = 1;
x.ver = 1;
x.addr = config->bind_address;
- x.undef = config->cluster_undefined_sessions + config->cluster_undefined_tunnels + config->cluster_undefined_bundles;
+ x.undef = config->cluster_undefined_sessions + config->cluster_undefined_tunnels +
+ config->cluster_undefined_groupes + config->cluster_undefined_bundles;
x.basetime = basetime;
add_type(&p, C_PING, basetime, (uint8_t *) &x, sizeof(x));
config->cluster_highest_bundleid = i;
}
+ //
+ // Go through and mark all the groupes as defined.
+ // Count the highest used groupe number as well.
+ //
+ config->cluster_highest_groupeid = 0;
+ for (i = 0; i < MAXGROUPE; ++i)
+ {
+ if (grpsession[i].state == GROUPEUNDEF)
+ grpsession[i].state = GROUPEFREE;
+
+ if (grpsession[i].state != GROUPEFREE && i > config->cluster_highest_groupeid)
+ config->cluster_highest_groupeid = i;
+ }
+
//
// Go through and mark all the sessions as being defined.
// reset the idle timeouts.
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].coutgrp_delta += sess_local[i].cout;
session[i].pin += sess_local[i].pin;
session[i].pout += sess_local[i].pout;
config->cluster_undefined_sessions = 0;
config->cluster_undefined_bundles = 0;
config->cluster_undefined_tunnels = 0;
+ config->cluster_undefined_groupes = 0;
config->cluster_iam_uptodate = 1; // assume all peers are up-to-date
// FIXME. We need to fix up the tunnel control message
// we fix it up here, and we ensure that the 'first free session'
// pointer is valid.
//
-static void cluster_check_sessions(int highsession, int freesession_ptr, int highbundle, int hightunnel)
+static void cluster_check_sessions(int highsession, int freesession_ptr, int highbundle, int hightunnel, int highgroupe)
{
int i;
if (config->cluster_iam_uptodate)
return;
- if (highsession > config->cluster_undefined_sessions && highbundle > config->cluster_undefined_bundles && hightunnel > config->cluster_undefined_tunnels)
+ if (highsession > config->cluster_undefined_sessions && highbundle > config->cluster_undefined_bundles &&
+ highgroupe > config->cluster_undefined_groupes && hightunnel > config->cluster_undefined_tunnels)
return;
// Clear out defined sessions, counting the number of
++config->cluster_undefined_tunnels;
}
+ // Clear out defined groupe, counting the number of
+ // undefs remaining.
+ config->cluster_undefined_groupes = 0;
+ for (i = 1 ; i < MAXGROUPE; ++i) {
+ if (i > highgroupe) {
+ if (grpsession[i].state == GROUPEUNDEF) grpsession[i].state = GROUPEFREE; // Defined.
+ continue;
+ }
+
+ if (grpsession[i].state == GROUPEUNDEF)
+ ++config->cluster_undefined_groupes;
+ }
- if (config->cluster_undefined_sessions || config->cluster_undefined_tunnels || config->cluster_undefined_bundles) {
- LOG(2, 0, 0, "Cleared undefined sessions/bundles/tunnels. %d sess (high %d), %d bund (high %d), %d tunn (high %d)\n",
- config->cluster_undefined_sessions, highsession, config->cluster_undefined_bundles, highbundle, config->cluster_undefined_tunnels, hightunnel);
+ if (config->cluster_undefined_sessions || config->cluster_undefined_tunnels || config->cluster_undefined_bundles || config->cluster_undefined_groupes) {
+ LOG(2, 0, 0, "Cleared undefined sessions/bundles/tunnels. %d sess (high %d), %d bund (high %d), %d grp (high %d), %d tunn (high %d)\n",
+ config->cluster_undefined_sessions, highsession, config->cluster_undefined_bundles, highbundle,
+ config->cluster_undefined_groupes, highgroupe, config->cluster_undefined_tunnels, hightunnel);
return;
}
add_type(p, C_BUNDLE, id, (uint8_t *) &bundle[id], sizeof(bundlet));
break;
+ case C_CGROUPE: { // Compressed C_GROUPE
+ uint8_t c[sizeof(groupsesst) * 2]; // Bigger than worst case.
+ uint8_t *d = (uint8_t *) &grpsession[id];
+ uint8_t *orig = d;
+ int size;
+
+ size = rle_compress( &d, sizeof(groupsesst), c, sizeof(c) );
+
+ // Did we compress the full structure, and is the size actually
+ // reduced??
+ if ( (d - orig) == sizeof(groupsesst) && size < sizeof(groupsesst) )
+ {
+ add_type(p, C_CGROUPE, id, c, size);
+ break;
+ }
+ // Failed to compress : Fall through.
+ }
+ case C_GROUPE:
+ add_type(p, C_GROUPE, id, (uint8_t *) &grpsession[id], sizeof(groupsesst));
+ break;
+
case C_CTUNNEL: { // Compressed C_TUNNEL
uint8_t c[sizeof(tunnelt) * 2]; // Bigger than worst case.
uint8_t *d = (uint8_t *) &tunnel[id];
//
void cluster_heartbeat()
{
- int i, count = 0, tcount = 0, bcount = 0;
+ int i, count = 0, tcount = 0, bcount = 0, gcount = 0;
uint8_t buff[MAX_HEART_SIZE + sizeof(heartt) + sizeof(int) ];
heartt h;
uint8_t *p = buff;
h.freesession = sessionfree;
h.hightunnel = config->cluster_highest_tunnelid;
h.highbundle = config->cluster_highest_bundleid;
+ h.highgroupe = config->cluster_highest_groupeid;
h.size_sess = sizeof(sessiont); // Just in case.
h.size_bund = sizeof(bundlet);
h.size_tunn = sizeof(tunnelt);
+ h.nextgrpid = gnextgrpid;
h.interval = config->cluster_hb_interval;
h.timeout = config->cluster_hb_timeout;
h.table_version = config->cluster_table_version;
//
// Fill out the packet with sessions from the session table...
- // (not forgetting to leave space so we can get some tunnels in too )
+ // (not forgetting to leave space so we can get some tunnels,bundle,groupe in too )
while ( (p + sizeof(uint32_t) * 2 + sizeof(sessiont) * 2 ) < (buff + MAX_HEART_SIZE) ) {
if (!walk_session_number) // session #0 isn't valid.
++count; // Count the number of extra sessions we're sending.
}
- //
- // Fill out the packet with tunnels from the tunnel table...
- // This effectively means we walk the tunnel table more quickly
- // than the session table. This is good because stuffing up a
- // tunnel is a much bigger deal than stuffing up a session.
- //
- while ( (p + sizeof(uint32_t) * 2 + sizeof(tunnelt) ) < (buff + MAX_HEART_SIZE) ) {
+ //
+ // Fill out the packet with tunnels from the tunnel table...
+ // This effectively means we walk the tunnel table more quickly
+ // than the session table. This is good because stuffing up a
+ // tunnel is a much bigger deal than stuffing up a session.
+ //
+ int maxsize = (sizeof(tunnelt) < sizeof(bundlet)) ? sizeof(bundlet):sizeof(tunnelt);
+ maxsize = (sizeof(groupsesst) < maxsize) ? maxsize:sizeof(groupsesst);
+ maxsize += (sizeof(uint32_t) * 2);
+
+ // Fill out the packet with tunnels,bundlets, groupes from the tables...
+ while ( (p + maxsize) < (buff + MAX_HEART_SIZE) )
+ {
+ if ((tcount >= config->cluster_highest_tunnelid) &&
+ (bcount >= config->cluster_highest_bundleid) &&
+ (gcount >= config->cluster_highest_groupeid))
+ break;
- if (!walk_tunnel_number) // tunnel #0 isn't valid.
- ++walk_tunnel_number;
+ if ( ((p + sizeof(uint32_t) * 2 + sizeof(tunnelt) ) < (buff + MAX_HEART_SIZE)) &&
+ (tcount < config->cluster_highest_tunnelid))
+ {
+ if (!walk_tunnel_number) // tunnel #0 isn't valid.
+ ++walk_tunnel_number;
- if (tcount >= config->cluster_highest_tunnelid)
- break;
+ hb_add_type(&p, C_CTUNNEL, walk_tunnel_number);
+ walk_tunnel_number = (1+walk_tunnel_number)%(config->cluster_highest_tunnelid+1); // +1 avoids divide by zero.
- hb_add_type(&p, C_CTUNNEL, walk_tunnel_number);
- walk_tunnel_number = (1+walk_tunnel_number)%(config->cluster_highest_tunnelid+1); // +1 avoids divide by zero.
+ ++tcount;
+ }
- ++tcount;
- }
+ if ( ((p + sizeof(uint32_t) * 2 + sizeof(bundlet) ) < (buff + MAX_HEART_SIZE)) &&
+ (bcount < config->cluster_highest_bundleid))
+ {
+ if (!walk_bundle_number) // bundle #0 isn't valid.
+ ++walk_bundle_number;
- //
- // Fill out the packet with bundles from the bundle table...
- while ( (p + sizeof(uint32_t) * 2 + sizeof(bundlet) ) < (buff + MAX_HEART_SIZE) ) {
+ hb_add_type(&p, C_CBUNDLE, walk_bundle_number);
+ walk_bundle_number = (1+walk_bundle_number)%(config->cluster_highest_bundleid+1); // +1 avoids divide by zero.
- if (!walk_bundle_number) // bundle #0 isn't valid.
- ++walk_bundle_number;
+ ++bcount;
+ }
- if (bcount >= config->cluster_highest_bundleid)
- break;
+ if ( ((p + sizeof(uint32_t) * 2 + sizeof(groupsesst) ) < (buff + MAX_HEART_SIZE)) &&
+ (gcount < config->cluster_highest_groupeid))
+ {
+ if (!walk_groupe_number) // groupe #0 isn't valid.
+ ++walk_groupe_number;
- hb_add_type(&p, C_CBUNDLE, walk_bundle_number);
- walk_bundle_number = (1+walk_bundle_number)%(config->cluster_highest_bundleid+1); // +1 avoids divide by zero.
- ++bcount;
- }
+ hb_add_type(&p, C_CGROUPE, walk_groupe_number);
+ walk_groupe_number = (1+walk_groupe_number)%(config->cluster_highest_groupeid+1); // +1 avoids divide by zero.
+ ++gcount;
+ }
+ }
//
// Did we do something wrong?
}
LOG(4, 0, 0, "Sending v%d heartbeat #%d, change #%" PRIu64 " with %d changes "
- "(%d x-sess, %d x-bundles, %d x-tunnels, %d highsess, %d highbund, %d hightun, size %d)\n",
+ "(%d x-sess, %d x-bundles, %d x-tunnels, %d x-groupes, %d highsess, %d highbund, %d hightun, %d highgrp, size %d)\n",
HB_VERSION, h.seq, h.table_version, config->cluster_num_changes,
- count, bcount, tcount, config->cluster_highest_sessionid, config->cluster_highest_bundleid,
- config->cluster_highest_tunnelid, (int) (p - buff));
+ count, bcount, tcount, gcount, config->cluster_highest_sessionid, config->cluster_highest_bundleid,
+ config->cluster_highest_tunnelid, config->cluster_highest_groupeid, (int) (p - buff));
config->cluster_num_changes = 0;
return type_changed(C_CBUNDLE, bid);
}
+// A particular groupe has been changed!
+int cluster_send_groupe(int gid)
+{
+ if (!config->cluster_iam_master)
+ {
+ LOG(0, 0, gid, "I'm not a master, but I just tried to change a groupe!\n");
+ return -1;
+ }
+
+ return type_changed(C_CGROUPE, gid);
+}
+
// A particular tunnel has been changed!
int cluster_send_tunnel(int tid)
{
session[b->sid].cin_delta += b->cin;
session[b->sid].cout_delta += b->cout;
- session[b->sid].coutgrp_delta += b->cout;
if (b->cin)
session[b->sid].last_packet = session[b->sid].last_data = time_now;
return 0;
}
+static int cluster_recv_groupe(int more, uint8_t *p)
+{
+ if (more >= MAXGROUPE) {
+ LOG(0, 0, 0, "DANGER: Received a group id > MAXGROUPE!\n");
+ return -1;
+ }
+
+ if (grpsession[more].state == GROUPEUNDEF) {
+ if (config->cluster_iam_uptodate) { // Sanity.
+ LOG(0, 0, 0, "I thought I was uptodate but I just found an undefined group!\n");
+ } else {
+ --config->cluster_undefined_groupes;
+ }
+ }
+
+ grp_cluster_load_groupe(more, (groupsesst *) p); // Copy groupe into groupe table..
+
+ LOG(5, 0, more, "Received group update (%d undef)\n", config->cluster_undefined_groupes);
+
+ if (!config->cluster_iam_uptodate)
+ cluster_uptodate(); // Check to see if we're up to date.
+
+ return 0;
+}
+
static int cluster_recv_tunnel(int more, uint8_t *p)
{
if (more >= MAXTUNNEL) {
memcpy(&past_hearts[i].data, data, size); // Save it.
- // Check that we don't have too many undefined sessions, and
- // that the free session pointer is correct.
- cluster_check_sessions(h->highsession, h->freesession, h->highbundle, h->hightunnel);
+ // Check that we don't have too many undefined sessions, and
+ // that the free session pointer is correct.
+ gnextgrpid = h->nextgrpid;
+ cluster_check_sessions(h->highsession, h->freesession, h->highbundle, h->hightunnel, h->highgroupe);
if (h->interval != config->cluster_hb_interval)
{
p += sizeof(bundle[more]);
s -= sizeof(bundle[more]);
break;
+
+ case C_CGROUPE:
+ { // Compressed Groupe structure.
+ uint8_t c[ sizeof(groupsesst) + 2];
+ int size;
+ uint8_t *orig_p = p;
+
+ size = rle_decompress((uint8_t **) &p, s, c, sizeof(c));
+ s -= (p - orig_p);
+
+ if (size != sizeof(groupsesst) )
+ { // Ouch! Very very bad!
+ LOG(0, 0, 0, "DANGER: Received a C_CGROUPE that didn't decompress correctly!\n");
+ // Now what? Should exit! No-longer up to date!
+ break;
+ }
+
+ cluster_recv_groupe(more, c);
+ break;
+ }
+ case C_GROUPE:
+ if ( s < sizeof(grpsession[more]))
+ goto shortpacket;
+
+ cluster_recv_groupe(more, p);
+
+ p += sizeof(grpsession[more]);
+ s -= sizeof(grpsession[more]);
+ break;
+
default:
LOG(0, 0, 0, "DANGER: I received a heartbeat element where I didn't understand the type! (%d)\n", type);
return -1; // can't process any more of the packet!!
cli_print(cli, "Next sequence number expected: %d", config->cluster_seq_number);
cli_print(cli, "%d sessions undefined of %d", config->cluster_undefined_sessions, config->cluster_highest_sessionid);
cli_print(cli, "%d bundles undefined of %d", config->cluster_undefined_bundles, config->cluster_highest_bundleid);
+ cli_print(cli, "%d groupes undefined of %d", config->cluster_undefined_groupes, config->cluster_highest_groupeid);
cli_print(cli, "%d tunnels undefined of %d", config->cluster_undefined_tunnels, config->cluster_highest_tunnelid);
} else {
cli_print(cli, "Table version # : %" PRIu64, config->cluster_table_version);
cli_print(cli, "Next heartbeat # : %d", config->cluster_seq_number);
cli_print(cli, "Highest session : %d", config->cluster_highest_sessionid);
cli_print(cli, "Highest bundle : %d", config->cluster_highest_bundleid);
+ cli_print(cli, "Highest groupe : %d", config->cluster_highest_groupeid);
cli_print(cli, "Highest tunnel : %d", config->cluster_highest_tunnelid);
cli_print(cli, "%d changes queued for sending", config->cluster_num_changes);
}
projects / l2tpns.git / blobdiff