src/utils/channel.c

   1 /*
   2  * SysDB - src/utils/channel.c
   3  * Copyright (C) 2013 Sebastian 'tokkee' Harl <sh@tokkee.org>
   4  * All rights reserved.
   5  *
   6  * Redistribution and use in source and binary forms, with or without
   7  * modification, are permitted provided that the following conditions
   8  * are met:
   9  * 1. Redistributions of source code must retain the above copyright
  10  *    notice, this list of conditions and the following disclaimer.
  11  * 2. Redistributions in binary form must reproduce the above copyright
  12  *    notice, this list of conditions and the following disclaimer in the
  13  *    documentation and/or other materials provided with the distribution.
  14  *
  15  * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
  16  * ``AS IS'' AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED
  17  * TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR
  18  * PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDERS OR
  19  * CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL,
  20  * EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO,
  21  * PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS;
  22  * OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY,
  23  * WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR
  24  * OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF
  25  * ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
  26  */
  27
  28 #include "utils/channel.h"
  29
  30 #include <assert.h>
  31
  32 #include <stdlib.h>
  33 #include <string.h>
  34
  35 #include <time.h>
  36
  37 #include <pthread.h>
  38
  39 /*
  40  * private data types
  41  */
  42
  43 struct sdb_channel {
  44         pthread_mutex_t lock;
  45
  46         /* signaling for select() operation */
  47         pthread_cond_t cond;
  48
  49         /* maybe TODO: add support for 'nil' values using a boolean area */
  50
  51         void  *data;
  52         size_t data_len;
  53         size_t elem_size;
  54
  55         size_t head;
  56         size_t tail;
  57         _Bool full;
  58 };
  59
  60 /*
  61  * private helper functions
  62  */
  63
  64 #define NEXT_WRITE(chan) (((chan)->tail + 1) % (chan)->data_len)
  65 #define NEXT_READ(chan) (((chan)->head + 1) % (chan)->data_len)
  66
  67 #define ELEM(chan, i) \
  68         (void *)((char *)(chan)->data + (i) * (chan)->elem_size)
  69 #define TAIL(chan) ELEM(chan, (chan)->tail)
  70 #define HEAD(chan) ELEM(chan, (chan)->head)
  71
  72 /* Insert a new element at the end; chan->lock must be held.
  73  * Returns 0 if data has been written or if data may be written
  74  * if 'data' is NULL. */
  75 static int
  76 channel_write(sdb_channel_t *chan, const void *data)
  77 {
  78         assert(chan);
  79
  80         if (chan->full)
  81                 return -1;
  82         else if (! data)
  83                 return 0;
  84
  85         memcpy(TAIL(chan), data, chan->elem_size);
  86         chan->tail = NEXT_WRITE(chan);
  87
  88         chan->full = chan->tail == chan->head;
  89         pthread_cond_broadcast(&chan->cond);
  90         return 0;
  91 } /* channel_write */
  92
  93 /* Retrieve the first element; chan->lock must be held.
  94  * Returns 0 if data has been read or if data is available
  95  * if 'data' is NULL. */
  96 static int
  97 channel_read(sdb_channel_t *chan, void *data)
  98 {
  99         assert(chan);
 100
 101         if ((chan->head == chan->tail) && (! chan->full))
 102                 return -1;
 103         else if (! data)
 104                 return 0;
 105
 106         memcpy(data, HEAD(chan), chan->elem_size);
 107         chan->head = NEXT_READ(chan);
 108
 109         chan->full = 0;
 110         pthread_cond_broadcast(&chan->cond);
 111         return 0;
 112 } /* channel_read */
 113
 114 /*
 115  * public API
 116  */
 117
 118 sdb_channel_t *
 119 sdb_channel_create(size_t size, size_t elem_size)
 120 {
 121         sdb_channel_t *chan;
 122
 123         if (! elem_size)
 124                 return NULL;
 125         if (! size)
 126                 size = 1;
 127
 128         chan = calloc(1, sizeof(*chan));
 129         if (! chan)
 130                 return NULL;
 131
 132         chan->data = calloc(size, elem_size);
 133         if (! chan->data) {
 134                 sdb_channel_destroy(chan);
 135                 return NULL;
 136         }
 137
 138         chan->data_len = size;
 139         chan->elem_size = elem_size;
 140
 141         pthread_mutex_init(&chan->lock, /* attr = */ NULL);
 142         pthread_cond_init(&chan->cond, /* attr = */ NULL);
 143
 144         chan->head = chan->tail = 0;
 145         return chan;
 146 } /* sdb_channel_create */
 147
 148 void
 149 sdb_channel_destroy(sdb_channel_t *chan)
 150 {
 151         if (! chan)
 152                 return;
 153
 154         pthread_mutex_lock(&chan->lock);
 155         free(chan->data);
 156         chan->data = NULL;
 157         chan->data_len = 0;
 158
 159         pthread_cond_destroy(&chan->cond);
 160
 161         pthread_mutex_unlock(&chan->lock);
 162         pthread_mutex_destroy(&chan->lock);
 163         free(chan);
 164 } /* sdb_channel_destroy */
 165
 166 int
 167 sdb_channel_select(sdb_channel_t *chan, int *wantread, void *read_data,
 168                 int *wantwrite, void *write_data, const struct timespec *timeout)
 169 {
 170         struct timespec abstime;
 171         int status = 0;
 172
 173         if (! chan)
 174                 return -1;
 175
 176         if ((! wantread) && (! read_data) && (! wantwrite) && (! write_data))
 177                 return -1;
 178
 179         if (timeout) {
 180                 if (clock_gettime(CLOCK_REALTIME, &abstime))
 181                         return -1;
 182
 183                 abstime.tv_sec += timeout->tv_sec;
 184                 abstime.tv_nsec += timeout->tv_nsec;
 185         }
 186
 187         pthread_mutex_lock(&chan->lock);
 188         while (! status) {
 189                 int read_status, write_status;
 190
 191                 read_status = channel_read(chan, read_data);
 192                 write_status = channel_write(chan, write_data);
 193
 194                 if ((! read_status) || (! write_status)) {
 195                         if (wantread)
 196                                 *wantread = read_status == 0;
 197                         if (wantwrite)
 198                                 *wantwrite = write_status == 0;
 199
 200                         if (((wantread || read_data) && (! read_status))
 201                                         || ((wantwrite || write_data) && (! write_status)))
 202                                 break;
 203                 }
 204
 205                 if (timeout)
 206                         status = pthread_cond_timedwait(&chan->cond, &chan->lock,
 207                                         &abstime);
 208                 else
 209                         status = pthread_cond_wait(&chan->cond, &chan->lock);
 210         }
 211
 212         pthread_mutex_unlock(&chan->lock);
 213         return status;
 214 } /* sdb_channel_select */
 215
 216 int
 217 sdb_channel_write(sdb_channel_t *chan, const void *data)
 218 {
 219         int status;
 220
 221         if ((! chan) || (! data))
 222                 return -1;
 223
 224         pthread_mutex_lock(&chan->lock);
 225         status = channel_write(chan, data);
 226         pthread_mutex_unlock(&chan->lock);
 227         return status;
 228 } /* sdb_channel_write */
 229
 230 int
 231 sdb_channel_read(sdb_channel_t *chan, void *data)
 232 {
 233         int status;
 234
 235         if ((! chan) || (! data))
 236                 return -1;
 237
 238         pthread_mutex_lock(&chan->lock);
 239         status = channel_read(chan, data);
 240         pthread_mutex_unlock(&chan->lock);
 241         return status;
 242 } /* sdb_channel_read */
 243
 244 /* vim: set tw=78 sw=4 ts=4 noexpandtab : */
 245