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>
29 #include <vppinfra/time.h>
35 * nels = number of elements on the queue
36 * elsize = element size, presumably 4 and cacheline-size will
40 * The idea is to call this function in the queue consumer,
41 * and e-mail the queue pointer to the producer(s).
43 * The vpp process / main thread allocates one of these
44 * at startup; its main input queue. The vpp main input queue
45 * has a pointer to it in the shared memory segment header.
47 * You probably want to be on an svm data heap before calling this
51 svm_queue_init (int nels,
52 int elsize, int consumer_pid, int signal_when_queue_non_empty)
55 pthread_mutexattr_t attr;
56 pthread_condattr_t cattr;
58 q = clib_mem_alloc_aligned (sizeof (svm_queue_t)
59 + nels * elsize, CLIB_CACHE_LINE_BYTES);
60 memset (q, 0, sizeof (*q));
64 q->consumer_pid = consumer_pid;
65 q->signal_when_queue_non_empty = signal_when_queue_non_empty;
67 memset (&attr, 0, sizeof (attr));
68 memset (&cattr, 0, sizeof (cattr));
70 if (pthread_mutexattr_init (&attr))
71 clib_unix_warning ("mutexattr_init");
72 if (pthread_mutexattr_setpshared (&attr, PTHREAD_PROCESS_SHARED))
73 clib_unix_warning ("pthread_mutexattr_setpshared");
74 if (pthread_mutex_init (&q->mutex, &attr))
75 clib_unix_warning ("mutex_init");
76 if (pthread_mutexattr_destroy (&attr))
77 clib_unix_warning ("mutexattr_destroy");
78 if (pthread_condattr_init (&cattr))
79 clib_unix_warning ("condattr_init");
80 /* prints funny-looking messages in the Linux target */
81 if (pthread_condattr_setpshared (&cattr, PTHREAD_PROCESS_SHARED))
82 clib_unix_warning ("condattr_setpshared");
83 if (pthread_cond_init (&q->condvar, &cattr))
84 clib_unix_warning ("cond_init1");
85 if (pthread_condattr_destroy (&cattr))
86 clib_unix_warning ("cond_init2");
95 svm_queue_free (svm_queue_t * q)
97 (void) pthread_mutex_destroy (&q->mutex);
98 (void) pthread_cond_destroy (&q->condvar);
103 svm_queue_lock (svm_queue_t * q)
105 pthread_mutex_lock (&q->mutex);
109 svm_queue_unlock (svm_queue_t * q)
111 pthread_mutex_unlock (&q->mutex);
115 svm_queue_is_full (svm_queue_t * q)
117 return q->cursize == q->maxsize;
121 * svm_queue_add_nolock
124 svm_queue_add_nolock (svm_queue_t * q, u8 * elem)
127 int need_broadcast = 0;
129 if (PREDICT_FALSE (q->cursize == q->maxsize))
131 while (q->cursize == q->maxsize)
133 (void) pthread_cond_wait (&q->condvar, &q->mutex);
137 tailp = (i8 *) (&q->data[0] + q->elsize * q->tail);
138 clib_memcpy (tailp, elem, q->elsize);
143 need_broadcast = (q->cursize == 1);
145 if (q->tail == q->maxsize)
150 (void) pthread_cond_broadcast (&q->condvar);
151 if (q->signal_when_queue_non_empty)
152 kill (q->consumer_pid, q->signal_when_queue_non_empty);
158 svm_queue_add_raw (svm_queue_t * q, u8 * elem)
162 if (PREDICT_FALSE (q->cursize == q->maxsize))
164 while (q->cursize == q->maxsize)
168 tailp = (i8 *) (&q->data[0] + q->elsize * q->tail);
169 clib_memcpy (tailp, elem, q->elsize);
174 if (q->tail == q->maxsize)
184 svm_queue_add (svm_queue_t * q, u8 * elem, int nowait)
187 int need_broadcast = 0;
191 /* zero on success */
192 if (pthread_mutex_trylock (&q->mutex))
198 pthread_mutex_lock (&q->mutex);
200 if (PREDICT_FALSE (q->cursize == q->maxsize))
204 pthread_mutex_unlock (&q->mutex);
207 while (q->cursize == q->maxsize)
209 (void) pthread_cond_wait (&q->condvar, &q->mutex);
213 tailp = (i8 *) (&q->data[0] + q->elsize * q->tail);
214 clib_memcpy (tailp, elem, q->elsize);
219 need_broadcast = (q->cursize == 1);
221 if (q->tail == q->maxsize)
226 (void) pthread_cond_broadcast (&q->condvar);
227 if (q->signal_when_queue_non_empty)
228 kill (q->consumer_pid, q->signal_when_queue_non_empty);
230 pthread_mutex_unlock (&q->mutex);
239 svm_queue_add2 (svm_queue_t * q, u8 * elem, u8 * elem2, int nowait)
242 int need_broadcast = 0;
246 /* zero on success */
247 if (pthread_mutex_trylock (&q->mutex))
253 pthread_mutex_lock (&q->mutex);
255 if (PREDICT_FALSE (q->cursize + 1 == q->maxsize))
259 pthread_mutex_unlock (&q->mutex);
262 while (q->cursize + 1 == q->maxsize)
264 (void) pthread_cond_wait (&q->condvar, &q->mutex);
268 tailp = (i8 *) (&q->data[0] + q->elsize * q->tail);
269 clib_memcpy (tailp, elem, q->elsize);
274 if (q->tail == q->maxsize)
277 need_broadcast = (q->cursize == 1);
279 tailp = (i8 *) (&q->data[0] + q->elsize * q->tail);
280 clib_memcpy (tailp, elem2, q->elsize);
285 if (q->tail == q->maxsize)
290 (void) pthread_cond_broadcast (&q->condvar);
291 if (q->signal_when_queue_non_empty)
292 kill (q->consumer_pid, q->signal_when_queue_non_empty);
294 pthread_mutex_unlock (&q->mutex);
303 svm_queue_sub (svm_queue_t * q, u8 * elem, svm_q_conditional_wait_t cond,
307 int need_broadcast = 0;
310 if (cond == SVM_Q_NOWAIT)
312 /* zero on success */
313 if (pthread_mutex_trylock (&q->mutex))
319 pthread_mutex_lock (&q->mutex);
321 if (PREDICT_FALSE (q->cursize == 0))
323 if (cond == SVM_Q_NOWAIT)
325 pthread_mutex_unlock (&q->mutex);
328 else if (cond == SVM_Q_TIMEDWAIT)
331 ts.tv_sec = unix_time_now () + time;
333 while (q->cursize == 0 && rc == 0)
335 rc = pthread_cond_timedwait (&q->condvar, &q->mutex, &ts);
339 pthread_mutex_unlock (&q->mutex);
345 while (q->cursize == 0)
347 (void) pthread_cond_wait (&q->condvar, &q->mutex);
352 headp = (i8 *) (&q->data[0] + q->elsize * q->head);
353 clib_memcpy (elem, headp, q->elsize);
356 /* $$$$ JFC shouldn't this be == 0? */
357 if (q->cursize == q->maxsize)
362 if (q->head == q->maxsize)
366 (void) pthread_cond_broadcast (&q->condvar);
368 pthread_mutex_unlock (&q->mutex);
374 svm_queue_sub2 (svm_queue_t * q, u8 * elem)
379 pthread_mutex_lock (&q->mutex);
382 pthread_mutex_unlock (&q->mutex);
386 headp = (i8 *) (&q->data[0] + q->elsize * q->head);
387 clib_memcpy (elem, headp, q->elsize);
390 need_broadcast = (q->cursize == q->maxsize / 2);
393 if (PREDICT_FALSE (q->head == q->maxsize))
395 pthread_mutex_unlock (&q->mutex);
398 (void) pthread_cond_broadcast (&q->condvar);
404 svm_queue_sub_raw (svm_queue_t * q, u8 * elem)
408 if (PREDICT_FALSE (q->cursize == 0))
410 while (q->cursize == 0)
414 headp = (i8 *) (&q->data[0] + q->elsize * q->head);
415 clib_memcpy (elem, headp, q->elsize);
420 if (q->head == q->maxsize)
426 * fd.io coding-style-patch-verification: ON
429 * eval: (c-set-style "gnu")