FD.io VPP  v20.05-21-gb1500e9ff
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 (mem_trace)
90  clib_mem_trace (1);
91 }
92 
93 __attribute__((destructor))
94 static void
96 {
97  if (mem_trace)
98  fformat(stderr, "TRACE: %s",
99  format (0, "%U\n",
101 }
102 
103 
104 static void
105 init (void)
106 {
107  vac_main_t *pm = &vac_main;
108  clib_memset(pm, 0, sizeof(*pm));
109  pthread_mutex_init(&pm->queue_lock, NULL);
110  pthread_cond_init(&pm->suspend_cv, NULL);
111  pthread_cond_init(&pm->resume_cv, NULL);
112  pthread_mutex_init(&pm->timeout_lock, NULL);
113  pm->timeout_loop = 1;
114  pthread_cond_init(&pm->timeout_cv, NULL);
115  pthread_cond_init(&pm->timeout_cancel_cv, NULL);
116  pthread_cond_init(&pm->terminate_cv, NULL);
117 }
118 
119 static void
120 cleanup (void)
121 {
122  vac_main_t *pm = &vac_main;
123  pthread_mutex_destroy(&pm->queue_lock);
124  pthread_cond_destroy(&pm->suspend_cv);
125  pthread_cond_destroy(&pm->resume_cv);
126  pthread_mutex_destroy(&pm->timeout_lock);
127  pthread_cond_destroy(&pm->timeout_cv);
128  pthread_cond_destroy(&pm->timeout_cancel_cv);
129  pthread_cond_destroy(&pm->terminate_cv);
130  clib_memset(pm, 0, sizeof(*pm));
131 }
132 
133 /*
134  * Satisfy external references when -lvlib is not available.
135  */
136 void vlib_cli_output (struct vlib_main_t * vm, char * fmt, ...)
137 {
138  clib_warning ("vlib_cli_output called...");
139 }
140 
141 void
142 vac_free (void * msg)
143 {
144  vl_msg_api_free (msg);
145 }
146 
147 static void
148 vac_api_handler (void *msg)
149 {
150  u16 id = ntohs(*((u16 *)msg));
151  msgbuf_t *msgbuf = (msgbuf_t *)(((u8 *)msg) - offsetof(msgbuf_t, data));
152  int l = ntohl(msgbuf->data_len);
153  if (l == 0)
154  clib_warning("Message ID %d has wrong length: %d\n", id, l);
155 
156  /* Call Python callback */
157  ASSERT(vac_callback);
158  (vac_callback)(msg, l);
159  vac_free(msg);
160 }
161 
162 static void *
163 vac_rx_thread_fn (void *arg)
164 {
165  svm_queue_t *q;
167  vl_api_memclnt_keepalive_reply_t *rmp;
168  vac_main_t *pm = &vac_main;
170  vl_shmem_hdr_t *shmem_hdr;
171  uword msg;
172 
173  q = am->vl_input_queue;
174 
175  while (1)
176  while (!svm_queue_sub(q, (u8 *)&msg, SVM_Q_WAIT, 0))
177  {
178  VL_MSG_API_UNPOISON((void *)msg);
179  u16 id = ntohs(*((u16 *)msg));
180  switch (id) {
181  case VL_API_RX_THREAD_EXIT:
182  vl_msg_api_free((void *) msg);
183  /* signal waiting threads that this thread is about to terminate */
184  pthread_mutex_lock(&pm->queue_lock);
185  rx_thread_done = true;
186  pthread_cond_signal(&pm->terminate_cv);
187  pthread_mutex_unlock(&pm->queue_lock);
188  pthread_exit(0);
189  return 0;
190  break;
191 
192  case VL_API_MEMCLNT_RX_THREAD_SUSPEND:
193  vl_msg_api_free((void * )msg);
194  /* Suspend thread and signal reader */
195  pthread_mutex_lock(&pm->queue_lock);
196  pthread_cond_signal(&pm->suspend_cv);
197  /* Wait for the resume signal */
198  pthread_cond_wait (&pm->resume_cv, &pm->queue_lock);
199  pthread_mutex_unlock(&pm->queue_lock);
200  break;
201 
202  case VL_API_MEMCLNT_READ_TIMEOUT:
203  clib_warning("Received read timeout in async thread\n");
204  vl_msg_api_free((void *) msg);
205  break;
206 
207  case VL_API_MEMCLNT_KEEPALIVE:
208  mp = (void *)msg;
209  rmp = vl_msg_api_alloc (sizeof (*rmp));
210  clib_memset (rmp, 0, sizeof (*rmp));
211  rmp->_vl_msg_id = ntohs(VL_API_MEMCLNT_KEEPALIVE_REPLY);
212  rmp->context = mp->context;
213  shmem_hdr = am->shmem_hdr;
214  vl_msg_api_send_shmem(shmem_hdr->vl_input_queue, (u8 *)&rmp);
215  vl_msg_api_free((void *) msg);
216  break;
217 
218  default:
219  vac_api_handler((void *)msg);
220  }
221  }
222 }
223 
224 static void *
226 {
228  vac_main_t *pm = &vac_main;
230  struct timespec ts;
231  struct timeval tv;
232  int rv;
233 
234  while (pm->timeout_loop)
235  {
236  /* Wait for poke */
237  pthread_mutex_lock(&pm->timeout_lock);
238  while (!timeout_in_progress)
239  pthread_cond_wait (&pm->timeout_cv, &pm->timeout_lock);
240 
241  /* Starting timer */
242  gettimeofday(&tv, NULL);
243  ts.tv_sec = tv.tv_sec + read_timeout;
244  ts.tv_nsec = 0;
245 
246  if (!timeout_cancelled) {
247  rv = pthread_cond_timedwait (&pm->timeout_cancel_cv,
248  &pm->timeout_lock, &ts);
249  if (rv == ETIMEDOUT && !timeout_thread_cancelled) {
250  ep = vl_msg_api_alloc (sizeof (*ep));
251  ep->_vl_msg_id = ntohs(VL_API_MEMCLNT_READ_TIMEOUT);
253  }
254  }
255 
256  pthread_mutex_unlock(&pm->timeout_lock);
257  }
258  pthread_exit(0);
259 }
260 
261 void
263 {
265  vac_main_t *pm = &vac_main;
267 
268  if (!pm->rx_thread_handle) return;
269  pthread_mutex_lock(&pm->queue_lock);
270  if (rx_is_running)
271  {
272  ep = vl_msg_api_alloc (sizeof (*ep));
273  ep->_vl_msg_id = ntohs(VL_API_MEMCLNT_RX_THREAD_SUSPEND);
275  /* Wait for RX thread to tell us it has suspended */
276  pthread_cond_wait(&pm->suspend_cv, &pm->queue_lock);
277  rx_is_running = false;
278  }
279  pthread_mutex_unlock(&pm->queue_lock);
280 }
281 
282 void
284 {
285  vac_main_t *pm = &vac_main;
286  if (!pm->rx_thread_handle) return;
287  pthread_mutex_lock(&pm->queue_lock);
288  if (rx_is_running) goto unlock;
289  pthread_cond_signal(&pm->resume_cv);
290  rx_is_running = true;
291  unlock:
292  pthread_mutex_unlock(&pm->queue_lock);
293 }
294 
295 static uword *
297 {
299  return (am->msg_index_by_name_and_crc);
300 }
301 
302 int
304 {
307 }
308 
309 int
310 vac_connect (char * name, char * chroot_prefix, vac_callback_t cb,
311  int rx_qlen)
312 {
313  rx_thread_done = false;
314  int rv = 0;
315  vac_main_t *pm = &vac_main;
316 
317  init();
318  if (chroot_prefix != NULL)
319  vl_set_memory_root_path (chroot_prefix);
320 
321  if ((rv = vl_client_api_map("/vpe-api"))) {
322  clib_warning ("vl_client_api_map returned %d", rv);
323  return rv;
324  }
325 
326  if (vl_client_connect(name, 0, rx_qlen) < 0) {
328  return (-1);
329  }
330 
331  if (cb) {
332  /* Start the rx queue thread */
333  rv = pthread_create(&pm->rx_thread_handle, NULL, vac_rx_thread_fn, 0);
334  if (rv) {
335  clib_warning("pthread_create returned %d", rv);
337  return (-1);
338  }
339  vac_callback = cb;
340  rx_is_running = true;
341  }
342 
343  /* Start read timeout thread */
344  rv = pthread_create(&pm->timeout_thread_handle, NULL,
346  if (rv) {
347  clib_warning("pthread_create returned %d", rv);
349  return (-1);
350  }
351 
352  pm->connected_to_vlib = 1;
353 
354  return (0);
355 }
356 static void
357 set_timeout (unsigned short timeout)
358 {
359  vac_main_t *pm = &vac_main;
360  pthread_mutex_lock(&pm->timeout_lock);
361  read_timeout = timeout;
362  timeout_in_progress = true;
363  timeout_cancelled = false;
364  pthread_cond_signal(&pm->timeout_cv);
365  pthread_mutex_unlock(&pm->timeout_lock);
366 }
367 
368 static void
370 {
371  vac_main_t *pm = &vac_main;
372  pthread_mutex_lock(&pm->timeout_lock);
373  timeout_in_progress = false;
374  timeout_cancelled = true;
375  pthread_cond_signal(&pm->timeout_cancel_cv);
376  pthread_mutex_unlock(&pm->timeout_lock);
377 }
378 
379 int
381 {
383  vac_main_t *pm = &vac_main;
384  uword junk;
385  int rv = 0;
386 
387  if (!pm->connected_to_vlib) return 0;
388 
389  if (pm->rx_thread_handle) {
391  ep = vl_msg_api_alloc (sizeof (*ep));
392  ep->_vl_msg_id = ntohs(VL_API_RX_THREAD_EXIT);
394 
395  /* wait (with timeout) until RX thread has finished */
396  struct timespec ts;
397  struct timeval tv;
398  gettimeofday(&tv, NULL);
399  ts.tv_sec = tv.tv_sec + 5;
400  ts.tv_nsec = 0;
401 
402  pthread_mutex_lock(&pm->queue_lock);
403  if (rx_thread_done == false)
404  rv = pthread_cond_timedwait(&pm->terminate_cv, &pm->queue_lock, &ts);
405  pthread_mutex_unlock(&pm->queue_lock);
406 
407  /* now join so we wait until thread has -really- finished */
408  if (rv == ETIMEDOUT)
409  pthread_cancel(pm->rx_thread_handle);
410  else
411  pthread_join(pm->rx_thread_handle, (void **) &junk);
412  }
413  if (pm->timeout_thread_handle) {
414  /* cancel, wake then join the timeout thread */
415  pm->timeout_loop = 0;
416  timeout_thread_cancelled = true;
417  set_timeout(0);
418  pthread_join(pm->timeout_thread_handle, (void **) &junk);
419  }
420 
423  vac_callback = 0;
424 
425  cleanup();
426 
427  return (0);
428 }
429 
430 int
431 vac_read (char **p, int *l, u16 timeout)
432 {
433  svm_queue_t *q;
435  vac_main_t *pm = &vac_main;
437  vl_api_memclnt_keepalive_reply_t *rmp;
438  uword msg;
439  msgbuf_t *msgbuf;
440  int rv;
441  vl_shmem_hdr_t *shmem_hdr;
442 
443  /* svm_queue_sub(below) returns {-1, -2} */
444  if (!pm->connected_to_vlib) return -3;
445 
446  *l = 0;
447 
448  /* svm_queue_sub(below) returns {-1, -2} */
449  if (am->our_pid == 0) return (-4);
450 
451  /* Poke timeout thread */
452  if (timeout)
453  set_timeout(timeout);
454 
455  q = am->vl_input_queue;
456 
457  again:
458  rv = svm_queue_sub(q, (u8 *)&msg, SVM_Q_WAIT, 0);
459 
460  if (rv == 0) {
461  VL_MSG_API_UNPOISON((void *)msg);
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  clib_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 (vlibapi_get_main()->my_client_index);
528 }
529 
530 int
531 vac_write (char *p, int l)
532 {
533  int rv = -1;
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:79
void vac_rx_resume(void)
Definition: client.c:283
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:247
pthread_t timeout_thread_handle
Definition: client.c:64
clib_memset(h->entries, 0, sizeof(h->entries[0]) *entries)
static void set_timeout(unsigned short timeout)
Definition: client.c:357
pthread_t rx_thread_handle
Definition: client.c:63
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:136
u8 * format(u8 *s, const char *fmt,...)
Definition: format.c:424
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
svm_queue_t * vl_input_queue
Peer input queue pointer.
Definition: api_common.h:328
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:276
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:289
unsigned int u32
Definition: types.h:88
void vac_rx_suspend(void)
Definition: client.c:262
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:369
vlib_main_t vlib_global_main
Definition: client.c:58
u8 * format_mheap(u8 *s, va_list *va)
Definition: mem_dlmalloc.c:388
int vac_connect(char *name, char *chroot_prefix, vac_callback_t cb, int rx_qlen)
Definition: client.c:310
void vl_set_memory_root_path(const char *name)
int vac_disconnect(void)
Definition: client.c:380
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:95
unsigned short u16
Definition: types.h:57
vec_header_t h
Definition: buffer.c:322
static void cleanup(void)
Definition: client.c:120
vac_main_t vac_main
Definition: client.c:75
static void * vac_timeout_thread_fn(void *arg)
Definition: client.c:225
vlib_main_t * vm
Definition: in2out_ed.c:1599
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:226
void * clib_mem_init(void *heap, uword size)
Definition: mem_dlmalloc.c:228
#define clib_warning(format, args...)
Definition: error.h:59
u32 vl_msg_api_get_msg_index(u8 *name_and_crc)
Definition: api_shared.c:1073
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:296
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
string name[64]
Definition: ip.api:44
static void * clib_mem_get_heap(void)
Definition: mem.h:262
static uword hash_elts(void *v)
Definition: hash.h:118
#define ASSERT(truth)
u8 data[128]
Definition: ipsec_types.api:89
u32 data_len
message length not including header
Definition: api_common.h:141
Message header structure.
Definition: api_common.h:138
int vac_read(char **p, int *l, u16 timeout)
Definition: client.c:431
static void init(void)
Definition: client.c:105
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:576
int vac_msg_table_size(void)
Definition: client.c:303
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
static api_main_t * vlibapi_get_main(void)
Definition: api_common.h:379
void vac_free(void *msg)
Definition: client.c:142
static void vac_api_handler(void *msg)
Definition: client.c:148
void clib_mem_trace(int enable)
Definition: mem_dlmalloc.c:454
uword * msg_index_by_name_and_crc
client message index hash table
Definition: api_common.h:346
typedef VL_API_PACKED(struct _vl_api_header { u16 _vl_msg_id;u32 client_index;})
Definition: client.c:519
static CLIB_NOSANITIZE_ADDR void VL_MSG_API_UNPOISON(const void *a)
Definition: api_common.h:147
static void * vac_rx_thread_fn(void *arg)
Definition: client.c:163