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