Centralized thread cancellation in processor_t

This ensures that no threads are active when plugins and the rest of the
daemon are unloaded.

callback_job_t was simplified a lot in the process as its main
functionality is now contained in processor_t.  The parent-child
relationships were abandoned as these were only needed to simplify job
cancellation.

src/libcharon/plugins/android/android_service.c

@@ -41,11 +41,6 @@ struct private_android_service_t {
 	 */
 	ike_sa_t *ike_sa;
 
-	/**
-	 * job that handles requests from the Android control socket
-	 */
-	callback_job_t *job;
-
 	/**
 	 * android credentials
 	 */
@@ -384,9 +379,9 @@ android_service_t *android_service_create(android_creds_t *creds)
 	}
 
 	charon->bus->add_listener(charon->bus, &this->public.listener);
-	this->job = callback_job_create((callback_job_cb_t)initiate, this,
-									NULL, NULL);
-	lib->processor->queue_job(lib->processor, (job_t*)this->job);
+	lib->processor->queue_job(lib->processor,
+		(job_t*)callback_job_create((callback_job_cb_t)initiate, this,
+									NULL, NULL));
 
 	return &this->public;
 }

src/libcharon/plugins/dhcp/dhcp_socket.c

@@ -105,11 +105,6 @@ struct private_dhcp_socket_t {
 	 * DHCP server address, or broadcast
 	 */
 	host_t *dst;
-
-	/**
-	 * Callback job receiving DHCP responses
-	 */
-	callback_job_t *job;
 };
 
 /**
@@ -613,10 +608,6 @@ static job_requeue_t receive_dhcp(private_dhcp_socket_t *this)
 METHOD(dhcp_socket_t, destroy, void,
 	private_dhcp_socket_t *this)
 {
-	if (this->job)
-	{
-		this->job->cancel(this->job);
-	}
 	while (this->waiting)
 	{
 		this->condvar->signal(this->condvar);
@@ -761,9 +752,9 @@ dhcp_socket_t *dhcp_socket_create()
 		return NULL;
 	}
 
-	this->job = callback_job_create_with_prio((callback_job_cb_t)receive_dhcp,
-									this, NULL, NULL, JOB_PRIO_CRITICAL);
-	lib->processor->queue_job(lib->processor, (job_t*)this->job);
+	lib->processor->queue_job(lib->processor,
+		(job_t*)callback_job_create_with_prio((callback_job_cb_t)receive_dhcp,
+			this, NULL, (callback_job_cancel_t)return_false, JOB_PRIO_CRITICAL));
 
 	return &this->public;
 }

src/libcharon/plugins/duplicheck/duplicheck_notify.c

@@ -42,11 +42,6 @@ struct private_duplicheck_notify_t {
 	 */
 	duplicheck_notify_t public;
 
-	/**
-	 * Callback job dispatching connections
-	 */
-	callback_job_t *job;
-
 	/**
 	 * Mutex to lock list
 	 */
@@ -167,10 +162,6 @@ METHOD(duplicheck_notify_t, destroy, void,
 	enumerator_t *enumerator;
 	uintptr_t fd;
 
-	if (this->job)
-	{
-		this->job->cancel(this->job);
-	}
 	enumerator = this->connected->create_enumerator(this->connected);
 	while (enumerator->enumerate(enumerator, &fd))
 	{
@@ -203,9 +194,9 @@ duplicheck_notify_t *duplicheck_notify_create()
 		destroy(this);
 		return NULL;
 	}
-	this->job = callback_job_create_with_prio((callback_job_cb_t)receive,
-									this, NULL, NULL, JOB_PRIO_CRITICAL);
-	lib->processor->queue_job(lib->processor, (job_t*)this->job);
+	lib->processor->queue_job(lib->processor,
+		(job_t*)callback_job_create_with_prio((callback_job_cb_t)receive, this,
+				NULL, (callback_job_cancel_t)return_false, JOB_PRIO_CRITICAL));
 
 	return &this->public;
 }

src/libcharon/plugins/eap_radius/eap_radius_dae.c

@@ -52,11 +52,6 @@ struct private_eap_radius_dae_t {
 	 */
 	int fd;
 
-	/**
-	 * Listen job
-	 */
-	callback_job_t *job;
-
 	/**
 	 * RADIUS shared secret for DAE exchanges
 	 */
@@ -481,10 +476,6 @@ static bool open_socket(private_eap_radius_dae_t *this)
 METHOD(eap_radius_dae_t, destroy, void,
 	private_eap_radius_dae_t *this)
 {
-	if (this->job)
-	{
-		this->job->cancel(this->job);
-	}
 	if (this->fd != -1)
 	{
 		close(this->fd);
@@ -538,9 +529,9 @@ eap_radius_dae_t *eap_radius_dae_create(eap_radius_accounting_t *accounting)
 		return NULL;
 	}
 
-	this->job = callback_job_create_with_prio((callback_job_cb_t)receive,
-										this, NULL, NULL, JOB_PRIO_CRITICAL);
-	lib->processor->queue_job(lib->processor, (job_t*)this->job);
+	lib->processor->queue_job(lib->processor,
+		(job_t*)callback_job_create_with_prio((callback_job_cb_t)receive,
+			this, NULL, (callback_job_cancel_t)return_false, JOB_PRIO_CRITICAL));
 
 	return &this->public;
 }

src/libcharon/plugins/farp/farp_spoofer.c

@@ -44,11 +44,6 @@ struct private_farp_spoofer_t {
 	 */
 	farp_listener_t *listener;
 
-	/**
-	 * Callback job to read ARP requests
-	 */
-	callback_job_t *job;
-
 	/**
 	 * RAW socket for ARP requests
 	 */
@@ -135,7 +130,6 @@ static job_requeue_t receive_arp(private_farp_spoofer_t *this)
 METHOD(farp_spoofer_t, destroy, void,
 	private_farp_spoofer_t *this)
 {
-	this->job->cancel(this->job);
 	close(this->skt);
 	free(this);
 }
@@ -189,9 +183,9 @@ farp_spoofer_t *farp_spoofer_create(farp_listener_t *listener)
 		return NULL;
 	}
 
-	this->job = callback_job_create_with_prio((callback_job_cb_t)receive_arp,
-									this, NULL, NULL, JOB_PRIO_CRITICAL);
-	lib->processor->queue_job(lib->processor, (job_t*)this->job);
+	lib->processor->queue_job(lib->processor,
+		(job_t*)callback_job_create_with_prio((callback_job_cb_t)receive_arp,
+			this, NULL, (callback_job_cancel_t)return_false, JOB_PRIO_CRITICAL));
 
 	return &this->public;
 }

src/libcharon/plugins/ha/ha_ctl.c

@@ -48,11 +48,6 @@ struct private_ha_ctl_t {
 	 * Resynchronization message cache
 	 */
 	ha_cache_t *cache;
-
-	/**
-	 * FIFO reader thread
-	 */
-	callback_job_t *job;
 };
 
 /**
@@ -105,7 +100,6 @@ static job_requeue_t dispatch_fifo(private_ha_ctl_t *this)
 METHOD(ha_ctl_t, destroy, void,
 	private_ha_ctl_t *this)
 {
-	this->job->cancel(this->job);
 	free(this);
 }
 
@@ -141,9 +135,9 @@ ha_ctl_t *ha_ctl_create(ha_segments_t *segments, ha_cache_t *cache)
 			 strerror(errno));
 	}
 
-	this->job = callback_job_create_with_prio((callback_job_cb_t)dispatch_fifo,
-										this, NULL, NULL, JOB_PRIO_CRITICAL);
-	lib->processor->queue_job(lib->processor, (job_t*)this->job);
+	lib->processor->queue_job(lib->processor,
+		(job_t*)callback_job_create_with_prio((callback_job_cb_t)dispatch_fifo,
+			this, NULL, (callback_job_cancel_t)return_false, JOB_PRIO_CRITICAL));
 	return &this->public;
 }
 

src/libcharon/plugins/ha/ha_dispatcher.c

@@ -58,11 +58,6 @@ struct private_ha_dispatcher_t {
 	 * HA enabled pool
 	 */
 	ha_attribute_t *attr;
-
-	/**
-	 * Dispatcher job
-	 */
-	callback_job_t *job;
 };
 
 /**
@@ -1032,7 +1027,6 @@ static job_requeue_t dispatch(private_ha_dispatcher_t *this)
 METHOD(ha_dispatcher_t, destroy, void,
 	private_ha_dispatcher_t *this)
 {
-	this->job->cancel(this->job);
 	free(this);
 }
 
@@ -1056,9 +1050,9 @@ ha_dispatcher_t *ha_dispatcher_create(ha_socket_t *socket,
 		.kernel = kernel,
 		.attr = attr,
 	);
-	this->job = callback_job_create_with_prio((callback_job_cb_t)dispatch,
-										this, NULL, NULL, JOB_PRIO_CRITICAL);
-	lib->processor->queue_job(lib->processor, (job_t*)this->job);
+	lib->processor->queue_job(lib->processor,
+		(job_t*)callback_job_create_with_prio((callback_job_cb_t)dispatch, this,
+				NULL, (callback_job_cancel_t)return_false, JOB_PRIO_CRITICAL));
 
 	return &this->public;
 }

src/libcharon/plugins/ha/ha_segments.c

@@ -61,11 +61,6 @@ struct private_ha_segments_t {
 	 */
 	condvar_t *condvar;
 
-	/**
-	 * Job checking for heartbeats
-	 */
-	callback_job_t *job;
-
 	/**
 	 * Total number of ClusterIP segments
 	 */
@@ -81,6 +76,11 @@ struct private_ha_segments_t {
 	 */
 	u_int node;
 
+	/**
+	 * Are we checking for heartbeats?
+	 */
+	bool heartbeat_active;
+
 	/**
 	 * Interval we send hearbeats
 	 */
@@ -237,7 +237,7 @@ METHOD(listener_t, alert_hook, bool,
 {
 	if (alert == ALERT_SHUTDOWN_SIGNAL)
 	{
-		if (this->job)
+		if (this->heartbeat_active)
 		{
 			DBG1(DBG_CFG, "HA heartbeat active, dropping all segments");
 			deactivate(this, 0, TRUE);
@@ -269,7 +269,7 @@ static job_requeue_t watchdog(private_ha_segments_t *this)
 		DBG1(DBG_CFG, "no heartbeat received, taking all segments");
 		activate(this, 0, TRUE);
 		/* disable heartbeat detection util we get one */
-		this->job = NULL;
+		this->heartbeat_active = FALSE;
 		return JOB_REQUEUE_NONE;
 	}
 	return JOB_REQUEUE_DIRECT;
@@ -280,9 +280,10 @@ static job_requeue_t watchdog(private_ha_segments_t *this)
  */
 static void start_watchdog(private_ha_segments_t *this)
 {
-	this->job = callback_job_create_with_prio((callback_job_cb_t)watchdog,
-									this, NULL, NULL, JOB_PRIO_CRITICAL);
-	lib->processor->queue_job(lib->processor, (job_t*)this->job);
+	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,
@@ -312,10 +313,10 @@ METHOD(ha_segments_t, handle_status, void,
 		}
 	}
 
-	this->mutex->unlock(this->mutex);
 	this->condvar->signal(this->condvar);
+	this->mutex->unlock(this->mutex);
 
-	if (!this->job)
+	if (!this->heartbeat_active)
 	{
 		DBG1(DBG_CFG, "received heartbeat, reenabling watchdog");
 		start_watchdog(this);
@@ -361,10 +362,6 @@ METHOD(ha_segments_t, is_active, bool,
 METHOD(ha_segments_t, destroy, void,
 	private_ha_segments_t *this)
 {
-	if (this->job)
-	{
-		this->job->cancel(this->job);
-	}
 	this->mutex->destroy(this->mutex);
 	this->condvar->destroy(this->condvar);
 	free(this);

src/libcharon/plugins/maemo/maemo_service.c

@@ -431,8 +431,10 @@ static job_requeue_t run(private_maemo_service_t *this)
 	return JOB_REQUEUE_NONE;
 }
 
-METHOD(maemo_service_t, destroy, void,
-	   private_maemo_service_t *this)
+/**
+ * Cancel the GLib Main Event Loop
+ */
+static bool cancel(private_maemo_service_t *this)
 {
 	if (this->loop)
 	{
@@ -442,6 +444,12 @@ METHOD(maemo_service_t, destroy, void,
 		}
 		g_main_loop_unref(this->loop);
 	}
+	return TRUE;
+}
+
+METHOD(maemo_service_t, destroy, void,
+	   private_maemo_service_t *this)
+{
 	if (this->context)
 	{
 		osso_rpc_unset_cb_f(this->context,
@@ -510,8 +518,8 @@ maemo_service_t *maemo_service_create()
 	}
 
 	lib->processor->queue_job(lib->processor,
-				(job_t*)callback_job_create_with_prio((callback_job_cb_t)run,
-										this, NULL, NULL, JOB_PRIO_CRITICAL));
+		(job_t*)callback_job_create_with_prio((callback_job_cb_t)run, this,
+				NULL, (callback_job_cancel_t)cancel, JOB_PRIO_CRITICAL));
 
 	return &this->public;
 }

src/libcharon/plugins/smp/smp.c

@@ -49,11 +49,6 @@ struct private_smp_t {
 	 * XML unix socket fd
 	 */
 	int socket;
-
-	/**
-	 * job accepting stroke messages
-	 */
-	callback_job_t *job;
 };
 
 ENUM(ike_sa_state_lower_names, IKE_CREATED, IKE_DELETING,
@@ -704,7 +699,8 @@ static job_requeue_t dispatch(private_smp_t *this)
 
 	fdp = malloc_thing(int);
 	*fdp = fd;
-	job = callback_job_create((callback_job_cb_t)process, fdp, free, this->job);
+	job = callback_job_create((callback_job_cb_t)process, fdp, free,
+							  (callback_job_cancel_t)return_false);
 	lib->processor->queue_job(lib->processor, (job_t*)job);
 
 	return JOB_REQUEUE_DIRECT;
@@ -719,7 +715,6 @@ METHOD(plugin_t, get_name, char*,
 METHOD(plugin_t, destroy, void,
 	private_smp_t *this)
 {
-	this->job->cancel(this->job);
 	close(this->socket);
 	free(this);
 }
@@ -775,9 +770,9 @@ plugin_t *smp_plugin_create()
 		return NULL;
 	}
 
-	this->job = callback_job_create_with_prio((callback_job_cb_t)dispatch,
-										this, NULL, NULL, JOB_PRIO_CRITICAL);
-	lib->processor->queue_job(lib->processor, (job_t*)this->job);
+	lib->processor->queue_job(lib->processor,
+		(job_t*)callback_job_create_with_prio((callback_job_cb_t)dispatch, this,
+				NULL, (callback_job_cancel_t)return_false, JOB_PRIO_CRITICAL));
 
 	return &this->public.plugin;
 }

src/libcharon/plugins/stroke/stroke_socket.c

@@ -63,16 +63,6 @@ struct private_stroke_socket_t {
 	 */
 	int socket;
 
-	/**
-	 * job accepting stroke messages
-	 */
-	callback_job_t *receiver;
-
-	/**
-	 * job handling stroke messages
-	 */
-	callback_job_t *handler;
-
 	/**
 	 * queued stroke commands
 	 */
@@ -702,7 +692,7 @@ static job_requeue_t handle(private_stroke_socket_t *this)
 	this->handling++;
 	thread_cleanup_pop(TRUE);
 	job = callback_job_create_with_prio((callback_job_cb_t)process, ctx,
-			(void*)stroke_job_context_destroy, this->handler, JOB_PRIO_HIGH);
+			(void*)stroke_job_context_destroy, NULL, JOB_PRIO_HIGH);
 	lib->processor->queue_job(lib->processor, (job_t*)job);
 	return JOB_REQUEUE_DIRECT;
 }
@@ -787,8 +777,6 @@ static bool open_socket(private_stroke_socket_t *this)
 METHOD(stroke_socket_t, destroy, void,
 	private_stroke_socket_t *this)
 {
-	this->handler->cancel(this->handler);
-	this->receiver->cancel(this->receiver);
 	this->commands->destroy_function(this->commands, (void*)stroke_job_context_destroy);
 	this->condvar->destroy(this->condvar);
 	this->mutex->destroy(this->mutex);
@@ -843,13 +831,13 @@ stroke_socket_t *stroke_socket_create()
 	charon->backends->add_backend(charon->backends, &this->config->backend);
 	hydra->attributes->add_provider(hydra->attributes, &this->attribute->provider);
 
-	this->receiver = callback_job_create_with_prio((callback_job_cb_t)receive,
-										this, NULL, NULL, JOB_PRIO_CRITICAL);
-	lib->processor->queue_job(lib->processor, (job_t*)this->receiver);
+	lib->processor->queue_job(lib->processor,
+		(job_t*)callback_job_create_with_prio((callback_job_cb_t)receive, this,
+				NULL, (callback_job_cancel_t)return_false, JOB_PRIO_CRITICAL));
 
-	this->handler = callback_job_create_with_prio((callback_job_cb_t)handle,
-										this, NULL, NULL, JOB_PRIO_CRITICAL);
-	lib->processor->queue_job(lib->processor, (job_t*)this->handler);
+	lib->processor->queue_job(lib->processor,
+		(job_t*)callback_job_create_with_prio((callback_job_cb_t)handle, this,
+				NULL, (callback_job_cancel_t)return_false, JOB_PRIO_CRITICAL));
 
 	return &this->public;
 }

src/libcharon/plugins/tnc_pdp/tnc_pdp.c

@@ -66,11 +66,6 @@ struct private_tnc_pdp_t {
 	 */
 	int ipv6;
 
-	/**
-	 * Callback job dispatching commands
-	 */
-	callback_job_t *job;
-
 	/**
 	 * RADIUS shared secret
 	 */
@@ -546,10 +541,6 @@ static job_requeue_t receive(private_tnc_pdp_t *this)
 METHOD(tnc_pdp_t, destroy, void,
 	private_tnc_pdp_t *this)
 {
-	if (this->job)
-	{
-		this->job->cancel(this->job);
-	}
 	if (this->ipv4)
 	{
 		close(this->ipv4);
@@ -639,9 +630,9 @@ tnc_pdp_t *tnc_pdp_create(u_int16_t port)
 	}
 	DBG1(DBG_IKE, "eap method %N selected", eap_type_names, this->type);
 
-	this->job = callback_job_create_with_prio((callback_job_cb_t)receive,
-										this, NULL, NULL, JOB_PRIO_CRITICAL);
-	lib->processor->queue_job(lib->processor, (job_t*)this->job);
+	lib->processor->queue_job(lib->processor,
+		(job_t*)callback_job_create_with_prio((callback_job_cb_t)receive, this,
+				NULL, (callback_job_cancel_t)return_false, JOB_PRIO_CRITICAL));
 
 	return &this->public;
 }

src/libcharon/plugins/uci/uci_control.c

@@ -42,11 +42,6 @@ struct private_uci_control_t {
 	 * Public part
 	 */
 	uci_control_t public;
-
-	/**
-	 * Job
-	 */
-	callback_job_t *job;
 };
 
 /**
@@ -269,7 +264,6 @@ static job_requeue_t receive(private_uci_control_t *this)
 METHOD(uci_control_t, destroy, void,
 	private_uci_control_t *this)
 {
-	this->job->cancel(this->job);
 	unlink(FIFO_FILE);
 	free(this);
 }
@@ -295,9 +289,10 @@ uci_control_t *uci_control_create()
 	}
 	else
 	{
-		this->job = callback_job_create_with_prio((callback_job_cb_t)receive,
-										this, NULL, NULL, JOB_PRIO_CRITICAL);
-		lib->processor->queue_job(lib->processor, (job_t*)this->job);
+		lib->processor->queue_job(lib->processor,
+			(job_t*)callback_job_create_with_prio((callback_job_cb_t)receive,
+							this, NULL, (callback_job_cancel_t)return_false,
+							JOB_PRIO_CRITICAL));
 	}
 	return &this->public;
 }

src/libcharon/plugins/whitelist/whitelist_control.c

@@ -49,11 +49,6 @@ struct private_whitelist_control_t {
 	 * Whitelist unix socket file descriptor
 	 */
 	int socket;
-
-	/**
-	 * Callback job dispatching commands
-	 */
-	callback_job_t *job;
 };
 
 /**
@@ -200,7 +195,6 @@ static job_requeue_t receive(private_whitelist_control_t *this)
 METHOD(whitelist_control_t, destroy, void,
 	private_whitelist_control_t *this)
 {
-	this->job->cancel(this->job);
 	close(this->socket);
 	free(this);
 }
@@ -225,9 +219,9 @@ whitelist_control_t *whitelist_control_create(whitelist_listener_t *listener)
 		return NULL;
 	}
 
-	this->job = callback_job_create_with_prio((callback_job_cb_t)receive,
-										this, NULL, NULL, JOB_PRIO_CRITICAL);
-	lib->processor->queue_job(lib->processor, (job_t*)this->job);
+	lib->processor->queue_job(lib->processor,
+		(job_t*)callback_job_create_with_prio((callback_job_cb_t)receive, this,
+				NULL, (callback_job_cancel_t)return_false, JOB_PRIO_CRITICAL));
 
 	return &this->public;
 }