[thirdparty/strongswan.git] / src / libcharon / plugins / ha / ha_segments.c

/*
 * Copyright (C) 2008 Martin Willi
 *
 * Copyright (C) secunet Security Networks AG
 *
 * This program is free software; you can redistribute it and/or modify it
 * under the terms of the GNU General Public License as published by the
 * Free Software Foundation; either version 2 of the License, or (at your
 * option) any later version.  See <http://www.fsf.org/copyleft/gpl.txt>.
 *
 * This program is distributed in the hope that it will be useful, but
 * WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY
 * or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License
 * for more details.
 */

#include "ha_segments.h"

#include <threading/mutex.h>
#include <threading/condvar.h>
#include <collections/linked_list.h>
#include <threading/thread.h>
#include <processing/jobs/callback_job.h>

#define DEFAULT_HEARTBEAT_DELAY 1000
#define DEFAULT_HEARTBEAT_TIMEOUT 2100

typedef struct private_ha_segments_t private_ha_segments_t;

/**
 * Private data of an ha_segments_t object.
 */
struct private_ha_segments_t {

	/**
	 * Public ha_segments_t interface.
	 */
	ha_segments_t public;

	/**
	 * communication socket
	 */
	ha_socket_t *socket;

	/**
	 * Sync tunnel, if any
	 */
	ha_tunnel_t *tunnel;

	/**
	 * Interface to control segments at kernel level
	 */
	ha_kernel_t *kernel;

	/**
	 * Mutex to lock segment manipulation
	 */
	mutex_t *mutex;

	/**
	 * Condvar to wait for heartbeats
	 */
	condvar_t *condvar;

	/**
	 * Total number of ClusterIP segments
	 */
	u_int count;

	/**
	 * mask of active segments
	 */
	segment_mask_t active;

	/**
	 * Node number
	 */
	u_int node;

	/**
	 * Are we checking for heartbeats?
	 */
	bool heartbeat_active;

	/**
	 * Interval we send heartbeats
	 */
	int heartbeat_delay;

	/**
	 * Timeout for heartbeats received from other node
	 */
	int heartbeat_timeout;

	/**
	 * Interval to check for autobalance, 0 to disable
	 */
	int autobalance;
};

/**
 * Log currently active segments
 */
static void log_segments(private_ha_segments_t *this, bool activated,
						 u_int segment)
{
	char buf[64] = "none", *pos = buf;
	int i;
	bool first = TRUE;

	for (i = 1; i <= this->count; i++)
	{
		if (this->active & SEGMENTS_BIT(i))
		{
			if (first)
			{
				first = FALSE;
			}
			else
			{
				pos += snprintf(pos, buf + sizeof(buf) - pos, ",");
			}
			pos += snprintf(pos, buf + sizeof(buf) - pos, "%d", i);
		}
	}
	DBG1(DBG_CFG, "HA segment %d %sactivated, now active: %s",
		 segment, activated ? "" : "de", buf);
}

/**
 * Enable/Disable a specific segment
 */
static void enable_disable(private_ha_segments_t *this, u_int segment,
						   bool enable, bool notify)
{
	ike_sa_t *ike_sa;
	enumerator_t *enumerator;
	ike_sa_state_t old, new;
	ha_message_t *message = NULL;
	ha_message_type_t type;
	bool changes = FALSE;

	if (segment > this->count)
	{
		return;
	}

	if (enable)
	{
		old = IKE_PASSIVE;
		new = IKE_ESTABLISHED;
		type = HA_SEGMENT_TAKE;
		if (!(this->active & SEGMENTS_BIT(segment)))
		{
			this->active |= SEGMENTS_BIT(segment);
			this->kernel->activate(this->kernel, segment);
			changes = TRUE;
		}
	}
	else
	{
		old = IKE_ESTABLISHED;
		new = IKE_PASSIVE;
		type = HA_SEGMENT_DROP;
		if (this->active & SEGMENTS_BIT(segment))
		{
			this->active &= ~SEGMENTS_BIT(segment);
			this->kernel->deactivate(this->kernel, segment);
			changes = TRUE;
		}
	}

	if (changes)
	{
		enumerator = charon->ike_sa_manager->create_enumerator(
												charon->ike_sa_manager, TRUE);
		while (enumerator->enumerate(enumerator, &ike_sa))
		{
			if (ike_sa->get_state(ike_sa) != old)
			{
				continue;
			}
			if (this->tunnel && this->tunnel->is_sa(this->tunnel, ike_sa))
			{
				continue;
			}
			if (this->kernel->get_segment(this->kernel,
									ike_sa->get_other_host(ike_sa)) == segment)
			{
				ike_sa->set_state(ike_sa, new);
			}
		}
		enumerator->destroy(enumerator);
		log_segments(this, enable, segment);
	}

	if (notify)
	{
		message = ha_message_create(type);
		message->add_attribute(message, HA_SEGMENT, segment);
		this->socket->push(this->socket, message);
		message->destroy(message);
	}
}

/**
 * Enable/Disable all or a specific segment, do locking
 */
static void enable_disable_all(private_ha_segments_t *this, u_int segment,
							   bool enable, bool notify)
{
	int i;

	this->mutex->lock(this->mutex);
	if (segment == 0)
	{
		for (i = 1; i <= this->count; i++)
		{
			enable_disable(this, i, enable, notify);
		}
	}
	else
	{
		enable_disable(this, segment, enable, notify);
	}
	this->mutex->unlock(this->mutex);
}

METHOD(ha_segments_t, activate, void,
	private_ha_segments_t *this, u_int segment, bool notify)
{
	enable_disable_all(this, segment, TRUE, notify);
}

METHOD(ha_segments_t, deactivate, void,
	private_ha_segments_t *this, u_int segment, bool notify)
{
	enable_disable_all(this, segment, FALSE, notify);
}

METHOD(listener_t, alert_hook, bool,
	private_ha_segments_t *this, ike_sa_t *ike_sa, alert_t alert, va_list args)
{
	if (alert == ALERT_SHUTDOWN_SIGNAL)
	{
		if (this->heartbeat_active)
		{
			DBG1(DBG_CFG, "HA heartbeat active, dropping all segments");
			deactivate(this, 0, TRUE);
		}
		else
		{
			DBG1(DBG_CFG, "no HA heartbeat active, closing IKE_SAs");
		}
	}
	return TRUE;
}

/**
 * Monitor heartbeat activity of remote node
 */
static job_requeue_t watchdog(private_ha_segments_t *this)
{
	bool timeout, oldstate;

	this->mutex->lock(this->mutex);
	thread_cleanup_push((void*)this->mutex->unlock, this->mutex);
	oldstate = thread_cancelability(TRUE);
	timeout = this->condvar->timed_wait(this->condvar, this->mutex,
										this->heartbeat_timeout);
	thread_cancelability(oldstate);
	thread_cleanup_pop(TRUE);
	if (timeout)
	{
		DBG1(DBG_CFG, "no heartbeat received, taking all segments");
		activate(this, 0, TRUE);
		/* disable heartbeat detection util we get one */
		this->heartbeat_active = FALSE;
		return JOB_REQUEUE_NONE;
	}
	return JOB_REQUEUE_DIRECT;
}

/**
 * Start the heartbeat detection thread
 */
static void start_watchdog(private_ha_segments_t *this)
{
	this->heartbeat_active = TRUE;
	lib->processor->queue_job(lib->processor,
		(job_t*)callback_job_create_with_prio((callback_job_cb_t)watchdog, this,
				NULL, (callback_job_cancel_t)return_false, JOB_PRIO_CRITICAL));
}

METHOD(ha_segments_t, handle_status, void,
	private_ha_segments_t *this, segment_mask_t mask)
{
	segment_mask_t missing, twice;
	int i;

	this->mutex->lock(this->mutex);

	missing = ~(this->active | mask);
	twice = this->active & mask;

	for (i = 1; i <= this->count; i++)
	{
		if (missing & SEGMENTS_BIT(i))
		{
			if (this->node == i % 2)
			{
				DBG1(DBG_CFG, "HA segment %d was not handled, taking", i);
				enable_disable(this, i, TRUE, TRUE);
			}
			else
			{
				DBG1(DBG_CFG, "HA segment %d was not handled, dropping", i);
				enable_disable(this, i, FALSE, TRUE);
			}
		}
		if (twice & SEGMENTS_BIT(i))
		{
			if (this->node == i % 2)
			{
				DBG1(DBG_CFG, "HA segment %d was handled twice, taking", i);
				enable_disable(this, i, TRUE, TRUE);
			}
			else
			{
				DBG1(DBG_CFG, "HA segment %d was handled twice, dropping", i);
				enable_disable(this, i, FALSE, TRUE);
			}
		}
	}

	this->condvar->signal(this->condvar);
	this->mutex->unlock(this->mutex);

	if (!this->heartbeat_active)
	{
		DBG1(DBG_CFG, "received heartbeat, reenabling watchdog");
		start_watchdog(this);
	}
}

/**
 * Send a status message with our active segments
 */
static job_requeue_t send_status(private_ha_segments_t *this)
{
	ha_message_t *message;
	int i;

	message = ha_message_create(HA_STATUS);

	this->mutex->lock(this->mutex);
	for (i = 1; i <= this->count; i++)
	{
		if (this->active & SEGMENTS_BIT(i))
		{
			message->add_attribute(message, HA_SEGMENT, i);
		}
	}
	this->mutex->unlock(this->mutex);

	this->socket->push(this->socket, message);
	message->destroy(message);

	/* schedule next invocation */
	return JOB_RESCHEDULE_MS(this->heartbeat_delay);
}

/**
 * Start the heartbeat sending task
 */
static void start_heartbeat(private_ha_segments_t *this)
{
	lib->processor->queue_job(lib->processor,
		(job_t*)callback_job_create_with_prio((callback_job_cb_t)send_status,
			this, NULL, (callback_job_cancel_t)return_false, JOB_PRIO_CRITICAL));
}

/**
 * Take a segment if we are handling less than half of segments
 */
static job_requeue_t autobalance(private_ha_segments_t *this)
{
	int i, active = 0;

	this->mutex->lock(this->mutex);

	for (i = 1; i <= this->count; i++)
	{
		if (this->active & SEGMENTS_BIT(i))
		{
			active++;
		}
	}
	if (active < this->count / 2)
	{
		for (i = 1; i <= this->count; i++)
		{
			if (!(this->active & SEGMENTS_BIT(i)))
			{
				DBG1(DBG_CFG, "autobalancing HA (%d/%d active), taking %d",
					 active, this->count, i);
				enable_disable(this, i, TRUE, TRUE);
				/* we claim only one in each interval */
				break;
			}
		}
	}

	this->mutex->unlock(this->mutex);

	return JOB_RESCHEDULE(this->autobalance);
}

/**
 * Schedule autobalancing
 */
static void start_autobalance(private_ha_segments_t *this)
{
	DBG1(DBG_CFG, "scheduling HA autobalance every %ds", this->autobalance);
	lib->scheduler->schedule_job(lib->scheduler,
		(job_t*)callback_job_create_with_prio((callback_job_cb_t)autobalance,
			this, NULL, (callback_job_cancel_t)return_false, JOB_PRIO_CRITICAL),
		this->autobalance);
}

METHOD(ha_segments_t, is_active, bool,
	private_ha_segments_t *this, u_int segment)
{
	return (this->active & SEGMENTS_BIT(segment)) != 0;
}

METHOD(ha_segments_t, count, u_int,
	private_ha_segments_t *this)
{
	return this->count;
}

METHOD(ha_segments_t, destroy, void,
	private_ha_segments_t *this)
{
	this->mutex->destroy(this->mutex);
	this->condvar->destroy(this->condvar);
	free(this);
}

/**
 * See header
 */
ha_segments_t *ha_segments_create(ha_socket_t *socket, ha_kernel_t *kernel,
								  ha_tunnel_t *tunnel, u_int count, u_int node,
								  bool monitor)
{
	private_ha_segments_t *this;

	INIT(this,
		.public = {
			.listener = {
				.alert = _alert_hook,
			},
			.activate = _activate,
			.deactivate = _deactivate,
			.handle_status = _handle_status,
			.is_active = _is_active,
			.count = _count,
			.destroy = _destroy,
		},
		.socket = socket,
		.tunnel = tunnel,
		.kernel = kernel,
		.count = count,
		.node = node,
		.mutex = mutex_create(MUTEX_TYPE_DEFAULT),
		.condvar = condvar_create(CONDVAR_TYPE_DEFAULT),
		.heartbeat_delay = lib->settings->get_int(lib->settings,
				"%s.plugins.ha.heartbeat_delay", DEFAULT_HEARTBEAT_DELAY,
				lib->ns),
		.heartbeat_timeout = lib->settings->get_int(lib->settings,
				"%s.plugins.ha.heartbeat_timeout", DEFAULT_HEARTBEAT_TIMEOUT,
				lib->ns),
		.autobalance = lib->settings->get_int(lib->settings,
				"%s.plugins.ha.autobalance", 0, lib->ns),
	);

	if (monitor)
	{
		DBG1(DBG_CFG, "starting HA heartbeat, delay %dms, timeout %dms",
			 this->heartbeat_delay, this->heartbeat_timeout);
		start_heartbeat(this);
		start_watchdog(this);
	}
	if (this->autobalance)
	{
		start_autobalance(this);
	}

	return &this->public;
}
Commit	Line	Data
765935c8 MW	1	/*
765935c8 MW	2	* Copyright (C) 2008 Martin Willi
19ef2aec TB	3	*
19ef2aec TB	4	* Copyright (C) secunet Security Networks AG
765935c8 MW	5	*
	6	* This program is free software; you can redistribute it and/or modify it
	7	* under the terms of the GNU General Public License as published by the
	8	* Free Software Foundation; either version 2 of the License, or (at your
	9	* option) any later version. See <http://www.fsf.org/copyleft/gpl.txt>.
	10	*
	11	* This program is distributed in the hope that it will be useful, but
	12	* WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY
	13	* or FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License
	14	* for more details.
765935c8 MW	15	*/
765935c8 MW	16
d8748966	17	#include "ha_segments.h"
765935c8	18
14041845 MW	19	#include <threading/mutex.h>
14041845 MW	20	#include <threading/condvar.h>
12642a68	21	#include <collections/linked_list.h>
60b71def	22	#include <threading/thread.h>
3912fdb1	23	#include <processing/jobs/callback_job.h>
765935c8	24
65d15aff MW	25	#define DEFAULT_HEARTBEAT_DELAY 1000
65d15aff MW	26	#define DEFAULT_HEARTBEAT_TIMEOUT 2100
e262f4e5	27
d8748966	28	typedef struct private_ha_segments_t private_ha_segments_t;
765935c8 MW	29
765935c8 MW	30	/**
d8748966	31	* Private data of an ha_segments_t object.
765935c8	32	*/
d8748966	33	struct private_ha_segments_t {
765935c8 MW	34
765935c8 MW	35	/**
d8748966	36	* Public ha_segments_t interface.
765935c8	37	*/
d8748966	38	ha_segments_t public;
aa98188a	39
37459ea9 MW	40	/**
	41	* communication socket
	42	*/
d8748966	43	ha_socket_t *socket;
37459ea9	44
b7f15be1 MW	45	/**
	46	* Sync tunnel, if any
	47	*/
d8748966	48	ha_tunnel_t *tunnel;
b7f15be1	49
3d672d4b	50	/**
dbc91f7c	51	* Interface to control segments at kernel level
3d672d4b	52	*/
d8748966	53	ha_kernel_t *kernel;
3d672d4b	54
aa98188a	55	/**
e262f4e5 MW	56	* Mutex to lock segment manipulation
	57	*/
	58	mutex_t *mutex;
	59
	60	/**
	61	* Condvar to wait for heartbeats
	62	*/
	63	condvar_t *condvar;
	64
aa98188a MW	65	/**
	66	* Total number of ClusterIP segments
	67	*/
3912fdb1	68	u_int count;
aa98188a MW	69
	70	/**
	71	* mask of active segments
	72	*/
6921e8d5	73	segment_mask_t active;
3912fdb1 MW	74
3912fdb1 MW	75	/**
3e8caf6a	76	* Node number
3912fdb1	77	*/
3e8caf6a	78	u_int node;
65d15aff	79
26d77eb3 TB	80	/**
	81	* Are we checking for heartbeats?
	82	*/
	83	bool heartbeat_active;
	84
65d15aff	85	/**
b3ab7a48	86	* Interval we send heartbeats
65d15aff MW	87	*/
	88	int heartbeat_delay;
	89
	90	/**
	91	* Timeout for heartbeats received from other node
	92	*/
	93	int heartbeat_timeout;
e2d2b542 MW	94
	95	/**
	96	* Interval to check for autobalance, 0 to disable
	97	*/
	98	int autobalance;
765935c8 MW	99	};
765935c8 MW	100
aa98188a MW	101	/**
	102	* Log currently active segments
	103	*/
d8748966	104	static void log_segments(private_ha_segments_t *this, bool activated,
aa98188a MW	105	u_int segment)
aa98188a MW	106	{
f8252385	107	char buf[64] = "none", *pos = buf;
aa98188a MW	108	int i;
	109	bool first = TRUE;
	110
3912fdb1	111	for (i = 1; i <= this->count; i++)
aa98188a	112	{
3912fdb1	113	if (this->active & SEGMENTS_BIT(i))
aa98188a MW	114	{
	115	if (first)
	116	{
	117	first = FALSE;
	118	}
	119	else
	120	{
	121	pos += snprintf(pos, buf + sizeof(buf) - pos, ",");
	122	}
3912fdb1	123	pos += snprintf(pos, buf + sizeof(buf) - pos, "%d", i);
aa98188a MW	124	}
aa98188a MW	125	}
d8748966	126	DBG1(DBG_CFG, "HA segment %d %sactivated, now active: %s",
aa98188a MW	127	segment, activated ? "" : "de", buf);
	128	}
	129
c573b11c	130	/**
874c0bd8	131	* Enable/Disable a specific segment
c573b11c	132	*/
d8748966	133	static void enable_disable(private_ha_segments_t *this, u_int segment,
874c0bd8	134	bool enable, bool notify)
765935c8 MW	135	{
765935c8 MW	136	ike_sa_t *ike_sa;
aa98188a	137	enumerator_t *enumerator;
874c0bd8	138	ike_sa_state_t old, new;
d8748966 MW	139	ha_message_t *message = NULL;
d8748966 MW	140	ha_message_type_t type;
874c0bd8	141	bool changes = FALSE;
aa98188a	142
874c0bd8 MW	143	if (segment > this->count)
	144	{
	145	return;
	146	}
3d672d4b	147
874c0bd8	148	if (enable)
aa98188a	149	{
874c0bd8 MW	150	old = IKE_PASSIVE;
874c0bd8 MW	151	new = IKE_ESTABLISHED;
d8748966	152	type = HA_SEGMENT_TAKE;
874c0bd8 MW	153	if (!(this->active & SEGMENTS_BIT(segment)))
	154	{
	155	this->active \|= SEGMENTS_BIT(segment);
	156	this->kernel->activate(this->kernel, segment);
	157	changes = TRUE;
c573b11c	158	}
874c0bd8 MW	159	}
	160	else
	161	{
	162	old = IKE_ESTABLISHED;
	163	new = IKE_PASSIVE;
d8748966	164	type = HA_SEGMENT_DROP;
874c0bd8 MW	165	if (this->active & SEGMENTS_BIT(segment))
	166	{
	167	this->active &= ~SEGMENTS_BIT(segment);
	168	this->kernel->deactivate(this->kernel, segment);
	169	changes = TRUE;
c573b11c	170	}
874c0bd8 MW	171	}
	172
	173	if (changes)
	174	{
69c3eca0 MW	175	enumerator = charon->ike_sa_manager->create_enumerator(
69c3eca0 MW	176	charon->ike_sa_manager, TRUE);
aa98188a MW	177	while (enumerator->enumerate(enumerator, &ike_sa))
aa98188a MW	178	{
b7f15be1 MW	179	if (ike_sa->get_state(ike_sa) != old)
	180	{
	181	continue;
	182	}
d8748966	183	if (this->tunnel && this->tunnel->is_sa(this->tunnel, ike_sa))
b7f15be1 MW	184	{
	185	continue;
	186	}
2cbc48ec MW	187	if (this->kernel->get_segment(this->kernel,
2cbc48ec MW	188	ike_sa->get_other_host(ike_sa)) == segment)
aa98188a	189	{
874c0bd8	190	ike_sa->set_state(ike_sa, new);
aa98188a MW	191	}
	192	}
	193	enumerator->destroy(enumerator);
c573b11c	194	log_segments(this, enable, segment);
aa98188a	195	}
3d672d4b	196
874c0bd8 MW	197	if (notify)
874c0bd8 MW	198	{
d8748966 MW	199	message = ha_message_create(type);
d8748966 MW	200	message->add_attribute(message, HA_SEGMENT, segment);
874c0bd8	201	this->socket->push(this->socket, message);
aa334daa	202	message->destroy(message);
874c0bd8 MW	203	}
	204	}
	205
	206	/**
	207	* Enable/Disable all or a specific segment, do locking
	208	*/
d8748966	209	static void enable_disable_all(private_ha_segments_t *this, u_int segment,
874c0bd8 MW	210	bool enable, bool notify)
	211	{
	212	int i;
	213
e262f4e5	214	this->mutex->lock(this->mutex);
874c0bd8 MW	215	if (segment == 0)
	216	{
	217	for (i = 1; i <= this->count; i++)
	218	{
	219	enable_disable(this, i, enable, notify);
	220	}
	221	}
	222	else
	223	{
	224	enable_disable(this, segment, enable, notify);
	225	}
e262f4e5	226	this->mutex->unlock(this->mutex);
aa98188a MW	227	}
aa98188a MW	228
00c1bd06 MW	229	METHOD(ha_segments_t, activate, void,
00c1bd06 MW	230	private_ha_segments_t *this, u_int segment, bool notify)
c573b11c	231	{
874c0bd8	232	enable_disable_all(this, segment, TRUE, notify);
c573b11c MW	233	}
c573b11c MW	234
00c1bd06 MW	235	METHOD(ha_segments_t, deactivate, void,
00c1bd06 MW	236	private_ha_segments_t *this, u_int segment, bool notify)
aa98188a	237	{
874c0bd8	238	enable_disable_all(this, segment, FALSE, notify);
765935c8 MW	239	}
765935c8 MW	240
aa334daa MW	241	METHOD(listener_t, alert_hook, bool,
aa334daa MW	242	private_ha_segments_t this, ike_sa_t ike_sa, alert_t alert, va_list args)
9ffcbea6	243	{
aa334daa	244	if (alert == ALERT_SHUTDOWN_SIGNAL)
9ffcbea6	245	{
26d77eb3	246	if (this->heartbeat_active)
9ffcbea6	247	{
aa334daa MW	248	DBG1(DBG_CFG, "HA heartbeat active, dropping all segments");
aa334daa MW	249	deactivate(this, 0, TRUE);
9ffcbea6	250	}
aa334daa	251	else
9ffcbea6	252	{
aa334daa	253	DBG1(DBG_CFG, "no HA heartbeat active, closing IKE_SAs");
9ffcbea6	254	}
310498f3 MW	255	}
	256	return TRUE;
	257	}
	258
ea249cc6	259	/**
1466af85	260	* Monitor heartbeat activity of remote node
ea249cc6	261	*/
1466af85	262	static job_requeue_t watchdog(private_ha_segments_t *this)
ea249cc6	263	{
60b71def	264	bool timeout, oldstate;
ea249cc6	265
1466af85	266	this->mutex->lock(this->mutex);
60b71def MW	267	thread_cleanup_push((void*)this->mutex->unlock, this->mutex);
60b71def MW	268	oldstate = thread_cancelability(TRUE);
1466af85	269	timeout = this->condvar->timed_wait(this->condvar, this->mutex,
65d15aff	270	this->heartbeat_timeout);
60b71def MW	271	thread_cancelability(oldstate);
60b71def MW	272	thread_cleanup_pop(TRUE);
1466af85	273	if (timeout)
ea249cc6	274	{
1466af85 MW	275	DBG1(DBG_CFG, "no heartbeat received, taking all segments");
	276	activate(this, 0, TRUE);
	277	/* disable heartbeat detection util we get one */
26d77eb3	278	this->heartbeat_active = FALSE;
1466af85	279	return JOB_REQUEUE_NONE;
ea249cc6	280	}
1466af85 MW	281	return JOB_REQUEUE_DIRECT;
	282	}
	283
	284	/**
	285	* Start the heartbeat detection thread
	286	*/
	287	static void start_watchdog(private_ha_segments_t *this)
	288	{
26d77eb3 TB	289	this->heartbeat_active = TRUE;
	290	lib->processor->queue_job(lib->processor,
	291	(job_t*)callback_job_create_with_prio((callback_job_cb_t)watchdog, this,
	292	NULL, (callback_job_cancel_t)return_false, JOB_PRIO_CRITICAL));
ea249cc6 MW	293	}
ea249cc6 MW	294
00c1bd06 MW	295	METHOD(ha_segments_t, handle_status, void,
00c1bd06 MW	296	private_ha_segments_t *this, segment_mask_t mask)
3912fdb1	297	{
33524f02	298	segment_mask_t missing, twice;
ea249cc6	299	int i;
3912fdb1	300
e262f4e5	301	this->mutex->lock(this->mutex);
874c0bd8	302
3912fdb1	303	missing = ~(this->active \| mask);
33524f02	304	twice = this->active & mask;
3912fdb1	305
3912fdb1 MW	306	for (i = 1; i <= this->count; i++)
	307	{
	308	if (missing & SEGMENTS_BIT(i))
	309	{
3e8caf6a	310	if (this->node == i % 2)
3912fdb1	311	{
1466af85 MW	312	DBG1(DBG_CFG, "HA segment %d was not handled, taking", i);
	313	enable_disable(this, i, TRUE, TRUE);
	314	}
	315	else
	316	{
	317	DBG1(DBG_CFG, "HA segment %d was not handled, dropping", i);
	318	enable_disable(this, i, FALSE, TRUE);
3912fdb1 MW	319	}
3912fdb1 MW	320	}
33524f02 MW	321	if (twice & SEGMENTS_BIT(i))
	322	{
	323	if (this->node == i % 2)
	324	{
	325	DBG1(DBG_CFG, "HA segment %d was handled twice, taking", i);
	326	enable_disable(this, i, TRUE, TRUE);
	327	}
	328	else
	329	{
	330	DBG1(DBG_CFG, "HA segment %d was handled twice, dropping", i);
	331	enable_disable(this, i, FALSE, TRUE);
	332	}
	333	}
3912fdb1	334	}
1466af85	335
e262f4e5	336	this->condvar->signal(this->condvar);
26d77eb3	337	this->mutex->unlock(this->mutex);
1466af85	338
26d77eb3	339	if (!this->heartbeat_active)
1466af85 MW	340	{
	341	DBG1(DBG_CFG, "received heartbeat, reenabling watchdog");
	342	start_watchdog(this);
	343	}
3912fdb1 MW	344	}
	345
	346	/**
	347	* Send a status message with our active segments
	348	*/
d8748966	349	static job_requeue_t send_status(private_ha_segments_t *this)
3912fdb1	350	{
d8748966	351	ha_message_t *message;
3912fdb1 MW	352	int i;
3912fdb1 MW	353
d8748966	354	message = ha_message_create(HA_STATUS);
3912fdb1	355
21f40fe8	356	this->mutex->lock(this->mutex);
3912fdb1 MW	357	for (i = 1; i <= this->count; i++)
	358	{
	359	if (this->active & SEGMENTS_BIT(i))
	360	{
d8748966	361	message->add_attribute(message, HA_SEGMENT, i);
3912fdb1 MW	362	}
3912fdb1 MW	363	}
21f40fe8	364	this->mutex->unlock(this->mutex);
3912fdb1 MW	365
3912fdb1 MW	366	this->socket->push(this->socket, message);
aa334daa	367	message->destroy(message);
3912fdb1 MW	368
3912fdb1 MW	369	/* schedule next invocation */
e0efd7c1	370	return JOB_RESCHEDULE_MS(this->heartbeat_delay);
3912fdb1 MW	371	}
3912fdb1 MW	372
2071dd63 MW	373	/**
	374	* Start the heartbeat sending task
	375	*/
	376	static void start_heartbeat(private_ha_segments_t *this)
	377	{
	378	lib->processor->queue_job(lib->processor,
	379	(job_t*)callback_job_create_with_prio((callback_job_cb_t)send_status,
	380	this, NULL, (callback_job_cancel_t)return_false, JOB_PRIO_CRITICAL));
	381	}
	382
e2d2b542 MW	383	/**
	384	* Take a segment if we are handling less than half of segments
	385	*/
	386	static job_requeue_t autobalance(private_ha_segments_t *this)
	387	{
	388	int i, active = 0;
	389
	390	this->mutex->lock(this->mutex);
	391
	392	for (i = 1; i <= this->count; i++)
	393	{
	394	if (this->active & SEGMENTS_BIT(i))
	395	{
	396	active++;
	397	}
	398	}
	399	if (active < this->count / 2)
	400	{
	401	for (i = 1; i <= this->count; i++)
	402	{
	403	if (!(this->active & SEGMENTS_BIT(i)))
	404	{
	405	DBG1(DBG_CFG, "autobalancing HA (%d/%d active), taking %d",
	406	active, this->count, i);
	407	enable_disable(this, i, TRUE, TRUE);
	408	/* we claim only one in each interval */
	409	break;
	410	}
	411	}
	412	}
	413
	414	this->mutex->unlock(this->mutex);
	415
	416	return JOB_RESCHEDULE(this->autobalance);
	417	}
	418
	419	/**
	420	* Schedule autobalancing
	421	*/
	422	static void start_autobalance(private_ha_segments_t *this)
	423	{
	424	DBG1(DBG_CFG, "scheduling HA autobalance every %ds", this->autobalance);
	425	lib->scheduler->schedule_job(lib->scheduler,
	426	(job_t*)callback_job_create_with_prio((callback_job_cb_t)autobalance,
	427	this, NULL, (callback_job_cancel_t)return_false, JOB_PRIO_CRITICAL),
	428	this->autobalance);
	429	}
	430
08e266a1 MW	431	METHOD(ha_segments_t, is_active, bool,
	432	private_ha_segments_t *this, u_int segment)
	433	{
	434	return (this->active & SEGMENTS_BIT(segment)) != 0;
	435	}
	436
16a898f5 TB	437	METHOD(ha_segments_t, count, u_int,
	438	private_ha_segments_t *this)
	439	{
	440	return this->count;
	441	}
	442
00c1bd06 MW	443	METHOD(ha_segments_t, destroy, void,
00c1bd06 MW	444	private_ha_segments_t *this)
765935c8	445	{
e262f4e5 MW	446	this->mutex->destroy(this->mutex);
e262f4e5 MW	447	this->condvar->destroy(this->condvar);
765935c8 MW	448	free(this);
	449	}
	450
	451	/**
	452	* See header
	453	*/
d8748966	454	ha_segments_t ha_segments_create(ha_socket_t socket, ha_kernel_t *kernel,
3e8caf6a	455	ha_tunnel_t *tunnel, u_int count, u_int node,
aa334daa	456	bool monitor)
765935c8	457	{
00c1bd06 MW	458	private_ha_segments_t *this;
	459
	460	INIT(this,
	461	.public = {
56bceda7 TB	462	.listener = {
	463	.alert = _alert_hook,
	464	},
00c1bd06 MW	465	.activate = _activate,
00c1bd06 MW	466	.deactivate = _deactivate,
00c1bd06	467	.handle_status = _handle_status,
08e266a1	468	.is_active = _is_active,
16a898f5	469	.count = _count,
00c1bd06 MW	470	.destroy = _destroy,
	471	},
	472	.socket = socket,
	473	.tunnel = tunnel,
	474	.kernel = kernel,
	475	.count = count,
	476	.node = node,
	477	.mutex = mutex_create(MUTEX_TYPE_DEFAULT),
	478	.condvar = condvar_create(CONDVAR_TYPE_DEFAULT),
65d15aff	479	.heartbeat_delay = lib->settings->get_int(lib->settings,
42500c27	480	"%s.plugins.ha.heartbeat_delay", DEFAULT_HEARTBEAT_DELAY,
d223fe80	481	lib->ns),
65d15aff	482	.heartbeat_timeout = lib->settings->get_int(lib->settings,
42500c27	483	"%s.plugins.ha.heartbeat_timeout", DEFAULT_HEARTBEAT_TIMEOUT,
d223fe80	484	lib->ns),
e2d2b542	485	.autobalance = lib->settings->get_int(lib->settings,
d223fe80	486	"%s.plugins.ha.autobalance", 0, lib->ns),
00c1bd06	487	);
3912fdb1	488
3e8caf6a MW	489	if (monitor)
3e8caf6a MW	490	{
65d15aff MW	491	DBG1(DBG_CFG, "starting HA heartbeat, delay %dms, timeout %dms",
65d15aff MW	492	this->heartbeat_delay, this->heartbeat_timeout);
2071dd63	493	start_heartbeat(this);
3e8caf6a MW	494	start_watchdog(this);
3e8caf6a MW	495	}
e2d2b542 MW	496	if (this->autobalance)
	497	{
	498	start_autobalance(this);
	499	}
e262f4e5	500
765935c8 MW	501	return &this->public;
765935c8 MW	502	}