api: api socket respect unix runtime directory
[vpp.git] / src / vlibmemory / memory_shared.c
1 /*
2  *------------------------------------------------------------------
3  * memclnt_shared.c - API message handling, common code for both clients
4  * and the vlib process itself.
5  *
6  *
7  * Copyright (c) 2009 Cisco and/or its affiliates.
8  * Licensed under the Apache License, Version 2.0 (the "License");
9  * you may not use this file except in compliance with the License.
10  * You may obtain a copy of the License at:
11  *
12  *     http://www.apache.org/licenses/LICENSE-2.0
13  *
14  * Unless required by applicable law or agreed to in writing, software
15  * distributed under the License is distributed on an "AS IS" BASIS,
16  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
17  * See the License for the specific language governing permissions and
18  * limitations under the License.
19  *------------------------------------------------------------------
20  */
21
22 #include <stdio.h>
23 #include <stdlib.h>
24 #include <stddef.h>
25 #include <string.h>
26 #include <unistd.h>
27 #include <signal.h>
28
29 #include <vppinfra/format.h>
30 #include <vppinfra/byte_order.h>
31 #include <vppinfra/error.h>
32 #include <vppinfra/elog.h>
33 #include <svm/queue.h>
34 #include <vlib/vlib.h>
35 #include <vlib/unix/unix.h>
36 #include <vlibmemory/memory_api.h>
37 #include <vlibmemory/vl_memory_msg_enum.h>
38
39 #define vl_typedefs
40 #include <vlibmemory/vl_memory_api_h.h>
41 #undef vl_typedefs
42
43 #define DEBUG_MESSAGE_BUFFER_OVERRUN 0
44
45 static inline void *
46 vl_msg_api_alloc_internal (int nbytes, int pool, int may_return_null)
47 {
48   int i;
49   msgbuf_t *rv;
50   ring_alloc_t *ap;
51   svm_queue_t *q;
52   void *oldheap;
53   vl_shmem_hdr_t *shmem_hdr;
54   api_main_t *am = &api_main;
55
56   shmem_hdr = am->shmem_hdr;
57
58 #if DEBUG_MESSAGE_BUFFER_OVERRUN > 0
59   nbytes += 4;
60 #endif
61
62   ASSERT (pool == 0 || vlib_get_thread_index () == 0);
63
64   if (shmem_hdr == 0)
65     {
66       clib_warning ("shared memory header NULL");
67       return 0;
68     }
69
70   /* account for the msgbuf_t header */
71   nbytes += sizeof (msgbuf_t);
72
73   if (shmem_hdr->vl_rings == 0)
74     {
75       clib_warning ("vl_rings NULL");
76       ASSERT (0);
77       abort ();
78     }
79
80   if (shmem_hdr->client_rings == 0)
81     {
82       clib_warning ("client_rings NULL");
83       ASSERT (0);
84       abort ();
85     }
86
87   ap = pool ? shmem_hdr->vl_rings : shmem_hdr->client_rings;
88   for (i = 0; i < vec_len (ap); i++)
89     {
90       /* Too big? */
91       if (nbytes > ap[i].size)
92         {
93           continue;
94         }
95
96       q = ap[i].rp;
97       if (pool == 0)
98         {
99           pthread_mutex_lock (&q->mutex);
100         }
101       rv = (msgbuf_t *) (&q->data[0] + q->head * q->elsize);
102       /*
103        * Is this item still in use?
104        */
105       if (rv->q)
106         {
107           u32 now = (u32) time (0);
108
109           if (PREDICT_TRUE (rv->gc_mark_timestamp == 0))
110             rv->gc_mark_timestamp = now;
111           else
112             {
113               if (now - rv->gc_mark_timestamp > 10)
114                 {
115                   if (CLIB_DEBUG > 0)
116                     {
117                       u16 *msg_idp, msg_id;
118                       clib_warning
119                         ("garbage collect pool %d ring %d index %d", pool, i,
120                          q->head);
121                       msg_idp = (u16 *) (rv->data);
122                       msg_id = clib_net_to_host_u16 (*msg_idp);
123                       if (msg_id < vec_len (api_main.msg_names))
124                         clib_warning ("msg id %d name %s", (u32) msg_id,
125                                       api_main.msg_names[msg_id]);
126                     }
127                   shmem_hdr->garbage_collects++;
128                   goto collected;
129                 }
130             }
131
132
133           /* yes, loser; try next larger pool */
134           ap[i].misses++;
135           if (pool == 0)
136             pthread_mutex_unlock (&q->mutex);
137           continue;
138         }
139     collected:
140
141       /* OK, we have a winner */
142       ap[i].hits++;
143       /*
144        * Remember the source queue, although we
145        * don't need to know the queue to free the item.
146        */
147       rv->q = q;
148       rv->gc_mark_timestamp = 0;
149       q->head++;
150       if (q->head == q->maxsize)
151         q->head = 0;
152
153       if (pool == 0)
154         pthread_mutex_unlock (&q->mutex);
155       goto out;
156     }
157
158   /*
159    * Request too big, or head element of all size-compatible rings
160    * still in use. Fall back to shared-memory malloc.
161    */
162   am->ring_misses++;
163
164   pthread_mutex_lock (&am->vlib_rp->mutex);
165   oldheap = svm_push_data_heap (am->vlib_rp);
166   if (may_return_null)
167     {
168       rv = clib_mem_alloc_or_null (nbytes);
169       if (PREDICT_FALSE (rv == 0))
170         {
171           svm_pop_heap (oldheap);
172           pthread_mutex_unlock (&am->vlib_rp->mutex);
173           return 0;
174         }
175     }
176   else
177     rv = clib_mem_alloc (nbytes);
178
179   rv->q = 0;
180   rv->gc_mark_timestamp = 0;
181   svm_pop_heap (oldheap);
182   pthread_mutex_unlock (&am->vlib_rp->mutex);
183
184 out:
185 #if DEBUG_MESSAGE_BUFFER_OVERRUN > 0
186   {
187     nbytes -= 4;
188     u32 *overrun;
189     overrun = (u32 *) (rv->data + nbytes - sizeof (msgbuf_t));
190     *overrun = 0x1badbabe;
191   }
192 #endif
193   rv->data_len = htonl (nbytes - sizeof (msgbuf_t));
194
195   return (rv->data);
196 }
197
198 void *
199 vl_msg_api_alloc (int nbytes)
200 {
201   int pool;
202   api_main_t *am = &api_main;
203   vl_shmem_hdr_t *shmem_hdr = am->shmem_hdr;
204
205   /*
206    * Clients use pool-0, vlib proc uses pool 1
207    */
208   pool = (am->our_pid == shmem_hdr->vl_pid);
209   return vl_msg_api_alloc_internal (nbytes, pool, 0 /* may_return_null */ );
210 }
211
212 void *
213 vl_msg_api_alloc_zero (int nbytes)
214 {
215   void *ret;
216
217   ret = vl_msg_api_alloc (nbytes);
218   clib_memset (ret, 0, nbytes);
219   return ret;
220 }
221
222 void *
223 vl_msg_api_alloc_or_null (int nbytes)
224 {
225   int pool;
226   api_main_t *am = &api_main;
227   vl_shmem_hdr_t *shmem_hdr = am->shmem_hdr;
228
229   pool = (am->our_pid == shmem_hdr->vl_pid);
230   return vl_msg_api_alloc_internal (nbytes, pool, 1 /* may_return_null */ );
231 }
232
233 void *
234 vl_msg_api_alloc_as_if_client (int nbytes)
235 {
236   return vl_msg_api_alloc_internal (nbytes, 0, 0 /* may_return_null */ );
237 }
238
239 void *
240 vl_msg_api_alloc_zero_as_if_client (int nbytes)
241 {
242   void *ret;
243
244   ret = vl_msg_api_alloc_as_if_client (nbytes);
245   clib_memset (ret, 0, nbytes);
246   return ret;
247 }
248
249 void *
250 vl_msg_api_alloc_as_if_client_or_null (int nbytes)
251 {
252   return vl_msg_api_alloc_internal (nbytes, 0, 1 /* may_return_null */ );
253 }
254
255 void *
256 vl_mem_api_alloc_as_if_client_w_reg (vl_api_registration_t * reg, int nbytes)
257 {
258   api_main_t *am = &api_main;
259   vl_shmem_hdr_t *save_shmem_hdr = am->shmem_hdr;
260   svm_region_t *vlib_rp, *save_vlib_rp = am->vlib_rp;
261   void *msg;
262
263   vlib_rp = am->vlib_rp = reg->vlib_rp;
264   am->shmem_hdr = (void *) vlib_rp->user_ctx;
265
266   msg = vl_msg_api_alloc_internal (nbytes, 0, 0 /* may_return_null */ );
267
268   am->shmem_hdr = save_shmem_hdr;
269   am->vlib_rp = save_vlib_rp;
270
271   return msg;
272 }
273
274 void
275 vl_msg_api_free (void *a)
276 {
277   msgbuf_t *rv;
278   void *oldheap;
279   api_main_t *am = &api_main;
280
281   rv = (msgbuf_t *) (((u8 *) a) - offsetof (msgbuf_t, data));
282
283   /*
284    * Here's the beauty of the scheme.  Only one proc/thread has
285    * control of a given message buffer. To free a buffer, we just clear the
286    * queue field, and leave. No locks, no hits, no errors...
287    */
288   if (rv->q)
289     {
290       rv->q = 0;
291       rv->gc_mark_timestamp = 0;
292 #if DEBUG_MESSAGE_BUFFER_OVERRUN > 0
293       {
294         u32 *overrun;
295         overrun = (u32 *) (rv->data + ntohl (rv->data_len));
296         ASSERT (*overrun == 0x1badbabe);
297       }
298 #endif
299       return;
300     }
301
302   pthread_mutex_lock (&am->vlib_rp->mutex);
303   oldheap = svm_push_data_heap (am->vlib_rp);
304
305 #if DEBUG_MESSAGE_BUFFER_OVERRUN > 0
306   {
307     u32 *overrun;
308     overrun = (u32 *) (rv->data + ntohl (rv->data_len));
309     ASSERT (*overrun == 0x1badbabe);
310   }
311 #endif
312
313   clib_mem_free (rv);
314   svm_pop_heap (oldheap);
315   pthread_mutex_unlock (&am->vlib_rp->mutex);
316 }
317
318 static void
319 vl_msg_api_free_nolock (void *a)
320 {
321   msgbuf_t *rv;
322   void *oldheap;
323   api_main_t *am = &api_main;
324
325   rv = (msgbuf_t *) (((u8 *) a) - offsetof (msgbuf_t, data));
326   /*
327    * Here's the beauty of the scheme.  Only one proc/thread has
328    * control of a given message buffer. To free a buffer, we just clear the
329    * queue field, and leave. No locks, no hits, no errors...
330    */
331   if (rv->q)
332     {
333       rv->q = 0;
334       return;
335     }
336
337   oldheap = svm_push_data_heap (am->vlib_rp);
338   clib_mem_free (rv);
339   svm_pop_heap (oldheap);
340 }
341
342 void
343 vl_set_memory_root_path (const char *name)
344 {
345   api_main_t *am = &api_main;
346
347   am->root_path = name;
348 }
349
350 void
351 vl_set_memory_uid (int uid)
352 {
353   api_main_t *am = &api_main;
354
355   am->api_uid = uid;
356 }
357
358 void
359 vl_set_memory_gid (int gid)
360 {
361   api_main_t *am = &api_main;
362
363   am->api_gid = gid;
364 }
365
366 void
367 vl_set_global_memory_baseva (u64 baseva)
368 {
369   api_main_t *am = &api_main;
370
371   am->global_baseva = baseva;
372 }
373
374 void
375 vl_set_global_memory_size (u64 size)
376 {
377   api_main_t *am = &api_main;
378
379   am->global_size = size;
380 }
381
382 void
383 vl_set_api_memory_size (u64 size)
384 {
385   api_main_t *am = &api_main;
386
387   am->api_size = size;
388 }
389
390 void
391 vl_set_global_pvt_heap_size (u64 size)
392 {
393   api_main_t *am = &api_main;
394
395   am->global_pvt_heap_size = size;
396 }
397
398 void
399 vl_set_api_pvt_heap_size (u64 size)
400 {
401   api_main_t *am = &api_main;
402
403   am->api_pvt_heap_size = size;
404 }
405
406 static void
407 vl_api_default_mem_config (vl_shmem_hdr_t * shmem_hdr)
408 {
409   api_main_t *am = &api_main;
410   u32 vlib_input_queue_length;
411
412   /* vlib main input queue */
413   vlib_input_queue_length = 1024;
414   if (am->vlib_input_queue_length)
415     vlib_input_queue_length = am->vlib_input_queue_length;
416
417   shmem_hdr->vl_input_queue =
418     svm_queue_alloc_and_init (vlib_input_queue_length, sizeof (uword),
419                               getpid ());
420
421 #define _(sz,n)                                                 \
422     do {                                                        \
423         ring_alloc_t _rp;                                       \
424         _rp.rp = svm_queue_alloc_and_init ((n), (sz), 0);       \
425         _rp.size = (sz);                                        \
426         _rp.nitems = n;                                         \
427         _rp.hits = 0;                                           \
428         _rp.misses = 0;                                         \
429         vec_add1(shmem_hdr->vl_rings, _rp);                     \
430     } while (0);
431
432   foreach_vl_aring_size;
433 #undef _
434
435 #define _(sz,n)                                                 \
436     do {                                                        \
437         ring_alloc_t _rp;                                       \
438         _rp.rp = svm_queue_alloc_and_init ((n), (sz), 0);       \
439         _rp.size = (sz);                                        \
440         _rp.nitems = n;                                         \
441         _rp.hits = 0;                                           \
442         _rp.misses = 0;                                         \
443         vec_add1(shmem_hdr->client_rings, _rp);                 \
444     } while (0);
445
446   foreach_clnt_aring_size;
447 #undef _
448 }
449
450 void
451 vl_api_mem_config (vl_shmem_hdr_t * hdr, vl_api_shm_elem_config_t * config)
452 {
453   vl_api_shm_elem_config_t *c;
454   ring_alloc_t *rp;
455   u32 size;
456
457   if (!config)
458     {
459       vl_api_default_mem_config (hdr);
460       return;
461     }
462
463   vec_foreach (c, config)
464   {
465     switch (c->type)
466       {
467       case VL_API_QUEUE:
468         hdr->vl_input_queue = svm_queue_alloc_and_init (c->count, c->size,
469                                                         getpid ());
470         continue;
471       case VL_API_VLIB_RING:
472         vec_add2 (hdr->vl_rings, rp, 1);
473         break;
474       case VL_API_CLIENT_RING:
475         vec_add2 (hdr->client_rings, rp, 1);
476         break;
477       default:
478         clib_warning ("unknown config type: %d", c->type);
479         continue;
480       }
481
482     size = sizeof (ring_alloc_t) + c->size;
483     rp->rp = svm_queue_alloc_and_init (c->count, size, 0);
484     rp->size = size;
485     rp->nitems = c->count;
486     rp->hits = 0;
487     rp->misses = 0;
488   }
489 }
490
491 void
492 vl_init_shmem (svm_region_t * vlib_rp, vl_api_shm_elem_config_t * config,
493                int is_vlib, int is_private_region)
494 {
495   api_main_t *am = &api_main;
496   vl_shmem_hdr_t *shmem_hdr = 0;
497   void *oldheap;
498   ASSERT (vlib_rp);
499
500   /* $$$$ need private region config parameters */
501
502   oldheap = svm_push_data_heap (vlib_rp);
503
504   vec_validate (shmem_hdr, 0);
505   shmem_hdr->version = VL_SHM_VERSION;
506   shmem_hdr->clib_file_index = VL_API_INVALID_FI;
507
508   /* Set up the queue and msg ring allocator */
509   vl_api_mem_config (shmem_hdr, config);
510
511   if (is_private_region == 0)
512     {
513       am->shmem_hdr = shmem_hdr;
514       am->vlib_rp = vlib_rp;
515       am->our_pid = getpid ();
516       if (is_vlib)
517         am->shmem_hdr->vl_pid = am->our_pid;
518     }
519   else
520     shmem_hdr->vl_pid = am->our_pid;
521
522   svm_pop_heap (oldheap);
523
524   /*
525    * After absolutely everything that a client might see is set up,
526    * declare the shmem region valid
527    */
528   vlib_rp->user_ctx = shmem_hdr;
529
530   pthread_mutex_unlock (&vlib_rp->mutex);
531 }
532
533 int
534 vl_map_shmem (const char *region_name, int is_vlib)
535 {
536   svm_map_region_args_t _a, *a = &_a;
537   svm_region_t *vlib_rp, *root_rp;
538   api_main_t *am = &api_main;
539   int i;
540   struct timespec ts, tsrem;
541   char *vpe_api_region_suffix = "-vpe-api";
542
543   clib_memset (a, 0, sizeof (*a));
544
545   if (strstr (region_name, vpe_api_region_suffix))
546     {
547       u8 *root_path = format (0, "%s", region_name);
548       _vec_len (root_path) = (vec_len (root_path) -
549                               strlen (vpe_api_region_suffix));
550       vec_terminate_c_string (root_path);
551       a->root_path = (const char *) root_path;
552       am->root_path = (const char *) root_path;
553     }
554
555   if (is_vlib == 0)
556     {
557       int tfd;
558       u8 *api_name;
559       /*
560        * Clients wait for vpp to set up the root / API regioins
561        */
562       if (am->root_path)
563         api_name = format (0, "/dev/shm/%s-%s%c", am->root_path,
564                            region_name + 1, 0);
565       else
566         api_name = format (0, "/dev/shm%s%c", region_name, 0);
567
568       /* Wait up to 100 seconds... */
569       for (i = 0; i < 10000; i++)
570         {
571           ts.tv_sec = 0;
572           ts.tv_nsec = 10000 * 1000;    /* 10 ms */
573           while (nanosleep (&ts, &tsrem) < 0)
574             ts = tsrem;
575           tfd = open ((char *) api_name, O_RDWR);
576           if (tfd >= 0)
577             break;
578         }
579       vec_free (api_name);
580       if (tfd < 0)
581         {
582           clib_warning ("region init fail");
583           return -2;
584         }
585       close (tfd);
586       svm_region_init_chroot_uid_gid (am->root_path, getuid (), getgid ());
587     }
588
589   if (a->root_path != NULL)
590     {
591       a->name = "/vpe-api";
592     }
593   else
594     a->name = region_name;
595   a->size = am->api_size ? am->api_size : (16 << 20);
596   a->flags = SVM_FLAGS_MHEAP;
597   a->uid = am->api_uid;
598   a->gid = am->api_gid;
599   a->pvt_heap_size = am->api_pvt_heap_size;
600
601   vlib_rp = svm_region_find_or_create (a);
602
603   if (vlib_rp == 0)
604     return (-2);
605
606   pthread_mutex_lock (&vlib_rp->mutex);
607   /* Has someone else set up the shared-memory variable table? */
608   if (vlib_rp->user_ctx)
609     {
610       am->shmem_hdr = (void *) vlib_rp->user_ctx;
611       am->our_pid = getpid ();
612       if (is_vlib)
613         {
614           svm_queue_t *q;
615           uword old_msg;
616           /*
617            * application restart. Reset cached pids, API message
618            * rings, list of clients; otherwise, various things
619            * fail. (e.g. queue non-empty notification)
620            */
621
622           /* ghosts keep the region from disappearing properly */
623           svm_client_scan_this_region_nolock (vlib_rp);
624           am->shmem_hdr->application_restarts++;
625           q = am->shmem_hdr->vl_input_queue;
626           am->shmem_hdr->vl_pid = getpid ();
627           q->consumer_pid = am->shmem_hdr->vl_pid;
628           /* Drain the input queue, freeing msgs */
629           for (i = 0; i < 10; i++)
630             {
631               if (pthread_mutex_trylock (&q->mutex) == 0)
632                 {
633                   pthread_mutex_unlock (&q->mutex);
634                   goto mutex_ok;
635                 }
636               ts.tv_sec = 0;
637               ts.tv_nsec = 10000 * 1000;        /* 10 ms */
638               while (nanosleep (&ts, &tsrem) < 0)
639                 ts = tsrem;
640             }
641           /* Mutex buggered, "fix" it */
642           clib_memset (&q->mutex, 0, sizeof (q->mutex));
643           clib_warning ("forcibly release main input queue mutex");
644
645         mutex_ok:
646           am->vlib_rp = vlib_rp;
647           while (svm_queue_sub (q, (u8 *) & old_msg, SVM_Q_NOWAIT, 0)
648                  != -2 /* queue underflow */ )
649             {
650               vl_msg_api_free_nolock ((void *) old_msg);
651               am->shmem_hdr->restart_reclaims++;
652             }
653           pthread_mutex_unlock (&vlib_rp->mutex);
654           root_rp = svm_get_root_rp ();
655           ASSERT (root_rp);
656           /* Clean up the root region client list */
657           pthread_mutex_lock (&root_rp->mutex);
658           svm_client_scan_this_region_nolock (root_rp);
659           pthread_mutex_unlock (&root_rp->mutex);
660         }
661       else
662         {
663           pthread_mutex_unlock (&vlib_rp->mutex);
664         }
665       am->vlib_rp = vlib_rp;
666       vec_add1 (am->mapped_shmem_regions, vlib_rp);
667       return 0;
668     }
669   /* Clients simply have to wait... */
670   if (!is_vlib)
671     {
672       pthread_mutex_unlock (&vlib_rp->mutex);
673
674       /* Wait up to 100 seconds... */
675       for (i = 0; i < 10000; i++)
676         {
677           ts.tv_sec = 0;
678           ts.tv_nsec = 10000 * 1000;    /* 10 ms */
679           while (nanosleep (&ts, &tsrem) < 0)
680             ts = tsrem;
681           if (vlib_rp->user_ctx)
682             goto ready;
683         }
684       /* Clean up and leave... */
685       svm_region_unmap (vlib_rp);
686       clib_warning ("region init fail");
687       return (-2);
688
689     ready:
690       am->shmem_hdr = (void *) vlib_rp->user_ctx;
691       am->our_pid = getpid ();
692       am->vlib_rp = vlib_rp;
693       vec_add1 (am->mapped_shmem_regions, vlib_rp);
694       return 0;
695     }
696
697   /* Nope, it's our problem... */
698   vl_init_shmem (vlib_rp, 0 /* default config */ , 1 /* is vlib */ ,
699                  0 /* is_private_region */ );
700
701   vec_add1 (am->mapped_shmem_regions, vlib_rp);
702   return 0;
703 }
704
705 void
706 vl_register_mapped_shmem_region (svm_region_t * rp)
707 {
708   api_main_t *am = &api_main;
709
710   vec_add1 (am->mapped_shmem_regions, rp);
711 }
712
713 static void
714 vl_unmap_shmem_internal (u8 is_client)
715 {
716   svm_region_t *rp;
717   int i;
718   api_main_t *am = &api_main;
719
720   if (!svm_get_root_rp ())
721     return;
722
723   for (i = 0; i < vec_len (am->mapped_shmem_regions); i++)
724     {
725       rp = am->mapped_shmem_regions[i];
726       is_client ? svm_region_unmap_client (rp) : svm_region_unmap (rp);
727     }
728
729   vec_free (am->mapped_shmem_regions);
730   am->shmem_hdr = 0;
731
732   is_client ? svm_region_exit_client () : svm_region_exit ();
733
734   /* $$$ more careful cleanup, valgrind run... */
735   vec_free (am->msg_handlers);
736   vec_free (am->msg_endian_handlers);
737   vec_free (am->msg_print_handlers);
738 }
739
740 void
741 vl_unmap_shmem (void)
742 {
743   vl_unmap_shmem_internal (0);
744 }
745
746 void
747 vl_unmap_shmem_client (void)
748 {
749   vl_unmap_shmem_internal (1);
750 }
751
752 void
753 vl_msg_api_send_shmem (svm_queue_t * q, u8 * elem)
754 {
755   api_main_t *am = &api_main;
756   uword *trace = (uword *) elem;
757
758   if (am->tx_trace && am->tx_trace->enabled)
759     vl_msg_api_trace (am, am->tx_trace, (void *) trace[0]);
760
761   /*
762    * Announce a probable binary API client bug:
763    * some client's input queue is stuffed.
764    * The situation may be recoverable, or not.
765    */
766   if (PREDICT_FALSE
767       (am->vl_clients /* vpp side */  && (q->cursize == q->maxsize)))
768     {
769       if (PREDICT_FALSE (am->elog_trace_api_messages))
770         {
771           /* *INDENT-OFF* */
772           ELOG_TYPE_DECLARE (e) =
773             {
774               .format = "api-client-queue-stuffed: %x%x",
775               .format_args = "i4i4",
776             };
777           /* *INDENT-ON* */
778           struct
779           {
780             u32 hi, low;
781           } *ed;
782           ed = ELOG_DATA (am->elog_main, e);
783           ed->hi = (uword) q >> 32;
784           ed->low = (uword) q & 0xFFFFFFFF;
785           clib_warning ("WARNING: client input queue at %llx is stuffed...",
786                         q);
787         }
788     }
789   (void) svm_queue_add (q, elem, 0 /* nowait */ );
790 }
791
792 int
793 vl_mem_api_can_send (svm_queue_t * q)
794 {
795   return (q->cursize < q->maxsize);
796 }
797
798 void
799 vl_msg_api_send_shmem_nolock (svm_queue_t * q, u8 * elem)
800 {
801   api_main_t *am = &api_main;
802   uword *trace = (uword *) elem;
803
804   if (am->tx_trace && am->tx_trace->enabled)
805     vl_msg_api_trace (am, am->tx_trace, (void *) trace[0]);
806
807   (void) svm_queue_add_nolock (q, elem);
808 }
809
810 /*
811  * fd.io coding-style-patch-verification: ON
812  *
813  * Local Variables:
814  * eval: (c-set-style "gnu")
815  * End:
816  */