Do not automatically take over segments, as we need to resync first

This commit is contained in:
Martin Willi
2010-04-07 13:55:16 +02:00
committed by Martin Willi
parent ea249cc6f0
commit 1466af8556
3 changed files with 57 additions and 80 deletions
+2
View File
@@ -591,10 +591,12 @@ static void process_segment(private_ha_dispatcher_t *this,
case HA_SEGMENT: case HA_SEGMENT:
if (take) if (take)
{ {
DBG1(DBG_CFG, "remote node takes segment %d", value.u16);
this->segments->deactivate(this->segments, value.u16, FALSE); this->segments->deactivate(this->segments, value.u16, FALSE);
} }
else else
{ {
DBG1(DBG_CFG, "remote node drops segment %d", value.u16);
this->segments->activate(this->segments, value.u16, FALSE); this->segments->activate(this->segments, value.u16, FALSE);
} }
break; break;
+5 -5
View File
@@ -175,9 +175,9 @@ static void deactivate(private_ha_kernel_t *this, u_int segment)
} }
/** /**
* Enable all not-yet enabled segments on all clusterip addresses * Disable all not-yet disabled segments on all clusterip addresses
*/ */
static void activate_all(private_ha_kernel_t *this) static void disable_all(private_ha_kernel_t *this)
{ {
enumerator_t *enumerator; enumerator_t *enumerator;
segment_mask_t active; segment_mask_t active;
@@ -190,9 +190,9 @@ static void activate_all(private_ha_kernel_t *this)
active = get_active(this, file); active = get_active(this, file);
for (i = 1; i <= this->count; i++) for (i = 1; i <= this->count; i++)
{ {
if (!(active & SEGMENTS_BIT(i))) if (active & SEGMENTS_BIT(i))
{ {
enable_disable(this, i, file, TRUE); enable_disable(this, i, file, FALSE);
} }
} }
} }
@@ -222,7 +222,7 @@ ha_kernel_t *ha_kernel_create(u_int count)
this->initval = 0; this->initval = 0;
this->count = count; this->count = count;
activate_all(this); disable_all(this);
return &this->public; return &this->public;
} }
+50 -75
View File
@@ -323,29 +323,39 @@ static bool alert_hook(private_ha_segments_t *this, ike_sa_t *ike_sa,
} }
/** /**
* Get the number of SAs in a segment. * Monitor heartbeat activity of remote node
*/ */
static u_int get_sa_count(private_ha_segments_t *this) static job_requeue_t watchdog(private_ha_segments_t *this)
{ {
enumerator_t *enumerator; int oldstate;
ike_sa_t *ike_sa; bool timeout;
u_int count = 0;
enumerator = charon->ike_sa_manager->create_enumerator(charon->ike_sa_manager); this->mutex->lock(this->mutex);
while (enumerator->enumerate(enumerator, &ike_sa)) pthread_cleanup_push((void*)this->mutex->unlock, this->mutex);
pthread_setcancelstate(PTHREAD_CANCEL_ENABLE, &oldstate);
timeout = this->condvar->timed_wait(this->condvar, this->mutex,
HEARTBEAT_TIMEOUT);
pthread_setcancelstate(oldstate, NULL);
pthread_cleanup_pop(TRUE);
if (timeout)
{ {
if (ike_sa->get_state(ike_sa) != IKE_ESTABLISHED) DBG1(DBG_CFG, "no heartbeat received, taking all segments");
{ activate(this, 0, TRUE);
continue; /* disable heartbeat detection util we get one */
} this->job = NULL;
if (this->tunnel && this->tunnel->is_sa(this->tunnel, ike_sa)) return JOB_REQUEUE_NONE;
{
continue;
}
count++;
} }
enumerator->destroy(enumerator); return JOB_REQUEUE_DIRECT;
return count; }
/**
* Start the heartbeat detection thread
*/
static void start_watchdog(private_ha_segments_t *this)
{
this->job = callback_job_create((callback_job_cb_t)watchdog,
this, NULL, NULL);
charon->processor->queue_job(charon->processor, (job_t*)this->job);
} }
/** /**
@@ -353,46 +363,38 @@ static u_int get_sa_count(private_ha_segments_t *this)
*/ */
static void handle_status(private_ha_segments_t *this, segment_mask_t mask) static void handle_status(private_ha_segments_t *this, segment_mask_t mask)
{ {
segment_mask_t missing, overlap; segment_mask_t missing;
int i; int i;
this->mutex->lock(this->mutex); this->mutex->lock(this->mutex);
missing = ~(this->active | mask); missing = ~(this->active | mask);
overlap = this->active & mask;
/* Activate any missing segment. The master will disable overlapping
* segments if both nodes activate the missing segments simultaneously. */
for (i = 1; i <= this->count; i++) for (i = 1; i <= this->count; i++)
{ {
if (missing & SEGMENTS_BIT(i)) if (missing & SEGMENTS_BIT(i))
{ {
DBG1(DBG_CFG, "HA segment %d was not handled", i); if (this->master != i % 2)
enable_disable(this, i, TRUE, TRUE);
}
}
if (this->master && overlap)
{
/* Disable overlapping segment on one node, controlled by master */
for (i = 1; i <= this->count; i++)
{
if (overlap & SEGMENTS_BIT(i))
{ {
if (get_sa_count(this)) DBG1(DBG_CFG, "HA segment %d was not handled, taking", i);
{ enable_disable(this, i, TRUE, TRUE);
DBG1(DBG_CFG, "HA segment %d overlaps, taking over", i); }
enable_disable(this, i, TRUE, TRUE); else
} {
else DBG1(DBG_CFG, "HA segment %d was not handled, dropping", i);
{ enable_disable(this, i, FALSE, TRUE);
DBG1(DBG_CFG, "HA segment %d overlaps, dropping", i);
enable_disable(this, i, FALSE, TRUE);
}
} }
} }
} }
this->mutex->unlock(this->mutex); this->mutex->unlock(this->mutex);
this->condvar->signal(this->condvar); this->condvar->signal(this->condvar);
if (!this->job)
{
DBG1(DBG_CFG, "received heartbeat, reenabling watchdog");
start_watchdog(this);
}
} }
/** /**
@@ -424,34 +426,15 @@ static job_requeue_t send_status(private_ha_segments_t *this)
return JOB_REQUEUE_NONE; return JOB_REQUEUE_NONE;
} }
/**
* Monitor heartbeat activity of remote node
*/
static job_requeue_t watchdog(private_ha_segments_t *this)
{
int oldstate;
bool timeout;
this->mutex->lock(this->mutex);
pthread_cleanup_push((void*)this->mutex->unlock, this->mutex);
pthread_setcancelstate(PTHREAD_CANCEL_ENABLE, &oldstate);
timeout = this->condvar->timed_wait(this->condvar, this->mutex,
HEARTBEAT_TIMEOUT);
pthread_setcancelstate(oldstate, NULL);
pthread_cleanup_pop(TRUE);
if (timeout)
{ /* didn't get a heartbeat, take all segments */
activate(this, 0, TRUE);
}
return JOB_REQUEUE_DIRECT;
}
/** /**
* Implementation of ha_segments_t.destroy. * Implementation of ha_segments_t.destroy.
*/ */
static void destroy(private_ha_segments_t *this) static void destroy(private_ha_segments_t *this)
{ {
this->job->cancel(this->job); if (this->job)
{
this->job->cancel(this->job);
}
this->mutex->destroy(this->mutex); this->mutex->destroy(this->mutex);
this->condvar->destroy(this->condvar); this->condvar->destroy(this->condvar);
free(this); free(this);
@@ -464,7 +447,6 @@ ha_segments_t *ha_segments_create(ha_socket_t *socket, ha_kernel_t *kernel,
ha_tunnel_t *tunnel, char *local, char *remote, u_int count) ha_tunnel_t *tunnel, char *local, char *remote, u_int count)
{ {
private_ha_segments_t *this = malloc_thing(private_ha_segments_t); private_ha_segments_t *this = malloc_thing(private_ha_segments_t);
int i;
memset(&this->public.listener, 0, sizeof(listener_t)); memset(&this->public.listener, 0, sizeof(listener_t));
this->public.listener.alert = (bool(*)(listener_t*, ike_sa_t *, alert_t, va_list))alert_hook; this->public.listener.alert = (bool(*)(listener_t*, ike_sa_t *, alert_t, va_list))alert_hook;
@@ -482,19 +464,12 @@ ha_segments_t *ha_segments_create(ha_socket_t *socket, ha_kernel_t *kernel,
this->count = count; this->count = count;
this->master = strcmp(local, remote) > 0; this->master = strcmp(local, remote) > 0;
/* initially all segments are active */ /* initially all segments are deactivated */
this->active = 0; this->active = 0;
for (i = 1; i <= count; i++)
{
this->active |= SEGMENTS_BIT(i);
}
send_status(this); send_status(this);
/* start heartbeat detection thread */ start_watchdog(this);
this->job = callback_job_create((callback_job_cb_t)watchdog,
this, NULL, NULL);
charon->processor->queue_job(charon->processor, (job_t*)this->job);
return &this->public; return &this->public;
} }