src/vpp-api/pneum/pneum.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 "pneum.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  *  pneum_write() -> suspends RX thread
  43  *  pneum_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 } pneum_main_t;
  69
  70 pneum_main_t pneum_main;
  71 pneum_callback_t pneum_callback;
  72 u16 read_timeout = 0;
  73 bool rx_is_running = false;
  74
  75 static void
  76 init (void)
  77 {
  78   pneum_main_t *pm = &pneum_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   pneum_main_t *pm = &pneum_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 pneum_free (void * msg)
 113 {
 114   vl_msg_api_free (msg);
 115 }
 116
 117 static void
 118 pneum_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(pneum_callback);
 128   (pneum_callback)(msg, l);
 129   pneum_free(msg);
 130 }
 131
 132 static void *
 133 pneum_rx_thread_fn (void *arg)
 134 {
 135   unix_shared_memory_queue_t *q;
 136   pneum_main_t *pm = &pneum_main;
 137   api_main_t *am = &api_main;
 138   uword msg;
 139
 140   q = am->vl_input_queue;
 141
 142   while (1)
 143     while (!unix_shared_memory_queue_sub(q, (u8 *)&msg, 0))
 144       {
 145         u16 id = ntohs(*((u16 *)msg));
 146         switch (id) {
 147         case VL_API_RX_THREAD_EXIT:
 148           vl_msg_api_free((void *) msg);
 149           /* signal waiting threads that this thread is about to terminate */
 150           pthread_mutex_lock(&pm->queue_lock);
 151           pthread_cond_signal(&pm->terminate_cv);
 152           pthread_mutex_unlock(&pm->queue_lock);
 153           pthread_exit(0);
 154           return 0;
 155           break;
 156
 157         case VL_API_MEMCLNT_RX_THREAD_SUSPEND:
 158           vl_msg_api_free((void * )msg);
 159           /* Suspend thread and signal reader */
 160           pthread_mutex_lock(&pm->queue_lock);
 161           pthread_cond_signal(&pm->suspend_cv);
 162           /* Wait for the resume signal */
 163           pthread_cond_wait (&pm->resume_cv, &pm->queue_lock);
 164           pthread_mutex_unlock(&pm->queue_lock);
 165           break;
 166
 167         case VL_API_MEMCLNT_READ_TIMEOUT:
 168           clib_warning("Received read timeout in async thread\n");
 169           vl_msg_api_free((void *) msg);
 170           break;
 171
 172         default:
 173           pneum_api_handler((void *)msg);
 174         }
 175       }
 176 }
 177
 178 static void *
 179 pneum_timeout_thread_fn (void *arg)
 180 {
 181   vl_api_memclnt_read_timeout_t *ep;
 182   pneum_main_t *pm = &pneum_main;
 183   api_main_t *am = &api_main;
 184   struct timespec ts;
 185   struct timeval tv;
 186   u16 timeout;
 187   int rv;
 188
 189   while (1)
 190     {
 191       /* Wait for poke */
 192       pthread_mutex_lock(&pm->timeout_lock);
 193       pthread_cond_wait (&pm->timeout_cv, &pm->timeout_lock);
 194       timeout = read_timeout;
 195       gettimeofday(&tv, NULL);
 196       ts.tv_sec = tv.tv_sec + timeout;
 197       ts.tv_nsec = 0;
 198       rv = pthread_cond_timedwait (&pm->timeout_cancel_cv,
 199                                    &pm->timeout_lock, &ts);
 200       pthread_mutex_unlock(&pm->timeout_lock);
 201       if (rv == ETIMEDOUT)
 202         {
 203           ep = vl_msg_api_alloc (sizeof (*ep));
 204           ep->_vl_msg_id = ntohs(VL_API_MEMCLNT_READ_TIMEOUT);
 205           vl_msg_api_send_shmem(am->vl_input_queue, (u8 *)&ep);
 206         }
 207     }
 208   pthread_exit(0);
 209 }
 210
 211 void
 212 pneum_rx_suspend (void)
 213 {
 214   api_main_t *am = &api_main;
 215   pneum_main_t *pm = &pneum_main;
 216   vl_api_memclnt_rx_thread_suspend_t *ep;
 217
 218   if (!pm->rx_thread_handle) return;
 219   pthread_mutex_lock(&pm->queue_lock);
 220   if (rx_is_running)
 221     {
 222       ep = vl_msg_api_alloc (sizeof (*ep));
 223       ep->_vl_msg_id = ntohs(VL_API_MEMCLNT_RX_THREAD_SUSPEND);
 224       vl_msg_api_send_shmem(am->vl_input_queue, (u8 *)&ep);
 225       /* Wait for RX thread to tell us it has suspendend */
 226       pthread_cond_wait(&pm->suspend_cv, &pm->queue_lock);
 227       rx_is_running = false;
 228     }
 229   pthread_mutex_unlock(&pm->queue_lock);
 230 }
 231
 232 void
 233 pneum_rx_resume (void)
 234 {
 235   pneum_main_t *pm = &pneum_main;
 236   if (!pm->rx_thread_handle) return;
 237   pthread_mutex_lock(&pm->queue_lock);
 238   if (rx_is_running) return;
 239   pthread_cond_signal(&pm->resume_cv);
 240   rx_is_running = true;
 241   pthread_mutex_unlock(&pm->queue_lock);
 242 }
 243
 244 static uword *
 245 pneum_msg_table_get_hash (void)
 246 {
 247   api_main_t *am = &api_main;
 248   return (am->msg_index_by_name_and_crc);
 249 }
 250
 251 int
 252 pneum_msg_table_size(void)
 253 {
 254   api_main_t *am = &api_main;
 255   return hash_elts(am->msg_index_by_name_and_crc);
 256 }
 257
 258 int
 259 pneum_connect (char * name, char * chroot_prefix, pneum_callback_t cb,
 260                int rx_qlen)
 261 {
 262   int rv = 0;
 263   pneum_main_t *pm = &pneum_main;
 264
 265   init();
 266   if (chroot_prefix != NULL)
 267     vl_set_memory_root_path (chroot_prefix);
 268
 269   if ((rv = vl_client_api_map("/vpe-api"))) {
 270     clib_warning ("vl_client_api map rv %d", rv);
 271     return rv;
 272   }
 273
 274   if (vl_client_connect(name, 0, rx_qlen) < 0) {
 275     vl_client_api_unmap();
 276     return (-1);
 277   }
 278
 279   if (cb) {
 280     /* Start the rx queue thread */
 281     rv = pthread_create(&pm->rx_thread_handle, NULL, pneum_rx_thread_fn, 0);
 282     if (rv) {
 283       clib_warning("pthread_create returned %d", rv);
 284       vl_client_api_unmap();
 285       return (-1);
 286     }
 287     pneum_callback = cb;
 288     rx_is_running = true;
 289   }
 290
 291   /* Start read timeout thread */
 292   rv = pthread_create(&pm->timeout_thread_handle, NULL,
 293                       pneum_timeout_thread_fn, 0);
 294   if (rv) {
 295     clib_warning("pthread_create returned %d", rv);
 296     vl_client_api_unmap();
 297     return (-1);
 298   }
 299
 300   pm->connected_to_vlib = 1;
 301
 302   return (0);
 303 }
 304
 305 int
 306 pneum_disconnect (void)
 307 {
 308   api_main_t *am = &api_main;
 309   pneum_main_t *pm = &pneum_main;
 310
 311   if (!pm->connected_to_vlib) return 0;
 312
 313   if (pm->rx_thread_handle) {
 314     vl_api_rx_thread_exit_t *ep;
 315     uword junk;
 316     ep = vl_msg_api_alloc (sizeof (*ep));
 317     ep->_vl_msg_id = ntohs(VL_API_RX_THREAD_EXIT);
 318     vl_msg_api_send_shmem(am->vl_input_queue, (u8 *)&ep);
 319
 320     /* wait (with timeout) until RX thread has finished */
 321     struct timespec ts;
 322     struct timeval tv;
 323     gettimeofday(&tv, NULL);
 324     ts.tv_sec = tv.tv_sec + 5;
 325     ts.tv_nsec = 0;
 326     pthread_mutex_lock(&pm->queue_lock);
 327     int rv = pthread_cond_timedwait(&pm->terminate_cv, &pm->queue_lock, &ts);
 328     pthread_mutex_unlock(&pm->queue_lock);
 329     /* now join so we wait until thread has -really- finished */
 330     if (rv == ETIMEDOUT)
 331       pthread_cancel(pm->rx_thread_handle);
 332     else
 333       pthread_join(pm->rx_thread_handle, (void **) &junk);
 334   }
 335   if (pm->timeout_thread_handle)
 336     pthread_cancel(pm->timeout_thread_handle);
 337
 338   vl_client_disconnect();
 339   vl_client_api_unmap();
 340   pneum_callback = 0;
 341
 342   cleanup();
 343
 344   return (0);
 345 }
 346
 347 static void
 348 set_timeout (unsigned short timeout)
 349 {
 350   pneum_main_t *pm = &pneum_main;
 351   pthread_mutex_lock(&pm->timeout_lock);
 352   read_timeout = timeout;
 353   pthread_cond_signal(&pm->timeout_cv);
 354   pthread_mutex_unlock(&pm->timeout_lock);
 355 }
 356
 357 static void
 358 unset_timeout (void)
 359 {
 360   pneum_main_t *pm = &pneum_main;
 361   pthread_mutex_lock(&pm->timeout_lock);
 362   pthread_cond_signal(&pm->timeout_cancel_cv);
 363   pthread_mutex_unlock(&pm->timeout_lock);
 364 }
 365
 366 int
 367 pneum_read (char **p, int *l, u16 timeout)
 368 {
 369   unix_shared_memory_queue_t *q;
 370   api_main_t *am = &api_main;
 371   pneum_main_t *pm = &pneum_main;
 372   uword msg;
 373   msgbuf_t *msgbuf;
 374
 375   if (!pm->connected_to_vlib) return -1;
 376
 377   *l = 0;
 378
 379   if (am->our_pid == 0) return (-1);
 380
 381   /* Poke timeout thread */
 382   if (timeout)
 383     set_timeout(timeout);
 384
 385   q = am->vl_input_queue;
 386   int rv = unix_shared_memory_queue_sub(q, (u8 *)&msg, 0);
 387   if (rv == 0) {
 388     u16 msg_id = ntohs(*((u16 *)msg));
 389     switch (msg_id) {
 390     case VL_API_RX_THREAD_EXIT:
 391       printf("Received thread exit\n");
 392       return -1;
 393     case VL_API_MEMCLNT_RX_THREAD_SUSPEND:
 394       printf("Received thread suspend\n");
 395       goto error;
 396     case VL_API_MEMCLNT_READ_TIMEOUT:
 397       printf("Received read timeout %ds\n", timeout);
 398       goto error;
 399
 400     default:
 401       msgbuf = (msgbuf_t *)(((u8 *)msg) - offsetof(msgbuf_t, data));
 402       *l = ntohl(msgbuf->data_len);
 403       if (*l == 0) {
 404         printf("Unregistered API message: %d\n", msg_id);
 405         goto error;
 406       }
 407     }
 408     *p = (char *)msg;
 409
 410     /* Let timeout notification thread know we're done */
 411     unset_timeout();
 412
 413   } else {
 414     printf("Read failed with %d\n", rv);
 415   }
 416   return (rv);
 417
 418  error:
 419   vl_msg_api_free((void *) msg);
 420   /* Client might forget to resume RX thread on failure */
 421   pneum_rx_resume ();
 422   return -1;
 423 }
 424
 425 /*
 426  * XXX: Makes the assumption that client_index is the first member
 427  */
 428 typedef VL_API_PACKED(struct _vl_api_header {
 429   u16 _vl_msg_id;
 430   u32 client_index;
 431 }) vl_api_header_t;
 432
 433 static unsigned int
 434 pneum_client_index (void)
 435 {
 436   return (api_main.my_client_index);
 437 }
 438
 439 int
 440 pneum_write (char *p, int l)
 441 {
 442   int rv = -1;
 443   api_main_t *am = &api_main;
 444   vl_api_header_t *mp = vl_msg_api_alloc(l);
 445   unix_shared_memory_queue_t *q;
 446   pneum_main_t *pm = &pneum_main;
 447
 448   if (!pm->connected_to_vlib) return -1;
 449   if (!mp) return (-1);
 450
 451   memcpy(mp, p, l);
 452   mp->client_index = pneum_client_index();
 453   q = am->shmem_hdr->vl_input_queue;
 454   rv = unix_shared_memory_queue_add(q, (u8 *)&mp, 0);
 455   if (rv != 0) {
 456     clib_warning("vpe_api_write fails: %d\n", rv);
 457     /* Clear message */
 458     pneum_free(mp);
 459   }
 460   return (rv);
 461 }
 462
 463 int
 464 pneum_get_msg_index (unsigned char * name)
 465 {
 466   return vl_api_get_msg_index (name);
 467 }
 468
 469 int
 470 pneum_msg_table_max_index(void)
 471 {
 472   int max = 0;
 473   hash_pair_t *hp;
 474   uword *h = pneum_msg_table_get_hash();
 475   hash_foreach_pair (hp, h,
 476   ({
 477     if (hp->value[0] > max)
 478       max = hp->value[0];
 479   }));
 480
 481   return max;
 482 }
 483
 484 void
 485 pneum_set_error_handler (pneum_error_callback_t cb)
 486 {
 487   if (cb) clib_error_register_handler (cb, 0);
 488 }