[PATCHv2,2/2] platform: implement odp_queue_destroy()

Signed-off-by: Taras Kondratiuk <taras.kondratiuk@linaro.org>
---
 .../linux-generic/include/odp_queue_internal.h     |   23 +++++--
 platform/linux-generic/odp_queue.c                 |   72 +++++++++++++++++++-
 platform/linux-generic/odp_schedule.c              |    3 +-
 3 files changed, 92 insertions(+), 6 deletions(-)

On Tue, Dec 2, 2014 at 10:47 AM, Taras Kondratiuk <
taras.kondratiuk@linaro.org> wrote:

> Signed-off-by: Taras Kondratiuk <taras.kondratiuk@linaro.org>
> ---
>  .../linux-generic/include/odp_queue_internal.h     |   23 +++++--
>  platform/linux-generic/odp_queue.c                 |   72
> +++++++++++++++++++-
>  platform/linux-generic/odp_schedule.c              |    3 +-
>  3 files changed, 92 insertions(+), 6 deletions(-)
>
> diff --git a/platform/linux-generic/include/odp_queue_internal.h
> b/platform/linux-generic/include/odp_queue_internal.h
> index 8b6c517..501f727 100644
> --- a/platform/linux-generic/include/odp_queue_internal.h
> +++ b/platform/linux-generic/include/odp_queue_internal.h
> @@ -34,10 +34,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;
> @@ -89,6 +90,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);
>
> @@ -113,6 +120,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 1318bcd..b87b354 100644
> --- a/platform/linux-generic/odp_queue.c
> +++ b/platform/linux-generic/odp_queue.c
> @@ -192,6 +192,49 @@ 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);
> +
>

Need to lock before this test.  Otherwise have race condition if two
threads try to call odp_queue_destroy() at the same time.  Both can decide
queue is not free and then both try to free it.


> +       if (queue->s.status == QUEUE_STATUS_FREE)
> +               return -1; /* Queue is alredy freed */
> +

+       LOCK(&queue->s.lock);
> +       if (queue->s.head != NULL) {
> +               UNLOCK(&queue->s.lock);
> +               return -1; /* Queue is 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)
>  {
> @@ -279,7 +322,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;
> @@ -314,6 +356,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)
>  {
> @@ -407,6 +461,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 cc84e11..c2aeb28 100644
> --- a/platform/linux-generic/odp_schedule.c
> +++ b/platform/linux-generic/odp_schedule.c
> @@ -302,7 +302,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;
> --
> 1.7.9.5
>
>
> _______________________________________________
> lng-odp mailing list
> lng-odp@lists.linaro.org
> http://lists.linaro.org/mailman/listinfo/lng-odp
>

On 12/02/2014 05:51 PM, Bill Fischofer wrote:
>
>
> On Tue, Dec 2, 2014 at 10:47 AM, Taras Kondratiuk
> <taras.kondratiuk@linaro.org <mailto:taras.kondratiuk@linaro.org>> wrote:
>
>     +int odp_queue_destroy(odp_queue_t handle)
>     +{
>     +       queue_entry_t *queue;
>     +       queue = queue_to_qentry(handle);
>     +
>
>
> Need to lock before this test.  Otherwise have race condition if two
> threads try to call odp_queue_destroy() at the same time.  Both can
> decide queue is not free and then both try to free it.

I assumed that queue destroying happens in a controlled manner from one
thread. Can move it under the lock.

>
>     +       if (queue->s.status == QUEUE_STATUS_FREE)
>     +               return -1; /* Queue is alredy freed */
>     +
>
>     +       LOCK(&queue->s.lock);
>     +       if (queue->s.head != NULL) {
>     +               UNLOCK(&queue->s.lock);
>     +               return -1; /* Queue is not empty */
>     +       }