libipsec: Use CoDel queues in processor for inbound/outbound queues

diff --git a/src/libipsec/ipsec_processor.c b/src/libipsec/ipsec_processor.c
index eae2ed2f159fa049a1b8238c4ad670dae7615e76..b58e1372ac516e2afc24cd0f85ea94dc23aeea6a 100644 (file)
--- a/src/libipsec/ipsec_processor.c
+++ b/src/libipsec/ipsec_processor.c
@@ -15,11 +15,11 @@
 
 #include "ipsec.h"
 #include "ipsec_processor.h"
+#include "codel_queue.h"
 
 #include <utils/debug.h>
 #include <library.h>
 #include <threading/rwlock.h>
-#include <collections/blocking_queue.h>
 #include <processing/jobs/callback_job.h>
 
 typedef struct private_ipsec_processor_t private_ipsec_processor_t;
@@ -37,12 +37,12 @@ struct private_ipsec_processor_t {
        /**
         * Queue for inbound packets (esp_packet_t*)
         */
-       blocking_queue_t *inbound_queue;
+       codel_queue_t *inbound_queue;
 
        /**
         * Queue for outbound packets (ip_packet_t*)
         */
-       blocking_queue_t *outbound_queue;
+       codel_queue_t *outbound_queue;
 
        /**
         * Registered inbound callback
@@ -235,13 +235,15 @@ static job_requeue_t process_outbound(private_ipsec_processor_t *this)
 METHOD(ipsec_processor_t, queue_inbound, void,
        private_ipsec_processor_t *this, esp_packet_t *packet)
 {
-       this->inbound_queue->enqueue(this->inbound_queue, packet);
+       this->inbound_queue->enqueue(this->inbound_queue, packet,
+                                                               packet->packet.get_data(&packet->packet).len);
 }
 
 METHOD(ipsec_processor_t, queue_outbound, void,
        private_ipsec_processor_t *this, ip_packet_t *packet)
 {
-       this->outbound_queue->enqueue(this->outbound_queue, packet);
+       this->outbound_queue->enqueue(this->outbound_queue, packet,
+                                                               packet->get_encoding(packet).len);
 }
 
 METHOD(ipsec_processor_t, register_inbound, void,
@@ -287,10 +289,8 @@ METHOD(ipsec_processor_t, unregister_outbound, void,
 METHOD(ipsec_processor_t, destroy, void,
        private_ipsec_processor_t *this)
 {
-       this->inbound_queue->destroy_offset(this->inbound_queue,
-                                                                               offsetof(esp_packet_t, destroy));
-       this->outbound_queue->destroy_offset(this->outbound_queue,
-                                                                                offsetof(ip_packet_t, destroy));
+       this->inbound_queue->destroy(this->inbound_queue);
+       this->outbound_queue->destroy(this->outbound_queue);
        this->lock->destroy(this->lock);
        free(this);
 }
@@ -312,8 +312,10 @@ ipsec_processor_t *ipsec_processor_create()
                        .unregister_outbound = _unregister_outbound,
                        .destroy = _destroy,
                },
-               .inbound_queue = blocking_queue_create(),
-               .outbound_queue = blocking_queue_create(),
+               .inbound_queue = codel_queue_create(
+                                                                               offsetof(esp_packet_t, destroy), 1500),
+               .outbound_queue = codel_queue_create(
+                                                                               offsetof(ip_packet_t, destroy), 1500),
                .lock = rwlock_create(RWLOCK_TYPE_DEFAULT),
        );