src/vcl/vppcom.c

   1 /*
   2  * Copyright (c) 2017 Cisco and/or its affiliates.
   3  * Licensed under the Apache License, Version 2.0 (the "License");
   4  * you may not use this
   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
  16 #include <stdio.h>
  17 #include <stdlib.h>
  18 #include <svm/svm_fifo_segment.h>
  19 #include <vcl/vppcom.h>
  20 #include <vcl/vcl_debug.h>
  21 #include <vcl/vcl_private.h>
  22
  23 __thread uword __vcl_worker_index = ~0;
  24
  25 static u8 not_ready;
  26
  27 void
  28 sigsegv_signal (int signum)
  29 {
  30   not_ready = 1;
  31 }
  32
  33 static void
  34 vcl_wait_for_memory (void *mem)
  35 {
  36   u8 __clib_unused test;
  37   if (vcm->mounting_segment)
  38     {
  39       while (vcm->mounting_segment)
  40         ;
  41       return;
  42     }
  43   if (1 || vcm->debug)
  44     {
  45       usleep (1e5);
  46       return;
  47     }
  48   if (signal (SIGSEGV, sigsegv_signal))
  49     {
  50       perror ("signal()");
  51       return;
  52     }
  53   not_ready = 0;
  54
  55 again:
  56   test = *(u8 *) mem;
  57   if (not_ready)
  58     {
  59       not_ready = 0;
  60       usleep (1);
  61       goto again;
  62     }
  63
  64   signal (SIGSEGV, SIG_DFL);
  65 }
  66
  67 const char *
  68 vppcom_session_state_str (session_state_t state)
  69 {
  70   char *st;
  71
  72   switch (state)
  73     {
  74     case STATE_START:
  75       st = "STATE_START";
  76       break;
  77
  78     case STATE_CONNECT:
  79       st = "STATE_CONNECT";
  80       break;
  81
  82     case STATE_LISTEN:
  83       st = "STATE_LISTEN";
  84       break;
  85
  86     case STATE_ACCEPT:
  87       st = "STATE_ACCEPT";
  88       break;
  89
  90     case STATE_CLOSE_ON_EMPTY:
  91       st = "STATE_CLOSE_ON_EMPTY";
  92       break;
  93
  94     case STATE_DISCONNECT:
  95       st = "STATE_DISCONNECT";
  96       break;
  97
  98     case STATE_FAILED:
  99       st = "STATE_FAILED";
 100       break;
 101
 102     default:
 103       st = "UNKNOWN_STATE";
 104       break;
 105     }
 106
 107   return st;
 108 }
 109
 110 u8 *
 111 format_ip4_address (u8 * s, va_list * args)
 112 {
 113   u8 *a = va_arg (*args, u8 *);
 114   return format (s, "%d.%d.%d.%d", a[0], a[1], a[2], a[3]);
 115 }
 116
 117 u8 *
 118 format_ip6_address (u8 * s, va_list * args)
 119 {
 120   ip6_address_t *a = va_arg (*args, ip6_address_t *);
 121   u32 i, i_max_n_zero, max_n_zeros, i_first_zero, n_zeros, last_double_colon;
 122
 123   i_max_n_zero = ARRAY_LEN (a->as_u16);
 124   max_n_zeros = 0;
 125   i_first_zero = i_max_n_zero;
 126   n_zeros = 0;
 127   for (i = 0; i < ARRAY_LEN (a->as_u16); i++)
 128     {
 129       u32 is_zero = a->as_u16[i] == 0;
 130       if (is_zero && i_first_zero >= ARRAY_LEN (a->as_u16))
 131         {
 132           i_first_zero = i;
 133           n_zeros = 0;
 134         }
 135       n_zeros += is_zero;
 136       if ((!is_zero && n_zeros > max_n_zeros)
 137           || (i + 1 >= ARRAY_LEN (a->as_u16) && n_zeros > max_n_zeros))
 138         {
 139           i_max_n_zero = i_first_zero;
 140           max_n_zeros = n_zeros;
 141           i_first_zero = ARRAY_LEN (a->as_u16);
 142           n_zeros = 0;
 143         }
 144     }
 145
 146   last_double_colon = 0;
 147   for (i = 0; i < ARRAY_LEN (a->as_u16); i++)
 148     {
 149       if (i == i_max_n_zero && max_n_zeros > 1)
 150         {
 151           s = format (s, "::");
 152           i += max_n_zeros - 1;
 153           last_double_colon = 1;
 154         }
 155       else
 156         {
 157           s = format (s, "%s%x",
 158                       (last_double_colon || i == 0) ? "" : ":",
 159                       clib_net_to_host_u16 (a->as_u16[i]));
 160           last_double_colon = 0;
 161         }
 162     }
 163
 164   return s;
 165 }
 166
 167 /* Format an IP46 address. */
 168 u8 *
 169 format_ip46_address (u8 * s, va_list * args)
 170 {
 171   ip46_address_t *ip46 = va_arg (*args, ip46_address_t *);
 172   ip46_type_t type = va_arg (*args, ip46_type_t);
 173   int is_ip4 = 1;
 174
 175   switch (type)
 176     {
 177     case IP46_TYPE_ANY:
 178       is_ip4 = ip46_address_is_ip4 (ip46);
 179       break;
 180     case IP46_TYPE_IP4:
 181       is_ip4 = 1;
 182       break;
 183     case IP46_TYPE_IP6:
 184       is_ip4 = 0;
 185       break;
 186     }
 187
 188   return is_ip4 ?
 189     format (s, "%U", format_ip4_address, &ip46->ip4) :
 190     format (s, "%U", format_ip6_address, &ip46->ip6);
 191 }
 192
 193 /*
 194  * VPPCOM Utility Functions
 195  */
 196
 197
 198 static svm_msg_q_t *
 199 vcl_session_vpp_evt_q (vcl_worker_t * wrk, vcl_session_t * s)
 200 {
 201   if (vcl_session_is_ct (s))
 202     return wrk->vpp_event_queues[0];
 203   else
 204     return wrk->vpp_event_queues[s->tx_fifo->master_thread_index];
 205 }
 206
 207 static void
 208 vcl_send_session_accepted_reply (svm_msg_q_t * mq, u32 context,
 209                                  session_handle_t handle, int retval)
 210 {
 211   app_session_evt_t _app_evt, *app_evt = &_app_evt;
 212   session_accepted_reply_msg_t *rmp;
 213   app_alloc_ctrl_evt_to_vpp (mq, app_evt, SESSION_CTRL_EVT_ACCEPTED_REPLY);
 214   rmp = (session_accepted_reply_msg_t *) app_evt->evt->data;
 215   rmp->handle = handle;
 216   rmp->context = context;
 217   rmp->retval = retval;
 218   app_send_ctrl_evt_to_vpp (mq, app_evt);
 219 }
 220
 221 static void
 222 vcl_send_session_disconnected_reply (svm_msg_q_t * mq, u32 context,
 223                                      session_handle_t handle, int retval)
 224 {
 225   app_session_evt_t _app_evt, *app_evt = &_app_evt;
 226   session_disconnected_reply_msg_t *rmp;
 227   app_alloc_ctrl_evt_to_vpp (mq, app_evt,
 228                              SESSION_CTRL_EVT_DISCONNECTED_REPLY);
 229   rmp = (session_disconnected_reply_msg_t *) app_evt->evt->data;
 230   rmp->handle = handle;
 231   rmp->context = context;
 232   rmp->retval = retval;
 233   app_send_ctrl_evt_to_vpp (mq, app_evt);
 234 }
 235
 236 static void
 237 vcl_send_session_reset_reply (svm_msg_q_t * mq, u32 context,
 238                               session_handle_t handle, int retval)
 239 {
 240   app_session_evt_t _app_evt, *app_evt = &_app_evt;
 241   session_reset_reply_msg_t *rmp;
 242   app_alloc_ctrl_evt_to_vpp (mq, app_evt, SESSION_CTRL_EVT_RESET_REPLY);
 243   rmp = (session_reset_reply_msg_t *) app_evt->evt->data;
 244   rmp->handle = handle;
 245   rmp->context = context;
 246   rmp->retval = retval;
 247   app_send_ctrl_evt_to_vpp (mq, app_evt);
 248 }
 249
 250 static u32
 251 vcl_session_accepted_handler (vcl_worker_t * wrk, session_accepted_msg_t * mp)
 252 {
 253   vcl_session_t *session, *listen_session;
 254   svm_fifo_t *rx_fifo, *tx_fifo;
 255   u32 vpp_wrk_index;
 256   svm_msg_q_t *evt_q;
 257
 258   session = vcl_session_alloc (wrk);
 259
 260   listen_session = vcl_session_table_lookup_listener (wrk,
 261                                                       mp->listener_handle);
 262   if (!listen_session)
 263     {
 264       svm_msg_q_t *evt_q;
 265       evt_q = uword_to_pointer (mp->vpp_event_queue_address, svm_msg_q_t *);
 266       clib_warning ("VCL<%d>: ERROR: couldn't find listen session: "
 267                     "unknown vpp listener handle %llx",
 268                     getpid (), mp->listener_handle);
 269       vcl_send_session_accepted_reply (evt_q, mp->context, mp->handle,
 270                                        VNET_API_ERROR_INVALID_ARGUMENT);
 271       vcl_session_free (wrk, session);
 272       return VCL_INVALID_SESSION_INDEX;
 273     }
 274
 275   rx_fifo = uword_to_pointer (mp->server_rx_fifo, svm_fifo_t *);
 276   tx_fifo = uword_to_pointer (mp->server_tx_fifo, svm_fifo_t *);
 277
 278   if (mp->server_event_queue_address)
 279     {
 280       session->vpp_evt_q = uword_to_pointer (mp->client_event_queue_address,
 281                                              svm_msg_q_t *);
 282       session->our_evt_q = uword_to_pointer (mp->server_event_queue_address,
 283                                              svm_msg_q_t *);
 284       vcl_wait_for_memory (session->vpp_evt_q);
 285       rx_fifo->master_session_index = session->session_index;
 286       tx_fifo->master_session_index = session->session_index;
 287       rx_fifo->master_thread_index = vcl_get_worker_index ();
 288       tx_fifo->master_thread_index = vcl_get_worker_index ();
 289       vec_validate (wrk->vpp_event_queues, 0);
 290       evt_q = uword_to_pointer (mp->vpp_event_queue_address, svm_msg_q_t *);
 291       wrk->vpp_event_queues[0] = evt_q;
 292     }
 293   else
 294     {
 295       session->vpp_evt_q = uword_to_pointer (mp->vpp_event_queue_address,
 296                                              svm_msg_q_t *);
 297       rx_fifo->client_session_index = session->session_index;
 298       tx_fifo->client_session_index = session->session_index;
 299       rx_fifo->client_thread_index = vcl_get_worker_index ();
 300       tx_fifo->client_thread_index = vcl_get_worker_index ();
 301       vpp_wrk_index = tx_fifo->master_thread_index;
 302       vec_validate (wrk->vpp_event_queues, vpp_wrk_index);
 303       wrk->vpp_event_queues[vpp_wrk_index] = session->vpp_evt_q;
 304     }
 305
 306   session->vpp_handle = mp->handle;
 307   session->client_context = mp->context;
 308   session->rx_fifo = rx_fifo;
 309   session->tx_fifo = tx_fifo;
 310
 311   session->session_state = STATE_ACCEPT;
 312   session->transport.rmt_port = mp->port;
 313   session->transport.is_ip4 = mp->is_ip4;
 314   clib_memcpy_fast (&session->transport.rmt_ip, mp->ip,
 315                     sizeof (ip46_address_t));
 316
 317   vcl_session_table_add_vpp_handle (wrk, mp->handle, session->session_index);
 318   session->transport.lcl_port = listen_session->transport.lcl_port;
 319   session->transport.lcl_ip = listen_session->transport.lcl_ip;
 320   session->session_type = listen_session->session_type;
 321   session->is_dgram = session->session_type == VPPCOM_PROTO_UDP;
 322
 323   VDBG (1, "VCL<%d>: vpp handle 0x%llx, sid %u: client accept request from %s"
 324         " address %U port %d queue %p!", getpid (), mp->handle,
 325         session->session_index,
 326         mp->is_ip4 ? "IPv4" : "IPv6", format_ip46_address, &mp->ip,
 327         mp->is_ip4 ? IP46_TYPE_IP4 : IP46_TYPE_IP6,
 328         clib_net_to_host_u16 (mp->port), session->vpp_evt_q);
 329   vcl_evt (VCL_EVT_ACCEPT, session, listen_session, session_index);
 330
 331   return session->session_index;
 332 }
 333
 334 static u32
 335 vcl_session_connected_handler (vcl_worker_t * wrk,
 336                                session_connected_msg_t * mp)
 337 {
 338   u32 session_index, vpp_wrk_index;
 339   svm_fifo_t *rx_fifo, *tx_fifo;
 340   vcl_session_t *session = 0;
 341   svm_msg_q_t *evt_q;
 342
 343   session_index = mp->context;
 344   session = vcl_session_get (wrk, session_index);
 345   if (!session)
 346     {
 347       clib_warning ("[%s] ERROR: vpp handle 0x%llx, sid %u: "
 348                     "Invalid session index (%u)!",
 349                     getpid (), mp->handle, session_index);
 350       return VCL_INVALID_SESSION_INDEX;
 351     }
 352   if (mp->retval)
 353     {
 354       clib_warning ("VCL<%d>: ERROR: sid %u: connect failed! %U", getpid (),
 355                     session_index, format_api_error, ntohl (mp->retval));
 356       session->session_state = STATE_FAILED;
 357       session->vpp_handle = mp->handle;
 358       return session_index;
 359     }
 360
 361   rx_fifo = uword_to_pointer (mp->server_rx_fifo, svm_fifo_t *);
 362   tx_fifo = uword_to_pointer (mp->server_tx_fifo, svm_fifo_t *);
 363   vcl_wait_for_memory (rx_fifo);
 364   rx_fifo->client_session_index = session_index;
 365   tx_fifo->client_session_index = session_index;
 366   rx_fifo->client_thread_index = vcl_get_worker_index ();
 367   tx_fifo->client_thread_index = vcl_get_worker_index ();
 368
 369   if (mp->client_event_queue_address)
 370     {
 371       session->vpp_evt_q = uword_to_pointer (mp->server_event_queue_address,
 372                                              svm_msg_q_t *);
 373       session->our_evt_q = uword_to_pointer (mp->client_event_queue_address,
 374                                              svm_msg_q_t *);
 375
 376       vec_validate (wrk->vpp_event_queues, 0);
 377       evt_q = uword_to_pointer (mp->vpp_event_queue_address, svm_msg_q_t *);
 378       wrk->vpp_event_queues[0] = evt_q;
 379     }
 380   else
 381     {
 382       session->vpp_evt_q = uword_to_pointer (mp->vpp_event_queue_address,
 383                                              svm_msg_q_t *);
 384       vpp_wrk_index = tx_fifo->master_thread_index;
 385       vec_validate (wrk->vpp_event_queues, vpp_wrk_index);
 386       wrk->vpp_event_queues[vpp_wrk_index] = session->vpp_evt_q;
 387     }
 388
 389   session->rx_fifo = rx_fifo;
 390   session->tx_fifo = tx_fifo;
 391   session->vpp_handle = mp->handle;
 392   session->transport.is_ip4 = mp->is_ip4;
 393   clib_memcpy_fast (&session->transport.lcl_ip, mp->lcl_ip,
 394                     sizeof (session->transport.lcl_ip));
 395   session->transport.lcl_port = mp->lcl_port;
 396   session->session_state = STATE_CONNECT;
 397
 398   /* Add it to lookup table */
 399   hash_set (wrk->session_index_by_vpp_handles, mp->handle, session_index);
 400
 401   VDBG (1, "VCL<%d>: vpp handle 0x%llx, sid %u: connect succeeded! "
 402         "session_rx_fifo %p, refcnt %d, session_tx_fifo %p, refcnt %d",
 403         getpid (), mp->handle, session_index, session->rx_fifo,
 404         session->rx_fifo->refcnt, session->tx_fifo, session->tx_fifo->refcnt);
 405
 406   return session_index;
 407 }
 408
 409 static u32
 410 vcl_session_reset_handler (vcl_worker_t * wrk,
 411                            session_reset_msg_t * reset_msg)
 412 {
 413   vcl_session_t *session;
 414   u32 sid;
 415
 416   sid = vcl_session_index_from_vpp_handle (wrk, reset_msg->handle);
 417   session = vcl_session_get (wrk, sid);
 418   if (!session)
 419     {
 420       VDBG (0, "request to reset unknown handle 0x%llx", reset_msg->handle);
 421       return VCL_INVALID_SESSION_INDEX;
 422     }
 423   session->session_state = STATE_CLOSE_ON_EMPTY;
 424   VDBG (0, "reset handle 0x%llx, sid %u ", reset_msg->handle, sid);
 425   vcl_send_session_reset_reply (vcl_session_vpp_evt_q (wrk, session),
 426                                 vcm->my_client_index, reset_msg->handle, 0);
 427   return sid;
 428 }
 429
 430 static u32
 431 vcl_session_bound_handler (vcl_worker_t * wrk, session_bound_msg_t * mp)
 432 {
 433   vcl_session_t *session;
 434   u32 sid = mp->context;
 435
 436   session = vcl_session_get (wrk, sid);
 437   if (mp->retval)
 438     {
 439       VDBG (0, "VCL<%d>: ERROR: vpp handle 0x%llx, sid %u: bind failed: %U",
 440             getpid (), mp->handle, sid, format_api_error, ntohl (mp->retval));
 441       if (session)
 442         {
 443           session->session_state = STATE_FAILED;
 444           session->vpp_handle = mp->handle;
 445           return sid;
 446         }
 447       else
 448         {
 449           clib_warning ("[%s] ERROR: vpp handle 0x%llx, sid %u: "
 450                         "Invalid session index (%u)!",
 451                         getpid (), mp->handle, sid);
 452           return VCL_INVALID_SESSION_INDEX;
 453         }
 454     }
 455
 456   session->vpp_handle = mp->handle;
 457   session->transport.is_ip4 = mp->lcl_is_ip4;
 458   clib_memcpy_fast (&session->transport.lcl_ip, mp->lcl_ip,
 459                     sizeof (ip46_address_t));
 460   session->transport.lcl_port = mp->lcl_port;
 461   vcl_session_table_add_listener (wrk, mp->handle, sid);
 462   session->session_state = STATE_LISTEN;
 463
 464   if (session->is_dgram)
 465     {
 466       svm_fifo_t *rx_fifo, *tx_fifo;
 467       session->vpp_evt_q = uword_to_pointer (mp->vpp_evt_q, svm_msg_q_t *);
 468       rx_fifo = uword_to_pointer (mp->rx_fifo, svm_fifo_t *);
 469       rx_fifo->client_session_index = sid;
 470       tx_fifo = uword_to_pointer (mp->tx_fifo, svm_fifo_t *);
 471       tx_fifo->client_session_index = sid;
 472       session->rx_fifo = rx_fifo;
 473       session->tx_fifo = tx_fifo;
 474     }
 475
 476   VDBG (1, "VCL<%d>: vpp handle 0x%llx, sid %u: bind succeeded!",
 477         getpid (), mp->handle, sid);
 478   return sid;
 479 }
 480
 481 static int
 482 vcl_handle_mq_event (vcl_worker_t * wrk, session_event_t * e)
 483 {
 484   session_accepted_msg_t *accepted_msg;
 485   session_disconnected_msg_t *disconnected_msg;
 486   vcl_session_msg_t *vcl_msg;
 487   vcl_session_t *session;
 488   u64 handle;
 489   u32 sid;
 490
 491   switch (e->event_type)
 492     {
 493     case FIFO_EVENT_APP_RX:
 494     case FIFO_EVENT_APP_TX:
 495     case SESSION_IO_EVT_CT_RX:
 496     case SESSION_IO_EVT_CT_TX:
 497       vec_add1 (wrk->unhandled_evts_vector, *e);
 498       break;
 499     case SESSION_CTRL_EVT_ACCEPTED:
 500       accepted_msg = (session_accepted_msg_t *) e->data;
 501       handle = accepted_msg->listener_handle;
 502       session = vcl_session_table_lookup_listener (wrk, handle);
 503       if (!session)
 504         {
 505           clib_warning ("VCL<%d>: ERROR: couldn't find listen session:"
 506                         "listener handle %llx", getpid (), handle);
 507           break;
 508         }
 509
 510       clib_fifo_add2 (session->accept_evts_fifo, vcl_msg);
 511       vcl_msg->accepted_msg = *accepted_msg;
 512       break;
 513     case SESSION_CTRL_EVT_CONNECTED:
 514       vcl_session_connected_handler (wrk,
 515                                      (session_connected_msg_t *) e->data);
 516       break;
 517     case SESSION_CTRL_EVT_DISCONNECTED:
 518       disconnected_msg = (session_disconnected_msg_t *) e->data;
 519       sid = vcl_session_index_from_vpp_handle (wrk, disconnected_msg->handle);
 520       session = vcl_session_get (wrk, sid);
 521       if (!session)
 522         {
 523           VDBG (0, "request to disconnect unknown handle 0x%llx",
 524                 disconnected_msg->handle);
 525           break;
 526         }
 527       session->session_state = STATE_DISCONNECT;
 528       VDBG (0, "disconnected handle 0x%llx, sid %u", disconnected_msg->handle,
 529             sid);
 530       break;
 531     case SESSION_CTRL_EVT_RESET:
 532       vcl_session_reset_handler (wrk, (session_reset_msg_t *) e->data);
 533       break;
 534     case SESSION_CTRL_EVT_BOUND:
 535       vcl_session_bound_handler (wrk, (session_bound_msg_t *) e->data);
 536       break;
 537     default:
 538       clib_warning ("unhandled %u", e->event_type);
 539     }
 540   return VPPCOM_OK;
 541 }
 542
 543 static inline int
 544 vppcom_wait_for_session_state_change (u32 session_index,
 545                                       session_state_t state,
 546                                       f64 wait_for_time)
 547 {
 548   vcl_worker_t *wrk = vcl_worker_get_current ();
 549   f64 timeout = clib_time_now (&wrk->clib_time) + wait_for_time;
 550   vcl_session_t *volatile session;
 551   svm_msg_q_msg_t msg;
 552   session_event_t *e;
 553
 554   do
 555     {
 556       session = vcl_session_get (wrk, session_index);
 557       if (PREDICT_FALSE (!session))
 558         {
 559           return VPPCOM_EBADFD;
 560         }
 561       if (session->session_state & state)
 562         {
 563           return VPPCOM_OK;
 564         }
 565       if (session->session_state & STATE_FAILED)
 566         {
 567           return VPPCOM_ECONNREFUSED;
 568         }
 569
 570       if (svm_msg_q_sub (wrk->app_event_queue, &msg, SVM_Q_NOWAIT, 0))
 571         continue;
 572       e = svm_msg_q_msg_data (wrk->app_event_queue, &msg);
 573       vcl_handle_mq_event (wrk, e);
 574       svm_msg_q_free_msg (wrk->app_event_queue, &msg);
 575     }
 576   while (clib_time_now (&wrk->clib_time) < timeout);
 577
 578   VDBG (0, "VCL<%d>: timeout waiting for state 0x%x (%s)", getpid (), state,
 579         vppcom_session_state_str (state));
 580   vcl_evt (VCL_EVT_SESSION_TIMEOUT, session, session_state);
 581
 582   return VPPCOM_ETIMEDOUT;
 583 }
 584
 585 static int
 586 vppcom_app_session_enable (void)
 587 {
 588   int rv;
 589
 590   if (vcm->app_state != STATE_APP_ENABLED)
 591     {
 592       vppcom_send_session_enable_disable (1 /* is_enabled == TRUE */ );
 593       rv = vcl_wait_for_app_state_change (STATE_APP_ENABLED);
 594       if (PREDICT_FALSE (rv))
 595         {
 596           VDBG (0, "VCL<%d>: application session enable timed out! "
 597                 "returning %d (%s)", getpid (), rv, vppcom_retval_str (rv));
 598           return rv;
 599         }
 600     }
 601   return VPPCOM_OK;
 602 }
 603
 604 static int
 605 vppcom_app_attach (void)
 606 {
 607   int rv;
 608
 609   vppcom_app_send_attach ();
 610   rv = vcl_wait_for_app_state_change (STATE_APP_ATTACHED);
 611   if (PREDICT_FALSE (rv))
 612     {
 613       VDBG (0, "VCL<%d>: application attach timed out! returning %d (%s)",
 614             getpid (), rv, vppcom_retval_str (rv));
 615       return rv;
 616     }
 617
 618   return VPPCOM_OK;
 619 }
 620
 621 static int
 622 vppcom_session_unbind (u32 session_handle)
 623 {
 624   vcl_worker_t *wrk = vcl_worker_get_current ();
 625   vcl_session_t *session = 0;
 626   u64 vpp_handle;
 627
 628   session = vcl_session_get_w_handle (wrk, session_handle);
 629   if (!session)
 630     return VPPCOM_EBADFD;
 631
 632   vpp_handle = session->vpp_handle;
 633   vcl_session_table_del_listener (wrk, vpp_handle);
 634   session->vpp_handle = ~0;
 635   session->session_state = STATE_DISCONNECT;
 636
 637   VDBG (1, "VCL<%d>: vpp handle 0x%llx, sid %u: sending unbind msg! new state"
 638         " 0x%x (%s)", getpid (), vpp_handle, session_handle, STATE_DISCONNECT,
 639         vppcom_session_state_str (STATE_DISCONNECT));
 640   vcl_evt (VCL_EVT_UNBIND, session);
 641   vppcom_send_unbind_sock (vpp_handle);
 642
 643   return VPPCOM_OK;
 644 }
 645
 646 static int
 647 vppcom_session_disconnect (u32 session_handle)
 648 {
 649   vcl_worker_t *wrk = vcl_worker_get_current ();
 650   svm_msg_q_t *vpp_evt_q;
 651   vcl_session_t *session;
 652   session_state_t state;
 653   u64 vpp_handle;
 654
 655   session = vcl_session_get_w_handle (wrk, session_handle);
 656   if (!session)
 657     return VPPCOM_EBADFD;
 658
 659   vpp_handle = session->vpp_handle;
 660   state = session->session_state;
 661
 662   VDBG (1, "VCL<%d>: vpp handle 0x%llx, sid %u state 0x%x (%s)", getpid (),
 663         vpp_handle, session_handle, state, vppcom_session_state_str (state));
 664
 665   if (PREDICT_FALSE (state & STATE_LISTEN))
 666     {
 667       clib_warning ("VCL<%d>: ERROR: vpp handle 0x%llx, sid %u: "
 668                     "Cannot disconnect a listen socket!",
 669                     getpid (), vpp_handle, session_handle);
 670       return VPPCOM_EBADFD;
 671     }
 672
 673   if (state & STATE_CLOSE_ON_EMPTY)
 674     {
 675       vpp_evt_q = vcl_session_vpp_evt_q (wrk, session);
 676       vcl_send_session_disconnected_reply (vpp_evt_q, vcm->my_client_index,
 677                                            vpp_handle, 0);
 678       VDBG (1, "VCL<%d>: vpp handle 0x%llx, sid %u: sending disconnect "
 679             "REPLY...", getpid (), vpp_handle, session_handle);
 680     }
 681   else
 682     {
 683       VDBG (1, "VCL<%d>: vpp handle 0x%llx, sid %u: sending disconnect...",
 684             getpid (), vpp_handle, session_handle);
 685       vppcom_send_disconnect_session (vpp_handle);
 686     }
 687
 688   return VPPCOM_OK;
 689 }
 690
 691 static void
 692 vcl_cleanup_bapi (void)
 693 {
 694   api_main_t *am = &api_main;
 695
 696   am->my_client_index = ~0;
 697   am->my_registration = 0;
 698   am->vl_input_queue = 0;
 699   am->msg_index_by_name_and_crc = 0;
 700
 701   vl_client_api_unmap ();
 702 }
 703
 704 void
 705 vcl_app_fork_child_handler (void)
 706 {
 707   u8 *child_name;
 708   int rv;
 709
 710   vcm->current_pid = getpid ();
 711   vcl_set_worker_index (0);
 712
 713   VDBG (0, "initializing forked child");
 714   child_name = format (0, "%v-child-%u%c", vcm->app_name, getpid (), 0);
 715
 716   vcl_cleanup_bapi ();
 717   vppcom_api_hookup ();
 718   vcm->app_state = STATE_APP_START;
 719   rv = vppcom_connect_to_vpp ((char *) child_name);
 720   vec_free (child_name);
 721   if (rv)
 722     {
 723       VERR ("couldn't connect to VPP!");
 724       return;
 725     }
 726
 727   vcm->app_state = STATE_APP_ADDING_WORKER;
 728   vcl_send_app_worker_add_del (1 /* is_add */ );
 729   if (vcl_wait_for_app_state_change (STATE_APP_READY))
 730     {
 731       VERR ("failed to add worker to vpp");
 732       return;
 733     }
 734   VDBG (0, "forked child main worker initialized");
 735 }
 736
 737 /*
 738  * VPPCOM Public API functions
 739  */
 740 int
 741 vppcom_app_create (char *app_name)
 742 {
 743   vppcom_cfg_t *vcl_cfg = &vcm->cfg;
 744   int rv;
 745
 746   if (!vcm->is_init)
 747     {
 748       vcm->is_init = 1;
 749       vppcom_cfg (&vcm->cfg);
 750       vcl_cfg = &vcm->cfg;
 751
 752       vcm->main_cpu = pthread_self ();
 753       vcm->main_pid = vcm->current_pid = getpid ();
 754       vcm->app_name = format (0, "%s", app_name);
 755       vppcom_init_error_string_table ();
 756       svm_fifo_segment_main_init (vcl_cfg->segment_baseva,
 757                                   20 /* timeout in secs */ );
 758       pool_init_fixed (vcm->workers, vcl_cfg->max_workers);
 759       clib_spinlock_init (&vcm->workers_lock);
 760       vcl_worker_alloc_and_init ();
 761       pthread_atfork (NULL, NULL, vcl_app_fork_child_handler);
 762     }
 763
 764   if (vcm->my_client_index == ~0)
 765     {
 766       /* API hookup and connect to VPP */
 767       vppcom_api_hookup ();
 768       vcl_elog_init (vcm);
 769       vcm->app_state = STATE_APP_START;
 770       rv = vppcom_connect_to_vpp (app_name);
 771       if (rv)
 772         {
 773           VERR ("couldn't connect to VPP!");
 774           return rv;
 775         }
 776       vcm->main_api_client_index = vcm->my_client_index;
 777       VDBG (0, "sending session enable");
 778       rv = vppcom_app_session_enable ();
 779       if (rv)
 780         {
 781           VERR ("vppcom_app_session_enable() failed!");
 782           return rv;
 783         }
 784
 785       VDBG (0, "sending app attach");
 786       rv = vppcom_app_attach ();
 787       if (rv)
 788         {
 789           VERR ("vppcom_app_attach() failed!");
 790           return rv;
 791         }
 792
 793       VDBG (0, "app_name '%s', my_client_index %d (0x%x)",
 794             app_name, vcm->my_client_index, vcm->my_client_index);
 795     }
 796
 797   return VPPCOM_OK;
 798 }
 799
 800 void
 801 vppcom_app_destroy (void)
 802 {
 803   int rv;
 804   f64 orig_app_timeout;
 805
 806   if (vcm->my_client_index == ~0)
 807     return;
 808
 809   VDBG (0, "detaching from VPP, my_client_index %d (0x%x)",
 810         vcm->my_client_index, vcm->my_client_index);
 811   vcl_evt (VCL_EVT_DETACH, vcm);
 812
 813   vppcom_app_send_detach ();
 814   orig_app_timeout = vcm->cfg.app_timeout;
 815   vcm->cfg.app_timeout = 2.0;
 816   rv = vcl_wait_for_app_state_change (STATE_APP_ENABLED);
 817   vcm->cfg.app_timeout = orig_app_timeout;
 818   if (PREDICT_FALSE (rv))
 819     VDBG (0, "application detach timed out! returning %d (%s)",
 820           rv, vppcom_retval_str (rv));
 821
 822   vcl_elog_stop (vcm);
 823   vl_client_disconnect_from_vlib ();
 824   vec_free (vcm->app_name);
 825   vcm->my_client_index = ~0;
 826   vcm->app_state = STATE_APP_START;
 827 }
 828
 829 int
 830 vppcom_session_create (u8 proto, u8 is_nonblocking)
 831 {
 832   vcl_worker_t *wrk = vcl_worker_get_current ();
 833   vcl_session_t *session;
 834
 835   session = vcl_session_alloc (wrk);
 836
 837   session->session_type = proto;
 838   session->session_state = STATE_START;
 839   session->vpp_handle = ~0;
 840   session->is_dgram = proto == VPPCOM_PROTO_UDP;
 841
 842   if (is_nonblocking)
 843     VCL_SESS_ATTR_SET (session->attr, VCL_SESS_ATTR_NONBLOCK);
 844
 845   vcl_evt (VCL_EVT_CREATE, session, session_type, session->session_state,
 846            is_nonblocking, session_index);
 847
 848   VDBG (0, "created sid %u", session->session_index);
 849
 850   return vcl_session_handle (session);
 851 }
 852
 853 int
 854 vppcom_session_close (uint32_t session_handle)
 855 {
 856   vcl_worker_t *wrk = vcl_worker_get_current ();
 857   vcl_session_t *session = 0;
 858   u8 is_vep, is_vep_session;
 859   session_state_t state;
 860   u32 next_sh, vep_sh;
 861   int rv = VPPCOM_OK;
 862   u64 vpp_handle;
 863
 864   session = vcl_session_get_w_handle (wrk, session_handle);
 865   if (!session)
 866     return VPPCOM_EBADFD;
 867
 868   is_vep = session->is_vep;
 869   is_vep_session = session->is_vep_session;
 870   next_sh = session->vep.next_sh;
 871   vep_sh = session->vep.vep_sh;
 872   state = session->session_state;
 873   vpp_handle = session->vpp_handle;
 874
 875   if (VPPCOM_DEBUG > 0)
 876     {
 877       if (is_vep)
 878         clib_warning ("VCL<%d>: vep_idx %u / sid %u: "
 879                       "closing epoll session...",
 880                       getpid (), session_handle, session_handle);
 881       else
 882         clib_warning ("VCL<%d>: vpp handle 0x%llx, sid %d: "
 883                       "closing session...",
 884                       getpid (), vpp_handle, session_handle);
 885     }
 886
 887   if (is_vep)
 888     {
 889       while (next_sh != ~0)
 890         {
 891           rv = vppcom_epoll_ctl (session_handle, EPOLL_CTL_DEL, next_sh, 0);
 892           if (PREDICT_FALSE (rv < 0))
 893             VDBG (0, "VCL<%d>: vpp handle 0x%llx, sid %u: EPOLL_CTL_DEL "
 894                   "vep_idx %u failed! rv %d (%s)",
 895                   getpid (), vpp_handle, next_sh, vep_sh,
 896                   rv, vppcom_retval_str (rv));
 897
 898           next_sh = session->vep.next_sh;
 899         }
 900     }
 901   else
 902     {
 903       if (is_vep_session)
 904         {
 905           rv = vppcom_epoll_ctl (vep_sh, EPOLL_CTL_DEL, session_handle, 0);
 906           if (rv < 0)
 907             VDBG (0, "VCL<%d>: vpp handle 0x%llx, sid %u: EPOLL_CTL_DEL "
 908                   "vep_idx %u failed! rv %d (%s)",
 909                   getpid (), vpp_handle, session_handle,
 910                   vep_sh, rv, vppcom_retval_str (rv));
 911         }
 912
 913       if (state & STATE_LISTEN)
 914         {
 915           rv = vppcom_session_unbind (session_handle);
 916           if (PREDICT_FALSE (rv < 0))
 917             VDBG (0, "VCL<%d>: vpp handle 0x%llx, sid %u: listener unbind "
 918                   "failed! rv %d (%s)",
 919                   getpid (), vpp_handle, session_handle,
 920                   rv, vppcom_retval_str (rv));
 921         }
 922       else if (state & STATE_OPEN)
 923         {
 924           rv = vppcom_session_disconnect (session_handle);
 925           if (PREDICT_FALSE (rv < 0))
 926             clib_warning ("VCL<%d>: ERROR: vpp handle 0x%llx, sid %u: "
 927                           "session disconnect failed! rv %d (%s)",
 928                           getpid (), vpp_handle, session_handle,
 929                           rv, vppcom_retval_str (rv));
 930         }
 931     }
 932
 933   if (vcl_session_is_ct (session))
 934     {
 935       vcl_cut_through_registration_t *ctr;
 936       uword mq_addr;
 937
 938       mq_addr = pointer_to_uword (session->our_evt_q);
 939       ctr = vcl_ct_registration_lock_and_lookup (wrk, mq_addr);
 940       ASSERT (ctr);
 941       if (ctr->epoll_evt_conn_index != ~0)
 942         vcl_mq_epoll_del_evfd (wrk, ctr->epoll_evt_conn_index);
 943       VDBG (0, "Removing ct registration %u",
 944             vcl_ct_registration_index (wrk, ctr));
 945       vcl_ct_registration_del (wrk, ctr);
 946       vcl_ct_registration_lookup_del (wrk, mq_addr);
 947       vcl_ct_registration_unlock (wrk);
 948     }
 949
 950   if (vpp_handle != ~0)
 951     {
 952       vcl_session_table_del_vpp_handle (wrk, vpp_handle);
 953     }
 954   vcl_session_free (wrk, session);
 955
 956   if (VPPCOM_DEBUG > 0)
 957     {
 958       if (is_vep)
 959         clib_warning ("VCL<%d>: vep_idx %u / sid %u: epoll session removed.",
 960                       getpid (), session_handle, session_handle);
 961       else
 962         clib_warning ("VCL<%d>: vpp handle 0x%llx, sid %u: session removed.",
 963                       getpid (), vpp_handle, session_handle);
 964     }
 965
 966   vcl_evt (VCL_EVT_CLOSE, session, rv);
 967
 968   return rv;
 969 }
 970
 971 int
 972 vppcom_session_bind (uint32_t session_handle, vppcom_endpt_t * ep)
 973 {
 974   vcl_worker_t *wrk = vcl_worker_get_current ();
 975   vcl_session_t *session = 0;
 976
 977   if (!ep || !ep->ip)
 978     return VPPCOM_EINVAL;
 979
 980   session = vcl_session_get_w_handle (wrk, session_handle);
 981   if (!session)
 982     return VPPCOM_EBADFD;
 983
 984   if (session->is_vep)
 985     {
 986       clib_warning ("VCL<%d>: ERROR: sid %u: cannot "
 987                     "bind to an epoll session!", getpid (), session_handle);
 988       return VPPCOM_EBADFD;
 989     }
 990
 991   session->transport.is_ip4 = ep->is_ip4;
 992   if (ep->is_ip4)
 993     clib_memcpy_fast (&session->transport.lcl_ip.ip4, ep->ip,
 994                       sizeof (ip4_address_t));
 995   else
 996     clib_memcpy_fast (&session->transport.lcl_ip.ip6, ep->ip,
 997                       sizeof (ip6_address_t));
 998   session->transport.lcl_port = ep->port;
 999
1000   VDBG (0, "VCL<%d>: sid %u: binding to local %s address %U port %u, "
1001         "proto %s", getpid (), session_handle,
1002         session->transport.is_ip4 ? "IPv4" : "IPv6",
1003         format_ip46_address, &session->transport.lcl_ip,
1004         session->transport.is_ip4 ? IP46_TYPE_IP4 : IP46_TYPE_IP6,
1005         clib_net_to_host_u16 (session->transport.lcl_port),
1006         session->session_type ? "UDP" : "TCP");
1007   vcl_evt (VCL_EVT_BIND, session);
1008
1009   if (session->session_type == VPPCOM_PROTO_UDP)
1010     vppcom_session_listen (session_handle, 10);
1011
1012   return VPPCOM_OK;
1013 }
1014
1015 int
1016 vppcom_session_listen (uint32_t listen_sh, uint32_t q_len)
1017 {
1018   vcl_worker_t *wrk = vcl_worker_get_current ();
1019   vcl_session_t *listen_session = 0;
1020   u64 listen_vpp_handle;
1021   int rv;
1022
1023   listen_session = vcl_session_get_w_handle (wrk, listen_sh);
1024   if (!listen_session)
1025     return VPPCOM_EBADFD;
1026
1027   if (q_len == 0 || q_len == ~0)
1028     q_len = vcm->cfg.listen_queue_size;
1029
1030   if (listen_session->is_vep)
1031     {
1032       clib_warning ("VCL<%d>: ERROR: sid %u: cannot listen on an "
1033                     "epoll session!", getpid (), listen_sh);
1034       return VPPCOM_EBADFD;
1035     }
1036
1037   listen_vpp_handle = listen_session->vpp_handle;
1038   if (listen_session->session_state & STATE_LISTEN)
1039     {
1040       VDBG (0, "VCL<%d>: vpp handle 0x%llx, sid %u: already in listen state!",
1041             getpid (), listen_vpp_handle, listen_sh);
1042       return VPPCOM_OK;
1043     }
1044
1045   VDBG (0, "VCL<%d>: vpp handle 0x%llx, sid %u: sending VPP bind+listen "
1046         "request...", getpid (), listen_vpp_handle, listen_sh);
1047
1048   /*
1049    * Send listen request to vpp and wait for reply
1050    */
1051   vppcom_send_bind_sock (listen_session);
1052   rv = vppcom_wait_for_session_state_change (listen_session->session_index,
1053                                              STATE_LISTEN,
1054                                              vcm->cfg.session_timeout);
1055
1056   if (PREDICT_FALSE (rv))
1057     {
1058       listen_session = vcl_session_get_w_handle (wrk, listen_sh);
1059       VDBG (0, "VCL<%d>: vpp handle 0x%llx, sid %u: bind+listen failed! "
1060             "returning %d (%s)", getpid (), listen_session->vpp_handle,
1061             listen_sh, rv, vppcom_retval_str (rv));
1062       return rv;
1063     }
1064
1065   return VPPCOM_OK;
1066 }
1067
1068 static int
1069 validate_args_session_accept_ (vcl_worker_t * wrk,
1070                                vcl_session_t * listen_session)
1071 {
1072   /* Input validation - expects spinlock on sessions_lockp */
1073   if (listen_session->is_vep)
1074     {
1075       clib_warning ("VCL<%d>: ERROR: sid %u: cannot accept on an "
1076                     "epoll session!", getpid (),
1077                     listen_session->session_index);
1078       return VPPCOM_EBADFD;
1079     }
1080
1081   if (listen_session->session_state != STATE_LISTEN)
1082     {
1083       clib_warning ("VCL<%d>: ERROR: vpp handle 0x%llx, sid %u: "
1084                     "not in listen state! state 0x%x (%s)", getpid (),
1085                     listen_session->vpp_handle, listen_session->session_index,
1086                     listen_session->session_state,
1087                     vppcom_session_state_str (listen_session->session_state));
1088       return VPPCOM_EBADFD;
1089     }
1090   return VPPCOM_OK;
1091 }
1092
1093 int
1094 vppcom_session_accept (uint32_t listen_session_handle, vppcom_endpt_t * ep,
1095                        uint32_t flags)
1096 {
1097   u32 client_session_index = ~0, listen_session_index;
1098   vcl_worker_t *wrk = vcl_worker_get_current ();
1099   session_accepted_msg_t accepted_msg;
1100   vcl_session_t *listen_session = 0;
1101   vcl_session_t *client_session = 0;
1102   svm_msg_q_t *vpp_evt_q;
1103   vcl_session_msg_t *evt;
1104   u64 listen_vpp_handle;
1105   svm_msg_q_msg_t msg;
1106   session_event_t *e;
1107   u8 is_nonblocking;
1108   int rv;
1109
1110   listen_session = vcl_session_get_w_handle (wrk, listen_session_handle);
1111   if (!listen_session)
1112     return VPPCOM_EBADFD;
1113
1114   listen_session_index = listen_session->session_index;
1115   if ((rv = validate_args_session_accept_ (wrk, listen_session)))
1116     return rv;
1117
1118   if (clib_fifo_elts (listen_session->accept_evts_fifo))
1119     {
1120       clib_fifo_sub2 (listen_session->accept_evts_fifo, evt);
1121       accepted_msg = evt->accepted_msg;
1122       goto handle;
1123     }
1124
1125   is_nonblocking = VCL_SESS_ATTR_TEST (listen_session->attr,
1126                                        VCL_SESS_ATTR_NONBLOCK);
1127   if (svm_msg_q_is_empty (wrk->app_event_queue) && is_nonblocking)
1128     return VPPCOM_EAGAIN;
1129
1130   while (1)
1131     {
1132       if (svm_msg_q_sub (wrk->app_event_queue, &msg, SVM_Q_WAIT, 0))
1133         return VPPCOM_EAGAIN;
1134
1135       e = svm_msg_q_msg_data (wrk->app_event_queue, &msg);
1136       if (e->event_type != SESSION_CTRL_EVT_ACCEPTED)
1137         {
1138           clib_warning ("discarded event: %u", e->event_type);
1139           svm_msg_q_free_msg (wrk->app_event_queue, &msg);
1140           continue;
1141         }
1142       clib_memcpy_fast (&accepted_msg, e->data, sizeof (accepted_msg));
1143       svm_msg_q_free_msg (wrk->app_event_queue, &msg);
1144       break;
1145     }
1146
1147 handle:
1148
1149   client_session_index = vcl_session_accepted_handler (wrk, &accepted_msg);
1150   listen_session = vcl_session_get (wrk, listen_session_index);
1151   client_session = vcl_session_get (wrk, client_session_index);
1152
1153   if (flags & O_NONBLOCK)
1154     VCL_SESS_ATTR_SET (client_session->attr, VCL_SESS_ATTR_NONBLOCK);
1155
1156   listen_vpp_handle = listen_session->vpp_handle;
1157   VDBG (0, "VCL<%d>: vpp handle 0x%llx, sid %u: Got a client request! "
1158         "vpp handle 0x%llx, sid %u, flags %d, is_nonblocking %u",
1159         getpid (), listen_vpp_handle, listen_session_handle,
1160         client_session->vpp_handle, client_session_index,
1161         flags, VCL_SESS_ATTR_TEST (client_session->attr,
1162                                    VCL_SESS_ATTR_NONBLOCK));
1163
1164   if (ep)
1165     {
1166       ep->is_ip4 = client_session->transport.is_ip4;
1167       ep->port = client_session->transport.rmt_port;
1168       if (client_session->transport.is_ip4)
1169         clib_memcpy_fast (ep->ip, &client_session->transport.rmt_ip.ip4,
1170                           sizeof (ip4_address_t));
1171       else
1172         clib_memcpy_fast (ep->ip, &client_session->transport.rmt_ip.ip6,
1173                           sizeof (ip6_address_t));
1174     }
1175
1176   if (accepted_msg.server_event_queue_address)
1177     vpp_evt_q = uword_to_pointer (accepted_msg.vpp_event_queue_address,
1178                                   svm_msg_q_t *);
1179   else
1180     vpp_evt_q = client_session->vpp_evt_q;
1181
1182   vcl_send_session_accepted_reply (vpp_evt_q, client_session->client_context,
1183                                    client_session->vpp_handle, 0);
1184
1185   VDBG (0, "VCL<%d>: vpp handle 0x%llx, sid %u: accepted vpp handle 0x%llx, "
1186         "sid %u connection from peer %s address %U port %u to local %s "
1187         "address %U port %u", getpid (), listen_vpp_handle,
1188         listen_session_handle, client_session->vpp_handle,
1189         client_session_index,
1190         client_session->transport.is_ip4 ? "IPv4" : "IPv6",
1191         format_ip46_address, &client_session->transport.rmt_ip,
1192         client_session->transport.is_ip4 ? IP46_TYPE_IP4 : IP46_TYPE_IP6,
1193         clib_net_to_host_u16 (client_session->transport.rmt_port),
1194         client_session->transport.is_ip4 ? "IPv4" : "IPv6",
1195         format_ip46_address, &client_session->transport.lcl_ip,
1196         client_session->transport.is_ip4 ? IP46_TYPE_IP4 : IP46_TYPE_IP6,
1197         clib_net_to_host_u16 (client_session->transport.lcl_port));
1198   vcl_evt (VCL_EVT_ACCEPT, client_session, listen_session,
1199            client_session_index);
1200
1201   return vcl_session_handle (client_session);
1202 }
1203
1204 int
1205 vppcom_session_connect (uint32_t session_handle, vppcom_endpt_t * server_ep)
1206 {
1207   vcl_worker_t *wrk = vcl_worker_get_current ();
1208   vcl_session_t *session = 0;
1209   u32 session_index;
1210   int rv;
1211
1212   session = vcl_session_get_w_handle (wrk, session_handle);
1213   if (!session)
1214     return VPPCOM_EBADFD;
1215   session_index = session->session_index;
1216
1217   if (PREDICT_FALSE (session->is_vep))
1218     {
1219       clib_warning ("VCL<%d>: ERROR: sid %u: cannot "
1220                     "connect on an epoll session!", getpid (),
1221                     session_handle);
1222       return VPPCOM_EBADFD;
1223     }
1224
1225   if (PREDICT_FALSE (session->session_state & CLIENT_STATE_OPEN))
1226     {
1227       VDBG (0, "VCL<%d>: vpp handle 0x%llx, sid %u: session already "
1228             "connected to %s %U port %d proto %s, state 0x%x (%s)",
1229             getpid (), session->vpp_handle, session_handle,
1230             session->transport.is_ip4 ? "IPv4" : "IPv6",
1231             format_ip46_address,
1232             &session->transport.rmt_ip, session->transport.is_ip4 ?
1233             IP46_TYPE_IP4 : IP46_TYPE_IP6,
1234             clib_net_to_host_u16 (session->transport.rmt_port),
1235             session->session_type ? "UDP" : "TCP", session->session_state,
1236             vppcom_session_state_str (session->session_state));
1237       return VPPCOM_OK;
1238     }
1239
1240   session->transport.is_ip4 = server_ep->is_ip4;
1241   if (session->transport.is_ip4)
1242     clib_memcpy_fast (&session->transport.rmt_ip.ip4, server_ep->ip,
1243                       sizeof (ip4_address_t));
1244   else
1245     clib_memcpy_fast (&session->transport.rmt_ip.ip6, server_ep->ip,
1246                       sizeof (ip6_address_t));
1247   session->transport.rmt_port = server_ep->port;
1248
1249   VDBG (0, "VCL<%d>: vpp handle 0x%llx, sid %u: connecting to server %s %U "
1250         "port %d proto %s",
1251         getpid (), session->vpp_handle, session_handle,
1252         session->transport.is_ip4 ? "IPv4" : "IPv6",
1253         format_ip46_address,
1254         &session->transport.rmt_ip, session->transport.is_ip4 ?
1255         IP46_TYPE_IP4 : IP46_TYPE_IP6,
1256         clib_net_to_host_u16 (session->transport.rmt_port),
1257         session->session_type ? "UDP" : "TCP");
1258
1259   /*
1260    * Send connect request and wait for reply from vpp
1261    */
1262   vppcom_send_connect_sock (session);
1263   rv = vppcom_wait_for_session_state_change (session_index, STATE_CONNECT,
1264                                              vcm->cfg.session_timeout);
1265
1266   session = vcl_session_get (wrk, session_index);
1267
1268   if (PREDICT_FALSE (rv))
1269     {
1270       if (VPPCOM_DEBUG > 0)
1271         {
1272           if (session)
1273             clib_warning ("VCL<%d>: vpp handle 0x%llx, sid %u: connect "
1274                           "failed! returning %d (%s)", getpid (),
1275                           session->vpp_handle, session_handle, rv,
1276                           vppcom_retval_str (rv));
1277           else
1278             clib_warning ("VCL<%d>: no session for sid %u: connect failed! "
1279                           "returning %d (%s)", getpid (),
1280                           session_handle, rv, vppcom_retval_str (rv));
1281         }
1282     }
1283   else
1284     VDBG (0, "VCL<%d>: vpp handle 0x%llx, sid %u: connected!",
1285           getpid (), session->vpp_handle, session_handle);
1286
1287   return rv;
1288 }
1289
1290 static u8
1291 vcl_is_rx_evt_for_session (session_event_t * e, u32 sid, u8 is_ct)
1292 {
1293   if (!is_ct)
1294     return (e->event_type == FIFO_EVENT_APP_RX
1295             && e->fifo->client_session_index == sid);
1296   else
1297     return (e->event_type == SESSION_IO_EVT_CT_TX);
1298 }
1299
1300 static inline u8
1301 vcl_session_is_readable (vcl_session_t * s)
1302 {
1303   return ((s->session_state & STATE_OPEN)
1304           || (s->session_state == STATE_LISTEN
1305               && s->session_type == VPPCOM_PROTO_UDP));
1306 }
1307
1308 static inline int
1309 vppcom_session_read_internal (uint32_t session_handle, void *buf, int n,
1310                               u8 peek)
1311 {
1312   vcl_worker_t *wrk = vcl_worker_get_current ();
1313   int n_read = 0, rv, is_nonblocking;
1314   vcl_session_t *s = 0;
1315   svm_fifo_t *rx_fifo;
1316   svm_msg_q_msg_t msg;
1317   session_event_t *e;
1318   svm_msg_q_t *mq;
1319   u8 is_ct;
1320
1321   if (PREDICT_FALSE (!buf))
1322     return VPPCOM_EINVAL;
1323
1324   s = vcl_session_get_w_handle (wrk, session_handle);
1325   if (PREDICT_FALSE (!s || s->is_vep))
1326     return VPPCOM_EBADFD;
1327
1328   if (PREDICT_FALSE (!vcl_session_is_readable (s)))
1329     {
1330       session_state_t state = s->session_state;
1331       rv = ((state & STATE_DISCONNECT) ? VPPCOM_ECONNRESET : VPPCOM_ENOTCONN);
1332
1333       VDBG (0, "VCL<%d>: vpp handle 0x%llx, sid %u: %s session is not open! "
1334             "state 0x%x (%s), returning %d (%s)",
1335             getpid (), s->vpp_handle, session_handle, state,
1336             vppcom_session_state_str (state), rv, vppcom_retval_str (rv));
1337       return rv;
1338     }
1339
1340   is_nonblocking = VCL_SESS_ATTR_TEST (s->attr, VCL_SESS_ATTR_NONBLOCK);
1341   is_ct = vcl_session_is_ct (s);
1342   mq = is_ct ? s->our_evt_q : wrk->app_event_queue;
1343   rx_fifo = s->rx_fifo;
1344   s->has_rx_evt = 0;
1345
1346   if (svm_fifo_is_empty (rx_fifo))
1347     {
1348       if (is_nonblocking)
1349         {
1350           svm_fifo_unset_event (rx_fifo);
1351           return VPPCOM_EWOULDBLOCK;
1352         }
1353       while (svm_fifo_is_empty (rx_fifo))
1354         {
1355           svm_fifo_unset_event (rx_fifo);
1356           svm_msg_q_lock (mq);
1357           if (svm_msg_q_is_empty (mq))
1358             svm_msg_q_wait (mq);
1359
1360           svm_msg_q_sub_w_lock (mq, &msg);
1361           e = svm_msg_q_msg_data (mq, &msg);
1362           svm_msg_q_unlock (mq);
1363           if (!vcl_is_rx_evt_for_session (e, s->session_index, is_ct))
1364             {
1365               vcl_handle_mq_event (wrk, e);
1366               svm_msg_q_free_msg (mq, &msg);
1367               continue;
1368             }
1369           svm_msg_q_free_msg (mq, &msg);
1370
1371           if (PREDICT_FALSE (s->session_state == STATE_CLOSE_ON_EMPTY))
1372             return 0;
1373         }
1374     }
1375
1376   if (s->is_dgram)
1377     n_read = app_recv_dgram_raw (rx_fifo, buf, n, &s->transport, 0, peek);
1378   else
1379     n_read = app_recv_stream_raw (rx_fifo, buf, n, 0, peek);
1380
1381   if (svm_fifo_is_empty (rx_fifo))
1382     svm_fifo_unset_event (rx_fifo);
1383
1384   if (is_ct && svm_fifo_want_tx_evt (rx_fifo))
1385     {
1386       svm_fifo_set_want_tx_evt (s->rx_fifo, 0);
1387       app_send_io_evt_to_vpp (s->vpp_evt_q, s->rx_fifo, SESSION_IO_EVT_CT_RX,
1388                               SVM_Q_WAIT);
1389     }
1390
1391   VDBG (2, "VCL<%d>: vpp handle 0x%llx, sid %u: read %d bytes from (%p)",
1392         getpid (), s->vpp_handle, session_handle, n_read, rx_fifo);
1393
1394   return n_read;
1395 }
1396
1397 int
1398 vppcom_session_read (uint32_t session_handle, void *buf, size_t n)
1399 {
1400   return (vppcom_session_read_internal (session_handle, buf, n, 0));
1401 }
1402
1403 static int
1404 vppcom_session_peek (uint32_t session_handle, void *buf, int n)
1405 {
1406   return (vppcom_session_read_internal (session_handle, buf, n, 1));
1407 }
1408
1409 int
1410 vppcom_session_read_segments (uint32_t session_handle,
1411                               vppcom_data_segments_t ds)
1412 {
1413   vcl_worker_t *wrk = vcl_worker_get_current ();
1414   int n_read = 0, rv, is_nonblocking;
1415   vcl_session_t *s = 0;
1416   svm_fifo_t *rx_fifo;
1417   svm_msg_q_msg_t msg;
1418   session_event_t *e;
1419   svm_msg_q_t *mq;
1420   u8 is_ct;
1421
1422   s = vcl_session_get_w_handle (wrk, session_handle);
1423   if (PREDICT_FALSE (!s || s->is_vep))
1424     return VPPCOM_EBADFD;
1425
1426   if (PREDICT_FALSE (!vcl_session_is_readable (s)))
1427     {
1428       session_state_t state = s->session_state;
1429       rv = ((state & STATE_DISCONNECT) ? VPPCOM_ECONNRESET : VPPCOM_ENOTCONN);
1430       return rv;
1431     }
1432
1433   is_nonblocking = VCL_SESS_ATTR_TEST (s->attr, VCL_SESS_ATTR_NONBLOCK);
1434   is_ct = vcl_session_is_ct (s);
1435   mq = is_ct ? s->our_evt_q : wrk->app_event_queue;
1436   rx_fifo = s->rx_fifo;
1437   s->has_rx_evt = 0;
1438
1439   if (svm_fifo_is_empty (rx_fifo))
1440     {
1441       if (is_nonblocking)
1442         {
1443           svm_fifo_unset_event (rx_fifo);
1444           return VPPCOM_EWOULDBLOCK;
1445         }
1446       while (svm_fifo_is_empty (rx_fifo))
1447         {
1448           svm_fifo_unset_event (rx_fifo);
1449           svm_msg_q_lock (mq);
1450           if (svm_msg_q_is_empty (mq))
1451             svm_msg_q_wait (mq);
1452
1453           svm_msg_q_sub_w_lock (mq, &msg);
1454           e = svm_msg_q_msg_data (mq, &msg);
1455           svm_msg_q_unlock (mq);
1456           if (!vcl_is_rx_evt_for_session (e, s->session_index, is_ct))
1457             {
1458               vcl_handle_mq_event (wrk, e);
1459               svm_msg_q_free_msg (mq, &msg);
1460               continue;
1461             }
1462           svm_msg_q_free_msg (mq, &msg);
1463
1464           if (PREDICT_FALSE (s->session_state == STATE_CLOSE_ON_EMPTY))
1465             return 0;
1466         }
1467     }
1468
1469   n_read = svm_fifo_segments (rx_fifo, (svm_fifo_segment_t *) ds);
1470   svm_fifo_unset_event (rx_fifo);
1471
1472   if (is_ct && n_read + svm_fifo_max_dequeue (rx_fifo) == rx_fifo->nitems)
1473     {
1474       /* If the peer is not polling send notification */
1475       if (!svm_fifo_has_event (s->rx_fifo))
1476         app_send_io_evt_to_vpp (s->vpp_evt_q, s->rx_fifo,
1477                                 SESSION_IO_EVT_CT_RX, SVM_Q_WAIT);
1478     }
1479
1480   return n_read;
1481 }
1482
1483 void
1484 vppcom_session_free_segments (uint32_t session_handle,
1485                               vppcom_data_segments_t ds)
1486 {
1487   vcl_worker_t *wrk = vcl_worker_get_current ();
1488   vcl_session_t *s;
1489
1490   s = vcl_session_get_w_handle (wrk, session_handle);
1491   if (PREDICT_FALSE (!s || s->is_vep))
1492     return;
1493
1494   svm_fifo_segments_free (s->rx_fifo, (svm_fifo_segment_t *) ds);
1495 }
1496
1497 static inline int
1498 vppcom_session_read_ready (vcl_session_t * session)
1499 {
1500   /* Assumes caller has acquired spinlock: vcm->sessions_lockp */
1501   if (PREDICT_FALSE (session->is_vep))
1502     {
1503       clib_warning ("VCL<%d>: ERROR: sid %u: cannot read from an "
1504                     "epoll session!", getpid (), session->session_index);
1505       return VPPCOM_EBADFD;
1506     }
1507
1508   if (PREDICT_FALSE (!(session->session_state & (STATE_OPEN | STATE_LISTEN))))
1509     {
1510       session_state_t state = session->session_state;
1511       int rv;
1512
1513       rv = ((state & STATE_DISCONNECT) ? VPPCOM_ECONNRESET : VPPCOM_ENOTCONN);
1514
1515       VDBG (1, "VCL<%d>: vpp handle 0x%llx, sid %u: session is not open!"
1516             " state 0x%x (%s), returning %d (%s)", getpid (),
1517             session->vpp_handle, session->session_index, state,
1518             vppcom_session_state_str (state), rv, vppcom_retval_str (rv));
1519       return rv;
1520     }
1521
1522   if (session->session_state & STATE_LISTEN)
1523     return clib_fifo_elts (session->accept_evts_fifo);
1524
1525   return svm_fifo_max_dequeue (session->rx_fifo);
1526 }
1527
1528 int
1529 vppcom_data_segment_copy (void *buf, vppcom_data_segments_t ds, u32 max_bytes)
1530 {
1531   u32 first_copy = clib_min (ds[0].len, max_bytes);
1532   clib_memcpy_fast (buf, ds[0].data, first_copy);
1533   if (first_copy < max_bytes)
1534     {
1535       clib_memcpy_fast (buf + first_copy, ds[1].data,
1536                         clib_min (ds[1].len, max_bytes - first_copy));
1537     }
1538   return 0;
1539 }
1540
1541 static u8
1542 vcl_is_tx_evt_for_session (session_event_t * e, u32 sid, u8 is_ct)
1543 {
1544   if (!is_ct)
1545     return (e->event_type == FIFO_EVENT_APP_TX
1546             && e->fifo->client_session_index == sid);
1547   else
1548     return (e->event_type == SESSION_IO_EVT_CT_RX);
1549 }
1550
1551 int
1552 vppcom_session_write (uint32_t session_handle, void *buf, size_t n)
1553 {
1554   vcl_worker_t *wrk = vcl_worker_get_current ();
1555   int rv, n_write, is_nonblocking;
1556   vcl_session_t *s = 0;
1557   svm_fifo_t *tx_fifo = 0;
1558   session_evt_type_t et;
1559   svm_msg_q_msg_t msg;
1560   session_event_t *e;
1561   svm_msg_q_t *mq;
1562   u8 is_ct;
1563
1564   if (PREDICT_FALSE (!buf))
1565     return VPPCOM_EINVAL;
1566
1567   s = vcl_session_get_w_handle (wrk, session_handle);
1568   if (PREDICT_FALSE (!s))
1569     return VPPCOM_EBADFD;
1570
1571   if (PREDICT_FALSE (s->is_vep))
1572     {
1573       clib_warning ("VCL<%d>: ERROR: vpp handle 0x%llx, sid %u: "
1574                     "cannot write to an epoll session!",
1575                     getpid (), s->vpp_handle, session_handle);
1576
1577       return VPPCOM_EBADFD;
1578     }
1579
1580   if (PREDICT_FALSE (!(s->session_state & STATE_OPEN)))
1581     {
1582       session_state_t state = s->session_state;
1583       rv = ((state & STATE_DISCONNECT) ? VPPCOM_ECONNRESET : VPPCOM_ENOTCONN);
1584       VDBG (1, "VCL<%d>: vpp handle 0x%llx, sid %u: session is not open! "
1585             "state 0x%x (%s)", getpid (), s->vpp_handle, session_handle,
1586             state, vppcom_session_state_str (state));
1587       return rv;
1588     }
1589
1590   tx_fifo = s->tx_fifo;
1591   is_ct = vcl_session_is_ct (s);
1592   is_nonblocking = VCL_SESS_ATTR_TEST (s->attr, VCL_SESS_ATTR_NONBLOCK);
1593   mq = is_ct ? s->our_evt_q : wrk->app_event_queue;
1594   if (svm_fifo_is_full (tx_fifo))
1595     {
1596       if (is_nonblocking)
1597         {
1598           return VPPCOM_EWOULDBLOCK;
1599         }
1600       while (svm_fifo_is_full (tx_fifo))
1601         {
1602           svm_fifo_set_want_tx_evt (tx_fifo, 1);
1603           svm_msg_q_lock (mq);
1604           if (svm_msg_q_is_empty (mq))
1605             svm_msg_q_wait (mq);
1606
1607           svm_msg_q_sub_w_lock (mq, &msg);
1608           e = svm_msg_q_msg_data (mq, &msg);
1609           svm_msg_q_unlock (mq);
1610
1611           if (!vcl_is_tx_evt_for_session (e, s->session_index, is_ct))
1612             vcl_handle_mq_event (wrk, e);
1613           svm_msg_q_free_msg (mq, &msg);
1614         }
1615     }
1616
1617   ASSERT (FIFO_EVENT_APP_TX + 1 == SESSION_IO_EVT_CT_TX);
1618   et = FIFO_EVENT_APP_TX + vcl_session_is_ct (s);
1619   if (s->is_dgram)
1620     n_write = app_send_dgram_raw (tx_fifo, &s->transport,
1621                                   s->vpp_evt_q, buf, n, et, SVM_Q_WAIT);
1622   else
1623     n_write = app_send_stream_raw (tx_fifo, s->vpp_evt_q, buf, n, et,
1624                                    SVM_Q_WAIT);
1625
1626   ASSERT (n_write > 0);
1627
1628   VDBG (2, "VCL<%d>: vpp handle 0x%llx, sid %u: wrote %d bytes", getpid (),
1629         s->vpp_handle, session_handle, n_write);
1630
1631   return n_write;
1632 }
1633
1634 static vcl_session_t *
1635 vcl_ct_session_get_from_fifo (vcl_worker_t * wrk, svm_fifo_t * f, u8 type)
1636 {
1637   vcl_session_t *s;
1638   s = vcl_session_get (wrk, f->client_session_index);
1639   if (s)
1640     {
1641       /* rx fifo */
1642       if (type == 0 && s->rx_fifo == f)
1643         return s;
1644       /* tx fifo */
1645       if (type == 1 && s->tx_fifo == f)
1646         return s;
1647     }
1648   s = vcl_session_get (wrk, f->master_session_index);
1649   if (s)
1650     {
1651       if (type == 0 && s->rx_fifo == f)
1652         return s;
1653       if (type == 1 && s->tx_fifo == f)
1654         return s;
1655     }
1656   return 0;
1657 }
1658
1659 static inline int
1660 vppcom_session_write_ready (vcl_session_t * session)
1661 {
1662   /* Assumes caller has acquired spinlock: vcm->sessions_lockp */
1663   if (PREDICT_FALSE (session->is_vep))
1664     {
1665       clib_warning ("VCL<%d>: ERROR: vpp handle 0x%llx, sid %u: "
1666                     "cannot write to an epoll session!",
1667                     getpid (), session->vpp_handle, session->session_index);
1668       return VPPCOM_EBADFD;
1669     }
1670
1671   if (PREDICT_FALSE (session->session_state & STATE_LISTEN))
1672     {
1673       clib_warning ("VCL<%d>: ERROR: vpp handle 0x%llx, sid %u: "
1674                     "cannot write to a listen session!",
1675                     getpid (), session->vpp_handle, session->session_index);
1676       return VPPCOM_EBADFD;
1677     }
1678
1679   if (PREDICT_FALSE (!(session->session_state & STATE_OPEN)))
1680     {
1681       session_state_t state = session->session_state;
1682       int rv;
1683
1684       rv = ((state & STATE_DISCONNECT) ? VPPCOM_ECONNRESET : VPPCOM_ENOTCONN);
1685       clib_warning ("VCL<%d>: ERROR: vpp handle 0x%llx, sid %u: "
1686                     "session is not open! state 0x%x (%s), "
1687                     "returning %d (%s)", getpid (), session->vpp_handle,
1688                     session->session_index,
1689                     state, vppcom_session_state_str (state),
1690                     rv, vppcom_retval_str (rv));
1691       return rv;
1692     }
1693
1694   VDBG (3, "VCL<%d>: vpp handle 0x%llx, sid %u: peek %s (%p), ready = %d",
1695         getpid (), session->vpp_handle, session->session_index,
1696         session->tx_fifo, svm_fifo_max_enqueue (session->tx_fifo));
1697
1698   return svm_fifo_max_enqueue (session->tx_fifo);
1699 }
1700
1701 static inline int
1702 vcl_mq_dequeue_batch (vcl_worker_t * wrk, svm_msg_q_t * mq)
1703 {
1704   svm_msg_q_msg_t *msg;
1705   u32 n_msgs;
1706   int i;
1707
1708   n_msgs = svm_msg_q_size (mq);
1709   for (i = 0; i < n_msgs; i++)
1710     {
1711       vec_add2 (wrk->mq_msg_vector, msg, 1);
1712       svm_msg_q_sub_w_lock (mq, msg);
1713     }
1714   return n_msgs;
1715 }
1716
1717 #define vcl_fifo_rx_evt_valid_or_break(_fifo)                   \
1718 if (PREDICT_FALSE (svm_fifo_is_empty (_fifo)))                  \
1719   {                                                             \
1720     svm_fifo_unset_event (_fifo);                               \
1721     if (svm_fifo_is_empty (_fifo))                              \
1722       break;                                                    \
1723   }                                                             \
1724
1725 static void
1726 vcl_select_handle_mq_event (vcl_worker_t * wrk, session_event_t * e,
1727                             unsigned long n_bits, unsigned long *read_map,
1728                             unsigned long *write_map,
1729                             unsigned long *except_map, u32 * bits_set)
1730 {
1731   session_disconnected_msg_t *disconnected_msg;
1732   session_connected_msg_t *connected_msg;
1733   session_accepted_msg_t *accepted_msg;
1734   vcl_session_msg_t *vcl_msg;
1735   vcl_session_t *session;
1736   u64 handle;
1737   u32 sid;
1738
1739   switch (e->event_type)
1740     {
1741     case FIFO_EVENT_APP_RX:
1742       vcl_fifo_rx_evt_valid_or_break (e->fifo);
1743       sid = e->fifo->client_session_index;
1744       session = vcl_session_get (wrk, sid);
1745       if (!session)
1746         break;
1747       if (sid < n_bits && read_map)
1748         {
1749           clib_bitmap_set_no_check (read_map, sid, 1);
1750           *bits_set += 1;
1751         }
1752       break;
1753     case FIFO_EVENT_APP_TX:
1754       sid = e->fifo->client_session_index;
1755       session = vcl_session_get (wrk, sid);
1756       if (!session)
1757         break;
1758       if (sid < n_bits && write_map)
1759         {
1760           clib_bitmap_set_no_check (write_map, sid, 1);
1761           *bits_set += 1;
1762         }
1763       break;
1764     case SESSION_IO_EVT_CT_TX:
1765       vcl_fifo_rx_evt_valid_or_break (e->fifo);
1766       session = vcl_ct_session_get_from_fifo (wrk, e->fifo, 0);
1767       if (!session)
1768         break;
1769       sid = session->session_index;
1770       if (sid < n_bits && read_map)
1771         {
1772           clib_bitmap_set_no_check (read_map, sid, 1);
1773           *bits_set += 1;
1774         }
1775       break;
1776     case SESSION_IO_EVT_CT_RX:
1777       session = vcl_ct_session_get_from_fifo (wrk, e->fifo, 1);
1778       if (!session)
1779         break;
1780       sid = session->session_index;
1781       if (sid < n_bits && write_map)
1782         {
1783           clib_bitmap_set_no_check (write_map, sid, 1);
1784           *bits_set += 1;
1785         }
1786       break;
1787     case SESSION_CTRL_EVT_ACCEPTED:
1788       accepted_msg = (session_accepted_msg_t *) e->data;
1789       handle = accepted_msg->listener_handle;
1790       session = vcl_session_table_lookup_listener (wrk, handle);
1791       if (!session)
1792         {
1793           clib_warning ("VCL<%d>: ERROR: couldn't find listen session:"
1794                         "listener handle %llx", getpid (), handle);
1795           break;
1796         }
1797
1798       clib_fifo_add2 (session->accept_evts_fifo, vcl_msg);
1799       vcl_msg->accepted_msg = *accepted_msg;
1800       sid = session->session_index;
1801       if (sid < n_bits && read_map)
1802         {
1803           clib_bitmap_set_no_check (read_map, sid, 1);
1804           *bits_set += 1;
1805         }
1806       break;
1807     case SESSION_CTRL_EVT_CONNECTED:
1808       connected_msg = (session_connected_msg_t *) e->data;
1809       vcl_session_connected_handler (wrk, connected_msg);
1810       break;
1811     case SESSION_CTRL_EVT_DISCONNECTED:
1812       disconnected_msg = (session_disconnected_msg_t *) e->data;
1813       sid = vcl_session_index_from_vpp_handle (wrk, disconnected_msg->handle);
1814       if (sid < n_bits && except_map)
1815         {
1816           clib_bitmap_set_no_check (except_map, sid, 1);
1817           *bits_set += 1;
1818         }
1819       break;
1820     case SESSION_CTRL_EVT_RESET:
1821       sid = vcl_session_reset_handler (wrk, (session_reset_msg_t *) e->data);
1822       if (sid < n_bits && except_map)
1823         {
1824           clib_bitmap_set_no_check (except_map, sid, 1);
1825           *bits_set += 1;
1826         }
1827       break;
1828     default:
1829       clib_warning ("unhandled: %u", e->event_type);
1830       break;
1831     }
1832 }
1833
1834 static int
1835 vcl_select_handle_mq (vcl_worker_t * wrk, svm_msg_q_t * mq,
1836                       unsigned long n_bits, unsigned long *read_map,
1837                       unsigned long *write_map, unsigned long *except_map,
1838                       double time_to_wait, u32 * bits_set)
1839 {
1840   svm_msg_q_msg_t *msg;
1841   session_event_t *e;
1842   u32 i;
1843
1844   svm_msg_q_lock (mq);
1845   if (svm_msg_q_is_empty (mq))
1846     {
1847       if (*bits_set)
1848         {
1849           svm_msg_q_unlock (mq);
1850           return 0;
1851         }
1852
1853       if (!time_to_wait)
1854         {
1855           svm_msg_q_unlock (mq);
1856           return 0;
1857         }
1858       else if (time_to_wait < 0)
1859         {
1860           svm_msg_q_wait (mq);
1861         }
1862       else
1863         {
1864           if (svm_msg_q_timedwait (mq, time_to_wait))
1865             {
1866               svm_msg_q_unlock (mq);
1867               return 0;
1868             }
1869         }
1870     }
1871   vcl_mq_dequeue_batch (wrk, mq);
1872   svm_msg_q_unlock (mq);
1873
1874   for (i = 0; i < vec_len (wrk->mq_msg_vector); i++)
1875     {
1876       msg = vec_elt_at_index (wrk->mq_msg_vector, i);
1877       e = svm_msg_q_msg_data (mq, msg);
1878       vcl_select_handle_mq_event (wrk, e, n_bits, read_map, write_map,
1879                                   except_map, bits_set);
1880       svm_msg_q_free_msg (mq, msg);
1881     }
1882   vec_reset_length (wrk->mq_msg_vector);
1883   return *bits_set;
1884 }
1885
1886 static int
1887 vppcom_select_condvar (vcl_worker_t * wrk, unsigned long n_bits,
1888                        unsigned long *read_map, unsigned long *write_map,
1889                        unsigned long *except_map, double time_to_wait,
1890                        u32 * bits_set)
1891 {
1892   double total_wait = 0, wait_slice;
1893   vcl_cut_through_registration_t *cr;
1894
1895   time_to_wait = (time_to_wait == -1) ? 10e9 : time_to_wait;
1896   wait_slice = wrk->cut_through_registrations ? 10e-6 : time_to_wait;
1897   do
1898     {
1899       vcl_ct_registration_lock (wrk);
1900       /* *INDENT-OFF* */
1901       pool_foreach (cr, wrk->cut_through_registrations, ({
1902         vcl_select_handle_mq (wrk, cr->mq, n_bits, read_map, write_map, except_map,
1903                               0, bits_set);
1904       }));
1905       /* *INDENT-ON* */
1906       vcl_ct_registration_unlock (wrk);
1907
1908       vcl_select_handle_mq (wrk, wrk->app_event_queue, n_bits, read_map,
1909                             write_map, except_map, time_to_wait, bits_set);
1910       total_wait += wait_slice;
1911       if (*bits_set)
1912         return *bits_set;
1913     }
1914   while (total_wait < time_to_wait);
1915
1916   return 0;
1917 }
1918
1919 static int
1920 vppcom_select_eventfd (vcl_worker_t * wrk, unsigned long n_bits,
1921                        unsigned long *read_map, unsigned long *write_map,
1922                        unsigned long *except_map, double time_to_wait,
1923                        u32 * bits_set)
1924 {
1925   vcl_mq_evt_conn_t *mqc;
1926   int __clib_unused n_read;
1927   int n_mq_evts, i;
1928   u64 buf;
1929
1930   vec_validate (wrk->mq_events, pool_elts (wrk->mq_evt_conns));
1931   n_mq_evts = epoll_wait (wrk->mqs_epfd, wrk->mq_events,
1932                           vec_len (wrk->mq_events), time_to_wait);
1933   for (i = 0; i < n_mq_evts; i++)
1934     {
1935       mqc = vcl_mq_evt_conn_get (wrk, wrk->mq_events[i].data.u32);
1936       n_read = read (mqc->mq_fd, &buf, sizeof (buf));
1937       vcl_select_handle_mq (wrk, mqc->mq, n_bits, read_map, write_map,
1938                             except_map, 0, bits_set);
1939     }
1940
1941   return (n_mq_evts > 0 ? (int) *bits_set : 0);
1942 }
1943
1944 int
1945 vppcom_select (unsigned long n_bits, unsigned long *read_map,
1946                unsigned long *write_map, unsigned long *except_map,
1947                double time_to_wait)
1948 {
1949   u32 sid, minbits = clib_max (n_bits, BITS (uword)), bits_set = 0;
1950   vcl_worker_t *wrk = vcl_worker_get_current ();
1951   vcl_session_t *session = 0;
1952   int rv, i;
1953
1954   ASSERT (sizeof (clib_bitmap_t) == sizeof (long int));
1955
1956   if (n_bits && read_map)
1957     {
1958       clib_bitmap_validate (wrk->rd_bitmap, minbits);
1959       clib_memcpy_fast (wrk->rd_bitmap, read_map,
1960                         vec_len (wrk->rd_bitmap) * sizeof (clib_bitmap_t));
1961       memset (read_map, 0, vec_len (wrk->rd_bitmap) * sizeof (clib_bitmap_t));
1962     }
1963   if (n_bits && write_map)
1964     {
1965       clib_bitmap_validate (wrk->wr_bitmap, minbits);
1966       clib_memcpy_fast (wrk->wr_bitmap, write_map,
1967                         vec_len (wrk->wr_bitmap) * sizeof (clib_bitmap_t));
1968       memset (write_map, 0,
1969               vec_len (wrk->wr_bitmap) * sizeof (clib_bitmap_t));
1970     }
1971   if (n_bits && except_map)
1972     {
1973       clib_bitmap_validate (wrk->ex_bitmap, minbits);
1974       clib_memcpy_fast (wrk->ex_bitmap, except_map,
1975                         vec_len (wrk->ex_bitmap) * sizeof (clib_bitmap_t));
1976       memset (except_map, 0,
1977               vec_len (wrk->ex_bitmap) * sizeof (clib_bitmap_t));
1978     }
1979
1980   if (!n_bits)
1981     return 0;
1982
1983   if (!write_map)
1984     goto check_rd;
1985
1986   /* *INDENT-OFF* */
1987   clib_bitmap_foreach (sid, wrk->wr_bitmap, ({
1988     if (!(session = vcl_session_get (wrk, sid)))
1989       {
1990         VDBG (0, "VCL<%d>: session %d specified in write_map is closed.",
1991               getpid (), sid);
1992         return VPPCOM_EBADFD;
1993       }
1994
1995     rv = svm_fifo_is_full (session->tx_fifo);
1996     if (!rv)
1997       {
1998         clib_bitmap_set_no_check (write_map, sid, 1);
1999         bits_set++;
2000       }
2001   }));
2002
2003 check_rd:
2004   if (!read_map)
2005     goto check_mq;
2006
2007   clib_bitmap_foreach (sid, wrk->rd_bitmap, ({
2008     if (!(session = vcl_session_get (wrk, sid)))
2009       {
2010         VDBG (0, "VCL<%d>: session %d specified in write_map is closed.",
2011               getpid (), sid);
2012         return VPPCOM_EBADFD;
2013       }
2014
2015     rv = vppcom_session_read_ready (session);
2016     if (rv)
2017       {
2018         clib_bitmap_set_no_check (read_map, sid, 1);
2019         bits_set++;
2020       }
2021   }));
2022   /* *INDENT-ON* */
2023
2024 check_mq:
2025
2026   for (i = 0; i < vec_len (wrk->unhandled_evts_vector); i++)
2027     {
2028       vcl_select_handle_mq_event (wrk, &wrk->unhandled_evts_vector[i], n_bits,
2029                                   read_map, write_map, except_map, &bits_set);
2030     }
2031   vec_reset_length (wrk->unhandled_evts_vector);
2032
2033   if (vcm->cfg.use_mq_eventfd)
2034     vppcom_select_eventfd (wrk, n_bits, read_map, write_map, except_map,
2035                            time_to_wait, &bits_set);
2036   else
2037     vppcom_select_condvar (wrk, n_bits, read_map, write_map, except_map,
2038                            time_to_wait, &bits_set);
2039
2040   return (bits_set);
2041 }
2042
2043 static inline void
2044 vep_verify_epoll_chain (vcl_worker_t * wrk, u32 vep_idx)
2045 {
2046   vcl_session_t *session;
2047   vppcom_epoll_t *vep;
2048   u32 sid = vep_idx;
2049
2050   if (VPPCOM_DEBUG <= 1)
2051     return;
2052
2053   /* Assumes caller has acquired spinlock: vcm->sessions_lockp */
2054   session = vcl_session_get (wrk, vep_idx);
2055   if (PREDICT_FALSE (!session))
2056     {
2057       clib_warning ("VCL<%d>: ERROR: Invalid vep_idx (%u)!",
2058                     getpid (), vep_idx);
2059       goto done;
2060     }
2061   if (PREDICT_FALSE (!session->is_vep))
2062     {
2063       clib_warning ("VCL<%d>: ERROR: vep_idx (%u) is not a vep!",
2064                     getpid (), vep_idx);
2065       goto done;
2066     }
2067   vep = &session->vep;
2068   clib_warning ("VCL<%d>: vep_idx (%u): Dumping epoll chain\n"
2069                 "{\n"
2070                 "   is_vep         = %u\n"
2071                 "   is_vep_session = %u\n"
2072                 "   next_sid       = 0x%x (%u)\n"
2073                 "   wait_cont_idx  = 0x%x (%u)\n"
2074                 "}\n", getpid (), vep_idx,
2075                 session->is_vep, session->is_vep_session,
2076                 vep->next_sh, vep->next_sh,
2077                 session->wait_cont_idx, session->wait_cont_idx);
2078
2079   for (sid = vep->next_sh; sid != ~0; sid = vep->next_sh)
2080     {
2081       session = vcl_session_get (wrk, sid);
2082       if (PREDICT_FALSE (!session))
2083         {
2084           clib_warning ("VCL<%d>: ERROR: Invalid sid (%u)!", getpid (), sid);
2085           goto done;
2086         }
2087       if (PREDICT_FALSE (session->is_vep))
2088         clib_warning ("VCL<%d>: ERROR: sid (%u) is a vep!",
2089                       getpid (), vep_idx);
2090       else if (PREDICT_FALSE (!session->is_vep_session))
2091         {
2092           clib_warning ("VCL<%d>: ERROR: session (%u) "
2093                         "is not a vep session!", getpid (), sid);
2094           goto done;
2095         }
2096       vep = &session->vep;
2097       if (PREDICT_FALSE (vep->vep_sh != vep_idx))
2098         clib_warning ("VCL<%d>: ERROR: session (%u) vep_idx (%u) != "
2099                       "vep_idx (%u)!", getpid (),
2100                       sid, session->vep.vep_sh, vep_idx);
2101       if (session->is_vep_session)
2102         {
2103           clib_warning ("vep_idx[%u]: sid 0x%x (%u)\n"
2104                         "{\n"
2105                         "   next_sid       = 0x%x (%u)\n"
2106                         "   prev_sid       = 0x%x (%u)\n"
2107                         "   vep_idx        = 0x%x (%u)\n"
2108                         "   ev.events      = 0x%x\n"
2109                         "   ev.data.u64    = 0x%llx\n"
2110                         "   et_mask        = 0x%x\n"
2111                         "}\n",
2112                         vep_idx, sid, sid,
2113                         vep->next_sh, vep->next_sh,
2114                         vep->prev_sh, vep->prev_sh,
2115                         vep->vep_sh, vep->vep_sh,
2116                         vep->ev.events, vep->ev.data.u64, vep->et_mask);
2117         }
2118     }
2119
2120 done:
2121   clib_warning ("VCL<%d>: vep_idx (%u): Dump complete!\n",
2122                 getpid (), vep_idx);
2123 }
2124
2125 int
2126 vppcom_epoll_create (void)
2127 {
2128   vcl_worker_t *wrk = vcl_worker_get_current ();
2129   vcl_session_t *vep_session;
2130
2131   vep_session = vcl_session_alloc (wrk);
2132
2133   vep_session->is_vep = 1;
2134   vep_session->vep.vep_sh = ~0;
2135   vep_session->vep.next_sh = ~0;
2136   vep_session->vep.prev_sh = ~0;
2137   vep_session->wait_cont_idx = ~0;
2138   vep_session->vpp_handle = ~0;
2139
2140   vcl_evt (VCL_EVT_EPOLL_CREATE, vep_session, vep_sh);
2141   VDBG (0, "VCL<%d>: Created vep_idx %u / sid %u!",
2142         getpid (), vep_session->session_index, vep_session->session_index);
2143
2144   return vcl_session_handle (vep_session);
2145 }
2146
2147 int
2148 vppcom_epoll_ctl (uint32_t vep_handle, int op, uint32_t session_handle,
2149                   struct epoll_event *event)
2150 {
2151   vcl_worker_t *wrk = vcl_worker_get_current ();
2152   vcl_session_t *vep_session;
2153   vcl_session_t *session;
2154   int rv = VPPCOM_OK;
2155
2156   if (vep_handle == session_handle)
2157     {
2158       clib_warning ("VCL<%d>: ERROR: vep_idx == session_index (%u)!",
2159                     getpid (), vep_handle);
2160       return VPPCOM_EINVAL;
2161     }
2162
2163   vep_session = vcl_session_get_w_handle (wrk, vep_handle);
2164   if (PREDICT_FALSE (!vep_session))
2165     {
2166       clib_warning ("VCL<%d>: ERROR: Invalid vep_idx (%u)!", vep_handle);
2167       return VPPCOM_EBADFD;
2168     }
2169   if (PREDICT_FALSE (!vep_session->is_vep))
2170     {
2171       clib_warning ("VCL<%d>: ERROR: vep_idx (%u) is not a vep!",
2172                     getpid (), vep_handle);
2173       return VPPCOM_EINVAL;
2174     }
2175
2176   ASSERT (vep_session->vep.vep_sh == ~0);
2177   ASSERT (vep_session->vep.prev_sh == ~0);
2178
2179   session = vcl_session_get_w_handle (wrk, session_handle);
2180   if (PREDICT_FALSE (!session))
2181     {
2182       VDBG (0, "VCL<%d>: ERROR: Invalid session_handle (%u)!",
2183             getpid (), session_handle);
2184       return VPPCOM_EBADFD;
2185     }
2186   if (PREDICT_FALSE (session->is_vep))
2187     {
2188       clib_warning ("ERROR: session_handle (%u) is a vep!", vep_handle);
2189       return VPPCOM_EINVAL;
2190     }
2191
2192   switch (op)
2193     {
2194     case EPOLL_CTL_ADD:
2195       if (PREDICT_FALSE (!event))
2196         {
2197           clib_warning ("VCL<%d>: ERROR: EPOLL_CTL_ADD: NULL pointer to "
2198                         "epoll_event structure!", getpid ());
2199           return VPPCOM_EINVAL;
2200         }
2201       if (vep_session->vep.next_sh != ~0)
2202         {
2203           vcl_session_t *next_session;
2204           next_session = vcl_session_get_w_handle (wrk,
2205                                                    vep_session->vep.next_sh);
2206           if (PREDICT_FALSE (!next_session))
2207             {
2208               clib_warning ("VCL<%d>: ERROR: EPOLL_CTL_ADD: Invalid "
2209                             "vep.next_sid (%u) on vep_idx (%u)!",
2210                             getpid (), vep_session->vep.next_sh, vep_handle);
2211               return VPPCOM_EBADFD;
2212             }
2213           ASSERT (next_session->vep.prev_sh == vep_handle);
2214           next_session->vep.prev_sh = session_handle;
2215         }
2216       session->vep.next_sh = vep_session->vep.next_sh;
2217       session->vep.prev_sh = vep_handle;
2218       session->vep.vep_sh = vep_handle;
2219       session->vep.et_mask = VEP_DEFAULT_ET_MASK;
2220       session->vep.ev = *event;
2221       session->is_vep = 0;
2222       session->is_vep_session = 1;
2223       vep_session->vep.next_sh = session_handle;
2224
2225       VDBG (1, "VCL<%d>: EPOLL_CTL_ADD: vep_idx %u, sid %u, events 0x%x, "
2226             "data 0x%llx!", getpid (), vep_handle, session_handle,
2227             event->events, event->data.u64);
2228       vcl_evt (VCL_EVT_EPOLL_CTLADD, session, event->events, event->data.u64);
2229       break;
2230
2231     case EPOLL_CTL_MOD:
2232       if (PREDICT_FALSE (!event))
2233         {
2234           clib_warning ("VCL<%d>: ERROR: EPOLL_CTL_MOD: NULL pointer to "
2235                         "epoll_event structure!", getpid ());
2236           rv = VPPCOM_EINVAL;
2237           goto done;
2238         }
2239       else if (PREDICT_FALSE (!session->is_vep_session))
2240         {
2241           clib_warning ("VCL<%d>: ERROR: sid %u EPOLL_CTL_MOD: "
2242                         "not a vep session!", getpid (), session_handle);
2243           rv = VPPCOM_EINVAL;
2244           goto done;
2245         }
2246       else if (PREDICT_FALSE (session->vep.vep_sh != vep_handle))
2247         {
2248           clib_warning ("VCL<%d>: ERROR: sid %u EPOLL_CTL_MOD: "
2249                         "vep_idx (%u) != vep_idx (%u)!",
2250                         getpid (), session_handle,
2251                         session->vep.vep_sh, vep_handle);
2252           rv = VPPCOM_EINVAL;
2253           goto done;
2254         }
2255       session->vep.et_mask = VEP_DEFAULT_ET_MASK;
2256       session->vep.ev = *event;
2257       VDBG (1, "VCL<%d>: EPOLL_CTL_MOD: vep_idx %u, sid %u, events 0x%x,"
2258             " data 0x%llx!", getpid (), vep_handle, session_handle,
2259             event->events, event->data.u64);
2260       break;
2261
2262     case EPOLL_CTL_DEL:
2263       if (PREDICT_FALSE (!session->is_vep_session))
2264         {
2265           clib_warning ("VCL<%d>: ERROR: sid %u EPOLL_CTL_DEL: "
2266                         "not a vep session!", getpid (), session_handle);
2267           rv = VPPCOM_EINVAL;
2268           goto done;
2269         }
2270       else if (PREDICT_FALSE (session->vep.vep_sh != vep_handle))
2271         {
2272           clib_warning ("VCL<%d>: ERROR: sid %u EPOLL_CTL_DEL: "
2273                         "vep_idx (%u) != vep_idx (%u)!",
2274                         getpid (), session_handle,
2275                         session->vep.vep_sh, vep_handle);
2276           rv = VPPCOM_EINVAL;
2277           goto done;
2278         }
2279
2280       vep_session->wait_cont_idx =
2281         (vep_session->wait_cont_idx == session_handle) ?
2282         session->vep.next_sh : vep_session->wait_cont_idx;
2283
2284       if (session->vep.prev_sh == vep_handle)
2285         vep_session->vep.next_sh = session->vep.next_sh;
2286       else
2287         {
2288           vcl_session_t *prev_session;
2289           prev_session = vcl_session_get_w_handle (wrk, session->vep.prev_sh);
2290           if (PREDICT_FALSE (!prev_session))
2291             {
2292               clib_warning ("VCL<%d>: ERROR: EPOLL_CTL_DEL: Invalid "
2293                             "vep.prev_sid (%u) on sid (%u)!",
2294                             getpid (), session->vep.prev_sh, session_handle);
2295               return VPPCOM_EBADFD;
2296             }
2297           ASSERT (prev_session->vep.next_sh == session_handle);
2298           prev_session->vep.next_sh = session->vep.next_sh;
2299         }
2300       if (session->vep.next_sh != ~0)
2301         {
2302           vcl_session_t *next_session;
2303           next_session = vcl_session_get_w_handle (wrk, session->vep.next_sh);
2304           if (PREDICT_FALSE (!next_session))
2305             {
2306               clib_warning ("VCL<%d>: ERROR: EPOLL_CTL_DEL: Invalid "
2307                             "vep.next_sid (%u) on sid (%u)!",
2308                             getpid (), session->vep.next_sh, session_handle);
2309               return VPPCOM_EBADFD;
2310             }
2311           ASSERT (next_session->vep.prev_sh == session_handle);
2312           next_session->vep.prev_sh = session->vep.prev_sh;
2313         }
2314
2315       memset (&session->vep, 0, sizeof (session->vep));
2316       session->vep.next_sh = ~0;
2317       session->vep.prev_sh = ~0;
2318       session->vep.vep_sh = ~0;
2319       session->is_vep_session = 0;
2320       VDBG (1, "VCL<%d>: EPOLL_CTL_DEL: vep_idx %u, sid %u!",
2321             getpid (), vep_handle, session_handle);
2322       vcl_evt (VCL_EVT_EPOLL_CTLDEL, session, vep_sh);
2323       break;
2324
2325     default:
2326       clib_warning ("VCL<%d>: ERROR: Invalid operation (%d)!", getpid (), op);
2327       rv = VPPCOM_EINVAL;
2328     }
2329
2330   vep_verify_epoll_chain (wrk, vep_handle);
2331
2332 done:
2333   return rv;
2334 }
2335
2336 static inline void
2337 vcl_epoll_wait_handle_mq_event (vcl_worker_t * wrk, session_event_t * e,
2338                                 struct epoll_event *events, u32 * num_ev)
2339 {
2340   session_disconnected_msg_t *disconnected_msg;
2341   session_connected_msg_t *connected_msg;
2342   session_accepted_msg_t *accepted_msg;
2343   u64 session_evt_data = ~0, handle;
2344   u32 sid = ~0, session_events;
2345   vcl_session_msg_t *vcl_msg;
2346   vcl_session_t *session;
2347   u8 add_event = 0;
2348
2349   switch (e->event_type)
2350     {
2351     case FIFO_EVENT_APP_RX:
2352       ASSERT (e->fifo->client_thread_index == vcl_get_worker_index ());
2353       vcl_fifo_rx_evt_valid_or_break (e->fifo);
2354       sid = e->fifo->client_session_index;
2355       session = vcl_session_get (wrk, sid);
2356       session_events = session->vep.ev.events;
2357       if (!(EPOLLIN & session->vep.ev.events) || session->has_rx_evt)
2358         break;
2359       add_event = 1;
2360       events[*num_ev].events |= EPOLLIN;
2361       session_evt_data = session->vep.ev.data.u64;
2362       session->has_rx_evt = 1;
2363       break;
2364     case FIFO_EVENT_APP_TX:
2365       sid = e->fifo->client_session_index;
2366       session = vcl_session_get (wrk, sid);
2367       session_events = session->vep.ev.events;
2368       if (!(EPOLLOUT & session_events))
2369         break;
2370       add_event = 1;
2371       events[*num_ev].events |= EPOLLOUT;
2372       session_evt_data = session->vep.ev.data.u64;
2373       break;
2374     case SESSION_IO_EVT_CT_TX:
2375       vcl_fifo_rx_evt_valid_or_break (e->fifo);
2376       session = vcl_ct_session_get_from_fifo (wrk, e->fifo, 0);
2377       sid = session->session_index;
2378       session_events = session->vep.ev.events;
2379       if (!(EPOLLIN & session->vep.ev.events) || session->has_rx_evt)
2380         break;
2381       add_event = 1;
2382       events[*num_ev].events |= EPOLLIN;
2383       session_evt_data = session->vep.ev.data.u64;
2384       session->has_rx_evt = 1;
2385       break;
2386     case SESSION_IO_EVT_CT_RX:
2387       session = vcl_ct_session_get_from_fifo (wrk, e->fifo, 1);
2388       sid = session->session_index;
2389       session_events = session->vep.ev.events;
2390       if (!(EPOLLOUT & session_events))
2391         break;
2392       add_event = 1;
2393       events[*num_ev].events |= EPOLLOUT;
2394       session_evt_data = session->vep.ev.data.u64;
2395       break;
2396     case SESSION_CTRL_EVT_ACCEPTED:
2397       accepted_msg = (session_accepted_msg_t *) e->data;
2398       handle = accepted_msg->listener_handle;
2399       session = vcl_session_table_lookup_listener (wrk, handle);
2400       if (!session)
2401         {
2402           clib_warning ("VCL<%d>: ERROR: couldn't find listen session:"
2403                         "listener handle %llx", getpid (), handle);
2404           break;
2405         }
2406
2407       clib_fifo_add2 (session->accept_evts_fifo, vcl_msg);
2408       vcl_msg->accepted_msg = *accepted_msg;
2409       session_events = session->vep.ev.events;
2410       if (!(EPOLLIN & session_events))
2411         break;
2412
2413       add_event = 1;
2414       events[*num_ev].events |= EPOLLIN;
2415       session_evt_data = session->vep.ev.data.u64;
2416       break;
2417     case SESSION_CTRL_EVT_CONNECTED:
2418       connected_msg = (session_connected_msg_t *) e->data;
2419       vcl_session_connected_handler (wrk, connected_msg);
2420       /* Generate EPOLLOUT because there's no connected event */
2421       sid = vcl_session_index_from_vpp_handle (wrk, connected_msg->handle);
2422       session = vcl_session_get (wrk, sid);
2423       session_events = session->vep.ev.events;
2424       if (EPOLLOUT & session_events)
2425         {
2426           add_event = 1;
2427           events[*num_ev].events |= EPOLLOUT;
2428           session_evt_data = session->vep.ev.data.u64;
2429         }
2430       break;
2431     case SESSION_CTRL_EVT_DISCONNECTED:
2432       disconnected_msg = (session_disconnected_msg_t *) e->data;
2433       sid = vcl_session_index_from_vpp_handle (wrk, disconnected_msg->handle);
2434       if (!(session = vcl_session_get (wrk, sid)))
2435         break;
2436       add_event = 1;
2437       events[*num_ev].events |= EPOLLHUP | EPOLLRDHUP;
2438       session_evt_data = session->vep.ev.data.u64;
2439       session_events = session->vep.ev.events;
2440       break;
2441     case SESSION_CTRL_EVT_RESET:
2442       sid = vcl_session_reset_handler (wrk, (session_reset_msg_t *) e->data);
2443       if (!(session = vcl_session_get (wrk, sid)))
2444         break;
2445       add_event = 1;
2446       events[*num_ev].events |= EPOLLHUP | EPOLLRDHUP;
2447       session_evt_data = session->vep.ev.data.u64;
2448       session_events = session->vep.ev.events;
2449       break;
2450     default:
2451       VDBG (0, "unhandled: %u", e->event_type);
2452       break;
2453     }
2454
2455   if (add_event)
2456     {
2457       events[*num_ev].data.u64 = session_evt_data;
2458       if (EPOLLONESHOT & session_events)
2459         {
2460           session = vcl_session_get (wrk, sid);
2461           session->vep.ev.events = 0;
2462         }
2463       *num_ev += 1;
2464     }
2465 }
2466
2467 static int
2468 vcl_epoll_wait_handle_mq (vcl_worker_t * wrk, svm_msg_q_t * mq,
2469                           struct epoll_event *events, u32 maxevents,
2470                           double wait_for_time, u32 * num_ev)
2471 {
2472   svm_msg_q_msg_t *msg;
2473   session_event_t *e;
2474   int i;
2475
2476   if (vec_len (wrk->mq_msg_vector) && svm_msg_q_is_empty (mq))
2477     goto handle_dequeued;
2478
2479   svm_msg_q_lock (mq);
2480   if (svm_msg_q_is_empty (mq))
2481     {
2482       if (!wait_for_time)
2483         {
2484           svm_msg_q_unlock (mq);
2485           return 0;
2486         }
2487       else if (wait_for_time < 0)
2488         {
2489           svm_msg_q_wait (mq);
2490         }
2491       else
2492         {
2493           if (svm_msg_q_timedwait (mq, wait_for_time / 1e3))
2494             {
2495               svm_msg_q_unlock (mq);
2496               return 0;
2497             }
2498         }
2499     }
2500   vcl_mq_dequeue_batch (wrk, mq);
2501   svm_msg_q_unlock (mq);
2502
2503 handle_dequeued:
2504   for (i = 0; i < vec_len (wrk->mq_msg_vector); i++)
2505     {
2506       msg = vec_elt_at_index (wrk->mq_msg_vector, i);
2507       e = svm_msg_q_msg_data (mq, msg);
2508       if (*num_ev < maxevents)
2509         vcl_epoll_wait_handle_mq_event (wrk, e, events, num_ev);
2510       else
2511         vec_add1 (wrk->unhandled_evts_vector, *e);
2512       svm_msg_q_free_msg (mq, msg);
2513     }
2514   vec_reset_length (wrk->mq_msg_vector);
2515
2516   return *num_ev;
2517 }
2518
2519 static int
2520 vppcom_epoll_wait_condvar (vcl_worker_t * wrk, struct epoll_event *events,
2521                            int maxevents, u32 n_evts, double wait_for_time)
2522 {
2523   vcl_cut_through_registration_t *cr;
2524   double total_wait = 0, wait_slice;
2525   int rv;
2526
2527   wait_for_time = (wait_for_time == -1) ? (double) 10e9 : wait_for_time;
2528   wait_slice = wrk->cut_through_registrations ? 10e-6 : wait_for_time;
2529
2530   do
2531     {
2532       vcl_ct_registration_lock (wrk);
2533       /* *INDENT-OFF* */
2534       pool_foreach (cr, wrk->cut_through_registrations, ({
2535         vcl_epoll_wait_handle_mq (wrk, cr->mq, events, maxevents, 0, &n_evts);
2536       }));
2537       /* *INDENT-ON* */
2538       vcl_ct_registration_unlock (wrk);
2539
2540       rv = vcl_epoll_wait_handle_mq (wrk, wrk->app_event_queue, events,
2541                                      maxevents, n_evts ? 0 : wait_slice,
2542                                      &n_evts);
2543       if (rv)
2544         total_wait += wait_slice;
2545       if (n_evts)
2546         return n_evts;
2547     }
2548   while (total_wait < wait_for_time);
2549   return n_evts;
2550 }
2551
2552 static int
2553 vppcom_epoll_wait_eventfd (vcl_worker_t * wrk, struct epoll_event *events,
2554                            int maxevents, u32 n_evts, double wait_for_time)
2555 {
2556   vcl_mq_evt_conn_t *mqc;
2557   int __clib_unused n_read;
2558   int n_mq_evts, i;
2559   u64 buf;
2560
2561   vec_validate (wrk->mq_events, pool_elts (wrk->mq_evt_conns));
2562 again:
2563   n_mq_evts = epoll_wait (wrk->mqs_epfd, wrk->mq_events,
2564                           vec_len (wrk->mq_events), wait_for_time);
2565   for (i = 0; i < n_mq_evts; i++)
2566     {
2567       mqc = vcl_mq_evt_conn_get (wrk, wrk->mq_events[i].data.u32);
2568       n_read = read (mqc->mq_fd, &buf, sizeof (buf));
2569       vcl_epoll_wait_handle_mq (wrk, mqc->mq, events, maxevents, 0, &n_evts);
2570     }
2571   if (!n_evts && n_mq_evts > 0)
2572     goto again;
2573
2574   return (int) n_evts;
2575 }
2576
2577 int
2578 vppcom_epoll_wait (uint32_t vep_handle, struct epoll_event *events,
2579                    int maxevents, double wait_for_time)
2580 {
2581   vcl_worker_t *wrk = vcl_worker_get_current ();
2582   vcl_session_t *vep_session;
2583   u32 n_evts = 0;
2584   int i;
2585
2586   if (PREDICT_FALSE (maxevents <= 0))
2587     {
2588       clib_warning ("VCL<%d>: ERROR: Invalid maxevents (%d)!",
2589                     getpid (), maxevents);
2590       return VPPCOM_EINVAL;
2591     }
2592
2593   vep_session = vcl_session_get_w_handle (wrk, vep_handle);
2594   if (!vep_session)
2595     return VPPCOM_EBADFD;
2596
2597   if (PREDICT_FALSE (!vep_session->is_vep))
2598     {
2599       clib_warning ("VCL<%d>: ERROR: vep_idx (%u) is not a vep!",
2600                     getpid (), vep_handle);
2601       return VPPCOM_EINVAL;
2602     }
2603
2604   memset (events, 0, sizeof (*events) * maxevents);
2605
2606   if (vec_len (wrk->unhandled_evts_vector))
2607     {
2608       for (i = 0; i < vec_len (wrk->unhandled_evts_vector); i++)
2609         {
2610           vcl_epoll_wait_handle_mq_event (wrk, &wrk->unhandled_evts_vector[i],
2611                                           events, &n_evts);
2612           if (n_evts == maxevents)
2613             {
2614               i += 1;
2615               break;
2616             }
2617         }
2618
2619       vec_delete (wrk->unhandled_evts_vector, i, 0);
2620     }
2621
2622   if (vcm->cfg.use_mq_eventfd)
2623     return vppcom_epoll_wait_eventfd (wrk, events, maxevents, n_evts,
2624                                       wait_for_time);
2625
2626   return vppcom_epoll_wait_condvar (wrk, events, maxevents, n_evts,
2627                                     wait_for_time);
2628 }
2629
2630 int
2631 vppcom_session_attr (uint32_t session_handle, uint32_t op,
2632                      void *buffer, uint32_t * buflen)
2633 {
2634   vcl_worker_t *wrk = vcl_worker_get_current ();
2635   vcl_session_t *session;
2636   int rv = VPPCOM_OK;
2637   u32 *flags = buffer;
2638   vppcom_endpt_t *ep = buffer;
2639
2640   session = vcl_session_get_w_handle (wrk, session_handle);
2641   if (!session)
2642     return VPPCOM_EBADFD;
2643
2644   switch (op)
2645     {
2646     case VPPCOM_ATTR_GET_NREAD:
2647       rv = vppcom_session_read_ready (session);
2648       VDBG (2, "VCL<%d>: VPPCOM_ATTR_GET_NREAD: sid %u, nread = %d",
2649             getpid (), rv);
2650       break;
2651
2652     case VPPCOM_ATTR_GET_NWRITE:
2653       rv = vppcom_session_write_ready (session);
2654       VDBG (2, "VCL<%d>: VPPCOM_ATTR_GET_NWRITE: sid %u, nwrite = %d",
2655             getpid (), session_handle, rv);
2656       break;
2657
2658     case VPPCOM_ATTR_GET_FLAGS:
2659       if (PREDICT_TRUE (buffer && buflen && (*buflen >= sizeof (*flags))))
2660         {
2661           *flags = O_RDWR | (VCL_SESS_ATTR_TEST (session->attr,
2662                                                  VCL_SESS_ATTR_NONBLOCK));
2663           *buflen = sizeof (*flags);
2664           VDBG (2, "VCL<%d>: VPPCOM_ATTR_GET_FLAGS: sid %u, flags = 0x%08x, "
2665                 "is_nonblocking = %u", getpid (),
2666                 session_handle, *flags,
2667                 VCL_SESS_ATTR_TEST (session->attr, VCL_SESS_ATTR_NONBLOCK));
2668         }
2669       else
2670         rv = VPPCOM_EINVAL;
2671       break;
2672
2673     case VPPCOM_ATTR_SET_FLAGS:
2674       if (PREDICT_TRUE (buffer && buflen && (*buflen == sizeof (*flags))))
2675         {
2676           if (*flags & O_NONBLOCK)
2677             VCL_SESS_ATTR_SET (session->attr, VCL_SESS_ATTR_NONBLOCK);
2678           else
2679             VCL_SESS_ATTR_CLR (session->attr, VCL_SESS_ATTR_NONBLOCK);
2680
2681           VDBG (2, "VCL<%d>: VPPCOM_ATTR_SET_FLAGS: sid %u, flags = 0x%08x,"
2682                 " is_nonblocking = %u",
2683                 getpid (), session_handle, *flags,
2684                 VCL_SESS_ATTR_TEST (session->attr, VCL_SESS_ATTR_NONBLOCK));
2685         }
2686       else
2687         rv = VPPCOM_EINVAL;
2688       break;
2689
2690     case VPPCOM_ATTR_GET_PEER_ADDR:
2691       if (PREDICT_TRUE (buffer && buflen &&
2692                         (*buflen >= sizeof (*ep)) && ep->ip))
2693         {
2694           ep->is_ip4 = session->transport.is_ip4;
2695           ep->port = session->transport.rmt_port;
2696           if (session->transport.is_ip4)
2697             clib_memcpy_fast (ep->ip, &session->transport.rmt_ip.ip4,
2698                               sizeof (ip4_address_t));
2699           else
2700             clib_memcpy_fast (ep->ip, &session->transport.rmt_ip.ip6,
2701                               sizeof (ip6_address_t));
2702           *buflen = sizeof (*ep);
2703           VDBG (1, "VCL<%d>: VPPCOM_ATTR_GET_PEER_ADDR: sid %u, is_ip4 = %u, "
2704                 "addr = %U, port %u", getpid (),
2705                 session_handle, ep->is_ip4, format_ip46_address,
2706                 &session->transport.rmt_ip,
2707                 ep->is_ip4 ? IP46_TYPE_IP4 : IP46_TYPE_IP6,
2708                 clib_net_to_host_u16 (ep->port));
2709         }
2710       else
2711         rv = VPPCOM_EINVAL;
2712       break;
2713
2714     case VPPCOM_ATTR_GET_LCL_ADDR:
2715       if (PREDICT_TRUE (buffer && buflen &&
2716                         (*buflen >= sizeof (*ep)) && ep->ip))
2717         {
2718           ep->is_ip4 = session->transport.is_ip4;
2719           ep->port = session->transport.lcl_port;
2720           if (session->transport.is_ip4)
2721             clib_memcpy_fast (ep->ip, &session->transport.lcl_ip.ip4,
2722                               sizeof (ip4_address_t));
2723           else
2724             clib_memcpy_fast (ep->ip, &session->transport.lcl_ip.ip6,
2725                               sizeof (ip6_address_t));
2726           *buflen = sizeof (*ep);
2727           VDBG (1, "VCL<%d>: VPPCOM_ATTR_GET_LCL_ADDR: sid %u, is_ip4 = %u,"
2728                 " addr = %U port %d", getpid (),
2729                 session_handle, ep->is_ip4, format_ip46_address,
2730                 &session->transport.lcl_ip,
2731                 ep->is_ip4 ? IP46_TYPE_IP4 : IP46_TYPE_IP6,
2732                 clib_net_to_host_u16 (ep->port));
2733         }
2734       else
2735         rv = VPPCOM_EINVAL;
2736       break;
2737
2738     case VPPCOM_ATTR_GET_LIBC_EPFD:
2739       rv = session->libc_epfd;
2740       VDBG (2, "VCL<%d>: VPPCOM_ATTR_GET_LIBC_EPFD: libc_epfd %d",
2741             getpid (), rv);
2742       break;
2743
2744     case VPPCOM_ATTR_SET_LIBC_EPFD:
2745       if (PREDICT_TRUE (buffer && buflen &&
2746                         (*buflen == sizeof (session->libc_epfd))))
2747         {
2748           session->libc_epfd = *(int *) buffer;
2749           *buflen = sizeof (session->libc_epfd);
2750
2751           VDBG (2, "VCL<%d>: VPPCOM_ATTR_SET_LIBC_EPFD: libc_epfd %d, "
2752                 "buflen %d", getpid (), session->libc_epfd, *buflen);
2753         }
2754       else
2755         rv = VPPCOM_EINVAL;
2756       break;
2757
2758     case VPPCOM_ATTR_GET_PROTOCOL:
2759       if (buffer && buflen && (*buflen >= sizeof (int)))
2760         {
2761           *(int *) buffer = session->session_type;
2762           *buflen = sizeof (int);
2763
2764           VDBG (2, "VCL<%d>: VPPCOM_ATTR_GET_PROTOCOL: %d (%s), buflen %d",
2765                 getpid (), *(int *) buffer, *(int *) buffer ? "UDP" : "TCP",
2766                 *buflen);
2767         }
2768       else
2769         rv = VPPCOM_EINVAL;
2770       break;
2771
2772     case VPPCOM_ATTR_GET_LISTEN:
2773       if (buffer && buflen && (*buflen >= sizeof (int)))
2774         {
2775           *(int *) buffer = VCL_SESS_ATTR_TEST (session->attr,
2776                                                 VCL_SESS_ATTR_LISTEN);
2777           *buflen = sizeof (int);
2778
2779           VDBG (2, "VCL<%d>: VPPCOM_ATTR_GET_LISTEN: %d, buflen %d",
2780                 getpid (), *(int *) buffer, *buflen);
2781         }
2782       else
2783         rv = VPPCOM_EINVAL;
2784       break;
2785
2786     case VPPCOM_ATTR_GET_ERROR:
2787       if (buffer && buflen && (*buflen >= sizeof (int)))
2788         {
2789           *(int *) buffer = 0;
2790           *buflen = sizeof (int);
2791
2792           VDBG (2, "VCL<%d>: VPPCOM_ATTR_GET_ERROR: %d, buflen %d, #VPP-TBD#",
2793                 getpid (), *(int *) buffer, *buflen);
2794         }
2795       else
2796         rv = VPPCOM_EINVAL;
2797       break;
2798
2799     case VPPCOM_ATTR_GET_TX_FIFO_LEN:
2800       if (buffer && buflen && (*buflen >= sizeof (u32)))
2801         {
2802
2803           /* VPP-TBD */
2804           *(size_t *) buffer = (session->sndbuf_size ? session->sndbuf_size :
2805                                 session->tx_fifo ? session->tx_fifo->nitems :
2806                                 vcm->cfg.tx_fifo_size);
2807           *buflen = sizeof (u32);
2808
2809           VDBG (2, "VCL<%d>: VPPCOM_ATTR_GET_TX_FIFO_LEN: %u (0x%x), "
2810                 "buflen %d, #VPP-TBD#", getpid (),
2811                 *(size_t *) buffer, *(size_t *) buffer, *buflen);
2812         }
2813       else
2814         rv = VPPCOM_EINVAL;
2815       break;
2816
2817     case VPPCOM_ATTR_SET_TX_FIFO_LEN:
2818       if (buffer && buflen && (*buflen == sizeof (u32)))
2819         {
2820           /* VPP-TBD */
2821           session->sndbuf_size = *(u32 *) buffer;
2822           VDBG (2, "VCL<%d>: VPPCOM_ATTR_SET_TX_FIFO_LEN: %u (0x%x), "
2823                 "buflen %d, #VPP-TBD#", getpid (),
2824                 session->sndbuf_size, session->sndbuf_size, *buflen);
2825         }
2826       else
2827         rv = VPPCOM_EINVAL;
2828       break;
2829
2830     case VPPCOM_ATTR_GET_RX_FIFO_LEN:
2831       if (buffer && buflen && (*buflen >= sizeof (u32)))
2832         {
2833
2834           /* VPP-TBD */
2835           *(size_t *) buffer = (session->rcvbuf_size ? session->rcvbuf_size :
2836                                 session->rx_fifo ? session->rx_fifo->nitems :
2837                                 vcm->cfg.rx_fifo_size);
2838           *buflen = sizeof (u32);
2839
2840           VDBG (2, "VCL<%d>: VPPCOM_ATTR_GET_RX_FIFO_LEN: %u (0x%x), "
2841                 "buflen %d, #VPP-TBD#", getpid (),
2842                 *(size_t *) buffer, *(size_t *) buffer, *buflen);
2843         }
2844       else
2845         rv = VPPCOM_EINVAL;
2846       break;
2847
2848     case VPPCOM_ATTR_SET_RX_FIFO_LEN:
2849       if (buffer && buflen && (*buflen == sizeof (u32)))
2850         {
2851           /* VPP-TBD */
2852           session->rcvbuf_size = *(u32 *) buffer;
2853           VDBG (2, "VCL<%d>: VPPCOM_ATTR_SET_RX_FIFO_LEN: %u (0x%x), "
2854                 "buflen %d, #VPP-TBD#", getpid (),
2855                 session->sndbuf_size, session->sndbuf_size, *buflen);
2856         }
2857       else
2858         rv = VPPCOM_EINVAL;
2859       break;
2860
2861     case VPPCOM_ATTR_GET_REUSEADDR:
2862       if (buffer && buflen && (*buflen >= sizeof (int)))
2863         {
2864           /* VPP-TBD */
2865           *(int *) buffer = VCL_SESS_ATTR_TEST (session->attr,
2866                                                 VCL_SESS_ATTR_REUSEADDR);
2867           *buflen = sizeof (int);
2868
2869           VDBG (2, "VCL<%d>: VPPCOM_ATTR_GET_REUSEADDR: %d, "
2870                 "buflen %d, #VPP-TBD#", getpid (), *(int *) buffer, *buflen);
2871         }
2872       else
2873         rv = VPPCOM_EINVAL;
2874       break;
2875
2876     case VPPCOM_ATTR_SET_REUSEADDR:
2877       if (buffer && buflen && (*buflen == sizeof (int)) &&
2878           !VCL_SESS_ATTR_TEST (session->attr, VCL_SESS_ATTR_LISTEN))
2879         {
2880           /* VPP-TBD */
2881           if (*(int *) buffer)
2882             VCL_SESS_ATTR_SET (session->attr, VCL_SESS_ATTR_REUSEADDR);
2883           else
2884             VCL_SESS_ATTR_CLR (session->attr, VCL_SESS_ATTR_REUSEADDR);
2885
2886           VDBG (2, "VCL<%d>: VPPCOM_ATTR_SET_REUSEADDR: %d, buflen %d,"
2887                 " #VPP-TBD#", getpid (),
2888                 VCL_SESS_ATTR_TEST (session->attr,
2889                                     VCL_SESS_ATTR_REUSEADDR), *buflen);
2890         }
2891       else
2892         rv = VPPCOM_EINVAL;
2893       break;
2894
2895     case VPPCOM_ATTR_GET_REUSEPORT:
2896       if (buffer && buflen && (*buflen >= sizeof (int)))
2897         {
2898           /* VPP-TBD */
2899           *(int *) buffer = VCL_SESS_ATTR_TEST (session->attr,
2900                                                 VCL_SESS_ATTR_REUSEPORT);
2901           *buflen = sizeof (int);
2902
2903           VDBG (2, "VCL<%d>: VPPCOM_ATTR_GET_REUSEPORT: %d, buflen %d,"
2904                 " #VPP-TBD#", getpid (), *(int *) buffer, *buflen);
2905         }
2906       else
2907         rv = VPPCOM_EINVAL;
2908       break;
2909
2910     case VPPCOM_ATTR_SET_REUSEPORT:
2911       if (buffer && buflen && (*buflen == sizeof (int)) &&
2912           !VCL_SESS_ATTR_TEST (session->attr, VCL_SESS_ATTR_LISTEN))
2913         {
2914           /* VPP-TBD */
2915           if (*(int *) buffer)
2916             VCL_SESS_ATTR_SET (session->attr, VCL_SESS_ATTR_REUSEPORT);
2917           else
2918             VCL_SESS_ATTR_CLR (session->attr, VCL_SESS_ATTR_REUSEPORT);
2919
2920           VDBG (2, "VCL<%d>: VPPCOM_ATTR_SET_REUSEPORT: %d, buflen %d,"
2921                 " #VPP-TBD#", getpid (),
2922                 VCL_SESS_ATTR_TEST (session->attr,
2923                                     VCL_SESS_ATTR_REUSEPORT), *buflen);
2924         }
2925       else
2926         rv = VPPCOM_EINVAL;
2927       break;
2928
2929     case VPPCOM_ATTR_GET_BROADCAST:
2930       if (buffer && buflen && (*buflen >= sizeof (int)))
2931         {
2932           /* VPP-TBD */
2933           *(int *) buffer = VCL_SESS_ATTR_TEST (session->attr,
2934                                                 VCL_SESS_ATTR_BROADCAST);
2935           *buflen = sizeof (int);
2936
2937           VDBG (2, "VCL<%d>: VPPCOM_ATTR_GET_BROADCAST: %d, buflen %d,"
2938                 " #VPP-TBD#", getpid (), *(int *) buffer, *buflen);
2939         }
2940       else
2941         rv = VPPCOM_EINVAL;
2942       break;
2943
2944     case VPPCOM_ATTR_SET_BROADCAST:
2945       if (buffer && buflen && (*buflen == sizeof (int)))
2946         {
2947           /* VPP-TBD */
2948           if (*(int *) buffer)
2949             VCL_SESS_ATTR_SET (session->attr, VCL_SESS_ATTR_BROADCAST);
2950           else
2951             VCL_SESS_ATTR_CLR (session->attr, VCL_SESS_ATTR_BROADCAST);
2952
2953           VDBG (2, "VCL<%d>: VPPCOM_ATTR_SET_BROADCAST: %d, buflen %d, "
2954                 "#VPP-TBD#", getpid (),
2955                 VCL_SESS_ATTR_TEST (session->attr,
2956                                     VCL_SESS_ATTR_BROADCAST), *buflen);
2957         }
2958       else
2959         rv = VPPCOM_EINVAL;
2960       break;
2961
2962     case VPPCOM_ATTR_GET_V6ONLY:
2963       if (buffer && buflen && (*buflen >= sizeof (int)))
2964         {
2965           /* VPP-TBD */
2966           *(int *) buffer = VCL_SESS_ATTR_TEST (session->attr,
2967                                                 VCL_SESS_ATTR_V6ONLY);
2968           *buflen = sizeof (int);
2969
2970           VDBG (2, "VCL<%d>: VPPCOM_ATTR_GET_V6ONLY: %d, buflen %d, "
2971                 "#VPP-TBD#", getpid (), *(int *) buffer, *buflen);
2972         }
2973       else
2974         rv = VPPCOM_EINVAL;
2975       break;
2976
2977     case VPPCOM_ATTR_SET_V6ONLY:
2978       if (buffer && buflen && (*buflen == sizeof (int)))
2979         {
2980           /* VPP-TBD */
2981           if (*(int *) buffer)
2982             VCL_SESS_ATTR_SET (session->attr, VCL_SESS_ATTR_V6ONLY);
2983           else
2984             VCL_SESS_ATTR_CLR (session->attr, VCL_SESS_ATTR_V6ONLY);
2985
2986           VDBG (2, "VCL<%d>: VPPCOM_ATTR_SET_V6ONLY: %d, buflen %d, "
2987                 "#VPP-TBD#", getpid (),
2988                 VCL_SESS_ATTR_TEST (session->attr,
2989                                     VCL_SESS_ATTR_V6ONLY), *buflen);
2990         }
2991       else
2992         rv = VPPCOM_EINVAL;
2993       break;
2994
2995     case VPPCOM_ATTR_GET_KEEPALIVE:
2996       if (buffer && buflen && (*buflen >= sizeof (int)))
2997         {
2998           /* VPP-TBD */
2999           *(int *) buffer = VCL_SESS_ATTR_TEST (session->attr,
3000                                                 VCL_SESS_ATTR_KEEPALIVE);
3001           *buflen = sizeof (int);
3002
3003           VDBG (2, "VCL<%d>: VPPCOM_ATTR_GET_KEEPALIVE: %d, buflen %d, "
3004                 "#VPP-TBD#", getpid (), *(int *) buffer, *buflen);
3005         }
3006       else
3007         rv = VPPCOM_EINVAL;
3008       break;
3009
3010     case VPPCOM_ATTR_SET_KEEPALIVE:
3011       if (buffer && buflen && (*buflen == sizeof (int)))
3012         {
3013           /* VPP-TBD */
3014           if (*(int *) buffer)
3015             VCL_SESS_ATTR_SET (session->attr, VCL_SESS_ATTR_KEEPALIVE);
3016           else
3017             VCL_SESS_ATTR_CLR (session->attr, VCL_SESS_ATTR_KEEPALIVE);
3018
3019           VDBG (2, "VCL<%d>: VPPCOM_ATTR_SET_KEEPALIVE: %d, buflen %d, "
3020                 "#VPP-TBD#", getpid (),
3021                 VCL_SESS_ATTR_TEST (session->attr,
3022                                     VCL_SESS_ATTR_KEEPALIVE), *buflen);
3023         }
3024       else
3025         rv = VPPCOM_EINVAL;
3026       break;
3027
3028     case VPPCOM_ATTR_GET_TCP_NODELAY:
3029       if (buffer && buflen && (*buflen >= sizeof (int)))
3030         {
3031           /* VPP-TBD */
3032           *(int *) buffer = VCL_SESS_ATTR_TEST (session->attr,
3033                                                 VCL_SESS_ATTR_TCP_NODELAY);
3034           *buflen = sizeof (int);
3035
3036           VDBG (2, "VCL<%d>: VPPCOM_ATTR_GET_TCP_NODELAY: %d, buflen %d, "
3037                 "#VPP-TBD#", getpid (), *(int *) buffer, *buflen);
3038         }
3039       else
3040         rv = VPPCOM_EINVAL;
3041       break;
3042
3043     case VPPCOM_ATTR_SET_TCP_NODELAY:
3044       if (buffer && buflen && (*buflen == sizeof (int)))
3045         {
3046           /* VPP-TBD */
3047           if (*(int *) buffer)
3048             VCL_SESS_ATTR_SET (session->attr, VCL_SESS_ATTR_TCP_NODELAY);
3049           else
3050             VCL_SESS_ATTR_CLR (session->attr, VCL_SESS_ATTR_TCP_NODELAY);
3051
3052           VDBG (2, "VCL<%d>: VPPCOM_ATTR_SET_TCP_NODELAY: %d, buflen %d, "
3053                 "#VPP-TBD#", getpid (),
3054                 VCL_SESS_ATTR_TEST (session->attr,
3055                                     VCL_SESS_ATTR_TCP_NODELAY), *buflen);
3056         }
3057       else
3058         rv = VPPCOM_EINVAL;
3059       break;
3060
3061     case VPPCOM_ATTR_GET_TCP_KEEPIDLE:
3062       if (buffer && buflen && (*buflen >= sizeof (int)))
3063         {
3064           /* VPP-TBD */
3065           *(int *) buffer = VCL_SESS_ATTR_TEST (session->attr,
3066                                                 VCL_SESS_ATTR_TCP_KEEPIDLE);
3067           *buflen = sizeof (int);
3068
3069           VDBG (2, "VCL<%d>: VPPCOM_ATTR_GET_TCP_KEEPIDLE: %d, buflen %d, "
3070                 "#VPP-TBD#", getpid (), *(int *) buffer, *buflen);
3071         }
3072       else
3073         rv = VPPCOM_EINVAL;
3074       break;
3075
3076     case VPPCOM_ATTR_SET_TCP_KEEPIDLE:
3077       if (buffer && buflen && (*buflen == sizeof (int)))
3078         {
3079           /* VPP-TBD */
3080           if (*(int *) buffer)
3081             VCL_SESS_ATTR_SET (session->attr, VCL_SESS_ATTR_TCP_KEEPIDLE);
3082           else
3083             VCL_SESS_ATTR_CLR (session->attr, VCL_SESS_ATTR_TCP_KEEPIDLE);
3084
3085           VDBG (2, "VCL<%d>: VPPCOM_ATTR_SET_TCP_KEEPIDLE: %d, buflen %d, "
3086                 "#VPP-TBD#", getpid (),
3087                 VCL_SESS_ATTR_TEST (session->attr,
3088                                     VCL_SESS_ATTR_TCP_KEEPIDLE), *buflen);
3089         }
3090       else
3091         rv = VPPCOM_EINVAL;
3092       break;
3093
3094     case VPPCOM_ATTR_GET_TCP_KEEPINTVL:
3095       if (buffer && buflen && (*buflen >= sizeof (int)))
3096         {
3097           /* VPP-TBD */
3098           *(int *) buffer = VCL_SESS_ATTR_TEST (session->attr,
3099                                                 VCL_SESS_ATTR_TCP_KEEPINTVL);
3100           *buflen = sizeof (int);
3101
3102           VDBG (2, "VCL<%d>: VPPCOM_ATTR_GET_TCP_KEEPINTVL: %d, buflen %d, "
3103                 "#VPP-TBD#", getpid (), *(int *) buffer, *buflen);
3104         }
3105       else
3106         rv = VPPCOM_EINVAL;
3107       break;
3108
3109     case VPPCOM_ATTR_SET_TCP_KEEPINTVL:
3110       if (buffer && buflen && (*buflen == sizeof (int)))
3111         {
3112           /* VPP-TBD */
3113           if (*(int *) buffer)
3114             VCL_SESS_ATTR_SET (session->attr, VCL_SESS_ATTR_TCP_KEEPINTVL);
3115           else
3116             VCL_SESS_ATTR_CLR (session->attr, VCL_SESS_ATTR_TCP_KEEPINTVL);
3117
3118           VDBG (2, "VCL<%d>: VPPCOM_ATTR_SET_TCP_KEEPINTVL: %d, buflen %d, "
3119                 "#VPP-TBD#", getpid (),
3120                 VCL_SESS_ATTR_TEST (session->attr,
3121                                     VCL_SESS_ATTR_TCP_KEEPINTVL), *buflen);
3122         }
3123       else
3124         rv = VPPCOM_EINVAL;
3125       break;
3126
3127     case VPPCOM_ATTR_GET_TCP_USER_MSS:
3128       if (buffer && buflen && (*buflen >= sizeof (u32)))
3129         {
3130           /* VPP-TBD */
3131           *(u32 *) buffer = session->user_mss;
3132           *buflen = sizeof (int);
3133
3134           VDBG (2, "VCL<%d>: VPPCOM_ATTR_GET_TCP_USER_MSS: %d, buflen %d,"
3135                 " #VPP-TBD#", getpid (), *(int *) buffer, *buflen);
3136         }
3137       else
3138         rv = VPPCOM_EINVAL;
3139       break;
3140
3141     case VPPCOM_ATTR_SET_TCP_USER_MSS:
3142       if (buffer && buflen && (*buflen == sizeof (u32)))
3143         {
3144           /* VPP-TBD */
3145           session->user_mss = *(u32 *) buffer;
3146
3147           VDBG (2, "VCL<%d>: VPPCOM_ATTR_SET_TCP_USER_MSS: %u, buflen %d, "
3148                 "#VPP-TBD#", getpid (), session->user_mss, *buflen);
3149         }
3150       else
3151         rv = VPPCOM_EINVAL;
3152       break;
3153
3154     default:
3155       rv = VPPCOM_EINVAL;
3156       break;
3157     }
3158
3159   return rv;
3160 }
3161
3162 int
3163 vppcom_session_recvfrom (uint32_t session_handle, void *buffer,
3164                          uint32_t buflen, int flags, vppcom_endpt_t * ep)
3165 {
3166   vcl_worker_t *wrk = vcl_worker_get_current ();
3167   int rv = VPPCOM_OK;
3168   vcl_session_t *session = 0;
3169
3170   if (ep)
3171     {
3172       session = vcl_session_get_w_handle (wrk, session_handle);
3173       if (PREDICT_FALSE (!session))
3174         {
3175           VDBG (0, "VCL<%d>: invalid session, sid (%u) has been closed!",
3176                 getpid (), session_handle);
3177           return VPPCOM_EBADFD;
3178         }
3179       ep->is_ip4 = session->transport.is_ip4;
3180       ep->port = session->transport.rmt_port;
3181     }
3182
3183   if (flags == 0)
3184     rv = vppcom_session_read (session_handle, buffer, buflen);
3185   else if (flags & MSG_PEEK)
3186     rv = vppcom_session_peek (session_handle, buffer, buflen);
3187   else
3188     {
3189       clib_warning ("VCL<%d>: Unsupport flags for recvfrom %d",
3190                     getpid (), flags);
3191       return VPPCOM_EAFNOSUPPORT;
3192     }
3193
3194   if (ep)
3195     {
3196       if (session->transport.is_ip4)
3197         clib_memcpy_fast (ep->ip, &session->transport.rmt_ip.ip4,
3198                           sizeof (ip4_address_t));
3199       else
3200         clib_memcpy_fast (ep->ip, &session->transport.rmt_ip.ip6,
3201                           sizeof (ip6_address_t));
3202     }
3203
3204   return rv;
3205 }
3206
3207 int
3208 vppcom_session_sendto (uint32_t session_handle, void *buffer,
3209                        uint32_t buflen, int flags, vppcom_endpt_t * ep)
3210 {
3211   if (!buffer)
3212     return VPPCOM_EINVAL;
3213
3214   if (ep)
3215     {
3216       // TBD
3217       return VPPCOM_EINVAL;
3218     }
3219
3220   if (flags)
3221     {
3222       // TBD check the flags and do the right thing
3223       VDBG (2, "VCL<%d>: handling flags 0x%u (%d) not implemented yet.",
3224             getpid (), flags, flags);
3225     }
3226
3227   return (vppcom_session_write (session_handle, buffer, buflen));
3228 }
3229
3230 int
3231 vppcom_poll (vcl_poll_t * vp, uint32_t n_sids, double wait_for_time)
3232 {
3233   vcl_worker_t *wrk = vcl_worker_get_current ();
3234   f64 timeout = clib_time_now (&wrk->clib_time) + wait_for_time;
3235   u32 i, keep_trying = 1;
3236   svm_msg_q_msg_t msg;
3237   session_event_t *e;
3238   int rv, num_ev = 0;
3239
3240   VDBG (3, "VCL<%d>: vp %p, nsids %u, wait_for_time %f",
3241         getpid (), vp, n_sids, wait_for_time);
3242
3243   if (!vp)
3244     return VPPCOM_EFAULT;
3245
3246   do
3247     {
3248       vcl_session_t *session;
3249
3250       /* Dequeue all events and drop all unhandled io events */
3251       while (svm_msg_q_sub (wrk->app_event_queue, &msg, SVM_Q_NOWAIT, 0) == 0)
3252         {
3253           e = svm_msg_q_msg_data (wrk->app_event_queue, &msg);
3254           vcl_handle_mq_event (wrk, e);
3255           svm_msg_q_free_msg (wrk->app_event_queue, &msg);
3256         }
3257       vec_reset_length (wrk->unhandled_evts_vector);
3258
3259       for (i = 0; i < n_sids; i++)
3260         {
3261           session = vcl_session_get (wrk, vp[i].sid);
3262           if (!session)
3263             {
3264               vp[i].revents = POLLHUP;
3265               num_ev++;
3266               continue;
3267             }
3268
3269           vp[i].revents = 0;
3270
3271           if (POLLIN & vp[i].events)
3272             {
3273               rv = vppcom_session_read_ready (session);
3274               if (rv > 0)
3275                 {
3276                   vp[i].revents |= POLLIN;
3277                   num_ev++;
3278                 }
3279               else if (rv < 0)
3280                 {
3281                   switch (rv)
3282                     {
3283                     case VPPCOM_ECONNRESET:
3284                       vp[i].revents = POLLHUP;
3285                       break;
3286
3287                     default:
3288                       vp[i].revents = POLLERR;
3289                       break;
3290                     }
3291                   num_ev++;
3292                 }
3293             }
3294
3295           if (POLLOUT & vp[i].events)
3296             {
3297               rv = vppcom_session_write_ready (session);
3298               if (rv > 0)
3299                 {
3300                   vp[i].revents |= POLLOUT;
3301                   num_ev++;
3302                 }
3303               else if (rv < 0)
3304                 {
3305                   switch (rv)
3306                     {
3307                     case VPPCOM_ECONNRESET:
3308                       vp[i].revents = POLLHUP;
3309                       break;
3310
3311                     default:
3312                       vp[i].revents = POLLERR;
3313                       break;
3314                     }
3315                   num_ev++;
3316                 }
3317             }
3318
3319           if (0)                // Note "done:" label used by VCL_SESSION_LOCK_AND_GET()
3320             {
3321               vp[i].revents = POLLNVAL;
3322               num_ev++;
3323             }
3324         }
3325       if (wait_for_time != -1)
3326         keep_trying = (clib_time_now (&wrk->clib_time) <= timeout) ? 1 : 0;
3327     }
3328   while ((num_ev == 0) && keep_trying);
3329
3330   if (VPPCOM_DEBUG > 3)
3331     {
3332       clib_warning ("VCL<%d>: returning %d", getpid (), num_ev);
3333       for (i = 0; i < n_sids; i++)
3334         {
3335           clib_warning ("VCL<%d>: vp[%d].sid %d (0x%x), .events 0x%x, "
3336                         ".revents 0x%x", getpid (), i, vp[i].sid, vp[i].sid,
3337                         vp[i].events, vp[i].revents);
3338         }
3339     }
3340   return num_ev;
3341 }
3342
3343 int
3344 vppcom_mq_epoll_fd (void)
3345 {
3346   vcl_worker_t *wrk = vcl_worker_get_current ();
3347   return wrk->mqs_epfd;
3348 }
3349
3350 int
3351 vppcom_session_index (uint32_t session_handle)
3352 {
3353   return session_handle & 0xFFFFFF;
3354 }
3355
3356 int
3357 vppcom_worker_register (void)
3358 {
3359   if (vcl_worker_alloc_and_init ())
3360     return VPPCOM_OK;
3361   return VPPCOM_EEXIST;
3362 }
3363
3364 /*
3365  * fd.io coding-style-patch-verification: ON
3366  *
3367  * Local Variables:
3368  * eval: (c-set-style "gnu")
3369  * End:
3370  */