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 bool timeout_cancelled;
  37 bool timeout_in_progress;
  38 bool rx_thread_done;
  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
  58 vlib_main_t vlib_global_main;
  59 vlib_main_t **vlib_mains;
  60
  61 typedef struct {
  62   u8 connected_to_vlib;
  63   pthread_t rx_thread_handle;
  64   pthread_t timeout_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;
  69   u8 timeout_loop;
  70   pthread_cond_t timeout_cv;
  71   pthread_cond_t timeout_cancel_cv;
  72   pthread_cond_t terminate_cv;
  73 } vac_main_t;
  74
  75 vac_main_t vac_main;
  76 vac_callback_t vac_callback;
  77 u16 read_timeout = 0;
  78 bool rx_is_running = false;
  79 bool timeout_thread_cancelled = false;
  80
  81 /* Set to true to enable memory tracing */
  82 bool mem_trace = false;
  83
  84 __attribute__((constructor))
  85 static void
  86 vac_client_constructor (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
  95 vac_client_destructor (void)
  96 {
  97   if (mem_trace)
  98     fformat(stderr, "TRACE: %s",
  99             format (0, "%U\n",
 100                     format_mheap, clib_mem_get_heap (), 1));
 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;
 166   vl_api_memclnt_keepalive_t *mp;
 167   vl_api_memclnt_keepalive_reply_t *rmp;
 168   vac_main_t *pm = &vac_main;
 169   api_main_t *am = vlibapi_get_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 *
 225 vac_timeout_thread_fn (void *arg)
 226 {
 227   vl_api_memclnt_read_timeout_t *ep;
 228   vac_main_t *pm = &vac_main;
 229   api_main_t *am = vlibapi_get_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);
 252           vl_msg_api_send_shmem(am->vl_input_queue, (u8 *)&ep);
 253         }
 254       }
 255
 256       pthread_mutex_unlock(&pm->timeout_lock);
 257     }
 258   pthread_exit(0);
 259 }
 260
 261 void
 262 vac_rx_suspend (void)
 263 {
 264   api_main_t *am = vlibapi_get_main();
 265   vac_main_t *pm = &vac_main;
 266   vl_api_memclnt_rx_thread_suspend_t *ep;
 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);
 274       vl_msg_api_send_shmem(am->vl_input_queue, (u8 *)&ep);
 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
 283 vac_rx_resume (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 *
 296 vac_msg_table_get_hash (void)
 297 {
 298   api_main_t *am = vlibapi_get_main();
 299   return (am->msg_index_by_name_and_crc);
 300 }
 301
 302 int
 303 vac_msg_table_size(void)
 304 {
 305   api_main_t *am = vlibapi_get_main();
 306   return hash_elts(am->msg_index_by_name_and_crc);
 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) {
 327     vl_client_api_unmap();
 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);
 336       vl_client_api_unmap();
 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,
 345                       vac_timeout_thread_fn, 0);
 346   if (rv) {
 347     clib_warning("pthread_create returned %d", rv);
 348     vl_client_api_unmap();
 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
 369 unset_timeout (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
 380 vac_disconnect (void)
 381 {
 382   api_main_t *am = vlibapi_get_main();
 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) {
 390     vl_api_rx_thread_exit_t *ep;
 391     ep = vl_msg_api_alloc (sizeof (*ep));
 392     ep->_vl_msg_id = ntohs(VL_API_RX_THREAD_EXIT);
 393     vl_msg_api_send_shmem(am->vl_input_queue, (u8 *)&ep);
 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
 421   vl_client_disconnect();
 422   vl_client_api_unmap();
 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;
 434   api_main_t *am = vlibapi_get_main();
 435   vac_main_t *pm = &vac_main;
 436   vl_api_memclnt_keepalive_t *mp;
 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;
 534   api_main_t *am = vlibapi_get_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
 561 vac_msg_table_max_index(void)
 562 {
 563   int max = 0;
 564   hash_pair_t *hp;
 565   uword *h = vac_msg_table_get_hash();
 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
 576 vac_set_error_handler (vac_error_callback_t cb)
 577 {
 578   if (cb) clib_error_register_handler (cb, 0);
 579 }