src/vpp-api/client/client.c

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