2 *------------------------------------------------------------------
3 * svm_queue.c - unidirectional shared-memory queues
5 * Copyright (c) 2009 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>
34 * nels = number of elements on the queue
35 * elsize = element size, presumably 4 and cacheline-size will
39 * The idea is to call this function in the queue consumer,
40 * and e-mail the queue pointer to the producer(s).
42 * The vpp process / main thread allocates one of these
43 * at startup; its main input queue. The vpp main input queue
44 * has a pointer to it in the shared memory segment header.
46 * You probably want to be on an svm data heap before calling this
50 svm_queue_init (int nels,
51 int elsize, int consumer_pid, int signal_when_queue_non_empty)
54 pthread_mutexattr_t attr;
55 pthread_condattr_t cattr;
57 q = clib_mem_alloc_aligned (sizeof (svm_queue_t)
58 + nels * elsize, CLIB_CACHE_LINE_BYTES);
59 memset (q, 0, sizeof (*q));
63 q->consumer_pid = consumer_pid;
64 q->signal_when_queue_non_empty = signal_when_queue_non_empty;
66 memset (&attr, 0, sizeof (attr));
67 memset (&cattr, 0, sizeof (cattr));
69 if (pthread_mutexattr_init (&attr))
70 clib_unix_warning ("mutexattr_init");
71 if (pthread_mutexattr_setpshared (&attr, PTHREAD_PROCESS_SHARED))
72 clib_unix_warning ("pthread_mutexattr_setpshared");
73 if (pthread_mutex_init (&q->mutex, &attr))
74 clib_unix_warning ("mutex_init");
75 if (pthread_mutexattr_destroy (&attr))
76 clib_unix_warning ("mutexattr_destroy");
77 if (pthread_condattr_init (&cattr))
78 clib_unix_warning ("condattr_init");
79 /* prints funny-looking messages in the Linux target */
80 if (pthread_condattr_setpshared (&cattr, PTHREAD_PROCESS_SHARED))
81 clib_unix_warning ("condattr_setpshared");
82 if (pthread_cond_init (&q->condvar, &cattr))
83 clib_unix_warning ("cond_init1");
84 if (pthread_condattr_destroy (&cattr))
85 clib_unix_warning ("cond_init2");
94 svm_queue_free (svm_queue_t * q)
96 (void) pthread_mutex_destroy (&q->mutex);
97 (void) pthread_cond_destroy (&q->condvar);
102 svm_queue_lock (svm_queue_t * q)
104 pthread_mutex_lock (&q->mutex);
108 svm_queue_unlock (svm_queue_t * q)
110 pthread_mutex_unlock (&q->mutex);
114 svm_queue_is_full (svm_queue_t * q)
116 return q->cursize == q->maxsize;
120 * svm_queue_add_nolock
123 svm_queue_add_nolock (svm_queue_t * q, u8 * elem)
126 int need_broadcast = 0;
128 if (PREDICT_FALSE (q->cursize == q->maxsize))
130 while (q->cursize == q->maxsize)
132 (void) pthread_cond_wait (&q->condvar, &q->mutex);
136 tailp = (i8 *) (&q->data[0] + q->elsize * q->tail);
137 clib_memcpy (tailp, elem, q->elsize);
142 need_broadcast = (q->cursize == 1);
144 if (q->tail == q->maxsize)
149 (void) pthread_cond_broadcast (&q->condvar);
150 if (q->signal_when_queue_non_empty)
151 kill (q->consumer_pid, q->signal_when_queue_non_empty);
157 svm_queue_add_raw (svm_queue_t * q, u8 * elem)
161 if (PREDICT_FALSE (q->cursize == q->maxsize))
163 while (q->cursize == q->maxsize)
167 tailp = (i8 *) (&q->data[0] + q->elsize * q->tail);
168 clib_memcpy (tailp, elem, q->elsize);
173 if (q->tail == q->maxsize)
183 svm_queue_add (svm_queue_t * q, u8 * elem, int nowait)
186 int need_broadcast = 0;
190 /* zero on success */
191 if (pthread_mutex_trylock (&q->mutex))
197 pthread_mutex_lock (&q->mutex);
199 if (PREDICT_FALSE (q->cursize == q->maxsize))
203 pthread_mutex_unlock (&q->mutex);
206 while (q->cursize == q->maxsize)
208 (void) pthread_cond_wait (&q->condvar, &q->mutex);
212 tailp = (i8 *) (&q->data[0] + q->elsize * q->tail);
213 clib_memcpy (tailp, elem, q->elsize);
218 need_broadcast = (q->cursize == 1);
220 if (q->tail == q->maxsize)
225 (void) pthread_cond_broadcast (&q->condvar);
226 if (q->signal_when_queue_non_empty)
227 kill (q->consumer_pid, q->signal_when_queue_non_empty);
229 pthread_mutex_unlock (&q->mutex);
238 svm_queue_add2 (svm_queue_t * q, u8 * elem, u8 * elem2, int nowait)
241 int need_broadcast = 0;
245 /* zero on success */
246 if (pthread_mutex_trylock (&q->mutex))
252 pthread_mutex_lock (&q->mutex);
254 if (PREDICT_FALSE (q->cursize + 1 == q->maxsize))
258 pthread_mutex_unlock (&q->mutex);
261 while (q->cursize + 1 == q->maxsize)
263 (void) pthread_cond_wait (&q->condvar, &q->mutex);
267 tailp = (i8 *) (&q->data[0] + q->elsize * q->tail);
268 clib_memcpy (tailp, elem, q->elsize);
273 if (q->tail == q->maxsize)
276 need_broadcast = (q->cursize == 1);
278 tailp = (i8 *) (&q->data[0] + q->elsize * q->tail);
279 clib_memcpy (tailp, elem2, q->elsize);
284 if (q->tail == q->maxsize)
289 (void) pthread_cond_broadcast (&q->condvar);
290 if (q->signal_when_queue_non_empty)
291 kill (q->consumer_pid, q->signal_when_queue_non_empty);
293 pthread_mutex_unlock (&q->mutex);
302 svm_queue_sub (svm_queue_t * q, u8 * elem, int nowait)
305 int need_broadcast = 0;
309 /* zero on success */
310 if (pthread_mutex_trylock (&q->mutex))
316 pthread_mutex_lock (&q->mutex);
318 if (PREDICT_FALSE (q->cursize == 0))
322 pthread_mutex_unlock (&q->mutex);
325 while (q->cursize == 0)
327 (void) pthread_cond_wait (&q->condvar, &q->mutex);
331 headp = (i8 *) (&q->data[0] + q->elsize * q->head);
332 clib_memcpy (elem, headp, q->elsize);
335 /* $$$$ JFC shouldn't this be == 0? */
336 if (q->cursize == q->maxsize)
341 if (q->head == q->maxsize)
345 (void) pthread_cond_broadcast (&q->condvar);
347 pthread_mutex_unlock (&q->mutex);
353 svm_queue_sub2 (svm_queue_t * q, u8 * elem)
358 pthread_mutex_lock (&q->mutex);
361 pthread_mutex_unlock (&q->mutex);
365 headp = (i8 *) (&q->data[0] + q->elsize * q->head);
366 clib_memcpy (elem, headp, q->elsize);
369 need_broadcast = (q->cursize == q->maxsize / 2);
372 if (PREDICT_FALSE (q->head == q->maxsize))
374 pthread_mutex_unlock (&q->mutex);
377 (void) pthread_cond_broadcast (&q->condvar);
383 svm_queue_sub_raw (svm_queue_t * q, u8 * elem)
387 if (PREDICT_FALSE (q->cursize == 0))
389 while (q->cursize == 0)
393 headp = (i8 *) (&q->data[0] + q->elsize * q->head);
394 clib_memcpy (elem, headp, q->elsize);
399 if (q->head == q->maxsize)
405 * fd.io coding-style-patch-verification: ON
408 * eval: (c-set-style "gnu")