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 */
28 #include "utils/channel.h"
30 #include <assert.h>
32 #include <errno.h>
34 #include <stdlib.h>
35 #include <string.h>
37 #include <time.h>
39 #include <pthread.h>
41 /*
42 * private data types
43 */
45 struct sdb_channel {
46 pthread_mutex_t lock;
48 /* signaling for select() operation */
49 pthread_cond_t cond;
51 /* maybe TODO: add support for 'nil' values using a boolean area */
53 void *data;
54 size_t data_len;
55 size_t elem_size;
57 size_t head;
58 size_t tail;
59 _Bool full;
61 _Bool shutdown;
62 };
64 /*
65 * private helper functions
66 */
68 #define NEXT_WRITE(chan) (((chan)->tail + 1) % (chan)->data_len)
69 #define NEXT_READ(chan) (((chan)->head + 1) % (chan)->data_len)
71 #define ELEM(chan, i) \
72 (void *)((char *)(chan)->data + (i) * (chan)->elem_size)
73 #define TAIL(chan) ELEM(chan, (chan)->tail)
74 #define HEAD(chan) ELEM(chan, (chan)->head)
76 /* Insert a new element at the end; chan->lock must be held.
77 * Returns 0 if data has been written or if data may be written
78 * if 'data' is NULL. */
79 static int
80 channel_write(sdb_channel_t *chan, const void *data)
81 {
82 assert(chan);
84 if (chan->full || chan->shutdown)
85 return -1;
86 else if (! data)
87 return 0;
89 memcpy(TAIL(chan), data, chan->elem_size);
90 chan->tail = NEXT_WRITE(chan);
92 chan->full = chan->tail == chan->head;
93 pthread_cond_broadcast(&chan->cond);
94 return 0;
95 } /* channel_write */
97 /* Retrieve the first element; chan->lock must be held.
98 * Returns 0 if data has been read or if data is available
99 * if 'data' is NULL. */
100 static int
101 channel_read(sdb_channel_t *chan, void *data)
102 {
103 assert(chan);
105 if ((chan->head == chan->tail) && (! chan->full))
106 return -1;
107 else if (! data)
108 return 0;
110 memcpy(data, HEAD(chan), chan->elem_size);
111 chan->head = NEXT_READ(chan);
113 chan->full = 0;
114 pthread_cond_broadcast(&chan->cond);
115 return 0;
116 } /* channel_read */
118 /*
119 * public API
120 */
122 sdb_channel_t *
123 sdb_channel_create(size_t size, size_t elem_size)
124 {
125 sdb_channel_t *chan;
127 if (! elem_size)
128 return NULL;
129 if (! size)
130 size = 1;
132 chan = calloc(1, sizeof(*chan));
133 if (! chan)
134 return NULL;
136 chan->data = calloc(size, elem_size);
137 if (! chan->data) {
138 sdb_channel_destroy(chan);
139 return NULL;
140 }
142 chan->data_len = size;
143 chan->elem_size = elem_size;
145 pthread_mutex_init(&chan->lock, /* attr = */ NULL);
146 pthread_cond_init(&chan->cond, /* attr = */ NULL);
148 chan->head = chan->tail = 0;
149 return chan;
150 } /* sdb_channel_create */
152 void
153 sdb_channel_destroy(sdb_channel_t *chan)
154 {
155 if (! chan)
156 return;
158 pthread_mutex_lock(&chan->lock);
159 free(chan->data);
160 chan->data = NULL;
161 chan->data_len = 0;
163 pthread_cond_destroy(&chan->cond);
165 pthread_mutex_unlock(&chan->lock);
166 pthread_mutex_destroy(&chan->lock);
167 free(chan);
168 } /* sdb_channel_destroy */
170 int
171 sdb_channel_select(sdb_channel_t *chan, int *wantread, void *read_data,
172 int *wantwrite, void *write_data, const struct timespec *timeout)
173 {
174 int status = 0;
176 if (! chan) {
177 errno = EINVAL;
178 return -1;
179 }
181 if ((! wantread) && (! read_data) && (! wantwrite) && (! write_data)) {
182 errno = EINVAL;
183 return -1;
184 }
186 pthread_mutex_lock(&chan->lock);
187 while (! status) {
188 int read_status, write_status;
190 read_status = channel_read(chan, read_data);
191 write_status = channel_write(chan, write_data);
193 if ((! read_status) || (! write_status)) {
194 if (wantread)
195 *wantread = read_status == 0;
196 if (wantwrite)
197 *wantwrite = write_status == 0;
199 if (((wantread || read_data) && (! read_status))
200 || ((wantwrite || write_data) && (! write_status)))
201 break;
202 }
204 if (chan->shutdown) {
205 if (read_status)
206 status = EBADF;
207 break;
208 }
210 if (timeout) {
211 struct timespec abstime;
213 if (clock_gettime(CLOCK_REALTIME, &abstime)) {
214 pthread_mutex_unlock(&chan->lock);
215 return -1;
216 }
218 abstime.tv_sec += timeout->tv_sec;
219 abstime.tv_nsec += timeout->tv_nsec;
221 if (abstime.tv_nsec > 1000000000) {
222 abstime.tv_nsec -= 1000000000;
223 abstime.tv_sec += 1;
224 }
226 status = pthread_cond_timedwait(&chan->cond, &chan->lock,
227 &abstime);
228 }
229 else
230 status = pthread_cond_wait(&chan->cond, &chan->lock);
231 }
232 pthread_mutex_unlock(&chan->lock);
234 if (status) {
235 errno = status;
236 return -1;
237 }
238 return 0;
239 } /* sdb_channel_select */
241 int
242 sdb_channel_write(sdb_channel_t *chan, const void *data)
243 {
244 int status;
246 if ((! chan) || (! data))
247 return -1;
249 pthread_mutex_lock(&chan->lock);
250 status = channel_write(chan, data);
251 pthread_mutex_unlock(&chan->lock);
252 return status;
253 } /* sdb_channel_write */
255 int
256 sdb_channel_read(sdb_channel_t *chan, void *data)
257 {
258 int status;
260 if ((! chan) || (! data))
261 return -1;
263 pthread_mutex_lock(&chan->lock);
264 status = channel_read(chan, data);
265 pthread_mutex_unlock(&chan->lock);
266 return status;
267 } /* sdb_channel_read */
269 int
270 sdb_channel_shutdown(sdb_channel_t *chan)
271 {
272 if (! chan)
273 return -1;
274 chan->shutdown = 1;
275 return 0;
276 } /* sdb_channel_shutdown */
278 /* vim: set tw=78 sw=4 ts=4 noexpandtab : */