2 *------------------------------------------------------------------
3 * unix_shared_memory_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 *------------------------------------------------------------------
24 #include <vppinfra/mem.h>
25 #include <vppinfra/format.h>
26 #include <vppinfra/cache.h>
27 #include <vlibmemory/unix_shared_memory_queue.h>
31 * unix_shared_memory_queue_init
33 * nels = number of elements on the queue
34 * elsize = element size, presumably 4 and cacheline-size will
36 * coid = consumer coid, from ChannelCreate
38 * pulse_code = pulse code consumer expects
39 * pulse_value = pulse value consumer expects
40 * consumer_prio = consumer's priority, so pulses won't change
41 * the consumer's priority.
43 * The idea is to call this function in the queue consumer,
44 * and e-mail the queue pointer to the producer(s).
46 * The spp process / main thread allocates one of these
47 * at startup; its main input queue. The spp main input queue
48 * has a pointer to it in the shared memory segment header.
50 * You probably want to be on an svm data heap before calling this
53 unix_shared_memory_queue_t *
54 unix_shared_memory_queue_init (int nels,
57 int signal_when_queue_non_empty)
59 unix_shared_memory_queue_t *q;
60 pthread_mutexattr_t attr;
61 pthread_condattr_t cattr;
63 q = clib_mem_alloc_aligned (sizeof (unix_shared_memory_queue_t)
64 + nels * elsize, CLIB_CACHE_LINE_BYTES);
65 memset (q, 0, sizeof (*q));
69 q->consumer_pid = consumer_pid;
70 q->signal_when_queue_non_empty = signal_when_queue_non_empty;
72 memset (&attr, 0, sizeof (attr));
73 memset (&cattr, 0, sizeof (attr));
75 if (pthread_mutexattr_init (&attr))
76 clib_unix_warning ("mutexattr_init");
77 if (pthread_mutexattr_setpshared (&attr, PTHREAD_PROCESS_SHARED))
78 clib_unix_warning ("pthread_mutexattr_setpshared");
79 if (pthread_mutex_init (&q->mutex, &attr))
80 clib_unix_warning ("mutex_init");
81 if (pthread_mutexattr_destroy (&attr))
82 clib_unix_warning ("mutexattr_destroy");
83 if (pthread_condattr_init (&cattr))
84 clib_unix_warning ("condattr_init");
85 /* prints funny-looking messages in the Linux target */
86 if (pthread_condattr_setpshared (&cattr, PTHREAD_PROCESS_SHARED))
87 clib_unix_warning ("condattr_setpshared");
88 if (pthread_cond_init (&q->condvar, &cattr))
89 clib_unix_warning ("cond_init1");
90 if (pthread_condattr_destroy (&cattr))
91 clib_unix_warning ("cond_init2");
97 * unix_shared_memory_queue_free
100 unix_shared_memory_queue_free (unix_shared_memory_queue_t * q)
102 (void) pthread_mutex_destroy (&q->mutex);
103 (void) pthread_cond_destroy (&q->condvar);
108 unix_shared_memory_queue_lock (unix_shared_memory_queue_t * q)
110 pthread_mutex_lock (&q->mutex);
114 unix_shared_memory_queue_unlock (unix_shared_memory_queue_t * q)
116 pthread_mutex_unlock (&q->mutex);
120 unix_shared_memory_queue_is_full (unix_shared_memory_queue_t * q)
122 return q->cursize == q->maxsize;
126 * unix_shared_memory_queue_add_nolock
129 unix_shared_memory_queue_add_nolock (unix_shared_memory_queue_t * q,
133 int need_broadcast = 0;
135 if (PREDICT_FALSE (q->cursize == q->maxsize))
137 while (q->cursize == q->maxsize)
139 (void) pthread_cond_wait (&q->condvar, &q->mutex);
143 tailp = (i8 *) (&q->data[0] + q->elsize * q->tail);
144 clib_memcpy (tailp, elem, q->elsize);
149 need_broadcast = (q->cursize == 1);
151 if (q->tail == q->maxsize)
156 (void) pthread_cond_broadcast (&q->condvar);
157 if (q->signal_when_queue_non_empty)
158 kill (q->consumer_pid, q->signal_when_queue_non_empty);
164 unix_shared_memory_queue_add_raw (unix_shared_memory_queue_t * q, u8 * elem)
168 if (PREDICT_FALSE (q->cursize == q->maxsize))
170 while (q->cursize == q->maxsize)
174 tailp = (i8 *) (&q->data[0] + q->elsize * q->tail);
175 clib_memcpy (tailp, elem, q->elsize);
180 if (q->tail == q->maxsize)
187 * unix_shared_memory_queue_add
190 unix_shared_memory_queue_add (unix_shared_memory_queue_t * q,
191 u8 * elem, int nowait)
194 int need_broadcast = 0;
198 /* zero on success */
199 if (pthread_mutex_trylock (&q->mutex))
205 pthread_mutex_lock (&q->mutex);
207 if (PREDICT_FALSE (q->cursize == q->maxsize))
211 pthread_mutex_unlock (&q->mutex);
214 while (q->cursize == q->maxsize)
216 (void) pthread_cond_wait (&q->condvar, &q->mutex);
220 tailp = (i8 *) (&q->data[0] + q->elsize * q->tail);
221 clib_memcpy (tailp, elem, q->elsize);
226 need_broadcast = (q->cursize == 1);
228 if (q->tail == q->maxsize)
233 (void) pthread_cond_broadcast (&q->condvar);
234 if (q->signal_when_queue_non_empty)
235 kill (q->consumer_pid, q->signal_when_queue_non_empty);
237 pthread_mutex_unlock (&q->mutex);
243 * unix_shared_memory_queue_sub
246 unix_shared_memory_queue_sub (unix_shared_memory_queue_t * q,
247 u8 * elem, int nowait)
250 int need_broadcast = 0;
254 /* zero on success */
255 if (pthread_mutex_trylock (&q->mutex))
261 pthread_mutex_lock (&q->mutex);
263 if (PREDICT_FALSE (q->cursize == 0))
267 pthread_mutex_unlock (&q->mutex);
270 while (q->cursize == 0)
272 (void) pthread_cond_wait (&q->condvar, &q->mutex);
276 headp = (i8 *) (&q->data[0] + q->elsize * q->head);
277 clib_memcpy (elem, headp, q->elsize);
280 if (q->cursize == q->maxsize)
285 if (q->head == q->maxsize)
289 (void) pthread_cond_broadcast (&q->condvar);
291 pthread_mutex_unlock (&q->mutex);
297 unix_shared_memory_queue_sub_raw (unix_shared_memory_queue_t * q, u8 * elem)
301 if (PREDICT_FALSE (q->cursize == 0))
303 while (q->cursize == 0)
307 headp = (i8 *) (&q->data[0] + q->elsize * q->head);
308 clib_memcpy (elem, headp, q->elsize);
313 if (q->head == q->maxsize)
319 * fd.io coding-style-patch-verification: ON
322 * eval: (c-set-style "gnu")