FD.io VPP  v19.08.3-2-gbabecb413
Vector Packet Processing
client.c
Go to the documentation of this file.
1 /*
2  * Copyright (c) 2016 Cisco and/or its affiliates.
3  * Licensed under the Apache License, Version 2.0 (the "License");
4  * you may not use this file except in compliance with the License.
5  * You may obtain a copy of the License at:
6  *
7  * http://www.apache.org/licenses/LICENSE-2.0
8  *
9  * Unless required by applicable law or agreed to in writing, software
10  * distributed under the License is distributed on an "AS IS" BASIS,
11  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12  * See the License for the specific language governing permissions and
13  * limitations under the License.
14  */
15 #include <stdio.h>
16 #include <stdlib.h>
17 #include <stddef.h>
18 #include <sys/types.h>
19 #include <sys/socket.h>
20 #include <sys/mman.h>
21 #include <sys/stat.h>
22 #include <netinet/in.h>
23 #include <netdb.h>
24 #include <signal.h>
25 #include <stdbool.h>
26 #include <vnet/vnet.h>
27 #include <vlib/vlib.h>
28 #include <vlib/unix/unix.h>
29 #include <vlibapi/api.h>
30 #include <vlibmemory/api.h>
31 
32 #include <vpp/api/vpe_msg_enum.h>
33 
34 #include "vppapiclient.h"
35 
39 
40 /*
41  * Asynchronous mode:
42  * Client registers a callback. All messages are sent to the callback.
43  * Synchronous mode:
44  * Client calls blocking read().
45  * Clients are expected to collate events on a queue.
46  * vac_write() -> suspends RX thread
47  * vac_read() -> resumes RX thread
48  */
49 
50 #define vl_typedefs /* define message structures */
51 #include <vpp/api/vpe_all_api_h.h>
52 #undef vl_typedefs
53 
54 #define vl_endianfun /* define message structures */
55 #include <vpp/api/vpe_all_api_h.h>
56 #undef vl_endianfun
57 
60 
61 typedef struct {
63  pthread_t rx_thread_handle;
65  pthread_mutex_t queue_lock;
66  pthread_cond_t suspend_cv;
67  pthread_cond_t resume_cv;
68  pthread_mutex_t timeout_lock;
70  pthread_cond_t timeout_cv;
71  pthread_cond_t timeout_cancel_cv;
72  pthread_cond_t terminate_cv;
73 } vac_main_t;
74 
78 bool rx_is_running = false;
80 
81 /* Set to true to enable memory tracing */
82 bool mem_trace = false;
83 
84 __attribute__((constructor))
85 static void
87 {
88  clib_mem_init (0, 1 << 30);
89 #if USE_DLMALLOC == 0
90  {
91  u8 *heap;
92  mheap_t *h;
93 
94  heap = clib_mem_get_per_cpu_heap ();
95  h = mheap_header (heap);
96  /* make the main heap thread-safe */
98  }
99 #endif
100  if (mem_trace)
101  clib_mem_trace (1);
102 }
103 
104 __attribute__((destructor))
105 static void
107 {
108  if (mem_trace)
109  fformat(stderr, "TRACE: %s",
110  format (0, "%U\n",
112 }
113 
114 
115 static void
116 init (void)
117 {
118  vac_main_t *pm = &vac_main;
119  clib_memset(pm, 0, sizeof(*pm));
120  pthread_mutex_init(&pm->queue_lock, NULL);
121  pthread_cond_init(&pm->suspend_cv, NULL);
122  pthread_cond_init(&pm->resume_cv, NULL);
123  pthread_mutex_init(&pm->timeout_lock, NULL);
124  pm->timeout_loop = 1;
125  pthread_cond_init(&pm->timeout_cv, NULL);
126  pthread_cond_init(&pm->timeout_cancel_cv, NULL);
127  pthread_cond_init(&pm->terminate_cv, NULL);
128 }
129 
130 static void
131 cleanup (void)
132 {
133  vac_main_t *pm = &vac_main;
134  pthread_mutex_destroy(&pm->queue_lock);
135  pthread_cond_destroy(&pm->suspend_cv);
136  pthread_cond_destroy(&pm->resume_cv);
137  pthread_mutex_destroy(&pm->timeout_lock);
138  pthread_cond_destroy(&pm->timeout_cv);
139  pthread_cond_destroy(&pm->timeout_cancel_cv);
140  pthread_cond_destroy(&pm->terminate_cv);
141  clib_memset(pm, 0, sizeof(*pm));
142 }
143 
144 /*
145  * Satisfy external references when -lvlib is not available.
146  */
147 void vlib_cli_output (struct vlib_main_t * vm, char * fmt, ...)
148 {
149  clib_warning ("vlib_cli_output called...");
150 }
151 
152 void
153 vac_free (void * msg)
154 {
155  vl_msg_api_free (msg);
156 }
157 
158 static void
159 vac_api_handler (void *msg)
160 {
161  u16 id = ntohs(*((u16 *)msg));
162  msgbuf_t *msgbuf = (msgbuf_t *)(((u8 *)msg) - offsetof(msgbuf_t, data));
163  int l = ntohl(msgbuf->data_len);
164  if (l == 0)
165  clib_warning("Message ID %d has wrong length: %d\n", id, l);
166 
167  /* Call Python callback */
168  ASSERT(vac_callback);
169  (vac_callback)(msg, l);
170  vac_free(msg);
171 }
172 
173 static void *
174 vac_rx_thread_fn (void *arg)
175 {
176  svm_queue_t *q;
178  vl_api_memclnt_keepalive_reply_t *rmp;
179  vac_main_t *pm = &vac_main;
180  api_main_t *am = &api_main;
181  vl_shmem_hdr_t *shmem_hdr;
182  uword msg;
183 
184  q = am->vl_input_queue;
185 
186  while (1)
187  while (!svm_queue_sub(q, (u8 *)&msg, SVM_Q_WAIT, 0))
188  {
189  u16 id = ntohs(*((u16 *)msg));
190  switch (id) {
191  case VL_API_RX_THREAD_EXIT:
192  vl_msg_api_free((void *) msg);
193  /* signal waiting threads that this thread is about to terminate */
194  pthread_mutex_lock(&pm->queue_lock);
195  rx_thread_done = true;
196  pthread_cond_signal(&pm->terminate_cv);
197  pthread_mutex_unlock(&pm->queue_lock);
198  pthread_exit(0);
199  return 0;
200  break;
201 
202  case VL_API_MEMCLNT_RX_THREAD_SUSPEND:
203  vl_msg_api_free((void * )msg);
204  /* Suspend thread and signal reader */
205  pthread_mutex_lock(&pm->queue_lock);
206  pthread_cond_signal(&pm->suspend_cv);
207  /* Wait for the resume signal */
208  pthread_cond_wait (&pm->resume_cv, &pm->queue_lock);
209  pthread_mutex_unlock(&pm->queue_lock);
210  break;
211 
212  case VL_API_MEMCLNT_READ_TIMEOUT:
213  clib_warning("Received read timeout in async thread\n");
214  vl_msg_api_free((void *) msg);
215  break;
216 
217  case VL_API_MEMCLNT_KEEPALIVE:
218  mp = (void *)msg;
219  rmp = vl_msg_api_alloc (sizeof (*rmp));
220  clib_memset (rmp, 0, sizeof (*rmp));
221  rmp->_vl_msg_id = ntohs(VL_API_MEMCLNT_KEEPALIVE_REPLY);
222  rmp->context = mp->context;
223  shmem_hdr = am->shmem_hdr;
224  vl_msg_api_send_shmem(shmem_hdr->vl_input_queue, (u8 *)&rmp);
225  vl_msg_api_free((void *) msg);
226  break;
227 
228  default:
229  vac_api_handler((void *)msg);
230  }
231  }
232 }
233 
234 static void *
236 {
238  vac_main_t *pm = &vac_main;
239  api_main_t *am = &api_main;
240  struct timespec ts;
241  struct timeval tv;
242  int rv;
243 
244  while (pm->timeout_loop)
245  {
246  /* Wait for poke */
247  pthread_mutex_lock(&pm->timeout_lock);
248  while (!timeout_in_progress)
249  pthread_cond_wait (&pm->timeout_cv, &pm->timeout_lock);
250 
251  /* Starting timer */
252  gettimeofday(&tv, NULL);
253  ts.tv_sec = tv.tv_sec + read_timeout;
254  ts.tv_nsec = 0;
255 
256  if (!timeout_cancelled) {
257  rv = pthread_cond_timedwait (&pm->timeout_cancel_cv,
258  &pm->timeout_lock, &ts);
259  if (rv == ETIMEDOUT && !timeout_thread_cancelled) {
260  ep = vl_msg_api_alloc (sizeof (*ep));
261  ep->_vl_msg_id = ntohs(VL_API_MEMCLNT_READ_TIMEOUT);
263  }
264  }
265 
266  pthread_mutex_unlock(&pm->timeout_lock);
267  }
268  pthread_exit(0);
269 }
270 
271 void
273 {
274  api_main_t *am = &api_main;
275  vac_main_t *pm = &vac_main;
277 
278  if (!pm->rx_thread_handle) return;
279  pthread_mutex_lock(&pm->queue_lock);
280  if (rx_is_running)
281  {
282  ep = vl_msg_api_alloc (sizeof (*ep));
283  ep->_vl_msg_id = ntohs(VL_API_MEMCLNT_RX_THREAD_SUSPEND);
285  /* Wait for RX thread to tell us it has suspended */
286  pthread_cond_wait(&pm->suspend_cv, &pm->queue_lock);
287  rx_is_running = false;
288  }
289  pthread_mutex_unlock(&pm->queue_lock);
290 }
291 
292 void
294 {
295  vac_main_t *pm = &vac_main;
296  if (!pm->rx_thread_handle) return;
297  pthread_mutex_lock(&pm->queue_lock);
298  if (rx_is_running) goto unlock;
299  pthread_cond_signal(&pm->resume_cv);
300  rx_is_running = true;
301  unlock:
302  pthread_mutex_unlock(&pm->queue_lock);
303 }
304 
305 static uword *
307 {
308  api_main_t *am = &api_main;
309  return (am->msg_index_by_name_and_crc);
310 }
311 
312 int
314 {
315  api_main_t *am = &api_main;
317 }
318 
319 int
320 vac_connect (char * name, char * chroot_prefix, vac_callback_t cb,
321  int rx_qlen)
322 {
323  rx_thread_done = false;
324  int rv = 0;
325  vac_main_t *pm = &vac_main;
326 
327  init();
328  if (chroot_prefix != NULL)
329  vl_set_memory_root_path (chroot_prefix);
330 
331  if ((rv = vl_client_api_map("/vpe-api"))) {
332  clib_warning ("vl_client_api_map returned %d", rv);
333  return rv;
334  }
335 
336  if (vl_client_connect(name, 0, rx_qlen) < 0) {
338  return (-1);
339  }
340 
341  if (cb) {
342  /* Start the rx queue thread */
343  rv = pthread_create(&pm->rx_thread_handle, NULL, vac_rx_thread_fn, 0);
344  if (rv) {
345  clib_warning("pthread_create returned %d", rv);
347  return (-1);
348  }
349  vac_callback = cb;
350  rx_is_running = true;
351  }
352 
353  /* Start read timeout thread */
354  rv = pthread_create(&pm->timeout_thread_handle, NULL,
356  if (rv) {
357  clib_warning("pthread_create returned %d", rv);
359  return (-1);
360  }
361 
362  pm->connected_to_vlib = 1;
363 
364  return (0);
365 }
366 static void
367 set_timeout (unsigned short timeout)
368 {
369  vac_main_t *pm = &vac_main;
370  pthread_mutex_lock(&pm->timeout_lock);
371  read_timeout = timeout;
372  timeout_in_progress = true;
373  timeout_cancelled = false;
374  pthread_cond_signal(&pm->timeout_cv);
375  pthread_mutex_unlock(&pm->timeout_lock);
376 }
377 
378 static void
380 {
381  vac_main_t *pm = &vac_main;
382  pthread_mutex_lock(&pm->timeout_lock);
383  timeout_in_progress = false;
384  timeout_cancelled = true;
385  pthread_cond_signal(&pm->timeout_cancel_cv);
386  pthread_mutex_unlock(&pm->timeout_lock);
387 }
388 
389 int
391 {
392  api_main_t *am = &api_main;
393  vac_main_t *pm = &vac_main;
394  uword junk;
395  int rv = 0;
396 
397  if (!pm->connected_to_vlib) return 0;
398 
399  if (pm->rx_thread_handle) {
401  ep = vl_msg_api_alloc (sizeof (*ep));
402  ep->_vl_msg_id = ntohs(VL_API_RX_THREAD_EXIT);
404 
405  /* wait (with timeout) until RX thread has finished */
406  struct timespec ts;
407  struct timeval tv;
408  gettimeofday(&tv, NULL);
409  ts.tv_sec = tv.tv_sec + 5;
410  ts.tv_nsec = 0;
411 
412  pthread_mutex_lock(&pm->queue_lock);
413  if (rx_thread_done == false)
414  rv = pthread_cond_timedwait(&pm->terminate_cv, &pm->queue_lock, &ts);
415  pthread_mutex_unlock(&pm->queue_lock);
416 
417  /* now join so we wait until thread has -really- finished */
418  if (rv == ETIMEDOUT)
419  pthread_cancel(pm->rx_thread_handle);
420  else
421  pthread_join(pm->rx_thread_handle, (void **) &junk);
422  }
423  if (pm->timeout_thread_handle) {
424  /* cancel, wake then join the timeout thread */
425  pm->timeout_loop = 0;
426  timeout_thread_cancelled = true;
427  set_timeout(0);
428  pthread_join(pm->timeout_thread_handle, (void **) &junk);
429  }
430 
433  vac_callback = 0;
434 
435  cleanup();
436 
437  return (0);
438 }
439 
440 int
441 vac_read (char **p, int *l, u16 timeout)
442 {
443  svm_queue_t *q;
444  api_main_t *am = &api_main;
445  vac_main_t *pm = &vac_main;
447  vl_api_memclnt_keepalive_reply_t *rmp;
448  uword msg;
449  msgbuf_t *msgbuf;
450  int rv;
451  vl_shmem_hdr_t *shmem_hdr;
452 
453  /* svm_queue_sub(below) returns {-1, -2} */
454  if (!pm->connected_to_vlib) return -3;
455 
456  *l = 0;
457 
458  /* svm_queue_sub(below) returns {-1, -2} */
459  if (am->our_pid == 0) return (-4);
460 
461  /* Poke timeout thread */
462  if (timeout)
463  set_timeout(timeout);
464 
465  q = am->vl_input_queue;
466 
467  again:
468  rv = svm_queue_sub(q, (u8 *)&msg, SVM_Q_WAIT, 0);
469 
470  if (rv == 0) {
471  u16 msg_id = ntohs(*((u16 *)msg));
472  switch (msg_id) {
473  case VL_API_RX_THREAD_EXIT:
474  vl_msg_api_free((void *) msg);
475  goto error;
476  case VL_API_MEMCLNT_RX_THREAD_SUSPEND:
477  goto error;
478  case VL_API_MEMCLNT_READ_TIMEOUT:
479  goto error;
480  case VL_API_MEMCLNT_KEEPALIVE:
481  /* Handle an alive-check ping from vpp. */
482  mp = (void *)msg;
483  rmp = vl_msg_api_alloc (sizeof (*rmp));
484  clib_memset (rmp, 0, sizeof (*rmp));
485  rmp->_vl_msg_id = ntohs(VL_API_MEMCLNT_KEEPALIVE_REPLY);
486  rmp->context = mp->context;
487  shmem_hdr = am->shmem_hdr;
488  vl_msg_api_send_shmem(shmem_hdr->vl_input_queue, (u8 *)&rmp);
489  vl_msg_api_free((void *) msg);
490  /*
491  * Python code is blissfully unaware of these pings, so
492  * act as if it never happened...
493  */
494  goto again;
495 
496  default:
497  msgbuf = (msgbuf_t *)(((u8 *)msg) - offsetof(msgbuf_t, data));
498  *l = ntohl(msgbuf->data_len);
499  if (*l == 0) {
500  fprintf(stderr, "Unregistered API message: %d\n", msg_id);
501  goto error;
502  }
503  }
504  *p = (char *)msg;
505 
506 
507  } else {
508  fprintf(stderr, "Read failed with %d\n", rv);
509  }
510  /* Let timeout notification thread know we're done */
511  if (timeout)
512  unset_timeout();
513 
514  return (rv);
515 
516  error:
517  if (timeout)
518  unset_timeout();
519  vl_msg_api_free((void *) msg);
520  /* Client might forget to resume RX thread on failure */
521  vac_rx_resume ();
522  return -1;
523 }
524 
525 /*
526  * XXX: Makes the assumption that client_index is the first member
527  */
528 typedef VL_API_PACKED(struct _vl_api_header {
529  u16 _vl_msg_id;
530  u32 client_index;
531 }) vl_api_header_t;
532 
533 static u32
534 vac_client_index (void)
535 {
536  return (api_main.my_client_index);
537 }
538 
539 int
540 vac_write (char *p, int l)
541 {
542  int rv = -1;
543  api_main_t *am = &api_main;
544  vl_api_header_t *mp = vl_msg_api_alloc(l);
545  svm_queue_t *q;
546  vac_main_t *pm = &vac_main;
547 
548  if (!pm->connected_to_vlib) return -1;
549  if (!mp) return (-1);
550 
551  memcpy(mp, p, l);
552  mp->client_index = vac_client_index();
553  q = am->shmem_hdr->vl_input_queue;
554  rv = svm_queue_add(q, (u8 *)&mp, 0);
555  if (rv != 0) {
556  fprintf(stderr, "vpe_api_write fails: %d\n", rv);
557  /* Clear message */
558  vac_free(mp);
559  }
560  return (rv);
561 }
562 
563 int
564 vac_get_msg_index (unsigned char * name)
565 {
566  return vl_msg_api_get_msg_index (name);
567 }
568 
569 int
571 {
572  int max = 0;
573  hash_pair_t *hp;
575  hash_foreach_pair (hp, h,
576  ({
577  if (hp->value[0] > max)
578  max = hp->value[0];
579  }));
580 
581  return max;
582 }
583 
584 void
586 {
587  if (cb) clib_error_register_handler (cb, 0);
588 }
bool timeout_thread_cancelled
Definition: client.c:79
void vac_rx_resume(void)
Definition: client.c:293
int vac_get_msg_index(unsigned char *name)
Definition: client.c:564
int svm_queue_add(svm_queue_t *q, u8 *elem, int nowait)
Definition: queue.c:247
pthread_t timeout_thread_handle
Definition: client.c:64
int my_client_index
All VLIB-side message handlers use my_client_index to identify the queue / client.
Definition: api_common.h:311
clib_memset(h->entries, 0, sizeof(h->entries[0]) *entries)
static void set_timeout(unsigned short timeout)
Definition: client.c:367
static mheap_t * mheap_header(u8 *v)
pthread_t rx_thread_handle
Definition: client.c:63
int vac_msg_table_max_index(void)
Definition: client.c:570
void vlib_cli_output(struct vlib_main_t *vm, char *fmt,...)
Definition: client.c:147
u8 * format(u8 *s, const char *fmt,...)
Definition: format.c:424
int vac_write(char *p, int l)
Definition: client.c:540
u8 data[128]
Definition: ipsec.api:251
void clib_error_register_handler(clib_error_handler_func_t func, void *arg)
Definition: error.c:75
#define MHEAP_FLAG_THREAD_SAFE
svm_queue_t * vl_input_queue
Peer input queue pointer.
Definition: api_common.h:305
void * vl_msg_api_alloc(int nbytes)
bool rx_is_running
Definition: client.c:78
unsigned char u8
Definition: types.h:56
uword value[0]
Definition: hash.h:165
void vl_client_api_unmap(void)
int our_pid
Current process PID.
Definition: api_common.h:253
u8 connected_to_vlib
Definition: client.c:62
pthread_cond_t timeout_cancel_cv
Definition: client.c:71
int svm_queue_sub(svm_queue_t *q, u8 *elem, svm_q_conditional_wait_t cond, u32 time)
Definition: queue.c:356
pthread_mutex_t queue_lock
Definition: client.c:65
static void vac_client_constructor(void)
Definition: client.c:86
vlib_main_t ** vlib_mains
Definition: client.c:59
bool rx_thread_done
Definition: client.c:38
bool timeout_cancelled
Definition: client.c:36
struct vl_shmem_hdr_ * shmem_hdr
Binary API shared-memory segment header pointer.
Definition: api_common.h:266
unsigned int u32
Definition: types.h:88
void vac_rx_suspend(void)
Definition: client.c:272
void vl_msg_api_send_shmem(svm_queue_t *q, u8 *elem)
bool timeout_in_progress
Definition: client.c:37
vac_callback_t vac_callback
Definition: client.c:76
static void unset_timeout(void)
Definition: client.c:379
vlib_main_t vlib_global_main
Definition: client.c:58
u8 * format_mheap(u8 *s, va_list *va)
Definition: mem_dlmalloc.c:354
int vac_connect(char *name, char *chroot_prefix, vac_callback_t cb, int rx_qlen)
Definition: client.c:320
void vl_set_memory_root_path(const char *name)
int vac_disconnect(void)
Definition: client.c:390
void(* vac_error_callback_t)(void *, unsigned char *, int)
Definition: vppapiclient.h:21
u8 timeout_loop
Definition: client.c:69
int vl_client_connect(const char *name, int ctx_quota, int input_queue_size)
bool mem_trace
Definition: client.c:82
static void vac_client_destructor(void)
Definition: client.c:106
unsigned short u16
Definition: types.h:57
static void * clib_mem_get_per_cpu_heap(void)
Definition: mem.h:85
static void cleanup(void)
Definition: client.c:131
vac_main_t vac_main
Definition: client.c:75
static void * vac_timeout_thread_fn(void *arg)
Definition: client.c:235
u8 name[64]
Definition: memclnt.api:152
u16 read_timeout
Definition: client.c:77
word fformat(FILE *f, char *fmt,...)
Definition: format.c:462
API main structure, used by both vpp and binary API clients.
Definition: api_common.h:203
void * clib_mem_init(void *heap, uword size)
Definition: mem_dlmalloc.c:205
vlib_main_t * vm
Definition: buffer.c:323
#define clib_warning(format, args...)
Definition: error.h:59
u32 vl_msg_api_get_msg_index(u8 *name_and_crc)
Definition: api_shared.c:1029
int vl_client_api_map(const char *region_name)
pthread_cond_t suspend_cv
Definition: client.c:66
static uword * vac_msg_table_get_hash(void)
Definition: client.c:306
blocking call - best used in combination with condvars, for eventfds we don&#39;t yield the cpu ...
Definition: queue.h:42
svm_queue_t * vl_input_queue
Definition: memory_shared.h:84
static void * clib_mem_get_heap(void)
Definition: mem.h:284
static uword hash_elts(void *v)
Definition: hash.h:118
#define ASSERT(truth)
u32 data_len
message length not including header
Definition: api_common.h:140
Message header structure.
Definition: api_common.h:137
int vac_read(char **p, int *l, u16 timeout)
Definition: client.c:441
static void init(void)
Definition: client.c:116
pthread_cond_t terminate_cv
Definition: client.c:72
void vl_msg_api_free(void *)
pthread_cond_t timeout_cv
Definition: client.c:70
pthread_cond_t resume_cv
Definition: client.c:67
#define hash_foreach_pair(p, v, body)
Iterate over hash pairs.
Definition: hash.h:373
int vl_client_disconnect(void)
u64 uword
Definition: types.h:112
void vac_set_error_handler(vac_error_callback_t cb)
Definition: client.c:585
int vac_msg_table_size(void)
Definition: client.c:313
struct _svm_queue svm_queue_t
void(* vac_callback_t)(unsigned char *data, int len)
Definition: vppapiclient.h:20
pthread_mutex_t timeout_lock
Definition: client.c:68
void vac_free(void *msg)
Definition: client.c:153
static void vac_api_handler(void *msg)
Definition: client.c:159
void clib_mem_trace(int enable)
Definition: mem_dlmalloc.c:420
uword * msg_index_by_name_and_crc
client message index hash table
Definition: api_common.h:323
api_main_t api_main
Definition: api_shared.c:35
typedef VL_API_PACKED(struct _vl_api_header { u16 _vl_msg_id;u32 client_index;})
Definition: client.c:528
static void * vac_rx_thread_fn(void *arg)
Definition: client.c:174