src/vcl/vppcom.c

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