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