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