2 *------------------------------------------------------------------
3 * svm_queue.c - unidirectional shared-memory queues
5 * Copyright (c) 2009-2019 Cisco and/or its affiliates.
6 * Licensed under the Apache License, Version 2.0 (the "License");
7 * you may not use this file except in compliance with the License.
8 * You may obtain a copy of the License at:
10 * http://www.apache.org/licenses/LICENSE-2.0
12 * Unless required by applicable law or agreed to in writing, software
13 * distributed under the License is distributed on an "AS IS" BASIS,
14 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
15 * See the License for the specific language governing permissions and
16 * limitations under the License.
17 *------------------------------------------------------------------
25 #include <vppinfra/mem.h>
26 #include <vppinfra/format.h>
27 #include <vppinfra/cache.h>
28 #include <svm/queue.h>
29 #include <vppinfra/time.h>
30 #include <vppinfra/lock.h>
33 svm_queue_init (void *base, int nels, int elsize)
36 pthread_mutexattr_t attr;
37 pthread_condattr_t cattr;
39 q = (svm_queue_t *) base;
40 clib_memset (q, 0, sizeof (*q));
44 q->producer_evtfd = -1;
45 q->consumer_evtfd = -1;
47 clib_memset (&attr, 0, sizeof (attr));
48 clib_memset (&cattr, 0, sizeof (cattr));
50 if (pthread_mutexattr_init (&attr))
51 clib_unix_warning ("mutexattr_init");
52 if (pthread_mutexattr_setpshared (&attr, PTHREAD_PROCESS_SHARED))
53 clib_unix_warning ("pthread_mutexattr_setpshared");
54 if (pthread_mutex_init (&q->mutex, &attr))
55 clib_unix_warning ("mutex_init");
56 if (pthread_mutexattr_destroy (&attr))
57 clib_unix_warning ("mutexattr_destroy");
58 if (pthread_condattr_init (&cattr))
59 clib_unix_warning ("condattr_init");
60 /* prints funny-looking messages in the Linux target */
61 if (pthread_condattr_setpshared (&cattr, PTHREAD_PROCESS_SHARED))
62 clib_unix_warning ("condattr_setpshared");
63 if (pthread_cond_init (&q->condvar, &cattr))
64 clib_unix_warning ("cond_init1");
65 if (pthread_condattr_destroy (&cattr))
66 clib_unix_warning ("cond_init2");
72 svm_queue_alloc_and_init (int nels, int elsize, int consumer_pid)
76 q = clib_mem_alloc_aligned (sizeof (svm_queue_t)
77 + nels * elsize, CLIB_CACHE_LINE_BYTES);
78 clib_memset (q, 0, sizeof (*q));
79 q = svm_queue_init (q, nels, elsize);
80 q->consumer_pid = consumer_pid;
89 svm_queue_free (svm_queue_t * q)
91 (void) pthread_mutex_destroy (&q->mutex);
92 (void) pthread_cond_destroy (&q->condvar);
97 svm_queue_lock (svm_queue_t * q)
99 pthread_mutex_lock (&q->mutex);
103 svm_queue_unlock (svm_queue_t * q)
105 pthread_mutex_unlock (&q->mutex);
109 svm_queue_is_full (svm_queue_t * q)
111 return q->cursize == q->maxsize;
115 svm_queue_send_signal_inline (svm_queue_t * q, u8 is_prod)
117 if (q->producer_evtfd == -1)
119 (void) pthread_cond_broadcast (&q->condvar);
123 int __clib_unused rv, fd;
125 ASSERT (q->consumer_evtfd > 0 && q->producer_evtfd > 0);
126 fd = is_prod ? q->producer_evtfd : q->consumer_evtfd;
127 rv = write (fd, &data, sizeof (data));
128 if (PREDICT_FALSE (rv < 0))
129 clib_unix_warning ("signal write on %d returned %d", fd, rv);
134 svm_queue_send_signal (svm_queue_t * q, u8 is_prod)
136 svm_queue_send_signal_inline (q, is_prod);
140 svm_queue_wait_inline (svm_queue_t * q)
142 if (q->producer_evtfd == -1)
144 pthread_cond_wait (&q->condvar, &q->mutex);
148 /* Fake a wait for event. We could use epoll but that would mean
149 * using yet another fd. Should do for now */
150 u32 cursize = q->cursize;
151 svm_queue_unlock (q);
152 while (q->cursize == cursize)
159 svm_queue_wait (svm_queue_t * q)
161 svm_queue_wait_inline (q);
165 svm_queue_timedwait_inline (svm_queue_t * q, double timeout)
168 ts.tv_sec = unix_time_now () + (u32) timeout;
169 ts.tv_nsec = (timeout - (u32) timeout) * 1e9;
171 if (q->producer_evtfd == -1)
173 return pthread_cond_timedwait (&q->condvar, &q->mutex, &ts);
177 double max_time = unix_time_now () + timeout;
178 u32 cursize = q->cursize;
181 svm_queue_unlock (q);
182 while (q->cursize == cursize && unix_time_now () < max_time)
184 rv = unix_time_now () < max_time ? 0 : ETIMEDOUT;
191 svm_queue_timedwait (svm_queue_t * q, double timeout)
193 return svm_queue_timedwait_inline (q, timeout);
197 * svm_queue_add_nolock
200 svm_queue_add_nolock (svm_queue_t * q, u8 * elem)
203 int need_broadcast = 0;
205 if (PREDICT_FALSE (q->cursize == q->maxsize))
207 while (q->cursize == q->maxsize)
208 svm_queue_wait_inline (q);
211 tailp = (i8 *) (&q->data[0] + q->elsize * q->tail);
212 clib_memcpy_fast (tailp, elem, q->elsize);
217 need_broadcast = (q->cursize == 1);
219 if (q->tail == q->maxsize)
223 svm_queue_send_signal_inline (q, 1);
228 svm_queue_add_raw (svm_queue_t * q, u8 * elem)
232 tailp = (i8 *) (&q->data[0] + q->elsize * q->tail);
233 clib_memcpy_fast (tailp, elem, q->elsize);
235 q->tail = (q->tail + 1) % q->maxsize;
239 svm_queue_send_signal_inline (q, 1);
247 svm_queue_add (svm_queue_t * q, u8 * elem, int nowait)
250 int need_broadcast = 0;
254 /* zero on success */
255 if (pthread_mutex_trylock (&q->mutex))
263 if (PREDICT_FALSE (q->cursize == q->maxsize))
267 svm_queue_unlock (q);
270 while (q->cursize == q->maxsize)
271 svm_queue_wait_inline (q);
274 tailp = (i8 *) (&q->data[0] + q->elsize * q->tail);
275 clib_memcpy_fast (tailp, elem, q->elsize);
280 need_broadcast = (q->cursize == 1);
282 if (q->tail == q->maxsize)
286 svm_queue_send_signal_inline (q, 1);
288 svm_queue_unlock (q);
297 svm_queue_add2 (svm_queue_t * q, u8 * elem, u8 * elem2, int nowait)
300 int need_broadcast = 0;
304 /* zero on success */
305 if (pthread_mutex_trylock (&q->mutex))
313 if (PREDICT_FALSE (q->cursize + 1 == q->maxsize))
317 svm_queue_unlock (q);
320 while (q->cursize + 1 == q->maxsize)
321 svm_queue_wait_inline (q);
324 tailp = (i8 *) (&q->data[0] + q->elsize * q->tail);
325 clib_memcpy_fast (tailp, elem, q->elsize);
330 if (q->tail == q->maxsize)
333 need_broadcast = (q->cursize == 1);
335 tailp = (i8 *) (&q->data[0] + q->elsize * q->tail);
336 clib_memcpy_fast (tailp, elem2, q->elsize);
341 if (q->tail == q->maxsize)
345 svm_queue_send_signal_inline (q, 1);
347 svm_queue_unlock (q);
356 svm_queue_sub (svm_queue_t * q, u8 * elem, svm_q_conditional_wait_t cond,
360 int need_broadcast = 0;
363 if (cond == SVM_Q_NOWAIT)
365 /* zero on success */
366 if (pthread_mutex_trylock (&q->mutex))
374 if (PREDICT_FALSE (q->cursize == 0))
376 if (cond == SVM_Q_NOWAIT)
378 svm_queue_unlock (q);
381 else if (cond == SVM_Q_TIMEDWAIT)
383 while (q->cursize == 0 && rc == 0)
384 rc = svm_queue_timedwait_inline (q, time);
388 svm_queue_unlock (q);
394 while (q->cursize == 0)
395 svm_queue_wait_inline (q);
399 headp = (i8 *) (&q->data[0] + q->elsize * q->head);
400 clib_memcpy_fast (elem, headp, q->elsize);
403 /* $$$$ JFC shouldn't this be == 0? */
404 if (q->cursize == q->maxsize)
409 if (q->head == q->maxsize)
413 svm_queue_send_signal_inline (q, 0);
415 svm_queue_unlock (q);
421 svm_queue_sub2 (svm_queue_t * q, u8 * elem)
429 svm_queue_unlock (q);
433 headp = (i8 *) (&q->data[0] + q->elsize * q->head);
434 clib_memcpy_fast (elem, headp, q->elsize);
437 need_broadcast = (q->cursize == q->maxsize / 2);
440 if (PREDICT_FALSE (q->head == q->maxsize))
442 svm_queue_unlock (q);
445 svm_queue_send_signal_inline (q, 0);
451 svm_queue_sub_raw (svm_queue_t * q, u8 * elem)
456 if (PREDICT_FALSE (q->cursize == 0))
458 while (q->cursize == 0)
462 headp = (i8 *) (&q->data[0] + q->elsize * q->head);
463 clib_memcpy_fast (elem, headp, q->elsize);
465 need_broadcast = q->cursize == q->maxsize;
467 q->head = (q->head + 1) % q->maxsize;
470 if (PREDICT_FALSE (need_broadcast))
471 svm_queue_send_signal_inline (q, 0);
477 svm_queue_set_producer_event_fd (svm_queue_t * q, int fd)
479 q->producer_evtfd = fd;
483 svm_queue_set_consumer_event_fd (svm_queue_t * q, int fd)
485 q->consumer_evtfd = fd;
489 * fd.io coding-style-patch-verification: ON
492 * eval: (c-set-style "gnu")