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