removed send_queue, handled internally in sender_t know

do header parsing in receiver, ready for cookie integration

src/charon/threads/receiver.c

@@ -31,7 +31,7 @@
 #include <network/packet.h>
 #include <queues/job_queue.h>
 #include <queues/jobs/job.h>
-#include <queues/jobs/incoming_packet_job.h>
+#include <queues/jobs/process_message_job.h>
 
 typedef struct block_t block_t;
 
@@ -87,75 +87,6 @@ struct private_receiver_t {
 	 pthread_mutex_t mutex;
 };
 
-/**
- * Implementation of receiver_t.receive_packets.
- */
-static void receive_packets(private_receiver_t * this)
-{
-	packet_t *packet;
-	job_t *job;
-	
-	/* cancellation disabled by default */
-	pthread_setcancelstate(PTHREAD_CANCEL_DISABLE, NULL);
-	DBG1(DBG_NET, "receiver thread running, thread_ID: %06u", 
-		 (int)pthread_self());
-	
-	while (TRUE)
-	{
-		if (charon->socket->receive(charon->socket, &packet) != SUCCESS)
-		{
-			DBG1(DBG_NET, "receiving from socket failed!");
-			continue;
-		}
-		
-		if (this->blocks->get_count(this->blocks))
-		{
-			iterator_t *iterator;
-			block_t *blocked;
-			bool found = FALSE;
-			u_int32_t now = time(NULL);
-			
-			pthread_mutex_lock(&this->mutex);
-			iterator = this->blocks->create_iterator(this->blocks, TRUE);
-			while (iterator->iterate(iterator, (void**)&blocked))
-			{
-				if (now > blocked->timeout)
-				{
-					/* block expired, remove */
-					iterator->remove(iterator);
-					block_destroy(blocked);
-					continue;
-				}
-			
-				if (!blocked->ip->ip_equals(blocked->ip, 
-											packet->get_source(packet)))
-				{
-					/* no match, get next */
-					continue;
-				}
-				
-				/* IP is blocked */
-				DBG2(DBG_NET, "received packets source address %H blocked", 
-					 blocked->ip);
-				packet->destroy(packet);
-				found = TRUE;
-				break;
-			}
-			iterator->destroy(iterator);
-			pthread_mutex_unlock(&this->mutex);
-			if (found)
-			{
-				/* get next packet */
-				continue;
-			}
-		}
-		
-		DBG2(DBG_NET, "creating job from packet");
-		job = (job_t *) incoming_packet_job_create(packet);
-		charon->job_queue->add(charon->job_queue, job);
-	}
-}
-
 /**
  * Implementation of receiver_t.block
  */
@@ -172,6 +103,100 @@ static void block(private_receiver_t *this, host_t *ip, u_int32_t seconds)
 	pthread_mutex_unlock(&this->mutex);
 }
 
+/**
+ * check if an IP is blocked
+ */
+static bool is_blocked(private_receiver_t *this, host_t *ip)
+{
+	bool found = FALSE;
+	
+	if (this->blocks->get_count(this->blocks))
+	{
+		iterator_t *iterator;
+		block_t *blocked;
+		u_int32_t now = time(NULL);
+		
+		pthread_mutex_lock(&this->mutex);
+		iterator = this->blocks->create_iterator(this->blocks, TRUE);
+		while (iterator->iterate(iterator, (void**)&blocked))
+		{
+			if (now > blocked->timeout)
+			{
+				/* blocking expired, remove */
+				iterator->remove(iterator);
+				block_destroy(blocked);
+				continue;
+			}
+		
+			if (!ip->ip_equals(ip, blocked->ip))
+			{
+				/* no match, get next */
+				continue;
+			}
+			
+			/* blocked */
+			DBG2(DBG_NET, "received packet source address %H blocked", ip);
+			found = TRUE;
+			break;
+		}
+		iterator->destroy(iterator);
+		pthread_mutex_unlock(&this->mutex);
+	}
+	return found;
+}
+
+/**
+ * Implementation of receiver_t.receive_packets.
+ */
+static void receive_packets(private_receiver_t * this)
+{
+	packet_t *packet;
+	message_t *message;
+	job_t *job;
+	
+	pthread_setcancelstate(PTHREAD_CANCEL_DISABLE, NULL);
+	DBG1(DBG_NET, "receiver thread running, thread_ID: %06u", 
+		 (int)pthread_self());
+	
+	while (TRUE)
+	{
+		if (charon->socket->receive(charon->socket, &packet) != SUCCESS)
+		{
+			DBG1(DBG_NET, "receiving from socket failed!");
+			continue;
+		}
+		
+		if (is_blocked(this, packet->get_source(packet)))
+		{
+			packet->destroy(packet);
+			continue;
+		}
+		
+		message = message_create_from_packet(packet);
+		if (message->parse_header(message) != SUCCESS)
+		{
+			DBG1(DBG_NET, "received invalid IKE header from %H, ignored",
+				 packet->get_source(packet));
+			message->destroy(message);
+			continue;
+		}
+		
+		if (message->get_major_version(message) != IKE_MAJOR_VERSION)
+		{
+			DBG1(DBG_NET, "received unsupported IKE version %d.%d from %H, "
+				 "ignored",	message->get_major_version(message), 
+				 message->get_minor_version(message), packet->get_source(packet));
+			message->destroy(message);
+			continue;
+		}
+			
+		
+		job = (job_t *)process_message_job_create(message);
+		charon->job_queue->add(charon->job_queue, job);
+	}
+}
+
+
 /**
  * Implementation of receiver_t.destroy.
  */

src/charon/threads/receiver.h

@@ -32,9 +32,8 @@ typedef struct receiver_t receiver_t;
 /**
  * @brief Receives packets from the socket and adds them to the job queue.
  * 
- * The receiver starts a thread, wich reads on the blocking socket. If 
- * data is available, a packet_t object is created, wrapped
- * in an incoming_packet_job_t and added to the job queue.
+ * The receiver starts a thread, wich reads on the blocking socket. A received
+ * packet is preparsed a process_message_job is queued in the job queue.
  * 
  * @b Constructors:
  *  - receiver_create()

src/charon/threads/sender.c

@@ -28,8 +28,6 @@
 
 #include <daemon.h>
 #include <network/socket.h>
-#include <network/packet.h>
-#include <queues/send_queue.h>
 
 
 typedef struct private_sender_t private_sender_t;
@@ -47,33 +45,72 @@ struct private_sender_t {
 	  * Assigned thread.
 	  */
 	 pthread_t assigned_thread;
+	 
+	/**
+	 * The packets are stored in a linked list
+	 */
+	linked_list_t *list;
 
+	/**
+	 * mutex to synchronize access to list
+	 */
+	pthread_mutex_t mutex;
+
+	/**
+	 * condvar to signal for packets in list
+	 */
+	pthread_cond_t condvar;
 };
 
+/**
+ * implements sender_t.send
+ */
+static void send_(private_sender_t *this, packet_t *packet)
+{
+	host_t *src, *dst;
+	
+	src = packet->get_source(packet);
+	dst = packet->get_destination(packet);
+	DBG1(DBG_NET, "sending packet: from %#H to %#H", src, dst);
+	
+	pthread_mutex_lock(&this->mutex);
+	this->list->insert_last(this->list, packet);
+	pthread_mutex_unlock(&this->mutex);
+	pthread_cond_signal(&this->condvar);
+}
+
 /**
  * Implementation of private_sender_t.send_packets.
  */
 static void send_packets(private_sender_t * this)
 {
-	packet_t *current_packet;
-	status_t status;
 	
 	/* cancellation disabled by default */
 	pthread_setcancelstate(PTHREAD_CANCEL_DISABLE, NULL);
-
-	DBG1(DBG_NET, "sender thread running, thread_ID: %06u",
-		 (int)pthread_self());
+	DBG1(DBG_NET, "sender thread running, thread_ID: %06u", (int)pthread_self());
 
 	while (TRUE)
 	{
-		current_packet = charon->send_queue->get(charon->send_queue);
-		DBG2(DBG_NET, "got a packet, sending it");
-		status = charon->socket->send(charon->socket, current_packet);
-		if (status != SUCCESS)
+		packet_t *packet;
+		int oldstate;
+	
+		pthread_mutex_lock(&this->mutex);
+		/* go to wait while no packets available */
+		while (this->list->get_count(this->list) == 0)
 		{
-			DBG1(DBG_NET, "sending packet failed");
+			/* add cleanup handler, wait for packet, remove cleanup handler */
+			pthread_cleanup_push((void(*)(void*))pthread_mutex_unlock, (void*)&this->mutex);
+			pthread_setcancelstate(PTHREAD_CANCEL_ENABLE, &oldstate);
+			pthread_cond_wait(&this->condvar, &this->mutex);
+			
+			pthread_setcancelstate(oldstate, NULL);
+			pthread_cleanup_pop(0);
 		}
-		current_packet->destroy(current_packet);
+		this->list->remove_first(this->list, (void**)&packet);
+		pthread_mutex_unlock(&this->mutex);
+		
+		charon->socket->send(charon->socket, packet);
+		packet->destroy(packet);
 	}
 }
 
@@ -84,6 +121,7 @@ static void destroy(private_sender_t *this)
 {
 	pthread_cancel(this->assigned_thread);
 	pthread_join(this->assigned_thread, NULL);
+	this->list->destroy_offset(this->list, offsetof(packet_t, destroy));
 	free(this);
 }
 
@@ -94,11 +132,16 @@ sender_t * sender_create()
 {
 	private_sender_t *this = malloc_thing(private_sender_t);
 
+	this->public.send = (void(*)(sender_t*,packet_t*))send_;
 	this->public.destroy = (void(*)(sender_t*)) destroy;
 
-	if (pthread_create(&(this->assigned_thread), NULL, (void*(*)(void*))send_packets, this) != 0)
+	this->list = linked_list_create();
+	pthread_mutex_init(&this->mutex, NULL);
+	pthread_cond_init(&this->condvar, NULL);
+
+	if (pthread_create(&this->assigned_thread, NULL,
+					   (void*)send_packets, this) != 0)
 	{
-		free(this);
 		charon->kill(charon, "unable to create sender thread");
 	}
 

src/charon/threads/sender.h

@@ -6,7 +6,7 @@
  */
 
 /*
- * Copyright (C) 2005-2006 Martin Willi
+ * Copyright (C) 2005-2007 Martin Willi
  * Copyright (C) 2005 Jan Hutter
  * Hochschule fuer Technik Rapperswil
  *
@@ -27,6 +27,7 @@
 typedef struct sender_t sender_t;
 
 #include <library.h>
+#include <network/packet.h>
 
 /**
  * @brief Thread responsible for sending packets over the socket.
@@ -37,25 +38,34 @@ typedef struct sender_t sender_t;
  * @ingroup threads
  */
 struct sender_t {
-
+	
+	/**
+	 * @brief Send a packet over the network.
+	 *
+	 * This function is non blocking and adds the packet to a queue.
+	 * Whenever the sender thread things it's good to send the packet,
+	 * it'll do so.
+	 *
+	 * @param this		calling object
+ 	 * @param packet	packet to send
+	 */
+	void (*send) (sender_t *this, packet_t *packet);
+	
 	/**
 	 * @brief Destroys a sender object.
 	 *
-	 * @param sender 	calling object
+	 * @param this	 	calling object
 	 */
-	void (*destroy) (sender_t *sender);
+	void (*destroy) (sender_t *this);
 };
 
-
 /**
  * @brief Create the sender thread.
  * 
  * The thread will start to work, getting packets
- * from the send queue and sends them out.
+ * from its queue and sends them out.
  * 
- * @return
- * 					- sender_t object
- * 					- NULL of thread could not be started
+ * @return		created sender object
  * 
  * @ingroup threads
  */