ported interfaces to new threading functions (incomplete)

src/charon/control/interfaces/dbus_interface.c

@@ -30,6 +30,7 @@
 
 #include <library.h>
 #include <daemon.h>
+#include <processing/jobs/callback_job.h>
 
 
 #define NM_DBUS_SERVICE_STRONG "org.freedesktop.NetworkManager.strongswan"
@@ -64,9 +65,9 @@ struct private_dbus_interface_t {
 	NMVPNState state;
 	
 	/**
-	 * dispatcher thread for DBUS messages
+	 * job accepting stroke messages
 	 */
-	pthread_t thread;
+	callback_job_t *job;
 	
 	/**
 	 * name of the currently active connection
@@ -392,14 +393,13 @@ static DBusHandlerResult signal_handler(DBusConnection *con, DBusMessage *msg,
 /**
  * dispatcher function processed by a seperate thread
  */
-static void dispatch(private_dbus_interface_t *this)
+static job_requeue_t dispatch(private_dbus_interface_t *this)
 {
-	charon->drop_capabilities(charon, TRUE);
+	if (dbus_connection_read_write_dispatch(this->conn, -1))
-
-	while (dbus_connection_read_write_dispatch(this->conn, -1))
 	{
-		/* nothing */
+		return JOB_REQUEUE_DIRECT;
 	}
+	return JOB_REQUEUE_NONE;
 }
 
 /**
@@ -407,8 +407,7 @@ static void dispatch(private_dbus_interface_t *this)
  */
 static void destroy(private_dbus_interface_t *this)
 {
-	pthread_cancel(this->thread);
+	this->job->cancel(this->job);
-	pthread_join(this->thread, NULL);
 	dbus_connection_close(this->conn);
 	dbus_error_free(&this->err);
 	dbus_shutdown();
@@ -469,10 +468,8 @@ interface_t *interface_create()
 	this->state = NM_VPN_STATE_INIT;
 	set_state(this, NM_VPN_STATE_STOPPED);
 
-	if (pthread_create(&this->thread, NULL, (void*(*)(void*))dispatch, this) != 0)
+	this->job = callback_job_create((callback_job_cb_t)dispatch, this, NULL, NULL);
-	{
+	charon->processor->queue_job(charon->processor, (job_t*)this->job);
-		charon->kill(charon, "unable to create stroke thread");
-	}
 
 	return &this->public.interface;
 }

src/charon/control/interfaces/xml_interface.c

@@ -37,7 +37,7 @@
 
 #include <library.h>
 #include <daemon.h>
-
+#include <processing/jobs/callback_job.h>
 
 static struct sockaddr_un socket_addr = { AF_UNIX, "/var/run/charon.xml"};
 
@@ -60,15 +60,16 @@ struct private_xml_interface_t {
 	int socket;
 	
 	/**
-	 * thread receiving messages
+	 * job accepting stroke messages
 	 */
-	pthread_t thread;
+	callback_job_t *job;
 };
 
-static void get(private_xml_interface_t *this, 
+/**
-				xmlTextReaderPtr reader, xmlTextWriterPtr writer)
+ * process a getRequest message
+ */
+static void process_get(xmlTextReaderPtr reader, xmlTextWriterPtr writer)
 {
-
 	if (/* <GetResponse> */
 		xmlTextWriterStartElement(writer, "GetResponse") < 0 ||
 		/*   <Status Code="200"><Message/></Status> */
@@ -85,48 +86,16 @@ static void get(private_xml_interface_t *this,
 	{
 		DBG1(DBG_CFG, "error writing XML document (GetResponse)");
 	}
-		
-
-/*
-					   DBG1(DBG_CFG, "%d %d %s %d %d %s", 
-						    xmlTextReaderDepth(reader),
-						    ,
-						    xmlTextReaderConstName(reader),
-						    xmlTextReaderIsEmptyElement(reader),
-						    xmlTextReaderHasValue(reader),
-						    xmlTextReaderConstValue(reader));
-		*/
 }
 
-static void receive(private_xml_interface_t *this)
+/**
+ * read from a opened connection and process it
+ */
+static job_requeue_t process(int *fdp)
 {
-	charon->drop_capabilities(charon, TRUE);
+	int oldstate, fd = *fdp;
-	
-	/* disable cancellation by default */
-	pthread_setcancelstate(PTHREAD_CANCEL_DISABLE, NULL);
-	
-	while (TRUE)
-	{
-		struct sockaddr_un strokeaddr;
-		int strokeaddrlen = sizeof(strokeaddr);
-		int oldstate;
-		int fd;
 	char buffer[4096];
 	size_t len;
-		
-		/* wait for connections, but allow thread to terminate */
-		pthread_setcancelstate(PTHREAD_CANCEL_ENABLE, &oldstate);
-		fd = accept(this->socket, (struct sockaddr *)&strokeaddr, &strokeaddrlen);
-		pthread_setcancelstate(oldstate, NULL);
-		
-		if (fd < 0)
-		{
-			DBG1(DBG_CFG, "accepting SMP XML socket failed: %s", strerror(errno));
-			continue;
-		}
-		DBG2(DBG_CFG, "SMP XML connection opened");
-		while (TRUE)
-		{
 	xmlTextReaderPtr reader;
 	xmlTextWriterPtr writer;
 	
@@ -137,14 +106,14 @@ static void receive(private_xml_interface_t *this)
 	{
 		close(fd);
 		DBG2(DBG_CFG, "SMP XML connection closed");
-				break;
+		return JOB_REQUEUE_NONE;
 	}
 	
 	reader = xmlReaderForMemory(buffer, len, NULL, NULL, 0);
 	if (reader == NULL)
 	{
 		DBG1(DBG_CFG, "opening SMP XML reader failed");
-				continue;
+		return JOB_REQUEUE_FAIR;;
 	}
 	
 	writer = xmlNewTextWriter(xmlOutputBufferCreateFd(fd, NULL));
@@ -152,7 +121,7 @@ static void receive(private_xml_interface_t *this)
 	{
 		xmlFreeTextReader(reader);
 		DBG1(DBG_CFG, "opening SMP XML writer failed");
-				continue;
+		return JOB_REQUEUE_FAIR;;
 	}
 	
 	/* create the standard message parts */
@@ -167,7 +136,7 @@ static void receive(private_xml_interface_t *this)
 		xmlFreeTextReader(reader);
 		xmlFreeTextWriter(writer);
 		DBG1(DBG_CFG, "creating SMP XML message failed");
-				continue;
+		return JOB_REQUEUE_FAIR;;
 	}
 	
     while (TRUE)
@@ -176,12 +145,11 @@ static void receive(private_xml_interface_t *this)
     	{
     		case 1:
     		{
-						if (xmlTextReaderNodeType(reader) ==
+				if (xmlTextReaderNodeType(reader) == XML_READER_TYPE_ELEMENT)
-							XML_READER_TYPE_ELEMENT)
 				{
 					if (streq(xmlTextReaderConstName(reader), "GetRequest"))
 					{
-								get(this, reader, writer);
+						process_get(reader, writer);
 						break;
 					}
 				}
@@ -203,10 +171,39 @@ static void receive(private_xml_interface_t *this)
 		DBG1(DBG_CFG, "completing SMP XML message failed");
     }
     xmlFreeTextWriter(writer);
+    
     /* write a newline to indicate end of xml */
     write(fd, "\n", 1);
+    return JOB_REQUEUE_FAIR;;
 }
+
+/**
+ * accept from XML socket and create jobs to process connections
+ */
+static job_requeue_t dispatch(private_xml_interface_t *this)
+{
+	struct sockaddr_un strokeaddr;
+	int oldstate, fd, *fdp, strokeaddrlen = sizeof(strokeaddr);
+	callback_job_t *job;
+	
+	/* wait for connections, but allow thread to terminate */
+	pthread_setcancelstate(PTHREAD_CANCEL_ENABLE, &oldstate);
+	fd = accept(this->socket, (struct sockaddr *)&strokeaddr, &strokeaddrlen);
+	pthread_setcancelstate(oldstate, NULL);
+	
+	if (fd < 0)
+	{
+		DBG1(DBG_CFG, "accepting SMP XML socket failed: %s", strerror(errno));
+		sleep(1);
+		return JOB_REQUEUE_FAIR;;
 	}
+	
+	fdp = malloc_thing(int);
+	*fdp = fd;
+	job = callback_job_create((callback_job_cb_t)process, fdp, free, this->job);
+	charon->processor->queue_job(charon->processor, (job_t*)job);
+	
+	return JOB_REQUEUE_DIRECT;
 }
 
 /**
@@ -214,9 +211,7 @@ static void receive(private_xml_interface_t *this)
  */
 static void destroy(private_xml_interface_t *this)
 {
-	pthread_cancel(this->thread);
+	this->job->cancel(this->job);
-	pthread_join(this->thread, NULL);
-	close(this->socket);
 	unlink(socket_addr.sun_path);
 	free(this);
 }
@@ -258,14 +253,8 @@ interface_t *interface_create()
 		return NULL;
 	}
 
-	if (pthread_create(&this->thread, NULL, (void*(*)(void*))receive, this) != 0)
+	this->job = callback_job_create((callback_job_cb_t)dispatch, this, NULL, NULL);
-	{
+	charon->processor->queue_job(charon->processor, (job_t*)this->job);
-		DBG1(DBG_CFG, "could not create XML socket thread: %s", strerror(errno));
-		close(this->socket);
-		unlink(socket_addr.sun_path);
-		free(this);
-		return NULL;
-	}
 	
 	return &this->public.interface;
 }