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   session_accepted_msg_t *accepted_msg;
 826   vcl_session_t *session = 0;
 827   vcl_session_msg_t *evt;
 828   u64 vpp_handle;
 829
 830   session = vcl_session_get_w_handle (wrk, session_handle);
 831   if (!session)
 832     return VPPCOM_EBADFD;
 833
 834   /* Flush pending accept events, if any */
 835   while (clib_fifo_elts (session->accept_evts_fifo))
 836     {
 837       clib_fifo_sub2 (session->accept_evts_fifo, evt);
 838       accepted_msg = &evt->accepted_msg;
 839       vcl_session_table_del_vpp_handle (wrk, accepted_msg->handle);
 840       vcl_send_session_accepted_reply (session->vpp_evt_q,
 841                                        accepted_msg->context,
 842                                        session->vpp_handle, -1);
 843     }
 844   clib_fifo_free (session->accept_evts_fifo);
 845
 846   vpp_handle = session->vpp_handle;
 847   session->vpp_handle = ~0;
 848   session->session_state = STATE_DISCONNECT;
 849
 850   VDBG (1, "session %u [0x%llx]: sending unbind!", session->session_index,
 851         vpp_handle);
 852   vcl_evt (VCL_EVT_UNBIND, session);
 853   vppcom_send_unbind_sock (wrk, vpp_handle);
 854
 855   return VPPCOM_OK;
 856 }
 857
 858 static int
 859 vppcom_session_disconnect (u32 session_handle)
 860 {
 861   vcl_worker_t *wrk = vcl_worker_get_current ();
 862   svm_msg_q_t *vpp_evt_q;
 863   vcl_session_t *session;
 864   vcl_session_state_t state;
 865   u64 vpp_handle;
 866
 867   session = vcl_session_get_w_handle (wrk, session_handle);
 868   if (!session)
 869     return VPPCOM_EBADFD;
 870
 871   vpp_handle = session->vpp_handle;
 872   state = session->session_state;
 873
 874   VDBG (1, "session %u [0x%llx] state 0x%x (%s)", session->session_index,
 875         vpp_handle, state, vppcom_session_state_str (state));
 876
 877   if (PREDICT_FALSE (state & STATE_LISTEN))
 878     {
 879       VDBG (0, "ERROR: Cannot disconnect a listen socket!");
 880       return VPPCOM_EBADFD;
 881     }
 882
 883   if (state & STATE_VPP_CLOSING)
 884     {
 885       vpp_evt_q = vcl_session_vpp_evt_q (wrk, session);
 886       vcl_send_session_disconnected_reply (vpp_evt_q, wrk->my_client_index,
 887                                            vpp_handle, 0);
 888       VDBG (1, "session %u [0x%llx]: sending disconnect REPLY...",
 889             session->session_index, vpp_handle);
 890     }
 891   else
 892     {
 893       VDBG (1, "session %u [0x%llx]: sending disconnect...",
 894             session->session_index, vpp_handle);
 895       vppcom_send_disconnect_session (vpp_handle);
 896     }
 897
 898   return VPPCOM_OK;
 899 }
 900
 901 /**
 902  * Handle app exit
 903  *
 904  * Notify vpp of the disconnect and mark the worker as free. If we're the
 905  * last worker, do a full cleanup otherwise, since we're probably a forked
 906  * child, avoid syscalls as much as possible. We might've lost privileges.
 907  */
 908 void
 909 vppcom_app_exit (void)
 910 {
 911   if (!pool_elts (vcm->workers))
 912     return;
 913   vcl_worker_cleanup (vcl_worker_get_current (), 1 /* notify vpp */ );
 914   vcl_set_worker_index (~0);
 915   vcl_elog_stop (vcm);
 916   if (vec_len (vcm->workers) == 1)
 917     vl_client_disconnect_from_vlib ();
 918   else
 919     vl_client_send_disconnect (1 /* vpp should cleanup */ );
 920 }
 921
 922 /*
 923  * VPPCOM Public API functions
 924  */
 925 int
 926 vppcom_app_create (char *app_name)
 927 {
 928   vppcom_cfg_t *vcl_cfg = &vcm->cfg;
 929   int rv;
 930
 931   if (vcm->is_init)
 932     {
 933       VDBG (1, "already initialized");
 934       return VPPCOM_EEXIST;
 935     }
 936
 937   vcm->is_init = 1;
 938   vppcom_cfg (&vcm->cfg);
 939   vcl_cfg = &vcm->cfg;
 940
 941   vcm->main_cpu = pthread_self ();
 942   vcm->main_pid = getpid ();
 943   vcm->app_name = format (0, "%s", app_name);
 944   vppcom_init_error_string_table ();
 945   fifo_segment_main_init (&vcm->segment_main, vcl_cfg->segment_baseva,
 946                           20 /* timeout in secs */ );
 947   pool_alloc (vcm->workers, vcl_cfg->max_workers);
 948   clib_spinlock_init (&vcm->workers_lock);
 949   clib_rwlock_init (&vcm->segment_table_lock);
 950   atexit (vppcom_app_exit);
 951
 952   /* Allocate default worker */
 953   vcl_worker_alloc_and_init ();
 954
 955   /* API hookup and connect to VPP */
 956   vppcom_api_hookup ();
 957   vcl_elog_init (vcm);
 958   vcm->app_state = STATE_APP_START;
 959   rv = vppcom_connect_to_vpp (app_name);
 960   if (rv)
 961     {
 962       VERR ("couldn't connect to VPP!");
 963       return rv;
 964     }
 965   VDBG (0, "sending session enable");
 966   rv = vppcom_app_session_enable ();
 967   if (rv)
 968     {
 969       VERR ("vppcom_app_session_enable() failed!");
 970       return rv;
 971     }
 972
 973   VDBG (0, "sending app attach");
 974   rv = vppcom_app_attach ();
 975   if (rv)
 976     {
 977       VERR ("vppcom_app_attach() failed!");
 978       return rv;
 979     }
 980
 981   VDBG (0, "app_name '%s', my_client_index %d (0x%x)", app_name,
 982         vcm->workers[0].my_client_index, vcm->workers[0].my_client_index);
 983
 984   return VPPCOM_OK;
 985 }
 986
 987 void
 988 vppcom_app_destroy (void)
 989 {
 990   int rv;
 991   f64 orig_app_timeout;
 992
 993   if (!pool_elts (vcm->workers))
 994     return;
 995
 996   vcl_evt (VCL_EVT_DETACH, vcm);
 997
 998   if (pool_elts (vcm->workers) == 1)
 999     {
1000       vppcom_app_send_detach ();
1001       orig_app_timeout = vcm->cfg.app_timeout;
1002       vcm->cfg.app_timeout = 2.0;
1003       rv = vcl_wait_for_app_state_change (STATE_APP_ENABLED);
1004       vcm->cfg.app_timeout = orig_app_timeout;
1005       if (PREDICT_FALSE (rv))
1006         VDBG (0, "application detach timed out! returning %d (%s)", rv,
1007               vppcom_retval_str (rv));
1008       vec_free (vcm->app_name);
1009       vcl_worker_cleanup (vcl_worker_get_current (), 0 /* notify vpp */ );
1010     }
1011   else
1012     {
1013       vcl_worker_cleanup (vcl_worker_get_current (), 1 /* notify vpp */ );
1014     }
1015
1016   vcl_set_worker_index (~0);
1017   vcl_elog_stop (vcm);
1018   vl_client_disconnect_from_vlib ();
1019 }
1020
1021 int
1022 vppcom_session_create (u8 proto, u8 is_nonblocking)
1023 {
1024   vcl_worker_t *wrk = vcl_worker_get_current ();
1025   vcl_session_t *session;
1026
1027   session = vcl_session_alloc (wrk);
1028
1029   session->session_type = proto;
1030   session->session_state = STATE_START;
1031   session->vpp_handle = ~0;
1032   session->is_dgram = proto == VPPCOM_PROTO_UDP;
1033
1034   if (is_nonblocking)
1035     VCL_SESS_ATTR_SET (session->attr, VCL_SESS_ATTR_NONBLOCK);
1036
1037   vcl_evt (VCL_EVT_CREATE, session, session_type, session->session_state,
1038            is_nonblocking, session_index);
1039
1040   VDBG (0, "created session %u", session->session_index);
1041
1042   return vcl_session_handle (session);
1043 }
1044
1045 int
1046 vcl_session_cleanup (vcl_worker_t * wrk, vcl_session_t * session,
1047                      vcl_session_handle_t sh, u8 do_disconnect)
1048 {
1049   vcl_session_state_t state;
1050   u32 next_sh, vep_sh;
1051   int rv = VPPCOM_OK;
1052   u64 vpp_handle;
1053   u8 is_vep;
1054
1055   is_vep = session->is_vep;
1056   next_sh = session->vep.next_sh;
1057   vep_sh = session->vep.vep_sh;
1058   state = session->session_state;
1059   vpp_handle = session->vpp_handle;
1060
1061   VDBG (1, "session %u [0x%llx] closing", session->session_index, vpp_handle);
1062
1063   if (is_vep)
1064     {
1065       while (next_sh != ~0)
1066         {
1067           rv = vppcom_epoll_ctl (sh, EPOLL_CTL_DEL, next_sh, 0);
1068           if (PREDICT_FALSE (rv < 0))
1069             VDBG (0, "vpp handle 0x%llx, sh %u: EPOLL_CTL_DEL vep_idx %u"
1070                   " failed! rv %d (%s)", vpp_handle, next_sh, vep_sh, rv,
1071                   vppcom_retval_str (rv));
1072
1073           next_sh = session->vep.next_sh;
1074         }
1075     }
1076   else
1077     {
1078       if (session->is_vep_session)
1079         {
1080           rv = vppcom_epoll_ctl (vep_sh, EPOLL_CTL_DEL, sh, 0);
1081           if (rv < 0)
1082             VDBG (0, "session %u [0x%llx]: EPOLL_CTL_DEL vep_idx %u "
1083                   "failed! rv %d (%s)", session->session_index, vpp_handle,
1084                   vep_sh, rv, vppcom_retval_str (rv));
1085         }
1086
1087       if (!do_disconnect)
1088         {
1089           VDBG (1, "session %u [0x%llx] disconnect skipped",
1090                 session->session_index, vpp_handle);
1091           goto cleanup;
1092         }
1093
1094       if (state & STATE_LISTEN)
1095         {
1096           rv = vppcom_session_unbind (sh);
1097           if (PREDICT_FALSE (rv < 0))
1098             VDBG (0, "session %u [0x%llx]: listener unbind failed! "
1099                   "rv %d (%s)", session->session_index, vpp_handle, rv,
1100                   vppcom_retval_str (rv));
1101           return rv;
1102         }
1103       else if (state & STATE_OPEN)
1104         {
1105           rv = vppcom_session_disconnect (sh);
1106           if (PREDICT_FALSE (rv < 0))
1107             VDBG (0, "ERROR: session %u [0x%llx]: disconnect failed!"
1108                   " rv %d (%s)", session->session_index, vpp_handle,
1109                   rv, vppcom_retval_str (rv));
1110         }
1111       else if (state == STATE_DISCONNECT)
1112         {
1113           svm_msg_q_t *mq = vcl_session_vpp_evt_q (wrk, session);
1114           vcl_send_session_reset_reply (mq, wrk->my_client_index,
1115                                         session->vpp_handle, 0);
1116         }
1117     }
1118
1119   VDBG (0, "session %u [0x%llx] removed", session->session_index, vpp_handle);
1120
1121 cleanup:
1122   vcl_session_table_del_vpp_handle (wrk, vpp_handle);
1123   vcl_session_free (wrk, session);
1124   vcl_evt (VCL_EVT_CLOSE, session, rv);
1125
1126   return rv;
1127 }
1128
1129 int
1130 vppcom_session_close (uint32_t session_handle)
1131 {
1132   vcl_worker_t *wrk = vcl_worker_get_current ();
1133   vcl_session_t *session;
1134
1135   session = vcl_session_get_w_handle (wrk, session_handle);
1136   if (!session)
1137     return VPPCOM_EBADFD;
1138   return vcl_session_cleanup (wrk, session, session_handle,
1139                               1 /* do_disconnect */ );
1140 }
1141
1142 int
1143 vppcom_session_bind (uint32_t session_handle, vppcom_endpt_t * ep)
1144 {
1145   vcl_worker_t *wrk = vcl_worker_get_current ();
1146   vcl_session_t *session = 0;
1147
1148   if (!ep || !ep->ip)
1149     return VPPCOM_EINVAL;
1150
1151   session = vcl_session_get_w_handle (wrk, session_handle);
1152   if (!session)
1153     return VPPCOM_EBADFD;
1154
1155   if (session->is_vep)
1156     {
1157       VDBG (0, "ERROR: cannot bind to epoll session %u!",
1158             session->session_index);
1159       return VPPCOM_EBADFD;
1160     }
1161
1162   session->transport.is_ip4 = ep->is_ip4;
1163   if (ep->is_ip4)
1164     clib_memcpy_fast (&session->transport.lcl_ip.ip4, ep->ip,
1165                       sizeof (ip4_address_t));
1166   else
1167     clib_memcpy_fast (&session->transport.lcl_ip.ip6, ep->ip,
1168                       sizeof (ip6_address_t));
1169   session->transport.lcl_port = ep->port;
1170
1171   VDBG (0, "session %u handle %u: binding to local %s address %U port %u, "
1172         "proto %s", session->session_index, session_handle,
1173         session->transport.is_ip4 ? "IPv4" : "IPv6",
1174         format_ip46_address, &session->transport.lcl_ip,
1175         session->transport.is_ip4 ? IP46_TYPE_IP4 : IP46_TYPE_IP6,
1176         clib_net_to_host_u16 (session->transport.lcl_port),
1177         vppcom_proto_str (session->session_type));
1178   vcl_evt (VCL_EVT_BIND, session);
1179
1180   if (session->session_type == VPPCOM_PROTO_UDP)
1181     vppcom_session_listen (session_handle, 10);
1182
1183   return VPPCOM_OK;
1184 }
1185
1186 int
1187 vppcom_session_listen (uint32_t listen_sh, uint32_t q_len)
1188 {
1189   vcl_worker_t *wrk = vcl_worker_get_current ();
1190   vcl_session_t *listen_session = 0;
1191   u64 listen_vpp_handle;
1192   int rv;
1193
1194   listen_session = vcl_session_get_w_handle (wrk, listen_sh);
1195   if (!listen_session || listen_session->is_vep)
1196     return VPPCOM_EBADFD;
1197
1198   if (q_len == 0 || q_len == ~0)
1199     q_len = vcm->cfg.listen_queue_size;
1200
1201   listen_vpp_handle = listen_session->vpp_handle;
1202   if (listen_session->session_state & STATE_LISTEN)
1203     {
1204       VDBG (0, "session %u [0x%llx]: already in listen state!",
1205             listen_sh, listen_vpp_handle);
1206       return VPPCOM_OK;
1207     }
1208
1209   VDBG (0, "session %u [0x%llx]: sending vpp listen request...",
1210         listen_sh, listen_vpp_handle);
1211
1212   /*
1213    * Send listen request to vpp and wait for reply
1214    */
1215   vppcom_send_bind_sock (listen_session);
1216   rv = vppcom_wait_for_session_state_change (listen_session->session_index,
1217                                              STATE_LISTEN,
1218                                              vcm->cfg.session_timeout);
1219
1220   if (PREDICT_FALSE (rv))
1221     {
1222       listen_session = vcl_session_get_w_handle (wrk, listen_sh);
1223       VDBG (0, "session %u [0x%llx]: listen failed! returning %d (%s)",
1224             listen_sh, listen_session->vpp_handle, rv,
1225             vppcom_retval_str (rv));
1226       return rv;
1227     }
1228
1229   return VPPCOM_OK;
1230 }
1231
1232 int
1233 vppcom_session_tls_add_cert (uint32_t session_handle, char *cert,
1234                              uint32_t cert_len)
1235 {
1236
1237   vcl_worker_t *wrk = vcl_worker_get_current ();
1238   vcl_session_t *session = 0;
1239
1240   session = vcl_session_get_w_handle (wrk, session_handle);
1241   if (!session)
1242     return VPPCOM_EBADFD;
1243
1244   if (cert_len == 0 || cert_len == ~0)
1245     return VPPCOM_EBADFD;
1246
1247   /*
1248    * Send listen request to vpp and wait for reply
1249    */
1250   vppcom_send_application_tls_cert_add (session, cert, cert_len);
1251
1252   return VPPCOM_OK;
1253
1254 }
1255
1256 int
1257 vppcom_session_tls_add_key (uint32_t session_handle, char *key,
1258                             uint32_t key_len)
1259 {
1260
1261   vcl_worker_t *wrk = vcl_worker_get_current ();
1262   vcl_session_t *session = 0;
1263
1264   session = vcl_session_get_w_handle (wrk, session_handle);
1265   if (!session)
1266     return VPPCOM_EBADFD;
1267
1268   if (key_len == 0 || key_len == ~0)
1269     return VPPCOM_EBADFD;
1270
1271   /*
1272    * Send listen request to vpp and wait for reply
1273    */
1274   vppcom_send_application_tls_key_add (session, key, key_len);
1275
1276   return VPPCOM_OK;
1277
1278
1279 }
1280
1281 static int
1282 validate_args_session_accept_ (vcl_worker_t * wrk, vcl_session_t * ls)
1283 {
1284   if (ls->is_vep)
1285     {
1286       VDBG (0, "ERROR: cannot accept on epoll session %u!",
1287             ls->session_index);
1288       return VPPCOM_EBADFD;
1289     }
1290
1291   if (ls->session_state != STATE_LISTEN)
1292     {
1293       VDBG (0, "ERROR: session [0x%llx]: not in listen state! state 0x%x"
1294             " (%s)", ls->vpp_handle, ls->session_index, ls->session_state,
1295             vppcom_session_state_str (ls->session_state));
1296       return VPPCOM_EBADFD;
1297     }
1298   return VPPCOM_OK;
1299 }
1300
1301 int
1302 vppcom_session_accept (uint32_t listen_session_handle, vppcom_endpt_t * ep,
1303                        uint32_t flags)
1304 {
1305   u32 client_session_index = ~0, listen_session_index, accept_flags = 0;
1306   vcl_worker_t *wrk = vcl_worker_get_current ();
1307   session_accepted_msg_t accepted_msg;
1308   vcl_session_t *listen_session = 0;
1309   vcl_session_t *client_session = 0;
1310   vcl_session_msg_t *evt;
1311   svm_msg_q_msg_t msg;
1312   session_event_t *e;
1313   u8 is_nonblocking;
1314   int rv;
1315
1316   listen_session = vcl_session_get_w_handle (wrk, listen_session_handle);
1317   if (!listen_session)
1318     return VPPCOM_EBADFD;
1319
1320   listen_session_index = listen_session->session_index;
1321   if ((rv = validate_args_session_accept_ (wrk, listen_session)))
1322     return rv;
1323
1324   if (clib_fifo_elts (listen_session->accept_evts_fifo))
1325     {
1326       clib_fifo_sub2 (listen_session->accept_evts_fifo, evt);
1327       accept_flags = evt->flags;
1328       accepted_msg = evt->accepted_msg;
1329       goto handle;
1330     }
1331
1332   is_nonblocking = VCL_SESS_ATTR_TEST (listen_session->attr,
1333                                        VCL_SESS_ATTR_NONBLOCK);
1334   if (svm_msg_q_is_empty (wrk->app_event_queue) && is_nonblocking)
1335     return VPPCOM_EAGAIN;
1336
1337   while (1)
1338     {
1339       if (svm_msg_q_sub (wrk->app_event_queue, &msg, SVM_Q_WAIT, 0))
1340         return VPPCOM_EAGAIN;
1341
1342       e = svm_msg_q_msg_data (wrk->app_event_queue, &msg);
1343       if (e->event_type != SESSION_CTRL_EVT_ACCEPTED)
1344         {
1345           VDBG (0, "discarded event: %u", e->event_type);
1346           svm_msg_q_free_msg (wrk->app_event_queue, &msg);
1347           continue;
1348         }
1349       clib_memcpy_fast (&accepted_msg, e->data, sizeof (accepted_msg));
1350       svm_msg_q_free_msg (wrk->app_event_queue, &msg);
1351       break;
1352     }
1353
1354 handle:
1355
1356   client_session_index = vcl_session_accepted_handler (wrk, &accepted_msg,
1357                                                        listen_session_index);
1358   if (client_session_index == VCL_INVALID_SESSION_INDEX)
1359     return VPPCOM_ECONNABORTED;
1360
1361   listen_session = vcl_session_get (wrk, listen_session_index);
1362   client_session = vcl_session_get (wrk, client_session_index);
1363
1364   if (flags & O_NONBLOCK)
1365     VCL_SESS_ATTR_SET (client_session->attr, VCL_SESS_ATTR_NONBLOCK);
1366
1367   VDBG (1, "listener %u [0x%llx]: Got a connect request! session %u [0x%llx],"
1368         " flags %d, is_nonblocking %u", listen_session->session_index,
1369         listen_session->vpp_handle, client_session_index,
1370         client_session->vpp_handle, flags,
1371         VCL_SESS_ATTR_TEST (client_session->attr, VCL_SESS_ATTR_NONBLOCK));
1372
1373   if (ep)
1374     {
1375       ep->is_ip4 = client_session->transport.is_ip4;
1376       ep->port = client_session->transport.rmt_port;
1377       if (client_session->transport.is_ip4)
1378         clib_memcpy_fast (ep->ip, &client_session->transport.rmt_ip.ip4,
1379                           sizeof (ip4_address_t));
1380       else
1381         clib_memcpy_fast (ep->ip, &client_session->transport.rmt_ip.ip6,
1382                           sizeof (ip6_address_t));
1383     }
1384
1385   VDBG (0, "listener %u [0x%llx] accepted %u [0x%llx] peer: %U:%u "
1386         "local: %U:%u", listen_session_handle, listen_session->vpp_handle,
1387         client_session_index, client_session->vpp_handle,
1388         format_ip46_address, &client_session->transport.rmt_ip,
1389         client_session->transport.is_ip4 ? IP46_TYPE_IP4 : IP46_TYPE_IP6,
1390         clib_net_to_host_u16 (client_session->transport.rmt_port),
1391         format_ip46_address, &client_session->transport.lcl_ip,
1392         client_session->transport.is_ip4 ? IP46_TYPE_IP4 : IP46_TYPE_IP6,
1393         clib_net_to_host_u16 (client_session->transport.lcl_port));
1394   vcl_evt (VCL_EVT_ACCEPT, client_session, listen_session,
1395            client_session_index);
1396
1397   /*
1398    * Session might have been closed already
1399    */
1400   if (accept_flags)
1401     {
1402       if (accept_flags & VCL_ACCEPTED_F_CLOSED)
1403         client_session->session_state = STATE_VPP_CLOSING;
1404       else if (accept_flags & VCL_ACCEPTED_F_RESET)
1405         client_session->session_state = STATE_DISCONNECT;
1406     }
1407   return vcl_session_handle (client_session);
1408 }
1409
1410 int
1411 vppcom_session_connect (uint32_t session_handle, vppcom_endpt_t * server_ep)
1412 {
1413   vcl_worker_t *wrk = vcl_worker_get_current ();
1414   vcl_session_t *session = 0;
1415   u32 session_index;
1416   int rv;
1417
1418   session = vcl_session_get_w_handle (wrk, session_handle);
1419   if (!session)
1420     return VPPCOM_EBADFD;
1421   session_index = session->session_index;
1422
1423   if (PREDICT_FALSE (session->is_vep))
1424     {
1425       VDBG (0, "ERROR: cannot connect epoll session %u!",
1426             session->session_index);
1427       return VPPCOM_EBADFD;
1428     }
1429
1430   if (PREDICT_FALSE (session->session_state & CLIENT_STATE_OPEN))
1431     {
1432       VDBG (0, "session handle %u [0x%llx]: session already "
1433             "connected to %s %U port %d proto %s, state 0x%x (%s)",
1434             session_handle, session->vpp_handle,
1435             session->transport.is_ip4 ? "IPv4" : "IPv6",
1436             format_ip46_address,
1437             &session->transport.rmt_ip, session->transport.is_ip4 ?
1438             IP46_TYPE_IP4 : IP46_TYPE_IP6,
1439             clib_net_to_host_u16 (session->transport.rmt_port),
1440             vppcom_proto_str (session->session_type), session->session_state,
1441             vppcom_session_state_str (session->session_state));
1442       return VPPCOM_OK;
1443     }
1444
1445   session->transport.is_ip4 = server_ep->is_ip4;
1446   if (session->transport.is_ip4)
1447     clib_memcpy_fast (&session->transport.rmt_ip.ip4, server_ep->ip,
1448                       sizeof (ip4_address_t));
1449   else
1450     clib_memcpy_fast (&session->transport.rmt_ip.ip6, server_ep->ip,
1451                       sizeof (ip6_address_t));
1452   session->transport.rmt_port = server_ep->port;
1453
1454   VDBG (0, "session handle %u [0x%llx]: connecting to server %s %U "
1455         "port %d proto %s", session_handle, session->vpp_handle,
1456         session->transport.is_ip4 ? "IPv4" : "IPv6",
1457         format_ip46_address,
1458         &session->transport.rmt_ip, session->transport.is_ip4 ?
1459         IP46_TYPE_IP4 : IP46_TYPE_IP6,
1460         clib_net_to_host_u16 (session->transport.rmt_port),
1461         vppcom_proto_str (session->session_type));
1462
1463   /*
1464    * Send connect request and wait for reply from vpp
1465    */
1466   vppcom_send_connect_sock (session);
1467   rv = vppcom_wait_for_session_state_change (session_index, STATE_CONNECT,
1468                                              vcm->cfg.session_timeout);
1469
1470   session = vcl_session_get (wrk, session_index);
1471   VDBG (0, "session %u [0x%llx]: connect %s!", session->session_index,
1472         session->vpp_handle, rv ? "failed" : "succeeded");
1473
1474   return rv;
1475 }
1476
1477 static u8
1478 vcl_is_rx_evt_for_session (session_event_t * e, u32 sid, u8 is_ct)
1479 {
1480   return (e->event_type == SESSION_IO_EVT_RX && e->session_index == sid);
1481 }
1482
1483 static inline int
1484 vppcom_session_read_internal (uint32_t session_handle, void *buf, int n,
1485                               u8 peek)
1486 {
1487   vcl_worker_t *wrk = vcl_worker_get_current ();
1488   int n_read = 0, is_nonblocking;
1489   vcl_session_t *s = 0;
1490   svm_fifo_t *rx_fifo;
1491   svm_msg_q_msg_t msg;
1492   session_event_t *e;
1493   svm_msg_q_t *mq;
1494   u8 is_ct;
1495
1496   if (PREDICT_FALSE (!buf))
1497     return VPPCOM_EINVAL;
1498
1499   s = vcl_session_get_w_handle (wrk, session_handle);
1500   if (PREDICT_FALSE (!s || s->is_vep))
1501     return VPPCOM_EBADFD;
1502
1503   if (PREDICT_FALSE (!vcl_session_is_open (s)))
1504     {
1505       VDBG (0, "session %u[0x%llx] is not open! state 0x%x (%s)",
1506             s->session_index, s->vpp_handle, s->session_state,
1507             vppcom_session_state_str (s->session_state));
1508       return vcl_session_closed_error (s);
1509     }
1510
1511   is_nonblocking = VCL_SESS_ATTR_TEST (s->attr, VCL_SESS_ATTR_NONBLOCK);
1512   is_ct = vcl_session_is_ct (s);
1513   mq = wrk->app_event_queue;
1514   rx_fifo = is_ct ? s->ct_rx_fifo : s->rx_fifo;
1515   s->has_rx_evt = 0;
1516
1517   if (svm_fifo_is_empty_cons (rx_fifo))
1518     {
1519       if (is_nonblocking)
1520         {
1521           svm_fifo_unset_event (s->rx_fifo);
1522           return VPPCOM_EWOULDBLOCK;
1523         }
1524       while (svm_fifo_is_empty_cons (rx_fifo))
1525         {
1526           if (vcl_session_is_closing (s))
1527             return vcl_session_closing_error (s);
1528
1529           svm_fifo_unset_event (s->rx_fifo);
1530           svm_msg_q_lock (mq);
1531           if (svm_msg_q_is_empty (mq))
1532             svm_msg_q_wait (mq);
1533
1534           svm_msg_q_sub_w_lock (mq, &msg);
1535           e = svm_msg_q_msg_data (mq, &msg);
1536           svm_msg_q_unlock (mq);
1537           if (!vcl_is_rx_evt_for_session (e, s->session_index, is_ct))
1538             vcl_handle_mq_event (wrk, e);
1539           svm_msg_q_free_msg (mq, &msg);
1540         }
1541     }
1542
1543   if (s->is_dgram)
1544     n_read = app_recv_dgram_raw (rx_fifo, buf, n, &s->transport, 0, peek);
1545   else
1546     n_read = app_recv_stream_raw (rx_fifo, buf, n, 0, peek);
1547
1548   if (svm_fifo_is_empty_cons (rx_fifo))
1549     svm_fifo_unset_event (s->rx_fifo);
1550
1551   /* Cut-through sessions might request tx notifications on rx fifos */
1552   if (PREDICT_FALSE (rx_fifo->want_deq_ntf))
1553     {
1554       app_send_io_evt_to_vpp (s->vpp_evt_q, s->rx_fifo->master_session_index,
1555                               SESSION_IO_EVT_RX, SVM_Q_WAIT);
1556       svm_fifo_reset_has_deq_ntf (s->rx_fifo);
1557     }
1558
1559   VDBG (2, "session %u[0x%llx]: read %d bytes from (%p)", s->session_index,
1560         s->vpp_handle, n_read, rx_fifo);
1561
1562   return n_read;
1563 }
1564
1565 int
1566 vppcom_session_read (uint32_t session_handle, void *buf, size_t n)
1567 {
1568   return (vppcom_session_read_internal (session_handle, buf, n, 0));
1569 }
1570
1571 static int
1572 vppcom_session_peek (uint32_t session_handle, void *buf, int n)
1573 {
1574   return (vppcom_session_read_internal (session_handle, buf, n, 1));
1575 }
1576
1577 int
1578 vppcom_session_read_segments (uint32_t session_handle,
1579                               vppcom_data_segments_t ds)
1580 {
1581   vcl_worker_t *wrk = vcl_worker_get_current ();
1582   int n_read = 0, is_nonblocking;
1583   vcl_session_t *s = 0;
1584   svm_fifo_t *rx_fifo;
1585   svm_msg_q_msg_t msg;
1586   session_event_t *e;
1587   svm_msg_q_t *mq;
1588   u8 is_ct;
1589
1590   s = vcl_session_get_w_handle (wrk, session_handle);
1591   if (PREDICT_FALSE (!s || s->is_vep))
1592     return VPPCOM_EBADFD;
1593
1594   if (PREDICT_FALSE (!vcl_session_is_open (s)))
1595     return vcl_session_closed_error (s);
1596
1597   is_nonblocking = VCL_SESS_ATTR_TEST (s->attr, VCL_SESS_ATTR_NONBLOCK);
1598   is_ct = vcl_session_is_ct (s);
1599   mq = is_ct ? s->our_evt_q : wrk->app_event_queue;
1600   rx_fifo = s->rx_fifo;
1601   s->has_rx_evt = 0;
1602
1603   if (is_ct)
1604     svm_fifo_unset_event (s->rx_fifo);
1605
1606   if (svm_fifo_is_empty_cons (rx_fifo))
1607     {
1608       if (is_nonblocking)
1609         {
1610           svm_fifo_unset_event (rx_fifo);
1611           return VPPCOM_EWOULDBLOCK;
1612         }
1613       while (svm_fifo_is_empty_cons (rx_fifo))
1614         {
1615           if (vcl_session_is_closing (s))
1616             return vcl_session_closing_error (s);
1617
1618           svm_fifo_unset_event (rx_fifo);
1619           svm_msg_q_lock (mq);
1620           if (svm_msg_q_is_empty (mq))
1621             svm_msg_q_wait (mq);
1622
1623           svm_msg_q_sub_w_lock (mq, &msg);
1624           e = svm_msg_q_msg_data (mq, &msg);
1625           svm_msg_q_unlock (mq);
1626           if (!vcl_is_rx_evt_for_session (e, s->session_index, is_ct))
1627             vcl_handle_mq_event (wrk, e);
1628           svm_msg_q_free_msg (mq, &msg);
1629         }
1630     }
1631
1632   n_read = svm_fifo_segments (rx_fifo, (svm_fifo_seg_t *) ds);
1633   svm_fifo_unset_event (rx_fifo);
1634
1635   return n_read;
1636 }
1637
1638 void
1639 vppcom_session_free_segments (uint32_t session_handle,
1640                               vppcom_data_segments_t ds)
1641 {
1642   vcl_worker_t *wrk = vcl_worker_get_current ();
1643   vcl_session_t *s;
1644
1645   s = vcl_session_get_w_handle (wrk, session_handle);
1646   if (PREDICT_FALSE (!s || s->is_vep))
1647     return;
1648
1649   svm_fifo_segments_free (s->rx_fifo, (svm_fifo_seg_t *) ds);
1650 }
1651
1652 int
1653 vppcom_data_segment_copy (void *buf, vppcom_data_segments_t ds, u32 max_bytes)
1654 {
1655   u32 first_copy = clib_min (ds[0].len, max_bytes);
1656   clib_memcpy_fast (buf, ds[0].data, first_copy);
1657   if (first_copy < max_bytes)
1658     {
1659       clib_memcpy_fast (buf + first_copy, ds[1].data,
1660                         clib_min (ds[1].len, max_bytes - first_copy));
1661     }
1662   return 0;
1663 }
1664
1665 static u8
1666 vcl_is_tx_evt_for_session (session_event_t * e, u32 sid, u8 is_ct)
1667 {
1668   return (e->event_type == SESSION_IO_EVT_TX && e->session_index == sid);
1669 }
1670
1671 static inline int
1672 vppcom_session_write_inline (uint32_t session_handle, void *buf, size_t n,
1673                              u8 is_flush)
1674 {
1675   vcl_worker_t *wrk = vcl_worker_get_current ();
1676   int n_write, is_nonblocking;
1677   vcl_session_t *s = 0;
1678   session_evt_type_t et;
1679   svm_msg_q_msg_t msg;
1680   svm_fifo_t *tx_fifo;
1681   session_event_t *e;
1682   svm_msg_q_t *mq;
1683   u8 is_ct;
1684
1685   if (PREDICT_FALSE (!buf))
1686     return VPPCOM_EINVAL;
1687
1688   s = vcl_session_get_w_handle (wrk, session_handle);
1689   if (PREDICT_FALSE (!s))
1690     return VPPCOM_EBADFD;
1691
1692   if (PREDICT_FALSE (s->is_vep))
1693     {
1694       VDBG (0, "ERROR: session %u [0x%llx]: cannot write to an epoll"
1695             " session!", s->session_index, s->vpp_handle);
1696       return VPPCOM_EBADFD;
1697     }
1698
1699   if (PREDICT_FALSE (!vcl_session_is_open (s)))
1700     {
1701       VDBG (1, "session %u [0x%llx]: is not open! state 0x%x (%s)",
1702             s->session_index, s->vpp_handle, s->session_state,
1703             vppcom_session_state_str (s->session_state));
1704       return vcl_session_closed_error (s);;
1705     }
1706
1707   is_ct = vcl_session_is_ct (s);
1708   tx_fifo = is_ct ? s->ct_tx_fifo : s->tx_fifo;
1709   is_nonblocking = VCL_SESS_ATTR_TEST (s->attr, VCL_SESS_ATTR_NONBLOCK);
1710
1711   mq = wrk->app_event_queue;
1712   if (svm_fifo_is_full_prod (tx_fifo))
1713     {
1714       if (is_nonblocking)
1715         {
1716           return VPPCOM_EWOULDBLOCK;
1717         }
1718       while (svm_fifo_is_full_prod (tx_fifo))
1719         {
1720           svm_fifo_add_want_deq_ntf (tx_fifo, SVM_FIFO_WANT_DEQ_NOTIF);
1721           if (vcl_session_is_closing (s))
1722             return vcl_session_closing_error (s);
1723           svm_msg_q_lock (mq);
1724           if (svm_msg_q_is_empty (mq))
1725             svm_msg_q_wait (mq);
1726
1727           svm_msg_q_sub_w_lock (mq, &msg);
1728           e = svm_msg_q_msg_data (mq, &msg);
1729           svm_msg_q_unlock (mq);
1730
1731           if (!vcl_is_tx_evt_for_session (e, s->session_index, is_ct))
1732             vcl_handle_mq_event (wrk, e);
1733           svm_msg_q_free_msg (mq, &msg);
1734         }
1735     }
1736
1737   et = SESSION_IO_EVT_TX;
1738   if (is_flush && !is_ct)
1739     et = SESSION_IO_EVT_TX_FLUSH;
1740
1741   if (s->is_dgram)
1742     n_write = app_send_dgram_raw (tx_fifo, &s->transport,
1743                                   s->vpp_evt_q, buf, n, et,
1744                                   0 /* do_evt */ , SVM_Q_WAIT);
1745   else
1746     n_write = app_send_stream_raw (tx_fifo, s->vpp_evt_q, buf, n, et,
1747                                    0 /* do_evt */ , SVM_Q_WAIT);
1748
1749   if (svm_fifo_set_event (s->tx_fifo))
1750     app_send_io_evt_to_vpp (s->vpp_evt_q, s->tx_fifo->master_session_index,
1751                             et, SVM_Q_WAIT);
1752
1753   ASSERT (n_write > 0);
1754
1755   VDBG (2, "session %u [0x%llx]: wrote %d bytes", s->session_index,
1756         s->vpp_handle, n_write);
1757
1758   return n_write;
1759 }
1760
1761 int
1762 vppcom_session_write (uint32_t session_handle, void *buf, size_t n)
1763 {
1764   return vppcom_session_write_inline (session_handle, buf, n,
1765                                       0 /* is_flush */ );
1766 }
1767
1768 int
1769 vppcom_session_write_msg (uint32_t session_handle, void *buf, size_t n)
1770 {
1771   return vppcom_session_write_inline (session_handle, buf, n,
1772                                       1 /* is_flush */ );
1773 }
1774
1775 #define vcl_fifo_rx_evt_valid_or_break(_s)                              \
1776 if (PREDICT_FALSE (svm_fifo_is_empty (_s->rx_fifo)))                    \
1777   {                                                                     \
1778     if (!vcl_session_is_ct (_s))                                        \
1779       {                                                                 \
1780         svm_fifo_unset_event (_s->rx_fifo);                             \
1781         if (svm_fifo_is_empty (_s->rx_fifo))                            \
1782           break;                                                        \
1783       }                                                                 \
1784     else if (svm_fifo_is_empty (_s->ct_rx_fifo))                        \
1785       {                                                                 \
1786         svm_fifo_unset_event (_s->ct_rx_fifo);                          \
1787         if (svm_fifo_is_empty (_s->ct_rx_fifo))                         \
1788           break;                                                        \
1789       }                                                                 \
1790   }                                                                     \
1791
1792 static void
1793 vcl_select_handle_mq_event (vcl_worker_t * wrk, session_event_t * e,
1794                             unsigned long n_bits, unsigned long *read_map,
1795                             unsigned long *write_map,
1796                             unsigned long *except_map, u32 * bits_set)
1797 {
1798   session_disconnected_msg_t *disconnected_msg;
1799   session_connected_msg_t *connected_msg;
1800   vcl_session_t *session;
1801   u32 sid;
1802
1803   switch (e->event_type)
1804     {
1805     case SESSION_IO_EVT_RX:
1806       sid = e->session_index;
1807       session = vcl_session_get (wrk, sid);
1808       if (!session)
1809         break;
1810       vcl_fifo_rx_evt_valid_or_break (session);
1811       if (sid < n_bits && read_map)
1812         {
1813           clib_bitmap_set_no_check ((uword *) read_map, sid, 1);
1814           *bits_set += 1;
1815         }
1816       break;
1817     case SESSION_IO_EVT_TX:
1818       sid = e->session_index;
1819       session = vcl_session_get (wrk, sid);
1820       if (!session)
1821         break;
1822       if (sid < n_bits && write_map)
1823         {
1824           clib_bitmap_set_no_check ((uword *) write_map, sid, 1);
1825           *bits_set += 1;
1826         }
1827       break;
1828     case SESSION_CTRL_EVT_ACCEPTED:
1829       session = vcl_session_accepted (wrk,
1830                                       (session_accepted_msg_t *) e->data);
1831       if (!session)
1832         break;
1833       sid = session->session_index;
1834       if (sid < n_bits && read_map)
1835         {
1836           clib_bitmap_set_no_check ((uword *) read_map, sid, 1);
1837           *bits_set += 1;
1838         }
1839       break;
1840     case SESSION_CTRL_EVT_CONNECTED:
1841       connected_msg = (session_connected_msg_t *) e->data;
1842       vcl_session_connected_handler (wrk, connected_msg);
1843       break;
1844     case SESSION_CTRL_EVT_DISCONNECTED:
1845       disconnected_msg = (session_disconnected_msg_t *) e->data;
1846       session = vcl_session_disconnected_handler (wrk, disconnected_msg);
1847       if (!session)
1848         break;
1849       sid = session->session_index;
1850       if (sid < n_bits && except_map)
1851         {
1852           clib_bitmap_set_no_check ((uword *) except_map, sid, 1);
1853           *bits_set += 1;
1854         }
1855       break;
1856     case SESSION_CTRL_EVT_RESET:
1857       sid = vcl_session_reset_handler (wrk, (session_reset_msg_t *) e->data);
1858       if (sid < n_bits && except_map)
1859         {
1860           clib_bitmap_set_no_check ((uword *) except_map, sid, 1);
1861           *bits_set += 1;
1862         }
1863       break;
1864     case SESSION_CTRL_EVT_UNLISTEN_REPLY:
1865       vcl_session_unlisten_reply_handler (wrk, e->data);
1866       break;
1867     case SESSION_CTRL_EVT_WORKER_UPDATE_REPLY:
1868       vcl_session_worker_update_reply_handler (wrk, e->data);
1869       break;
1870     case SESSION_CTRL_EVT_REQ_WORKER_UPDATE:
1871       vcl_session_req_worker_update_handler (wrk, e->data);
1872       break;
1873     default:
1874       clib_warning ("unhandled: %u", e->event_type);
1875       break;
1876     }
1877 }
1878
1879 static int
1880 vcl_select_handle_mq (vcl_worker_t * wrk, svm_msg_q_t * mq,
1881                       unsigned long n_bits, unsigned long *read_map,
1882                       unsigned long *write_map, unsigned long *except_map,
1883                       double time_to_wait, u32 * bits_set)
1884 {
1885   svm_msg_q_msg_t *msg;
1886   session_event_t *e;
1887   u32 i;
1888
1889   svm_msg_q_lock (mq);
1890   if (svm_msg_q_is_empty (mq))
1891     {
1892       if (*bits_set)
1893         {
1894           svm_msg_q_unlock (mq);
1895           return 0;
1896         }
1897
1898       if (!time_to_wait)
1899         {
1900           svm_msg_q_unlock (mq);
1901           return 0;
1902         }
1903       else if (time_to_wait < 0)
1904         {
1905           svm_msg_q_wait (mq);
1906         }
1907       else
1908         {
1909           if (svm_msg_q_timedwait (mq, time_to_wait))
1910             {
1911               svm_msg_q_unlock (mq);
1912               return 0;
1913             }
1914         }
1915     }
1916   vcl_mq_dequeue_batch (wrk, mq, ~0);
1917   svm_msg_q_unlock (mq);
1918
1919   for (i = 0; i < vec_len (wrk->mq_msg_vector); i++)
1920     {
1921       msg = vec_elt_at_index (wrk->mq_msg_vector, i);
1922       e = svm_msg_q_msg_data (mq, msg);
1923       vcl_select_handle_mq_event (wrk, e, n_bits, read_map, write_map,
1924                                   except_map, bits_set);
1925       svm_msg_q_free_msg (mq, msg);
1926     }
1927   vec_reset_length (wrk->mq_msg_vector);
1928   vcl_handle_pending_wrk_updates (wrk);
1929   return *bits_set;
1930 }
1931
1932 static int
1933 vppcom_select_condvar (vcl_worker_t * wrk, int n_bits,
1934                        vcl_si_set * read_map, vcl_si_set * write_map,
1935                        vcl_si_set * except_map, double time_to_wait,
1936                        u32 * bits_set)
1937 {
1938   double wait = 0, start = 0;
1939
1940   if (!*bits_set)
1941     {
1942       wait = time_to_wait;
1943       start = clib_time_now (&wrk->clib_time);
1944     }
1945
1946   do
1947     {
1948       vcl_select_handle_mq (wrk, wrk->app_event_queue, n_bits, read_map,
1949                             write_map, except_map, wait, bits_set);
1950       if (*bits_set)
1951         return *bits_set;
1952       if (wait == -1)
1953         continue;
1954
1955       wait = wait - (clib_time_now (&wrk->clib_time) - start);
1956     }
1957   while (wait > 0);
1958
1959   return 0;
1960 }
1961
1962 static int
1963 vppcom_select_eventfd (vcl_worker_t * wrk, int n_bits,
1964                        vcl_si_set * read_map, vcl_si_set * write_map,
1965                        vcl_si_set * except_map, double time_to_wait,
1966                        u32 * bits_set)
1967 {
1968   vcl_mq_evt_conn_t *mqc;
1969   int __clib_unused n_read;
1970   int n_mq_evts, i;
1971   u64 buf;
1972
1973   vec_validate (wrk->mq_events, pool_elts (wrk->mq_evt_conns));
1974   n_mq_evts = epoll_wait (wrk->mqs_epfd, wrk->mq_events,
1975                           vec_len (wrk->mq_events), time_to_wait);
1976   for (i = 0; i < n_mq_evts; i++)
1977     {
1978       mqc = vcl_mq_evt_conn_get (wrk, wrk->mq_events[i].data.u32);
1979       n_read = read (mqc->mq_fd, &buf, sizeof (buf));
1980       vcl_select_handle_mq (wrk, mqc->mq, n_bits, read_map, write_map,
1981                             except_map, 0, bits_set);
1982     }
1983
1984   return (n_mq_evts > 0 ? (int) *bits_set : 0);
1985 }
1986
1987 int
1988 vppcom_select (int n_bits, vcl_si_set * read_map, vcl_si_set * write_map,
1989                vcl_si_set * except_map, double time_to_wait)
1990 {
1991   u32 sid, minbits = clib_max (n_bits, BITS (uword)), bits_set = 0;
1992   vcl_worker_t *wrk = vcl_worker_get_current ();
1993   vcl_session_t *session = 0;
1994   int rv, i;
1995
1996   if (n_bits && read_map)
1997     {
1998       clib_bitmap_validate (wrk->rd_bitmap, minbits);
1999       clib_memcpy_fast (wrk->rd_bitmap, read_map,
2000                         vec_len (wrk->rd_bitmap) * sizeof (vcl_si_set));
2001       memset (read_map, 0, vec_len (wrk->rd_bitmap) * sizeof (vcl_si_set));
2002     }
2003   if (n_bits && write_map)
2004     {
2005       clib_bitmap_validate (wrk->wr_bitmap, minbits);
2006       clib_memcpy_fast (wrk->wr_bitmap, write_map,
2007                         vec_len (wrk->wr_bitmap) * sizeof (vcl_si_set));
2008       memset (write_map, 0, vec_len (wrk->wr_bitmap) * sizeof (vcl_si_set));
2009     }
2010   if (n_bits && except_map)
2011     {
2012       clib_bitmap_validate (wrk->ex_bitmap, minbits);
2013       clib_memcpy_fast (wrk->ex_bitmap, except_map,
2014                         vec_len (wrk->ex_bitmap) * sizeof (vcl_si_set));
2015       memset (except_map, 0, vec_len (wrk->ex_bitmap) * sizeof (vcl_si_set));
2016     }
2017
2018   if (!n_bits)
2019     return 0;
2020
2021   if (!write_map)
2022     goto check_rd;
2023
2024   /* *INDENT-OFF* */
2025   clib_bitmap_foreach (sid, wrk->wr_bitmap, ({
2026     if (!(session = vcl_session_get (wrk, sid)))
2027       {
2028         if (except_map && sid < minbits)
2029           clib_bitmap_set_no_check (except_map, sid, 1);
2030         continue;
2031       }
2032
2033     rv = svm_fifo_is_full_prod (session->tx_fifo);
2034     if (!rv)
2035       {
2036         clib_bitmap_set_no_check ((uword*)write_map, sid, 1);
2037         bits_set++;
2038       }
2039     else
2040       svm_fifo_add_want_deq_ntf (session->tx_fifo, SVM_FIFO_WANT_DEQ_NOTIF);
2041   }));
2042
2043 check_rd:
2044   if (!read_map)
2045     goto check_mq;
2046
2047   clib_bitmap_foreach (sid, wrk->rd_bitmap, ({
2048     if (!(session = vcl_session_get (wrk, sid)))
2049       {
2050         if (except_map && sid < minbits)
2051           clib_bitmap_set_no_check (except_map, sid, 1);
2052         continue;
2053       }
2054
2055     rv = vcl_session_read_ready (session);
2056     if (rv)
2057       {
2058         clib_bitmap_set_no_check ((uword*)read_map, sid, 1);
2059         bits_set++;
2060       }
2061   }));
2062   /* *INDENT-ON* */
2063
2064 check_mq:
2065
2066   for (i = 0; i < vec_len (wrk->unhandled_evts_vector); i++)
2067     {
2068       vcl_select_handle_mq_event (wrk, &wrk->unhandled_evts_vector[i], n_bits,
2069                                   read_map, write_map, except_map, &bits_set);
2070     }
2071   vec_reset_length (wrk->unhandled_evts_vector);
2072
2073   if (vcm->cfg.use_mq_eventfd)
2074     vppcom_select_eventfd (wrk, n_bits, read_map, write_map, except_map,
2075                            time_to_wait, &bits_set);
2076   else
2077     vppcom_select_condvar (wrk, n_bits, read_map, write_map, except_map,
2078                            time_to_wait, &bits_set);
2079
2080   return (bits_set);
2081 }
2082
2083 static inline void
2084 vep_verify_epoll_chain (vcl_worker_t * wrk, u32 vep_idx)
2085 {
2086   vcl_session_t *session;
2087   vppcom_epoll_t *vep;
2088   u32 sid = vep_idx;
2089
2090   if (VPPCOM_DEBUG <= 2)
2091     return;
2092
2093   session = vcl_session_get (wrk, vep_idx);
2094   if (PREDICT_FALSE (!session))
2095     {
2096       VDBG (0, "ERROR: Invalid vep_idx (%u)!", vep_idx);
2097       goto done;
2098     }
2099   if (PREDICT_FALSE (!session->is_vep))
2100     {
2101       VDBG (0, "ERROR: vep_idx (%u) is not a vep!", vep_idx);
2102       goto done;
2103     }
2104   vep = &session->vep;
2105   VDBG (0, "vep_idx (%u): Dumping epoll chain\n"
2106         "{\n"
2107         "   is_vep         = %u\n"
2108         "   is_vep_session = %u\n"
2109         "   next_sid       = 0x%x (%u)\n"
2110         "}\n", vep_idx, session->is_vep, session->is_vep_session,
2111         vep->next_sh, vep->next_sh);
2112
2113   for (sid = vep->next_sh; sid != ~0; sid = vep->next_sh)
2114     {
2115       session = vcl_session_get (wrk, sid);
2116       if (PREDICT_FALSE (!session))
2117         {
2118           VDBG (0, "ERROR: Invalid sid (%u)!", sid);
2119           goto done;
2120         }
2121       if (PREDICT_FALSE (session->is_vep))
2122         {
2123           VDBG (0, "ERROR: sid (%u) is a vep!", vep_idx);
2124         }
2125       else if (PREDICT_FALSE (!session->is_vep_session))
2126         {
2127           VDBG (0, "ERROR: session (%u) is not a vep session!", sid);
2128           goto done;
2129         }
2130       vep = &session->vep;
2131       if (PREDICT_FALSE (vep->vep_sh != vep_idx))
2132         VDBG (0, "ERROR: session (%u) vep_idx (%u) != vep_idx (%u)!",
2133               sid, session->vep.vep_sh, vep_idx);
2134       if (session->is_vep_session)
2135         {
2136           VDBG (0, "vep_idx[%u]: sid 0x%x (%u)\n"
2137                 "{\n"
2138                 "   next_sid       = 0x%x (%u)\n"
2139                 "   prev_sid       = 0x%x (%u)\n"
2140                 "   vep_idx        = 0x%x (%u)\n"
2141                 "   ev.events      = 0x%x\n"
2142                 "   ev.data.u64    = 0x%llx\n"
2143                 "   et_mask        = 0x%x\n"
2144                 "}\n",
2145                 vep_idx, sid, sid, vep->next_sh, vep->next_sh, vep->prev_sh,
2146                 vep->prev_sh, vep->vep_sh, vep->vep_sh, vep->ev.events,
2147                 vep->ev.data.u64, vep->et_mask);
2148         }
2149     }
2150
2151 done:
2152   VDBG (0, "vep_idx (%u): Dump complete!\n", vep_idx);
2153 }
2154
2155 int
2156 vppcom_epoll_create (void)
2157 {
2158   vcl_worker_t *wrk = vcl_worker_get_current ();
2159   vcl_session_t *vep_session;
2160
2161   vep_session = vcl_session_alloc (wrk);
2162
2163   vep_session->is_vep = 1;
2164   vep_session->vep.vep_sh = ~0;
2165   vep_session->vep.next_sh = ~0;
2166   vep_session->vep.prev_sh = ~0;
2167   vep_session->vpp_handle = ~0;
2168
2169   vcl_evt (VCL_EVT_EPOLL_CREATE, vep_session, vep_session->session_index);
2170   VDBG (0, "Created vep_idx %u", vep_session->session_index);
2171
2172   return vcl_session_handle (vep_session);
2173 }
2174
2175 int
2176 vppcom_epoll_ctl (uint32_t vep_handle, int op, uint32_t session_handle,
2177                   struct epoll_event *event)
2178 {
2179   vcl_worker_t *wrk = vcl_worker_get_current ();
2180   vcl_session_t *vep_session;
2181   vcl_session_t *session;
2182   int rv = VPPCOM_OK;
2183
2184   if (vep_handle == session_handle)
2185     {
2186       VDBG (0, "vep_sh == session handle (%u)!", vep_handle);
2187       return VPPCOM_EINVAL;
2188     }
2189
2190   vep_session = vcl_session_get_w_handle (wrk, vep_handle);
2191   if (PREDICT_FALSE (!vep_session))
2192     {
2193       VDBG (0, "Invalid vep_sh (%u)!", vep_handle);
2194       return VPPCOM_EBADFD;
2195     }
2196   if (PREDICT_FALSE (!vep_session->is_vep))
2197     {
2198       VDBG (0, "vep_sh (%u) is not a vep!", vep_handle);
2199       return VPPCOM_EINVAL;
2200     }
2201
2202   ASSERT (vep_session->vep.vep_sh == ~0);
2203   ASSERT (vep_session->vep.prev_sh == ~0);
2204
2205   session = vcl_session_get_w_handle (wrk, session_handle);
2206   if (PREDICT_FALSE (!session))
2207     {
2208       VDBG (0, "Invalid session_handle (%u)!", session_handle);
2209       return VPPCOM_EBADFD;
2210     }
2211   if (PREDICT_FALSE (session->is_vep))
2212     {
2213       VDBG (0, "session_handle (%u) is a vep!", vep_handle);
2214       return VPPCOM_EINVAL;
2215     }
2216
2217   switch (op)
2218     {
2219     case EPOLL_CTL_ADD:
2220       if (PREDICT_FALSE (!event))
2221         {
2222           VDBG (0, "EPOLL_CTL_ADD: NULL pointer to epoll_event structure!");
2223           return VPPCOM_EINVAL;
2224         }
2225       if (vep_session->vep.next_sh != ~0)
2226         {
2227           vcl_session_t *next_session;
2228           next_session = vcl_session_get_w_handle (wrk,
2229                                                    vep_session->vep.next_sh);
2230           if (PREDICT_FALSE (!next_session))
2231             {
2232               VDBG (0, "EPOLL_CTL_ADD: Invalid vep.next_sh (%u) on "
2233                     "vep_idx (%u)!", vep_session->vep.next_sh, vep_handle);
2234               return VPPCOM_EBADFD;
2235             }
2236           ASSERT (next_session->vep.prev_sh == vep_handle);
2237           next_session->vep.prev_sh = session_handle;
2238         }
2239       session->vep.next_sh = vep_session->vep.next_sh;
2240       session->vep.prev_sh = vep_handle;
2241       session->vep.vep_sh = vep_handle;
2242       session->vep.et_mask = VEP_DEFAULT_ET_MASK;
2243       session->vep.ev = *event;
2244       session->is_vep = 0;
2245       session->is_vep_session = 1;
2246       vep_session->vep.next_sh = session_handle;
2247
2248       if (session->tx_fifo)
2249         svm_fifo_add_want_deq_ntf (session->tx_fifo,
2250                                    SVM_FIFO_WANT_DEQ_NOTIF_IF_FULL);
2251
2252       VDBG (1, "EPOLL_CTL_ADD: vep_sh %u, sh %u, events 0x%x, data 0x%llx!",
2253             vep_handle, session_handle, event->events, event->data.u64);
2254       vcl_evt (VCL_EVT_EPOLL_CTLADD, session, event->events, event->data.u64);
2255       break;
2256
2257     case EPOLL_CTL_MOD:
2258       if (PREDICT_FALSE (!event))
2259         {
2260           VDBG (0, "EPOLL_CTL_MOD: NULL pointer to epoll_event structure!");
2261           rv = VPPCOM_EINVAL;
2262           goto done;
2263         }
2264       else if (PREDICT_FALSE (!session->is_vep_session))
2265         {
2266           VDBG (0, "sh %u EPOLL_CTL_MOD: not a vep session!", session_handle);
2267           rv = VPPCOM_EINVAL;
2268           goto done;
2269         }
2270       else if (PREDICT_FALSE (session->vep.vep_sh != vep_handle))
2271         {
2272           VDBG (0, "EPOLL_CTL_MOD: sh %u vep_sh (%u) != vep_sh (%u)!",
2273                 session_handle, session->vep.vep_sh, vep_handle);
2274           rv = VPPCOM_EINVAL;
2275           goto done;
2276         }
2277       session->vep.et_mask = VEP_DEFAULT_ET_MASK;
2278       session->vep.ev = *event;
2279       VDBG (1, "EPOLL_CTL_MOD: vep_sh %u, sh %u, events 0x%x, data 0x%llx!",
2280             vep_handle, session_handle, event->events, event->data.u64);
2281       break;
2282
2283     case EPOLL_CTL_DEL:
2284       if (PREDICT_FALSE (!session->is_vep_session))
2285         {
2286           VDBG (0, "EPOLL_CTL_DEL: %u not a vep session!", session_handle);
2287           rv = VPPCOM_EINVAL;
2288           goto done;
2289         }
2290       else if (PREDICT_FALSE (session->vep.vep_sh != vep_handle))
2291         {
2292           VDBG (0, "EPOLL_CTL_DEL: sh %u vep_sh (%u) != vep_sh (%u)!",
2293                 session_handle, session->vep.vep_sh, vep_handle);
2294           rv = VPPCOM_EINVAL;
2295           goto done;
2296         }
2297
2298       if (session->vep.prev_sh == vep_handle)
2299         vep_session->vep.next_sh = session->vep.next_sh;
2300       else
2301         {
2302           vcl_session_t *prev_session;
2303           prev_session = vcl_session_get_w_handle (wrk, session->vep.prev_sh);
2304           if (PREDICT_FALSE (!prev_session))
2305             {
2306               VDBG (0, "EPOLL_CTL_DEL: Invalid prev_sh (%u) on sh (%u)!",
2307                     session->vep.prev_sh, session_handle);
2308               return VPPCOM_EBADFD;
2309             }
2310           ASSERT (prev_session->vep.next_sh == session_handle);
2311           prev_session->vep.next_sh = session->vep.next_sh;
2312         }
2313       if (session->vep.next_sh != ~0)
2314         {
2315           vcl_session_t *next_session;
2316           next_session = vcl_session_get_w_handle (wrk, session->vep.next_sh);
2317           if (PREDICT_FALSE (!next_session))
2318             {
2319               VDBG (0, "EPOLL_CTL_DEL: Invalid next_sh (%u) on sh (%u)!",
2320                     session->vep.next_sh, session_handle);
2321               return VPPCOM_EBADFD;
2322             }
2323           ASSERT (next_session->vep.prev_sh == session_handle);
2324           next_session->vep.prev_sh = session->vep.prev_sh;
2325         }
2326
2327       memset (&session->vep, 0, sizeof (session->vep));
2328       session->vep.next_sh = ~0;
2329       session->vep.prev_sh = ~0;
2330       session->vep.vep_sh = ~0;
2331       session->is_vep_session = 0;
2332
2333       if (session->tx_fifo)
2334         svm_fifo_del_want_deq_ntf (session->tx_fifo, SVM_FIFO_NO_DEQ_NOTIF);
2335
2336       VDBG (1, "EPOLL_CTL_DEL: vep_idx %u, sh %u!", vep_handle,
2337             session_handle);
2338       vcl_evt (VCL_EVT_EPOLL_CTLDEL, session, vep_sh);
2339       break;
2340
2341     default:
2342       VDBG (0, "Invalid operation (%d)!", op);
2343       rv = VPPCOM_EINVAL;
2344     }
2345
2346   vep_verify_epoll_chain (wrk, vep_handle);
2347
2348 done:
2349   return rv;
2350 }
2351
2352 static inline void
2353 vcl_epoll_wait_handle_mq_event (vcl_worker_t * wrk, session_event_t * e,
2354                                 struct epoll_event *events, u32 * num_ev)
2355 {
2356   session_disconnected_msg_t *disconnected_msg;
2357   session_connected_msg_t *connected_msg;
2358   u32 sid = ~0, session_events;
2359   u64 session_evt_data = ~0;
2360   vcl_session_t *session;
2361   u8 add_event = 0;
2362
2363   switch (e->event_type)
2364     {
2365     case SESSION_IO_EVT_RX:
2366       sid = e->session_index;
2367       if (!(session = vcl_session_get (wrk, sid)))
2368         break;
2369       vcl_fifo_rx_evt_valid_or_break (session);
2370       session_events = session->vep.ev.events;
2371       if (!(EPOLLIN & session->vep.ev.events) || session->has_rx_evt)
2372         break;
2373       add_event = 1;
2374       events[*num_ev].events |= EPOLLIN;
2375       session_evt_data = session->vep.ev.data.u64;
2376       session->has_rx_evt = 1;
2377       break;
2378     case SESSION_IO_EVT_TX:
2379       sid = e->session_index;
2380       if (!(session = vcl_session_get (wrk, sid)))
2381         break;
2382       session_events = session->vep.ev.events;
2383       if (!(EPOLLOUT & session_events))
2384         break;
2385       add_event = 1;
2386       events[*num_ev].events |= EPOLLOUT;
2387       session_evt_data = session->vep.ev.data.u64;
2388       svm_fifo_reset_has_deq_ntf (session->tx_fifo);
2389       break;
2390     case SESSION_CTRL_EVT_ACCEPTED:
2391       session = vcl_session_accepted (wrk,
2392                                       (session_accepted_msg_t *) e->data);
2393       if (!session)
2394         break;
2395
2396       session_events = session->vep.ev.events;
2397       if (!(EPOLLIN & session_events))
2398         break;
2399
2400       add_event = 1;
2401       events[*num_ev].events |= EPOLLIN;
2402       session_evt_data = session->vep.ev.data.u64;
2403       break;
2404     case SESSION_CTRL_EVT_CONNECTED:
2405       connected_msg = (session_connected_msg_t *) e->data;
2406       vcl_session_connected_handler (wrk, connected_msg);
2407       /* Generate EPOLLOUT because there's no connected event */
2408       sid = vcl_session_index_from_vpp_handle (wrk, connected_msg->handle);
2409       if (!(session = vcl_session_get (wrk, sid)))
2410         break;
2411       session_events = session->vep.ev.events;
2412       if (!(EPOLLOUT & session_events))
2413         break;
2414       add_event = 1;
2415       events[*num_ev].events |= EPOLLOUT;
2416       session_evt_data = session->vep.ev.data.u64;
2417       break;
2418     case SESSION_CTRL_EVT_DISCONNECTED:
2419       disconnected_msg = (session_disconnected_msg_t *) e->data;
2420       session = vcl_session_disconnected_handler (wrk, disconnected_msg);
2421       if (!session)
2422         break;
2423       session_events = session->vep.ev.events;
2424       if (!((EPOLLHUP | EPOLLRDHUP) & session_events))
2425         break;
2426       add_event = 1;
2427       events[*num_ev].events |= EPOLLHUP | EPOLLRDHUP;
2428       session_evt_data = session->vep.ev.data.u64;
2429       break;
2430     case SESSION_CTRL_EVT_RESET:
2431       sid = vcl_session_reset_handler (wrk, (session_reset_msg_t *) e->data);
2432       if (!(session = vcl_session_get (wrk, sid)))
2433         break;
2434       session_events = session->vep.ev.events;
2435       if (!((EPOLLHUP | EPOLLRDHUP) & session_events))
2436         break;
2437       add_event = 1;
2438       events[*num_ev].events |= EPOLLHUP | EPOLLRDHUP;
2439       session_evt_data = session->vep.ev.data.u64;
2440       break;
2441     case SESSION_CTRL_EVT_UNLISTEN_REPLY:
2442       vcl_session_unlisten_reply_handler (wrk, e->data);
2443       break;
2444     case SESSION_CTRL_EVT_REQ_WORKER_UPDATE:
2445       vcl_session_req_worker_update_handler (wrk, e->data);
2446       break;
2447     case SESSION_CTRL_EVT_WORKER_UPDATE_REPLY:
2448       vcl_session_worker_update_reply_handler (wrk, e->data);
2449       break;
2450     default:
2451       VDBG (0, "unhandled: %u", e->event_type);
2452       break;
2453     }
2454
2455   if (add_event)
2456     {
2457       events[*num_ev].data.u64 = session_evt_data;
2458       if (EPOLLONESHOT & session_events)
2459         {
2460           session = vcl_session_get (wrk, sid);
2461           session->vep.ev.events = 0;
2462         }
2463       *num_ev += 1;
2464     }
2465 }
2466
2467 static int
2468 vcl_epoll_wait_handle_mq (vcl_worker_t * wrk, svm_msg_q_t * mq,
2469                           struct epoll_event *events, u32 maxevents,
2470                           double wait_for_time, u32 * num_ev)
2471 {
2472   svm_msg_q_msg_t *msg;
2473   session_event_t *e;
2474   int i;
2475
2476   if (vec_len (wrk->mq_msg_vector) && svm_msg_q_is_empty (mq))
2477     goto handle_dequeued;
2478
2479   svm_msg_q_lock (mq);
2480   if (svm_msg_q_is_empty (mq))
2481     {
2482       if (!wait_for_time)
2483         {
2484           svm_msg_q_unlock (mq);
2485           return 0;
2486         }
2487       else if (wait_for_time < 0)
2488         {
2489           svm_msg_q_wait (mq);
2490         }
2491       else
2492         {
2493           if (svm_msg_q_timedwait (mq, wait_for_time / 1e3))
2494             {
2495               svm_msg_q_unlock (mq);
2496               return 0;
2497             }
2498         }
2499     }
2500   ASSERT (maxevents > *num_ev);
2501   vcl_mq_dequeue_batch (wrk, mq, maxevents - *num_ev);
2502   svm_msg_q_unlock (mq);
2503
2504 handle_dequeued:
2505   for (i = 0; i < vec_len (wrk->mq_msg_vector); i++)
2506     {
2507       msg = vec_elt_at_index (wrk->mq_msg_vector, i);
2508       e = svm_msg_q_msg_data (mq, msg);
2509       vcl_epoll_wait_handle_mq_event (wrk, e, events, num_ev);
2510       svm_msg_q_free_msg (mq, msg);
2511     }
2512   vec_reset_length (wrk->mq_msg_vector);
2513   vcl_handle_pending_wrk_updates (wrk);
2514   return *num_ev;
2515 }
2516
2517 static int
2518 vppcom_epoll_wait_condvar (vcl_worker_t * wrk, struct epoll_event *events,
2519                            int maxevents, u32 n_evts, double wait_for_time)
2520 {
2521   double wait = 0, start = 0, now;
2522
2523   if (!n_evts)
2524     {
2525       wait = wait_for_time;
2526       start = clib_time_now (&wrk->clib_time);
2527     }
2528
2529   do
2530     {
2531       vcl_epoll_wait_handle_mq (wrk, wrk->app_event_queue, events, maxevents,
2532                                 wait, &n_evts);
2533       if (n_evts)
2534         return n_evts;
2535       if (wait == -1)
2536         continue;
2537
2538       now = clib_time_now (&wrk->clib_time);
2539       wait -= now - start;
2540       start = now;
2541     }
2542   while (wait > 0);
2543
2544   return 0;
2545 }
2546
2547 static int
2548 vppcom_epoll_wait_eventfd (vcl_worker_t * wrk, struct epoll_event *events,
2549                            int maxevents, u32 n_evts, double wait_for_time)
2550 {
2551   vcl_mq_evt_conn_t *mqc;
2552   int __clib_unused n_read;
2553   int n_mq_evts, i;
2554   u64 buf;
2555
2556   vec_validate (wrk->mq_events, pool_elts (wrk->mq_evt_conns));
2557 again:
2558   n_mq_evts = epoll_wait (wrk->mqs_epfd, wrk->mq_events,
2559                           vec_len (wrk->mq_events), wait_for_time);
2560   for (i = 0; i < n_mq_evts; i++)
2561     {
2562       mqc = vcl_mq_evt_conn_get (wrk, wrk->mq_events[i].data.u32);
2563       n_read = read (mqc->mq_fd, &buf, sizeof (buf));
2564       vcl_epoll_wait_handle_mq (wrk, mqc->mq, events, maxevents, 0, &n_evts);
2565     }
2566   if (!n_evts && n_mq_evts > 0)
2567     goto again;
2568
2569   return (int) n_evts;
2570 }
2571
2572 int
2573 vppcom_epoll_wait (uint32_t vep_handle, struct epoll_event *events,
2574                    int maxevents, double wait_for_time)
2575 {
2576   vcl_worker_t *wrk = vcl_worker_get_current ();
2577   vcl_session_t *vep_session;
2578   u32 n_evts = 0;
2579   int i;
2580
2581   if (PREDICT_FALSE (maxevents <= 0))
2582     {
2583       VDBG (0, "ERROR: Invalid maxevents (%d)!", maxevents);
2584       return VPPCOM_EINVAL;
2585     }
2586
2587   vep_session = vcl_session_get_w_handle (wrk, vep_handle);
2588   if (!vep_session)
2589     return VPPCOM_EBADFD;
2590
2591   if (PREDICT_FALSE (!vep_session->is_vep))
2592     {
2593       VDBG (0, "ERROR: vep_idx (%u) is not a vep!", vep_handle);
2594       return VPPCOM_EINVAL;
2595     }
2596
2597   memset (events, 0, sizeof (*events) * maxevents);
2598
2599   if (vec_len (wrk->unhandled_evts_vector))
2600     {
2601       for (i = 0; i < vec_len (wrk->unhandled_evts_vector); i++)
2602         {
2603           vcl_epoll_wait_handle_mq_event (wrk, &wrk->unhandled_evts_vector[i],
2604                                           events, &n_evts);
2605           if (n_evts == maxevents)
2606             {
2607               vec_delete (wrk->unhandled_evts_vector, i + 1, 0);
2608               return n_evts;
2609             }
2610         }
2611       vec_reset_length (wrk->unhandled_evts_vector);
2612     }
2613
2614   if (vcm->cfg.use_mq_eventfd)
2615     return vppcom_epoll_wait_eventfd (wrk, events, maxevents, n_evts,
2616                                       wait_for_time);
2617
2618   return vppcom_epoll_wait_condvar (wrk, events, maxevents, n_evts,
2619                                     wait_for_time);
2620 }
2621
2622 int
2623 vppcom_session_attr (uint32_t session_handle, uint32_t op,
2624                      void *buffer, uint32_t * buflen)
2625 {
2626   vcl_worker_t *wrk = vcl_worker_get_current ();
2627   vcl_session_t *session;
2628   int rv = VPPCOM_OK;
2629   u32 *flags = buffer, tmp_flags = 0;
2630   vppcom_endpt_t *ep = buffer;
2631
2632   session = vcl_session_get_w_handle (wrk, session_handle);
2633   if (!session)
2634     return VPPCOM_EBADFD;
2635
2636   switch (op)
2637     {
2638     case VPPCOM_ATTR_GET_NREAD:
2639       rv = vcl_session_read_ready (session);
2640       VDBG (2, "VPPCOM_ATTR_GET_NREAD: sh %u, nread = %d", session_handle,
2641             rv);
2642       break;
2643
2644     case VPPCOM_ATTR_GET_NWRITE:
2645       rv = vcl_session_write_ready (session);
2646       VDBG (2, "VPPCOM_ATTR_GET_NWRITE: sh %u, nwrite = %d", session_handle,
2647             rv);
2648       break;
2649
2650     case VPPCOM_ATTR_GET_FLAGS:
2651       if (PREDICT_TRUE (buffer && buflen && (*buflen >= sizeof (*flags))))
2652         {
2653           *flags = O_RDWR | (VCL_SESS_ATTR_TEST (session->attr,
2654                                                  VCL_SESS_ATTR_NONBLOCK));
2655           *buflen = sizeof (*flags);
2656           VDBG (2, "VPPCOM_ATTR_GET_FLAGS: sh %u, flags = 0x%08x, "
2657                 "is_nonblocking = %u", session_handle, *flags,
2658                 VCL_SESS_ATTR_TEST (session->attr, VCL_SESS_ATTR_NONBLOCK));
2659         }
2660       else
2661         rv = VPPCOM_EINVAL;
2662       break;
2663
2664     case VPPCOM_ATTR_SET_FLAGS:
2665       if (PREDICT_TRUE (buffer && buflen && (*buflen == sizeof (*flags))))
2666         {
2667           if (*flags & O_NONBLOCK)
2668             VCL_SESS_ATTR_SET (session->attr, VCL_SESS_ATTR_NONBLOCK);
2669           else
2670             VCL_SESS_ATTR_CLR (session->attr, VCL_SESS_ATTR_NONBLOCK);
2671
2672           VDBG (2, "VPPCOM_ATTR_SET_FLAGS: sh %u, flags = 0x%08x,"
2673                 " is_nonblocking = %u", session_handle, *flags,
2674                 VCL_SESS_ATTR_TEST (session->attr, VCL_SESS_ATTR_NONBLOCK));
2675         }
2676       else
2677         rv = VPPCOM_EINVAL;
2678       break;
2679
2680     case VPPCOM_ATTR_GET_PEER_ADDR:
2681       if (PREDICT_TRUE (buffer && buflen &&
2682                         (*buflen >= sizeof (*ep)) && ep->ip))
2683         {
2684           ep->is_ip4 = session->transport.is_ip4;
2685           ep->port = session->transport.rmt_port;
2686           if (session->transport.is_ip4)
2687             clib_memcpy_fast (ep->ip, &session->transport.rmt_ip.ip4,
2688                               sizeof (ip4_address_t));
2689           else
2690             clib_memcpy_fast (ep->ip, &session->transport.rmt_ip.ip6,
2691                               sizeof (ip6_address_t));
2692           *buflen = sizeof (*ep);
2693           VDBG (1, "VPPCOM_ATTR_GET_PEER_ADDR: sh %u, is_ip4 = %u, "
2694                 "addr = %U, port %u", session_handle, ep->is_ip4,
2695                 format_ip46_address, &session->transport.rmt_ip,
2696                 ep->is_ip4 ? IP46_TYPE_IP4 : IP46_TYPE_IP6,
2697                 clib_net_to_host_u16 (ep->port));
2698         }
2699       else
2700         rv = VPPCOM_EINVAL;
2701       break;
2702
2703     case VPPCOM_ATTR_GET_LCL_ADDR:
2704       if (PREDICT_TRUE (buffer && buflen &&
2705                         (*buflen >= sizeof (*ep)) && ep->ip))
2706         {
2707           ep->is_ip4 = session->transport.is_ip4;
2708           ep->port = session->transport.lcl_port;
2709           if (session->transport.is_ip4)
2710             clib_memcpy_fast (ep->ip, &session->transport.lcl_ip.ip4,
2711                               sizeof (ip4_address_t));
2712           else
2713             clib_memcpy_fast (ep->ip, &session->transport.lcl_ip.ip6,
2714                               sizeof (ip6_address_t));
2715           *buflen = sizeof (*ep);
2716           VDBG (1, "VPPCOM_ATTR_GET_LCL_ADDR: sh %u, is_ip4 = %u, addr = %U"
2717                 " port %d", session_handle, ep->is_ip4, format_ip46_address,
2718                 &session->transport.lcl_ip,
2719                 ep->is_ip4 ? IP46_TYPE_IP4 : IP46_TYPE_IP6,
2720                 clib_net_to_host_u16 (ep->port));
2721         }
2722       else
2723         rv = VPPCOM_EINVAL;
2724       break;
2725
2726     case VPPCOM_ATTR_GET_LIBC_EPFD:
2727       rv = session->libc_epfd;
2728       VDBG (2, "VPPCOM_ATTR_GET_LIBC_EPFD: libc_epfd %d", rv);
2729       break;
2730
2731     case VPPCOM_ATTR_SET_LIBC_EPFD:
2732       if (PREDICT_TRUE (buffer && buflen &&
2733                         (*buflen == sizeof (session->libc_epfd))))
2734         {
2735           session->libc_epfd = *(int *) buffer;
2736           *buflen = sizeof (session->libc_epfd);
2737
2738           VDBG (2, "VPPCOM_ATTR_SET_LIBC_EPFD: libc_epfd %d, buflen %d",
2739                 session->libc_epfd, *buflen);
2740         }
2741       else
2742         rv = VPPCOM_EINVAL;
2743       break;
2744
2745     case VPPCOM_ATTR_GET_PROTOCOL:
2746       if (buffer && buflen && (*buflen >= sizeof (int)))
2747         {
2748           *(int *) buffer = session->session_type;
2749           *buflen = sizeof (int);
2750
2751           VDBG (2, "VPPCOM_ATTR_GET_PROTOCOL: %d (%s), buflen %d",
2752                 *(int *) buffer, *(int *) buffer ? "UDP" : "TCP", *buflen);
2753         }
2754       else
2755         rv = VPPCOM_EINVAL;
2756       break;
2757
2758     case VPPCOM_ATTR_GET_LISTEN:
2759       if (buffer && buflen && (*buflen >= sizeof (int)))
2760         {
2761           *(int *) buffer = VCL_SESS_ATTR_TEST (session->attr,
2762                                                 VCL_SESS_ATTR_LISTEN);
2763           *buflen = sizeof (int);
2764
2765           VDBG (2, "VPPCOM_ATTR_GET_LISTEN: %d, buflen %d", *(int *) buffer,
2766                 *buflen);
2767         }
2768       else
2769         rv = VPPCOM_EINVAL;
2770       break;
2771
2772     case VPPCOM_ATTR_GET_ERROR:
2773       if (buffer && buflen && (*buflen >= sizeof (int)))
2774         {
2775           *(int *) buffer = 0;
2776           *buflen = sizeof (int);
2777
2778           VDBG (2, "VPPCOM_ATTR_GET_ERROR: %d, buflen %d, #VPP-TBD#",
2779                 *(int *) buffer, *buflen);
2780         }
2781       else
2782         rv = VPPCOM_EINVAL;
2783       break;
2784
2785     case VPPCOM_ATTR_GET_TX_FIFO_LEN:
2786       if (buffer && buflen && (*buflen >= sizeof (u32)))
2787         {
2788
2789           /* VPP-TBD */
2790           *(size_t *) buffer = (session->sndbuf_size ? session->sndbuf_size :
2791                                 session->tx_fifo ? session->tx_fifo->nitems :
2792                                 vcm->cfg.tx_fifo_size);
2793           *buflen = sizeof (u32);
2794
2795           VDBG (2, "VPPCOM_ATTR_GET_TX_FIFO_LEN: %u (0x%x), buflen %d,"
2796                 " #VPP-TBD#", *(size_t *) buffer, *(size_t *) buffer,
2797                 *buflen);
2798         }
2799       else
2800         rv = VPPCOM_EINVAL;
2801       break;
2802
2803     case VPPCOM_ATTR_SET_TX_FIFO_LEN:
2804       if (buffer && buflen && (*buflen == sizeof (u32)))
2805         {
2806           /* VPP-TBD */
2807           session->sndbuf_size = *(u32 *) buffer;
2808           VDBG (2, "VPPCOM_ATTR_SET_TX_FIFO_LEN: %u (0x%x), buflen %d,"
2809                 " #VPP-TBD#", session->sndbuf_size, session->sndbuf_size,
2810                 *buflen);
2811         }
2812       else
2813         rv = VPPCOM_EINVAL;
2814       break;
2815
2816     case VPPCOM_ATTR_GET_RX_FIFO_LEN:
2817       if (buffer && buflen && (*buflen >= sizeof (u32)))
2818         {
2819
2820           /* VPP-TBD */
2821           *(size_t *) buffer = (session->rcvbuf_size ? session->rcvbuf_size :
2822                                 session->rx_fifo ? session->rx_fifo->nitems :
2823                                 vcm->cfg.rx_fifo_size);
2824           *buflen = sizeof (u32);
2825
2826           VDBG (2, "VPPCOM_ATTR_GET_RX_FIFO_LEN: %u (0x%x), buflen %d, "
2827                 "#VPP-TBD#", *(size_t *) buffer, *(size_t *) buffer, *buflen);
2828         }
2829       else
2830         rv = VPPCOM_EINVAL;
2831       break;
2832
2833     case VPPCOM_ATTR_SET_RX_FIFO_LEN:
2834       if (buffer && buflen && (*buflen == sizeof (u32)))
2835         {
2836           /* VPP-TBD */
2837           session->rcvbuf_size = *(u32 *) buffer;
2838           VDBG (2, "VPPCOM_ATTR_SET_RX_FIFO_LEN: %u (0x%x), buflen %d,"
2839                 " #VPP-TBD#", session->sndbuf_size, session->sndbuf_size,
2840                 *buflen);
2841         }
2842       else
2843         rv = VPPCOM_EINVAL;
2844       break;
2845
2846     case VPPCOM_ATTR_GET_REUSEADDR:
2847       if (buffer && buflen && (*buflen >= sizeof (int)))
2848         {
2849           /* VPP-TBD */
2850           *(int *) buffer = VCL_SESS_ATTR_TEST (session->attr,
2851                                                 VCL_SESS_ATTR_REUSEADDR);
2852           *buflen = sizeof (int);
2853
2854           VDBG (2, "VPPCOM_ATTR_GET_REUSEADDR: %d, buflen %d, #VPP-TBD#",
2855                 *(int *) buffer, *buflen);
2856         }
2857       else
2858         rv = VPPCOM_EINVAL;
2859       break;
2860
2861     case VPPCOM_ATTR_SET_REUSEADDR:
2862       if (buffer && buflen && (*buflen == sizeof (int)) &&
2863           !VCL_SESS_ATTR_TEST (session->attr, VCL_SESS_ATTR_LISTEN))
2864         {
2865           /* VPP-TBD */
2866           if (*(int *) buffer)
2867             VCL_SESS_ATTR_SET (session->attr, VCL_SESS_ATTR_REUSEADDR);
2868           else
2869             VCL_SESS_ATTR_CLR (session->attr, VCL_SESS_ATTR_REUSEADDR);
2870
2871           VDBG (2, "VPPCOM_ATTR_SET_REUSEADDR: %d, buflen %d, #VPP-TBD#",
2872                 VCL_SESS_ATTR_TEST (session->attr, VCL_SESS_ATTR_REUSEADDR),
2873                 *buflen);
2874         }
2875       else
2876         rv = VPPCOM_EINVAL;
2877       break;
2878
2879     case VPPCOM_ATTR_GET_REUSEPORT:
2880       if (buffer && buflen && (*buflen >= sizeof (int)))
2881         {
2882           /* VPP-TBD */
2883           *(int *) buffer = VCL_SESS_ATTR_TEST (session->attr,
2884                                                 VCL_SESS_ATTR_REUSEPORT);
2885           *buflen = sizeof (int);
2886
2887           VDBG (2, "VPPCOM_ATTR_GET_REUSEPORT: %d, buflen %d, #VPP-TBD#",
2888                 *(int *) buffer, *buflen);
2889         }
2890       else
2891         rv = VPPCOM_EINVAL;
2892       break;
2893
2894     case VPPCOM_ATTR_SET_REUSEPORT:
2895       if (buffer && buflen && (*buflen == sizeof (int)) &&
2896           !VCL_SESS_ATTR_TEST (session->attr, VCL_SESS_ATTR_LISTEN))
2897         {
2898           /* VPP-TBD */
2899           if (*(int *) buffer)
2900             VCL_SESS_ATTR_SET (session->attr, VCL_SESS_ATTR_REUSEPORT);
2901           else
2902             VCL_SESS_ATTR_CLR (session->attr, VCL_SESS_ATTR_REUSEPORT);
2903
2904           VDBG (2, "VPPCOM_ATTR_SET_REUSEPORT: %d, buflen %d, #VPP-TBD#",
2905                 VCL_SESS_ATTR_TEST (session->attr, VCL_SESS_ATTR_REUSEPORT),
2906                 *buflen);
2907         }
2908       else
2909         rv = VPPCOM_EINVAL;
2910       break;
2911
2912     case VPPCOM_ATTR_GET_BROADCAST:
2913       if (buffer && buflen && (*buflen >= sizeof (int)))
2914         {
2915           /* VPP-TBD */
2916           *(int *) buffer = VCL_SESS_ATTR_TEST (session->attr,
2917                                                 VCL_SESS_ATTR_BROADCAST);
2918           *buflen = sizeof (int);
2919
2920           VDBG (2, "VPPCOM_ATTR_GET_BROADCAST: %d, buflen %d, #VPP-TBD#",
2921                 *(int *) buffer, *buflen);
2922         }
2923       else
2924         rv = VPPCOM_EINVAL;
2925       break;
2926
2927     case VPPCOM_ATTR_SET_BROADCAST:
2928       if (buffer && buflen && (*buflen == sizeof (int)))
2929         {
2930           /* VPP-TBD */
2931           if (*(int *) buffer)
2932             VCL_SESS_ATTR_SET (session->attr, VCL_SESS_ATTR_BROADCAST);
2933           else
2934             VCL_SESS_ATTR_CLR (session->attr, VCL_SESS_ATTR_BROADCAST);
2935
2936           VDBG (2, "VPPCOM_ATTR_SET_BROADCAST: %d, buflen %d, #VPP-TBD#",
2937                 VCL_SESS_ATTR_TEST (session->attr, VCL_SESS_ATTR_BROADCAST),
2938                 *buflen);
2939         }
2940       else
2941         rv = VPPCOM_EINVAL;
2942       break;
2943
2944     case VPPCOM_ATTR_GET_V6ONLY:
2945       if (buffer && buflen && (*buflen >= sizeof (int)))
2946         {
2947           /* VPP-TBD */
2948           *(int *) buffer = VCL_SESS_ATTR_TEST (session->attr,
2949                                                 VCL_SESS_ATTR_V6ONLY);
2950           *buflen = sizeof (int);
2951
2952           VDBG (2, "VPPCOM_ATTR_GET_V6ONLY: %d, buflen %d, #VPP-TBD#",
2953                 *(int *) buffer, *buflen);
2954         }
2955       else
2956         rv = VPPCOM_EINVAL;
2957       break;
2958
2959     case VPPCOM_ATTR_SET_V6ONLY:
2960       if (buffer && buflen && (*buflen == sizeof (int)))
2961         {
2962           /* VPP-TBD */
2963           if (*(int *) buffer)
2964             VCL_SESS_ATTR_SET (session->attr, VCL_SESS_ATTR_V6ONLY);
2965           else
2966             VCL_SESS_ATTR_CLR (session->attr, VCL_SESS_ATTR_V6ONLY);
2967
2968           VDBG (2, "VPPCOM_ATTR_SET_V6ONLY: %d, buflen %d, #VPP-TBD#",
2969                 VCL_SESS_ATTR_TEST (session->attr, VCL_SESS_ATTR_V6ONLY),
2970                 *buflen);
2971         }
2972       else
2973         rv = VPPCOM_EINVAL;
2974       break;
2975
2976     case VPPCOM_ATTR_GET_KEEPALIVE:
2977       if (buffer && buflen && (*buflen >= sizeof (int)))
2978         {
2979           /* VPP-TBD */
2980           *(int *) buffer = VCL_SESS_ATTR_TEST (session->attr,
2981                                                 VCL_SESS_ATTR_KEEPALIVE);
2982           *buflen = sizeof (int);
2983
2984           VDBG (2, "VPPCOM_ATTR_GET_KEEPALIVE: %d, buflen %d, #VPP-TBD#",
2985                 *(int *) buffer, *buflen);
2986         }
2987       else
2988         rv = VPPCOM_EINVAL;
2989       break;
2990
2991     case VPPCOM_ATTR_SET_KEEPALIVE:
2992       if (buffer && buflen && (*buflen == sizeof (int)))
2993         {
2994           /* VPP-TBD */
2995           if (*(int *) buffer)
2996             VCL_SESS_ATTR_SET (session->attr, VCL_SESS_ATTR_KEEPALIVE);
2997           else
2998             VCL_SESS_ATTR_CLR (session->attr, VCL_SESS_ATTR_KEEPALIVE);
2999
3000           VDBG (2, "VPPCOM_ATTR_SET_KEEPALIVE: %d, buflen %d, #VPP-TBD#",
3001                 VCL_SESS_ATTR_TEST (session->attr, VCL_SESS_ATTR_KEEPALIVE),
3002                 *buflen);
3003         }
3004       else
3005         rv = VPPCOM_EINVAL;
3006       break;
3007
3008     case VPPCOM_ATTR_GET_TCP_NODELAY:
3009       if (buffer && buflen && (*buflen >= sizeof (int)))
3010         {
3011           /* VPP-TBD */
3012           *(int *) buffer = VCL_SESS_ATTR_TEST (session->attr,
3013                                                 VCL_SESS_ATTR_TCP_NODELAY);
3014           *buflen = sizeof (int);
3015
3016           VDBG (2, "VPPCOM_ATTR_GET_TCP_NODELAY: %d, buflen %d, #VPP-TBD#",
3017                 *(int *) buffer, *buflen);
3018         }
3019       else
3020         rv = VPPCOM_EINVAL;
3021       break;
3022
3023     case VPPCOM_ATTR_SET_TCP_NODELAY:
3024       if (buffer && buflen && (*buflen == sizeof (int)))
3025         {
3026           /* VPP-TBD */
3027           if (*(int *) buffer)
3028             VCL_SESS_ATTR_SET (session->attr, VCL_SESS_ATTR_TCP_NODELAY);
3029           else
3030             VCL_SESS_ATTR_CLR (session->attr, VCL_SESS_ATTR_TCP_NODELAY);
3031
3032           VDBG (2, "VPPCOM_ATTR_SET_TCP_NODELAY: %d, buflen %d, #VPP-TBD#",
3033                 VCL_SESS_ATTR_TEST (session->attr, VCL_SESS_ATTR_TCP_NODELAY),
3034                 *buflen);
3035         }
3036       else
3037         rv = VPPCOM_EINVAL;
3038       break;
3039
3040     case VPPCOM_ATTR_GET_TCP_KEEPIDLE:
3041       if (buffer && buflen && (*buflen >= sizeof (int)))
3042         {
3043           /* VPP-TBD */
3044           *(int *) buffer = VCL_SESS_ATTR_TEST (session->attr,
3045                                                 VCL_SESS_ATTR_TCP_KEEPIDLE);
3046           *buflen = sizeof (int);
3047
3048           VDBG (2, "VPPCOM_ATTR_GET_TCP_KEEPIDLE: %d, buflen %d, #VPP-TBD#",
3049                 *(int *) buffer, *buflen);
3050         }
3051       else
3052         rv = VPPCOM_EINVAL;
3053       break;
3054
3055     case VPPCOM_ATTR_SET_TCP_KEEPIDLE:
3056       if (buffer && buflen && (*buflen == sizeof (int)))
3057         {
3058           /* VPP-TBD */
3059           if (*(int *) buffer)
3060             VCL_SESS_ATTR_SET (session->attr, VCL_SESS_ATTR_TCP_KEEPIDLE);
3061           else
3062             VCL_SESS_ATTR_CLR (session->attr, VCL_SESS_ATTR_TCP_KEEPIDLE);
3063
3064           VDBG (2, "VPPCOM_ATTR_SET_TCP_KEEPIDLE: %d, buflen %d, #VPP-TBD#",
3065                 VCL_SESS_ATTR_TEST (session->attr,
3066                                     VCL_SESS_ATTR_TCP_KEEPIDLE), *buflen);
3067         }
3068       else
3069         rv = VPPCOM_EINVAL;
3070       break;
3071
3072     case VPPCOM_ATTR_GET_TCP_KEEPINTVL:
3073       if (buffer && buflen && (*buflen >= sizeof (int)))
3074         {
3075           /* VPP-TBD */
3076           *(int *) buffer = VCL_SESS_ATTR_TEST (session->attr,
3077                                                 VCL_SESS_ATTR_TCP_KEEPINTVL);
3078           *buflen = sizeof (int);
3079
3080           VDBG (2, "VPPCOM_ATTR_GET_TCP_KEEPINTVL: %d, buflen %d, #VPP-TBD#",
3081                 *(int *) buffer, *buflen);
3082         }
3083       else
3084         rv = VPPCOM_EINVAL;
3085       break;
3086
3087     case VPPCOM_ATTR_SET_TCP_KEEPINTVL:
3088       if (buffer && buflen && (*buflen == sizeof (int)))
3089         {
3090           /* VPP-TBD */
3091           if (*(int *) buffer)
3092             VCL_SESS_ATTR_SET (session->attr, VCL_SESS_ATTR_TCP_KEEPINTVL);
3093           else
3094             VCL_SESS_ATTR_CLR (session->attr, VCL_SESS_ATTR_TCP_KEEPINTVL);
3095
3096           VDBG (2, "VPPCOM_ATTR_SET_TCP_KEEPINTVL: %d, buflen %d, #VPP-TBD#",
3097                 VCL_SESS_ATTR_TEST (session->attr,
3098                                     VCL_SESS_ATTR_TCP_KEEPINTVL), *buflen);
3099         }
3100       else
3101         rv = VPPCOM_EINVAL;
3102       break;
3103
3104     case VPPCOM_ATTR_GET_TCP_USER_MSS:
3105       if (buffer && buflen && (*buflen >= sizeof (u32)))
3106         {
3107           /* VPP-TBD */
3108           *(u32 *) buffer = session->user_mss;
3109           *buflen = sizeof (int);
3110
3111           VDBG (2, "VPPCOM_ATTR_GET_TCP_USER_MSS: %d, buflen %d, #VPP-TBD#",
3112                 *(int *) buffer, *buflen);
3113         }
3114       else
3115         rv = VPPCOM_EINVAL;
3116       break;
3117
3118     case VPPCOM_ATTR_SET_TCP_USER_MSS:
3119       if (buffer && buflen && (*buflen == sizeof (u32)))
3120         {
3121           /* VPP-TBD */
3122           session->user_mss = *(u32 *) buffer;
3123
3124           VDBG (2, "VPPCOM_ATTR_SET_TCP_USER_MSS: %u, buflen %d, #VPP-TBD#",
3125                 session->user_mss, *buflen);
3126         }
3127       else
3128         rv = VPPCOM_EINVAL;
3129       break;
3130
3131     case VPPCOM_ATTR_SET_SHUT:
3132       if (*flags == SHUT_RD || *flags == SHUT_RDWR)
3133         VCL_SESS_ATTR_SET (session->attr, VCL_SESS_ATTR_SHUT_RD);
3134       if (*flags == SHUT_WR || *flags == SHUT_RDWR)
3135         VCL_SESS_ATTR_SET (session->attr, VCL_SESS_ATTR_SHUT_WR);
3136       break;
3137
3138     case VPPCOM_ATTR_GET_SHUT:
3139       if (VCL_SESS_ATTR_TEST (session->attr, VCL_SESS_ATTR_SHUT_RD))
3140         tmp_flags = 1;
3141       if (VCL_SESS_ATTR_TEST (session->attr, VCL_SESS_ATTR_SHUT_WR))
3142         tmp_flags |= 2;
3143       if (tmp_flags == 1)
3144         *(int *) buffer = SHUT_RD;
3145       else if (tmp_flags == 2)
3146         *(int *) buffer = SHUT_WR;
3147       else if (tmp_flags == 3)
3148         *(int *) buffer = SHUT_RDWR;
3149       *buflen = sizeof (int);
3150       break;
3151     default:
3152       rv = VPPCOM_EINVAL;
3153       break;
3154     }
3155
3156   return rv;
3157 }
3158
3159 int
3160 vppcom_session_recvfrom (uint32_t session_handle, void *buffer,
3161                          uint32_t buflen, int flags, vppcom_endpt_t * ep)
3162 {
3163   vcl_worker_t *wrk = vcl_worker_get_current ();
3164   int rv = VPPCOM_OK;
3165   vcl_session_t *session = 0;
3166
3167   if (ep)
3168     {
3169       session = vcl_session_get_w_handle (wrk, session_handle);
3170       if (PREDICT_FALSE (!session))
3171         {
3172           VDBG (0, "sh 0x%llx is closed!", session_handle);
3173           return VPPCOM_EBADFD;
3174         }
3175       ep->is_ip4 = session->transport.is_ip4;
3176       ep->port = session->transport.rmt_port;
3177     }
3178
3179   if (flags == 0)
3180     rv = vppcom_session_read (session_handle, buffer, buflen);
3181   else if (flags & MSG_PEEK)
3182     rv = vppcom_session_peek (session_handle, buffer, buflen);
3183   else
3184     {
3185       VDBG (0, "Unsupport flags for recvfrom %d", flags);
3186       return VPPCOM_EAFNOSUPPORT;
3187     }
3188
3189   if (ep)
3190     {
3191       if (session->transport.is_ip4)
3192         clib_memcpy_fast (ep->ip, &session->transport.rmt_ip.ip4,
3193                           sizeof (ip4_address_t));
3194       else
3195         clib_memcpy_fast (ep->ip, &session->transport.rmt_ip.ip6,
3196                           sizeof (ip6_address_t));
3197     }
3198
3199   return rv;
3200 }
3201
3202 int
3203 vppcom_session_sendto (uint32_t session_handle, void *buffer,
3204                        uint32_t buflen, int flags, vppcom_endpt_t * ep)
3205 {
3206   if (!buffer)
3207     return VPPCOM_EINVAL;
3208
3209   if (ep)
3210     {
3211       // TBD
3212       return VPPCOM_EINVAL;
3213     }
3214
3215   if (flags)
3216     {
3217       // TBD check the flags and do the right thing
3218       VDBG (2, "handling flags 0x%u (%d) not implemented yet.", flags, flags);
3219     }
3220
3221   return (vppcom_session_write_inline (session_handle, buffer, buflen, 1));
3222 }
3223
3224 int
3225 vppcom_poll (vcl_poll_t * vp, uint32_t n_sids, double wait_for_time)
3226 {
3227   vcl_worker_t *wrk = vcl_worker_get_current ();
3228   f64 timeout = clib_time_now (&wrk->clib_time) + wait_for_time;
3229   u32 i, keep_trying = 1;
3230   svm_msg_q_msg_t msg;
3231   session_event_t *e;
3232   int rv, num_ev = 0;
3233
3234   VDBG (3, "vp %p, nsids %u, wait_for_time %f", vp, n_sids, wait_for_time);
3235
3236   if (!vp)
3237     return VPPCOM_EFAULT;
3238
3239   do
3240     {
3241       vcl_session_t *session;
3242
3243       /* Dequeue all events and drop all unhandled io events */
3244       while (svm_msg_q_sub (wrk->app_event_queue, &msg, SVM_Q_NOWAIT, 0) == 0)
3245         {
3246           e = svm_msg_q_msg_data (wrk->app_event_queue, &msg);
3247           vcl_handle_mq_event (wrk, e);
3248           svm_msg_q_free_msg (wrk->app_event_queue, &msg);
3249         }
3250       vec_reset_length (wrk->unhandled_evts_vector);
3251
3252       for (i = 0; i < n_sids; i++)
3253         {
3254           session = vcl_session_get (wrk, vp[i].sh);
3255           if (!session)
3256             {
3257               vp[i].revents = POLLHUP;
3258               num_ev++;
3259               continue;
3260             }
3261
3262           vp[i].revents = 0;
3263
3264           if (POLLIN & vp[i].events)
3265             {
3266               rv = vcl_session_read_ready (session);
3267               if (rv > 0)
3268                 {
3269                   vp[i].revents |= POLLIN;
3270                   num_ev++;
3271                 }
3272               else if (rv < 0)
3273                 {
3274                   switch (rv)
3275                     {
3276                     case VPPCOM_ECONNRESET:
3277                       vp[i].revents = POLLHUP;
3278                       break;
3279
3280                     default:
3281                       vp[i].revents = POLLERR;
3282                       break;
3283                     }
3284                   num_ev++;
3285                 }
3286             }
3287
3288           if (POLLOUT & vp[i].events)
3289             {
3290               rv = vcl_session_write_ready (session);
3291               if (rv > 0)
3292                 {
3293                   vp[i].revents |= POLLOUT;
3294                   num_ev++;
3295                 }
3296               else if (rv < 0)
3297                 {
3298                   switch (rv)
3299                     {
3300                     case VPPCOM_ECONNRESET:
3301                       vp[i].revents = POLLHUP;
3302                       break;
3303
3304                     default:
3305                       vp[i].revents = POLLERR;
3306                       break;
3307                     }
3308                   num_ev++;
3309                 }
3310             }
3311
3312           if (0)                // Note "done:" label used by VCL_SESSION_LOCK_AND_GET()
3313             {
3314               vp[i].revents = POLLNVAL;
3315               num_ev++;
3316             }
3317         }
3318       if (wait_for_time != -1)
3319         keep_trying = (clib_time_now (&wrk->clib_time) <= timeout) ? 1 : 0;
3320     }
3321   while ((num_ev == 0) && keep_trying);
3322
3323   return num_ev;
3324 }
3325
3326 int
3327 vppcom_mq_epoll_fd (void)
3328 {
3329   vcl_worker_t *wrk = vcl_worker_get_current ();
3330   return wrk->mqs_epfd;
3331 }
3332
3333 int
3334 vppcom_session_index (vcl_session_handle_t session_handle)
3335 {
3336   return session_handle & 0xFFFFFF;
3337 }
3338
3339 int
3340 vppcom_session_worker (vcl_session_handle_t session_handle)
3341 {
3342   return session_handle >> 24;
3343 }
3344
3345 int
3346 vppcom_worker_register (void)
3347 {
3348   if (!vcl_worker_alloc_and_init ())
3349     return VPPCOM_EEXIST;
3350
3351   if (vcl_worker_set_bapi ())
3352     return VPPCOM_EEXIST;
3353
3354   if (vcl_worker_register_with_vpp ())
3355     return VPPCOM_EEXIST;
3356
3357   return VPPCOM_OK;
3358 }
3359
3360 void
3361 vppcom_worker_unregister (void)
3362 {
3363   vcl_worker_cleanup (vcl_worker_get_current (), 1 /* notify vpp */ );
3364   vcl_set_worker_index (~0);
3365 }
3366
3367 int
3368 vppcom_worker_index (void)
3369 {
3370   return vcl_get_worker_index ();
3371 }
3372
3373 int
3374 vppcom_worker_mqs_epfd (void)
3375 {
3376   vcl_worker_t *wrk = vcl_worker_get_current ();
3377   if (!vcm->cfg.use_mq_eventfd)
3378     return -1;
3379   return wrk->mqs_epfd;
3380 }
3381
3382 /*
3383  * fd.io coding-style-patch-verification: ON
3384  *
3385  * Local Variables:
3386  * eval: (c-set-style "gnu")
3387  * End:
3388  */