linux-generic: implement odp_queue_destroy()

Signed-off-by: Taras Kondratiuk <taras.kondratiuk@linaro.org>
Reviewed-by: Bill Fischofer <bill.fischofer@linaro.org>
Signed-off-by: Maxim Uvarov <maxim.uvarov@linaro.org>

3 files changed, 89 insertions, 6 deletions

diff --git a/platform/linux-generic/include/odp_queue_internal.h b/platform/linux-generic/include/odp_queue_internal.h
index 1254763ba..d5c8e4e85 100644
--- a/platform/linux-generic/include/odp_queue_internal.h
+++ b/platform/linux-generic/include/odp_queue_internal.h
@@ -35,10 +35,11 @@ extern "C" {
 
 #define QUEUE_MULTI_MAX 8
 
-#define QUEUE_STATUS_FREE     0
-#define QUEUE_STATUS_READY    1
-#define QUEUE_STATUS_NOTSCHED 2
-#define QUEUE_STATUS_SCHED    3
+#define QUEUE_STATUS_FREE         0
+#define QUEUE_STATUS_READY        1
+#define QUEUE_STATUS_NOTSCHED     2
+#define QUEUE_STATUS_SCHED        3
+#define QUEUE_STATUS_DESTROYED    4
 
 /* forward declaration */
 union queue_entry_u;
@@ -90,6 +91,12 @@ odp_buffer_hdr_t *queue_deq(queue_entry_t *queue);
 int queue_enq_multi(queue_entry_t *queue, odp_buffer_hdr_t *buf_hdr[], int num);
 int queue_deq_multi(queue_entry_t *queue, odp_buffer_hdr_t *buf_hdr[], int num);
 
+int queue_enq_dummy(queue_entry_t *queue, odp_buffer_hdr_t *buf_hdr);
+int queue_enq_multi_dummy(queue_entry_t *queue, odp_buffer_hdr_t *buf_hdr[],
+			  int num);
+int queue_deq_multi_destroy(queue_entry_t *queue, odp_buffer_hdr_t *buf_hdr[],
+			    int num);
+
 void queue_lock(queue_entry_t *queue);
 void queue_unlock(queue_entry_t *queue);
 
@@ -114,6 +121,14 @@ static inline queue_entry_t *queue_to_qentry(odp_queue_t handle)
 	return get_qentry(queue_id);
 }
 
+static inline int queue_is_destroyed(odp_queue_t handle)
+{
+	queue_entry_t *queue;
+
+	queue = queue_to_qentry(handle);
+
+	return queue->s.status == QUEUE_STATUS_DESTROYED;
+}
 #ifdef __cplusplus
 }
 #endif
diff --git a/platform/linux-generic/odp_queue.c b/platform/linux-generic/odp_queue.c
index a7c5e4272..1462b4108 100644
--- a/platform/linux-generic/odp_queue.c
+++ b/platform/linux-generic/odp_queue.c
@@ -193,6 +193,46 @@ odp_queue_t odp_queue_create(const char *name, odp_queue_type_t type,
 	return handle;
 }
 
+int odp_queue_destroy(odp_queue_t handle)
+{
+	queue_entry_t *queue;
+	queue = queue_to_qentry(handle);
+
+	LOCK(&queue->s.lock);
+	if (queue->s.status == QUEUE_STATUS_FREE || queue->s.head != NULL) {
+		UNLOCK(&queue->s.lock);
+		return -1; /* Queue is already free or not empty */
+	}
+
+	queue->s.enqueue = queue_enq_dummy;
+	queue->s.enqueue_multi = queue_enq_multi_dummy;
+
+	if (queue->s.type == ODP_QUEUE_TYPE_POLL ||
+	    queue->s.type == ODP_QUEUE_TYPE_PKTOUT) {
+		queue->s.status = QUEUE_STATUS_FREE;
+		queue->s.head = NULL;
+		queue->s.tail = NULL;
+	} else if (queue->s.type == ODP_QUEUE_TYPE_SCHED) {
+		if (queue->s.status == QUEUE_STATUS_SCHED)  {
+			/*
+			 * Override dequeue_multi to destroy queue when it will
+			 * be scheduled next time.
+			 */
+			queue->s.status = QUEUE_STATUS_DESTROYED;
+			queue->s.dequeue_multi = queue_deq_multi_destroy;
+		} else {
+			/* Queue won't be scheduled anymore */
+			odp_buffer_free(queue->s.sched_buf);
+			queue->s.sched_buf = ODP_BUFFER_INVALID;
+			queue->s.status = QUEUE_STATUS_FREE;
+			queue->s.head = NULL;
+			queue->s.tail = NULL;
+		}
+	}
+	UNLOCK(&queue->s.lock);
+
+	return 0;
+}
 
 odp_buffer_t queue_sched_buf(odp_queue_t handle)
 {
@@ -280,7 +320,6 @@ int queue_enq(queue_entry_t *queue, odp_buffer_hdr_t *buf_hdr)
 	return 0;
 }
 
-
 int queue_enq_multi(queue_entry_t *queue, odp_buffer_hdr_t *buf_hdr[], int num)
 {
 	int sched = 0;
@@ -315,6 +354,18 @@ int queue_enq_multi(queue_entry_t *queue, odp_buffer_hdr_t *buf_hdr[], int num)
 	return 0;
 }
 
+int queue_enq_dummy(queue_entry_t *queue ODP_UNUSED,
+		    odp_buffer_hdr_t *buf_hdr ODP_UNUSED)
+{
+	return -1;
+}
+
+int queue_enq_multi_dummy(queue_entry_t *queue ODP_UNUSED,
+			  odp_buffer_hdr_t *buf_hdr[] ODP_UNUSED,
+			  int num ODP_UNUSED)
+{
+	return -1;
+}
 
 int odp_queue_enq_multi(odp_queue_t handle, odp_buffer_t buf[], int num)
 {
@@ -408,6 +459,22 @@ int queue_deq_multi(queue_entry_t *queue, odp_buffer_hdr_t *buf_hdr[], int num)
 	return i;
 }
 
+int queue_deq_multi_destroy(queue_entry_t *queue,
+			    odp_buffer_hdr_t *buf_hdr[] ODP_UNUSED,
+			    int num ODP_UNUSED)
+{
+	LOCK(&queue->s.lock);
+
+	odp_buffer_free(queue->s.sched_buf);
+	queue->s.sched_buf = ODP_BUFFER_INVALID;
+	queue->s.status = QUEUE_STATUS_FREE;
+	queue->s.head = NULL;
+	queue->s.tail = NULL;
+
+	UNLOCK(&queue->s.lock);
+
+	return 0;
+}
 
 int odp_queue_deq_multi(odp_queue_t handle, odp_buffer_t buf[], int num)
 {
diff --git a/platform/linux-generic/odp_schedule.c b/platform/linux-generic/odp_schedule.c
index ac7624025..a14de4fa1 100644
--- a/platform/linux-generic/odp_schedule.c
+++ b/platform/linux-generic/odp_schedule.c
@@ -295,7 +295,8 @@ static int schedule(odp_queue_t *out_queue, odp_buffer_t out_buf[],
 					 * except packet input queues
 					 */
 					if (odp_queue_type(queue) ==
-					    ODP_QUEUE_TYPE_PKTIN)
+					    ODP_QUEUE_TYPE_PKTIN &&
+					    !queue_is_destroyed(queue))
 						odp_queue_enq(pri_q, desc_buf);
 
 					continue;