code.grnet.gr Git - archipelago/blob - xseg/xtypes/xwaitq.c

   1 /*
   2  * Copyright 2013 GRNET S.A. All rights reserved.
   3  *
   4  * Redistribution and use in source and binary forms, with or
   5  * without modification, are permitted provided that the following
   6  * conditions are met:
   7  *
   8  *   1. Redistributions of source code must retain the above
   9  *      copyright notice, this list of conditions and the following
  10  *      disclaimer.
  11  *   2. Redistributions in binary form must reproduce the above
  12  *      copyright notice, this list of conditions and the following
  13  *      disclaimer in the documentation and/or other materials
  14  *      provided with the distribution.
  15  *
  16  * THIS SOFTWARE IS PROVIDED BY GRNET S.A. ``AS IS'' AND ANY EXPRESS
  17  * OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
  18  * WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR
  19  * PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL GRNET S.A OR
  20  * CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
  21  * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
  22  * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF
  23  * USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED
  24  * AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
  25  * LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN
  26  * ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
  27  * POSSIBILITY OF SUCH DAMAGE.
  28  *
  29  * The views and conclusions contained in the software and
  30  * documentation are those of the authors and should not be
  31  * interpreted as representing official policies, either expressed
  32  * or implied, of GRNET S.A.
  33  */
  34
  35 #include <xtypes/domain.h>
  36 #include <xtypes/xwaitq.h>
  37
  38 static int __check_cond(struct xwaitq *wq)
  39 {
  40         return wq->cond_fn(wq->cond_arg);
  41 }
  42
  43 int xwaitq_init(struct xwaitq *wq, int (*cond_fn)(void *arg), void *arg, uint32_t flags)
  44 {
  45         wq->cond_fn = cond_fn;
  46         wq->cond_arg = arg;
  47         wq->flags = flags;
  48         wq->q = xtypes_malloc(sizeof(struct xq));
  49         if (!wq->q)
  50                 return -1;
  51         xlock_release(&wq->lock);
  52         if (!xq_alloc_empty(wq->q, 8)){
  53                 xtypes_free(wq->q);
  54                 return -1;
  55         }
  56         return 0;
  57 }
  58
  59 void xwaitq_destroy(struct xwaitq *wq)
  60 {
  61         xq_free(wq->q);
  62         xtypes_free(wq->q);
  63 }
  64
  65 int __xwaitq_enqueue(struct xwaitq *wq, struct work *w)
  66 {
  67         //enqueure and resize if necessary
  68         xqindex r;
  69         struct xq *newq;
  70         r = __xq_append_tail(wq->q, (xqindex)w);
  71         if (r == Noneidx){
  72                 newq = xtypes_malloc(sizeof(struct xq));
  73                 if (!newq){
  74                         return -1;
  75                 }
  76                 if (!xq_alloc_empty(newq, wq->q->size*2)){
  77                         xtypes_free(newq);
  78                         return -1;
  79                 }
  80                 if (__xq_resize(wq->q, newq) == Noneidx){
  81                         xq_free(newq);
  82                         xtypes_free(newq);
  83                         return -1;
  84                 }
  85                 xtypes_free(wq->q);
  86                 wq->q = newq;
  87                 r = __xq_append_tail(wq->q, (xqindex)w);
  88         }
  89
  90         return ((r == Noneidx)? -1 : 0);
  91 }
  92
  93 int xwaitq_enqueue(struct xwaitq *wq, struct work *w)
  94 {
  95         int r;
  96         if (__check_cond(wq)){
  97                 w->job_fn(wq, w->job);
  98                 return 0;
  99         }
 100         xlock_acquire(&wq->lock, 1);
 101         r = __xwaitq_enqueue(wq, w);
 102         xlock_release(&wq->lock);
 103         xwaitq_signal(wq);
 104         return r;
 105 }
 106
 107 void xwaitq_signal(struct xwaitq *wq)
 108 {
 109         xqindex xqi;
 110         struct work *w;
 111
 112         if (!xq_count(wq->q))
 113                 return;
 114
 115         if (wq->flags & XWAIT_SIGNAL_ONE){
 116                 if (!xlock_try_lock(&wq->lock, 1))
 117                         return;
 118         } else {
 119                 xlock_acquire(&wq->lock, 1);
 120         }
 121         while (xq_count(wq->q) && __check_cond(wq)){
 122                 xqi = __xq_pop_head(wq->q);
 123                 if (xqi == Noneidx){
 124                         break;
 125                 }
 126                 xlock_release(&wq->lock);
 127                 w = (struct work *)xqi;
 128                 w->job_fn(wq, w->job);
 129                 if (wq->flags & XWAIT_SIGNAL_ONE){
 130                         if (!xlock_try_lock(&wq->lock, 1))
 131                                 return;
 132                 } else {
 133                         xlock_acquire(&wq->lock, 1);
 134                 }
 135         }
 136         xlock_release(&wq->lock);
 137 }
 138
 139 #ifdef __KERNEL__
 140 #include <linux/module.h>
 141 #include <xtypes/xwaitq_exports.h>
 142 #endif