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