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->port;
 313   session->transport.is_ip4 = mp->is_ip4;
 314   clib_memcpy_fast (&session->transport.rmt_ip, mp->ip,
 315                     sizeof (ip46_address_t));
 316
 317   vcl_session_table_add_vpp_handle (wrk, mp->handle, session->session_index);
 318   session->transport.lcl_port = listen_session->transport.lcl_port;
 319   session->transport.lcl_ip = listen_session->transport.lcl_ip;
 320   session->session_type = listen_session->session_type;
 321   session->is_dgram = session->session_type == VPPCOM_PROTO_UDP;
 322
 323   VDBG (1, "session %u [0x%llx]: client accept request from %s address %U"
 324         " port %d queue %p!", session->session_index, mp->handle,
 325         mp->is_ip4 ? "IPv4" : "IPv6", format_ip46_address, &mp->ip,
 326         mp->is_ip4 ? IP46_TYPE_IP4 : IP46_TYPE_IP6,
 327         clib_net_to_host_u16 (mp->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->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 (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 (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 (rx_fifo))
1531     svm_fifo_unset_event (s->rx_fifo);
1532
1533   VDBG (2, "session %u[0x%llx]: read %d bytes from (%p)", s->session_index,
1534         s->vpp_handle, n_read, rx_fifo);
1535
1536   return n_read;
1537 }
1538
1539 int
1540 vppcom_session_read (uint32_t session_handle, void *buf, size_t n)
1541 {
1542   return (vppcom_session_read_internal (session_handle, buf, n, 0));
1543 }
1544
1545 static int
1546 vppcom_session_peek (uint32_t session_handle, void *buf, int n)
1547 {
1548   return (vppcom_session_read_internal (session_handle, buf, n, 1));
1549 }
1550
1551 int
1552 vppcom_session_read_segments (uint32_t session_handle,
1553                               vppcom_data_segments_t ds)
1554 {
1555   vcl_worker_t *wrk = vcl_worker_get_current ();
1556   int n_read = 0, is_nonblocking;
1557   vcl_session_t *s = 0;
1558   svm_fifo_t *rx_fifo;
1559   svm_msg_q_msg_t msg;
1560   session_event_t *e;
1561   svm_msg_q_t *mq;
1562   u8 is_ct;
1563
1564   s = vcl_session_get_w_handle (wrk, session_handle);
1565   if (PREDICT_FALSE (!s || s->is_vep))
1566     return VPPCOM_EBADFD;
1567
1568   if (PREDICT_FALSE (!vcl_session_is_open (s)))
1569     return vcl_session_closed_error (s);
1570
1571   is_nonblocking = VCL_SESS_ATTR_TEST (s->attr, VCL_SESS_ATTR_NONBLOCK);
1572   is_ct = vcl_session_is_ct (s);
1573   mq = is_ct ? s->our_evt_q : wrk->app_event_queue;
1574   rx_fifo = s->rx_fifo;
1575   s->has_rx_evt = 0;
1576
1577   if (is_ct)
1578     svm_fifo_unset_event (s->rx_fifo);
1579
1580   if (svm_fifo_is_empty (rx_fifo))
1581     {
1582       if (is_nonblocking)
1583         {
1584           svm_fifo_unset_event (rx_fifo);
1585           return VPPCOM_EWOULDBLOCK;
1586         }
1587       while (svm_fifo_is_empty (rx_fifo))
1588         {
1589           if (vcl_session_is_closing (s))
1590             return vcl_session_closing_error (s);
1591
1592           svm_fifo_unset_event (rx_fifo);
1593           svm_msg_q_lock (mq);
1594           if (svm_msg_q_is_empty (mq))
1595             svm_msg_q_wait (mq);
1596
1597           svm_msg_q_sub_w_lock (mq, &msg);
1598           e = svm_msg_q_msg_data (mq, &msg);
1599           svm_msg_q_unlock (mq);
1600           if (!vcl_is_rx_evt_for_session (e, s->session_index, is_ct))
1601             vcl_handle_mq_event (wrk, e);
1602           svm_msg_q_free_msg (mq, &msg);
1603         }
1604     }
1605
1606   n_read = svm_fifo_segments (rx_fifo, (svm_fifo_segment_t *) ds);
1607   svm_fifo_unset_event (rx_fifo);
1608
1609   return n_read;
1610 }
1611
1612 void
1613 vppcom_session_free_segments (uint32_t session_handle,
1614                               vppcom_data_segments_t ds)
1615 {
1616   vcl_worker_t *wrk = vcl_worker_get_current ();
1617   vcl_session_t *s;
1618
1619   s = vcl_session_get_w_handle (wrk, session_handle);
1620   if (PREDICT_FALSE (!s || s->is_vep))
1621     return;
1622
1623   svm_fifo_segments_free (s->rx_fifo, (svm_fifo_segment_t *) ds);
1624 }
1625
1626 int
1627 vppcom_data_segment_copy (void *buf, vppcom_data_segments_t ds, u32 max_bytes)
1628 {
1629   u32 first_copy = clib_min (ds[0].len, max_bytes);
1630   clib_memcpy_fast (buf, ds[0].data, first_copy);
1631   if (first_copy < max_bytes)
1632     {
1633       clib_memcpy_fast (buf + first_copy, ds[1].data,
1634                         clib_min (ds[1].len, max_bytes - first_copy));
1635     }
1636   return 0;
1637 }
1638
1639 static u8
1640 vcl_is_tx_evt_for_session (session_event_t * e, u32 sid, u8 is_ct)
1641 {
1642   return (e->event_type == SESSION_IO_EVT_TX && e->session_index == sid);
1643 }
1644
1645 static inline int
1646 vppcom_session_write_inline (uint32_t session_handle, void *buf, size_t n,
1647                              u8 is_flush)
1648 {
1649   vcl_worker_t *wrk = vcl_worker_get_current ();
1650   int n_write, is_nonblocking;
1651   vcl_session_t *s = 0;
1652   session_evt_type_t et;
1653   svm_msg_q_msg_t msg;
1654   svm_fifo_t *tx_fifo;
1655   session_event_t *e;
1656   svm_msg_q_t *mq;
1657   u8 is_ct;
1658
1659   if (PREDICT_FALSE (!buf))
1660     return VPPCOM_EINVAL;
1661
1662   s = vcl_session_get_w_handle (wrk, session_handle);
1663   if (PREDICT_FALSE (!s))
1664     return VPPCOM_EBADFD;
1665
1666   if (PREDICT_FALSE (s->is_vep))
1667     {
1668       VDBG (0, "ERROR: session %u [0x%llx]: cannot write to an epoll"
1669             " session!", s->session_index, s->vpp_handle);
1670       return VPPCOM_EBADFD;
1671     }
1672
1673   if (PREDICT_FALSE (!vcl_session_is_open (s)))
1674     {
1675       VDBG (1, "session %u [0x%llx]: is not open! state 0x%x (%s)",
1676             s->session_index, s->vpp_handle, s->session_state,
1677             vppcom_session_state_str (s->session_state));
1678       return vcl_session_closed_error (s);;
1679     }
1680
1681   is_ct = vcl_session_is_ct (s);
1682   tx_fifo = is_ct ? s->ct_tx_fifo : s->tx_fifo;
1683   is_nonblocking = VCL_SESS_ATTR_TEST (s->attr, VCL_SESS_ATTR_NONBLOCK);
1684   mq = wrk->app_event_queue;
1685   if (svm_fifo_is_full (tx_fifo))
1686     {
1687       if (is_nonblocking)
1688         {
1689           return VPPCOM_EWOULDBLOCK;
1690         }
1691       while (svm_fifo_is_full (tx_fifo))
1692         {
1693           svm_fifo_add_want_tx_ntf (tx_fifo, SVM_FIFO_WANT_TX_NOTIF);
1694           if (vcl_session_is_closing (s))
1695             return vcl_session_closing_error (s);
1696           svm_msg_q_lock (mq);
1697           if (svm_msg_q_is_empty (mq))
1698             svm_msg_q_wait (mq);
1699
1700           svm_msg_q_sub_w_lock (mq, &msg);
1701           e = svm_msg_q_msg_data (mq, &msg);
1702           svm_msg_q_unlock (mq);
1703
1704           if (!vcl_is_tx_evt_for_session (e, s->session_index, is_ct))
1705             vcl_handle_mq_event (wrk, e);
1706           svm_msg_q_free_msg (mq, &msg);
1707         }
1708     }
1709
1710   et = SESSION_IO_EVT_TX;
1711   if (is_flush && !is_ct)
1712     et = SESSION_IO_EVT_TX_FLUSH;
1713
1714   if (s->is_dgram)
1715     n_write = app_send_dgram_raw (tx_fifo, &s->transport,
1716                                   s->vpp_evt_q, buf, n, et,
1717                                   0 /* do_evt */ , SVM_Q_WAIT);
1718   else
1719     n_write = app_send_stream_raw (tx_fifo, s->vpp_evt_q, buf, n, et,
1720                                    0 /* do_evt */ , SVM_Q_WAIT);
1721
1722   if (svm_fifo_set_event (s->tx_fifo))
1723     app_send_io_evt_to_vpp (s->vpp_evt_q, s->tx_fifo->master_session_index,
1724                             et, SVM_Q_WAIT);
1725
1726   ASSERT (n_write > 0);
1727
1728   VDBG (2, "session %u [0x%llx]: wrote %d bytes", s->session_index,
1729         s->vpp_handle, n_write);
1730
1731   return n_write;
1732 }
1733
1734 int
1735 vppcom_session_write (uint32_t session_handle, void *buf, size_t n)
1736 {
1737   return vppcom_session_write_inline (session_handle, buf, n,
1738                                       0 /* is_flush */ );
1739 }
1740
1741 int
1742 vppcom_session_write_msg (uint32_t session_handle, void *buf, size_t n)
1743 {
1744   return vppcom_session_write_inline (session_handle, buf, n,
1745                                       1 /* is_flush */ );
1746 }
1747
1748 #define vcl_fifo_rx_evt_valid_or_break(_s)                              \
1749 if (PREDICT_FALSE (svm_fifo_is_empty (_s->rx_fifo)))                    \
1750   {                                                                     \
1751     if (!vcl_session_is_ct (_s))                                        \
1752       {                                                                 \
1753         svm_fifo_unset_event (_s->rx_fifo);                             \
1754         if (svm_fifo_is_empty (_s->rx_fifo))                            \
1755           break;                                                        \
1756       }                                                                 \
1757     else if (svm_fifo_is_empty (_s->ct_rx_fifo))                        \
1758       {                                                                 \
1759         svm_fifo_unset_event (_s->ct_rx_fifo);                          \
1760         if (svm_fifo_is_empty (_s->ct_rx_fifo))                         \
1761           break;                                                        \
1762       }                                                                 \
1763   }                                                                     \
1764
1765 static void
1766 vcl_select_handle_mq_event (vcl_worker_t * wrk, session_event_t * e,
1767                             unsigned long n_bits, unsigned long *read_map,
1768                             unsigned long *write_map,
1769                             unsigned long *except_map, u32 * bits_set)
1770 {
1771   session_disconnected_msg_t *disconnected_msg;
1772   session_connected_msg_t *connected_msg;
1773   vcl_session_t *session;
1774   u32 sid;
1775
1776   switch (e->event_type)
1777     {
1778     case SESSION_IO_EVT_RX:
1779       sid = e->session_index;
1780       session = vcl_session_get (wrk, sid);
1781       if (!session)
1782         break;
1783       vcl_fifo_rx_evt_valid_or_break (session);
1784       if (sid < n_bits && read_map)
1785         {
1786           clib_bitmap_set_no_check ((uword *) read_map, sid, 1);
1787           *bits_set += 1;
1788         }
1789       break;
1790     case SESSION_IO_EVT_TX:
1791       sid = e->session_index;
1792       session = vcl_session_get (wrk, sid);
1793       if (!session)
1794         break;
1795       if (sid < n_bits && write_map)
1796         {
1797           clib_bitmap_set_no_check ((uword *) write_map, sid, 1);
1798           *bits_set += 1;
1799         }
1800       break;
1801     case SESSION_CTRL_EVT_ACCEPTED:
1802       session = vcl_session_accepted (wrk,
1803                                       (session_accepted_msg_t *) e->data);
1804       if (!session)
1805         break;
1806       sid = session->session_index;
1807       if (sid < n_bits && read_map)
1808         {
1809           clib_bitmap_set_no_check ((uword *) read_map, sid, 1);
1810           *bits_set += 1;
1811         }
1812       break;
1813     case SESSION_CTRL_EVT_CONNECTED:
1814       connected_msg = (session_connected_msg_t *) e->data;
1815       vcl_session_connected_handler (wrk, connected_msg);
1816       break;
1817     case SESSION_CTRL_EVT_DISCONNECTED:
1818       disconnected_msg = (session_disconnected_msg_t *) e->data;
1819       session = vcl_session_disconnected_handler (wrk, disconnected_msg);
1820       if (!session)
1821         break;
1822       sid = session->session_index;
1823       if (sid < n_bits && except_map)
1824         {
1825           clib_bitmap_set_no_check ((uword *) except_map, sid, 1);
1826           *bits_set += 1;
1827         }
1828       break;
1829     case SESSION_CTRL_EVT_RESET:
1830       sid = vcl_session_reset_handler (wrk, (session_reset_msg_t *) e->data);
1831       if (sid < n_bits && except_map)
1832         {
1833           clib_bitmap_set_no_check ((uword *) except_map, sid, 1);
1834           *bits_set += 1;
1835         }
1836       break;
1837     case SESSION_CTRL_EVT_UNLISTEN_REPLY:
1838       vcl_session_unlisten_reply_handler (wrk, e->data);
1839       break;
1840     case SESSION_CTRL_EVT_WORKER_UPDATE_REPLY:
1841       vcl_session_worker_update_reply_handler (wrk, e->data);
1842       break;
1843     case SESSION_CTRL_EVT_REQ_WORKER_UPDATE:
1844       vcl_session_req_worker_update_handler (wrk, e->data);
1845       break;
1846     default:
1847       clib_warning ("unhandled: %u", e->event_type);
1848       break;
1849     }
1850 }
1851
1852 static int
1853 vcl_select_handle_mq (vcl_worker_t * wrk, svm_msg_q_t * mq,
1854                       unsigned long n_bits, unsigned long *read_map,
1855                       unsigned long *write_map, unsigned long *except_map,
1856                       double time_to_wait, u32 * bits_set)
1857 {
1858   svm_msg_q_msg_t *msg;
1859   session_event_t *e;
1860   u32 i;
1861
1862   svm_msg_q_lock (mq);
1863   if (svm_msg_q_is_empty (mq))
1864     {
1865       if (*bits_set)
1866         {
1867           svm_msg_q_unlock (mq);
1868           return 0;
1869         }
1870
1871       if (!time_to_wait)
1872         {
1873           svm_msg_q_unlock (mq);
1874           return 0;
1875         }
1876       else if (time_to_wait < 0)
1877         {
1878           svm_msg_q_wait (mq);
1879         }
1880       else
1881         {
1882           if (svm_msg_q_timedwait (mq, time_to_wait))
1883             {
1884               svm_msg_q_unlock (mq);
1885               return 0;
1886             }
1887         }
1888     }
1889   vcl_mq_dequeue_batch (wrk, mq);
1890   svm_msg_q_unlock (mq);
1891
1892   for (i = 0; i < vec_len (wrk->mq_msg_vector); i++)
1893     {
1894       msg = vec_elt_at_index (wrk->mq_msg_vector, i);
1895       e = svm_msg_q_msg_data (mq, msg);
1896       vcl_select_handle_mq_event (wrk, e, n_bits, read_map, write_map,
1897                                   except_map, bits_set);
1898       svm_msg_q_free_msg (mq, msg);
1899     }
1900   vec_reset_length (wrk->mq_msg_vector);
1901   vcl_handle_pending_wrk_updates (wrk);
1902   return *bits_set;
1903 }
1904
1905 static int
1906 vppcom_select_condvar (vcl_worker_t * wrk, int n_bits,
1907                        vcl_si_set * read_map, vcl_si_set * write_map,
1908                        vcl_si_set * except_map, double time_to_wait,
1909                        u32 * bits_set)
1910 {
1911   double wait = 0, start = 0;
1912
1913   if (!*bits_set)
1914     {
1915       wait = time_to_wait;
1916       start = clib_time_now (&wrk->clib_time);
1917     }
1918
1919   do
1920     {
1921       vcl_select_handle_mq (wrk, wrk->app_event_queue, n_bits, read_map,
1922                             write_map, except_map, wait, bits_set);
1923       if (*bits_set)
1924         return *bits_set;
1925       if (wait == -1)
1926         continue;
1927
1928       wait = wait - (clib_time_now (&wrk->clib_time) - start);
1929     }
1930   while (wait > 0);
1931
1932   return 0;
1933 }
1934
1935 static int
1936 vppcom_select_eventfd (vcl_worker_t * wrk, int n_bits,
1937                        vcl_si_set * read_map, vcl_si_set * write_map,
1938                        vcl_si_set * except_map, double time_to_wait,
1939                        u32 * bits_set)
1940 {
1941   vcl_mq_evt_conn_t *mqc;
1942   int __clib_unused n_read;
1943   int n_mq_evts, i;
1944   u64 buf;
1945
1946   vec_validate (wrk->mq_events, pool_elts (wrk->mq_evt_conns));
1947   n_mq_evts = epoll_wait (wrk->mqs_epfd, wrk->mq_events,
1948                           vec_len (wrk->mq_events), time_to_wait);
1949   for (i = 0; i < n_mq_evts; i++)
1950     {
1951       mqc = vcl_mq_evt_conn_get (wrk, wrk->mq_events[i].data.u32);
1952       n_read = read (mqc->mq_fd, &buf, sizeof (buf));
1953       vcl_select_handle_mq (wrk, mqc->mq, n_bits, read_map, write_map,
1954                             except_map, 0, bits_set);
1955     }
1956
1957   return (n_mq_evts > 0 ? (int) *bits_set : 0);
1958 }
1959
1960 int
1961 vppcom_select (int n_bits, vcl_si_set * read_map, vcl_si_set * write_map,
1962                vcl_si_set * except_map, double time_to_wait)
1963 {
1964   u32 sid, minbits = clib_max (n_bits, BITS (uword)), bits_set = 0;
1965   vcl_worker_t *wrk = vcl_worker_get_current ();
1966   vcl_session_t *session = 0;
1967   int rv, i;
1968
1969   if (n_bits && read_map)
1970     {
1971       clib_bitmap_validate (wrk->rd_bitmap, minbits);
1972       clib_memcpy_fast (wrk->rd_bitmap, read_map,
1973                         vec_len (wrk->rd_bitmap) * sizeof (vcl_si_set));
1974       memset (read_map, 0, vec_len (wrk->rd_bitmap) * sizeof (vcl_si_set));
1975     }
1976   if (n_bits && write_map)
1977     {
1978       clib_bitmap_validate (wrk->wr_bitmap, minbits);
1979       clib_memcpy_fast (wrk->wr_bitmap, write_map,
1980                         vec_len (wrk->wr_bitmap) * sizeof (vcl_si_set));
1981       memset (write_map, 0, vec_len (wrk->wr_bitmap) * sizeof (vcl_si_set));
1982     }
1983   if (n_bits && except_map)
1984     {
1985       clib_bitmap_validate (wrk->ex_bitmap, minbits);
1986       clib_memcpy_fast (wrk->ex_bitmap, except_map,
1987                         vec_len (wrk->ex_bitmap) * sizeof (vcl_si_set));
1988       memset (except_map, 0, vec_len (wrk->ex_bitmap) * sizeof (vcl_si_set));
1989     }
1990
1991   if (!n_bits)
1992     return 0;
1993
1994   if (!write_map)
1995     goto check_rd;
1996
1997   /* *INDENT-OFF* */
1998   clib_bitmap_foreach (sid, wrk->wr_bitmap, ({
1999     if (!(session = vcl_session_get (wrk, sid)))
2000       {
2001         if (except_map && sid < minbits)
2002           clib_bitmap_set_no_check (except_map, sid, 1);
2003         continue;
2004       }
2005
2006     rv = svm_fifo_is_full (session->tx_fifo);
2007     if (!rv)
2008       {
2009         clib_bitmap_set_no_check ((uword*)write_map, sid, 1);
2010         bits_set++;
2011       }
2012     else
2013       svm_fifo_add_want_tx_ntf (session->tx_fifo, SVM_FIFO_WANT_TX_NOTIF);
2014   }));
2015
2016 check_rd:
2017   if (!read_map)
2018     goto check_mq;
2019
2020   clib_bitmap_foreach (sid, wrk->rd_bitmap, ({
2021     if (!(session = vcl_session_get (wrk, sid)))
2022       {
2023         if (except_map && sid < minbits)
2024           clib_bitmap_set_no_check (except_map, sid, 1);
2025         continue;
2026       }
2027
2028     rv = vcl_session_read_ready (session);
2029     if (rv)
2030       {
2031         clib_bitmap_set_no_check ((uword*)read_map, sid, 1);
2032         bits_set++;
2033       }
2034   }));
2035   /* *INDENT-ON* */
2036
2037 check_mq:
2038
2039   for (i = 0; i < vec_len (wrk->unhandled_evts_vector); i++)
2040     {
2041       vcl_select_handle_mq_event (wrk, &wrk->unhandled_evts_vector[i], n_bits,
2042                                   read_map, write_map, except_map, &bits_set);
2043     }
2044   vec_reset_length (wrk->unhandled_evts_vector);
2045
2046   if (vcm->cfg.use_mq_eventfd)
2047     vppcom_select_eventfd (wrk, n_bits, read_map, write_map, except_map,
2048                            time_to_wait, &bits_set);
2049   else
2050     vppcom_select_condvar (wrk, n_bits, read_map, write_map, except_map,
2051                            time_to_wait, &bits_set);
2052
2053   return (bits_set);
2054 }
2055
2056 static inline void
2057 vep_verify_epoll_chain (vcl_worker_t * wrk, u32 vep_idx)
2058 {
2059   vcl_session_t *session;
2060   vppcom_epoll_t *vep;
2061   u32 sid = vep_idx;
2062
2063   if (VPPCOM_DEBUG <= 2)
2064     return;
2065
2066   session = vcl_session_get (wrk, vep_idx);
2067   if (PREDICT_FALSE (!session))
2068     {
2069       VDBG (0, "ERROR: Invalid vep_idx (%u)!", vep_idx);
2070       goto done;
2071     }
2072   if (PREDICT_FALSE (!session->is_vep))
2073     {
2074       VDBG (0, "ERROR: vep_idx (%u) is not a vep!", vep_idx);
2075       goto done;
2076     }
2077   vep = &session->vep;
2078   VDBG (0, "vep_idx (%u): Dumping epoll chain\n"
2079         "{\n"
2080         "   is_vep         = %u\n"
2081         "   is_vep_session = %u\n"
2082         "   next_sid       = 0x%x (%u)\n"
2083         "}\n", vep_idx, session->is_vep, session->is_vep_session,
2084         vep->next_sh, vep->next_sh);
2085
2086   for (sid = vep->next_sh; sid != ~0; sid = vep->next_sh)
2087     {
2088       session = vcl_session_get (wrk, sid);
2089       if (PREDICT_FALSE (!session))
2090         {
2091           VDBG (0, "ERROR: Invalid sid (%u)!", sid);
2092           goto done;
2093         }
2094       if (PREDICT_FALSE (session->is_vep))
2095         {
2096           VDBG (0, "ERROR: sid (%u) is a vep!", vep_idx);
2097         }
2098       else if (PREDICT_FALSE (!session->is_vep_session))
2099         {
2100           VDBG (0, "ERROR: session (%u) is not a vep session!", sid);
2101           goto done;
2102         }
2103       vep = &session->vep;
2104       if (PREDICT_FALSE (vep->vep_sh != vep_idx))
2105         VDBG (0, "ERROR: session (%u) vep_idx (%u) != vep_idx (%u)!",
2106               sid, session->vep.vep_sh, vep_idx);
2107       if (session->is_vep_session)
2108         {
2109           VDBG (0, "vep_idx[%u]: sid 0x%x (%u)\n"
2110                 "{\n"
2111                 "   next_sid       = 0x%x (%u)\n"
2112                 "   prev_sid       = 0x%x (%u)\n"
2113                 "   vep_idx        = 0x%x (%u)\n"
2114                 "   ev.events      = 0x%x\n"
2115                 "   ev.data.u64    = 0x%llx\n"
2116                 "   et_mask        = 0x%x\n"
2117                 "}\n",
2118                 vep_idx, sid, sid, vep->next_sh, vep->next_sh, vep->prev_sh,
2119                 vep->prev_sh, vep->vep_sh, vep->vep_sh, vep->ev.events,
2120                 vep->ev.data.u64, vep->et_mask);
2121         }
2122     }
2123
2124 done:
2125   VDBG (0, "vep_idx (%u): Dump complete!\n", vep_idx);
2126 }
2127
2128 int
2129 vppcom_epoll_create (void)
2130 {
2131   vcl_worker_t *wrk = vcl_worker_get_current ();
2132   vcl_session_t *vep_session;
2133
2134   vep_session = vcl_session_alloc (wrk);
2135
2136   vep_session->is_vep = 1;
2137   vep_session->vep.vep_sh = ~0;
2138   vep_session->vep.next_sh = ~0;
2139   vep_session->vep.prev_sh = ~0;
2140   vep_session->vpp_handle = ~0;
2141
2142   vcl_evt (VCL_EVT_EPOLL_CREATE, vep_session, vep_session->session_index);
2143   VDBG (0, "Created vep_idx %u", vep_session->session_index);
2144
2145   return vcl_session_handle (vep_session);
2146 }
2147
2148 int
2149 vppcom_epoll_ctl (uint32_t vep_handle, int op, uint32_t session_handle,
2150                   struct epoll_event *event)
2151 {
2152   vcl_worker_t *wrk = vcl_worker_get_current ();
2153   vcl_session_t *vep_session;
2154   vcl_session_t *session;
2155   int rv = VPPCOM_OK;
2156
2157   if (vep_handle == session_handle)
2158     {
2159       VDBG (0, "vep_sh == session handle (%u)!", vep_handle);
2160       return VPPCOM_EINVAL;
2161     }
2162
2163   vep_session = vcl_session_get_w_handle (wrk, vep_handle);
2164   if (PREDICT_FALSE (!vep_session))
2165     {
2166       VDBG (0, "Invalid vep_sh (%u)!", vep_handle);
2167       return VPPCOM_EBADFD;
2168     }
2169   if (PREDICT_FALSE (!vep_session->is_vep))
2170     {
2171       VDBG (0, "vep_sh (%u) is not a vep!", vep_handle);
2172       return VPPCOM_EINVAL;
2173     }
2174
2175   ASSERT (vep_session->vep.vep_sh == ~0);
2176   ASSERT (vep_session->vep.prev_sh == ~0);
2177
2178   session = vcl_session_get_w_handle (wrk, session_handle);
2179   if (PREDICT_FALSE (!session))
2180     {
2181       VDBG (0, "Invalid session_handle (%u)!", session_handle);
2182       return VPPCOM_EBADFD;
2183     }
2184   if (PREDICT_FALSE (session->is_vep))
2185     {
2186       VDBG (0, "session_handle (%u) is a vep!", vep_handle);
2187       return VPPCOM_EINVAL;
2188     }
2189
2190   switch (op)
2191     {
2192     case EPOLL_CTL_ADD:
2193       if (PREDICT_FALSE (!event))
2194         {
2195           VDBG (0, "EPOLL_CTL_ADD: NULL pointer to epoll_event structure!");
2196           return VPPCOM_EINVAL;
2197         }
2198       if (vep_session->vep.next_sh != ~0)
2199         {
2200           vcl_session_t *next_session;
2201           next_session = vcl_session_get_w_handle (wrk,
2202                                                    vep_session->vep.next_sh);
2203           if (PREDICT_FALSE (!next_session))
2204             {
2205               VDBG (0, "EPOLL_CTL_ADD: Invalid vep.next_sh (%u) on "
2206                     "vep_idx (%u)!", vep_session->vep.next_sh, vep_handle);
2207               return VPPCOM_EBADFD;
2208             }
2209           ASSERT (next_session->vep.prev_sh == vep_handle);
2210           next_session->vep.prev_sh = session_handle;
2211         }
2212       session->vep.next_sh = vep_session->vep.next_sh;
2213       session->vep.prev_sh = vep_handle;
2214       session->vep.vep_sh = vep_handle;
2215       session->vep.et_mask = VEP_DEFAULT_ET_MASK;
2216       session->vep.ev = *event;
2217       session->is_vep = 0;
2218       session->is_vep_session = 1;
2219       vep_session->vep.next_sh = session_handle;
2220
2221       if (session->tx_fifo)
2222         svm_fifo_add_want_tx_ntf (session->tx_fifo,
2223                                   SVM_FIFO_WANT_TX_NOTIF_IF_FULL);
2224
2225       VDBG (1, "EPOLL_CTL_ADD: vep_sh %u, sh %u, events 0x%x, data 0x%llx!",
2226             vep_handle, session_handle, event->events, event->data.u64);
2227       vcl_evt (VCL_EVT_EPOLL_CTLADD, session, event->events, event->data.u64);
2228       break;
2229
2230     case EPOLL_CTL_MOD:
2231       if (PREDICT_FALSE (!event))
2232         {
2233           VDBG (0, "EPOLL_CTL_MOD: NULL pointer to epoll_event structure!");
2234           rv = VPPCOM_EINVAL;
2235           goto done;
2236         }
2237       else if (PREDICT_FALSE (!session->is_vep_session))
2238         {
2239           VDBG (0, "sh %u EPOLL_CTL_MOD: not a vep session!", session_handle);
2240           rv = VPPCOM_EINVAL;
2241           goto done;
2242         }
2243       else if (PREDICT_FALSE (session->vep.vep_sh != vep_handle))
2244         {
2245           VDBG (0, "EPOLL_CTL_MOD: sh %u vep_sh (%u) != vep_sh (%u)!",
2246                 session_handle, session->vep.vep_sh, vep_handle);
2247           rv = VPPCOM_EINVAL;
2248           goto done;
2249         }
2250       session->vep.et_mask = VEP_DEFAULT_ET_MASK;
2251       session->vep.ev = *event;
2252       VDBG (1, "EPOLL_CTL_MOD: vep_sh %u, sh %u, events 0x%x, data 0x%llx!",
2253             vep_handle, session_handle, event->events, event->data.u64);
2254       break;
2255
2256     case EPOLL_CTL_DEL:
2257       if (PREDICT_FALSE (!session->is_vep_session))
2258         {
2259           VDBG (0, "EPOLL_CTL_DEL: %u not a vep session!", session_handle);
2260           rv = VPPCOM_EINVAL;
2261           goto done;
2262         }
2263       else if (PREDICT_FALSE (session->vep.vep_sh != vep_handle))
2264         {
2265           VDBG (0, "EPOLL_CTL_DEL: sh %u vep_sh (%u) != vep_sh (%u)!",
2266                 session_handle, session->vep.vep_sh, vep_handle);
2267           rv = VPPCOM_EINVAL;
2268           goto done;
2269         }
2270
2271       if (session->vep.prev_sh == vep_handle)
2272         vep_session->vep.next_sh = session->vep.next_sh;
2273       else
2274         {
2275           vcl_session_t *prev_session;
2276           prev_session = vcl_session_get_w_handle (wrk, session->vep.prev_sh);
2277           if (PREDICT_FALSE (!prev_session))
2278             {
2279               VDBG (0, "EPOLL_CTL_DEL: Invalid prev_sh (%u) on sh (%u)!",
2280                     session->vep.prev_sh, session_handle);
2281               return VPPCOM_EBADFD;
2282             }
2283           ASSERT (prev_session->vep.next_sh == session_handle);
2284           prev_session->vep.next_sh = session->vep.next_sh;
2285         }
2286       if (session->vep.next_sh != ~0)
2287         {
2288           vcl_session_t *next_session;
2289           next_session = vcl_session_get_w_handle (wrk, session->vep.next_sh);
2290           if (PREDICT_FALSE (!next_session))
2291             {
2292               VDBG (0, "EPOLL_CTL_DEL: Invalid next_sh (%u) on sh (%u)!",
2293                     session->vep.next_sh, session_handle);
2294               return VPPCOM_EBADFD;
2295             }
2296           ASSERT (next_session->vep.prev_sh == session_handle);
2297           next_session->vep.prev_sh = session->vep.prev_sh;
2298         }
2299
2300       memset (&session->vep, 0, sizeof (session->vep));
2301       session->vep.next_sh = ~0;
2302       session->vep.prev_sh = ~0;
2303       session->vep.vep_sh = ~0;
2304       session->is_vep_session = 0;
2305
2306       if (session->tx_fifo)
2307         svm_fifo_del_want_tx_ntf (session->tx_fifo, SVM_FIFO_NO_TX_NOTIF);
2308
2309       VDBG (1, "EPOLL_CTL_DEL: vep_idx %u, sh %u!", vep_handle,
2310             session_handle);
2311       vcl_evt (VCL_EVT_EPOLL_CTLDEL, session, vep_sh);
2312       break;
2313
2314     default:
2315       VDBG (0, "Invalid operation (%d)!", op);
2316       rv = VPPCOM_EINVAL;
2317     }
2318
2319   vep_verify_epoll_chain (wrk, vep_handle);
2320
2321 done:
2322   return rv;
2323 }
2324
2325 static inline void
2326 vcl_epoll_wait_handle_mq_event (vcl_worker_t * wrk, session_event_t * e,
2327                                 struct epoll_event *events, u32 * num_ev)
2328 {
2329   session_disconnected_msg_t *disconnected_msg;
2330   session_connected_msg_t *connected_msg;
2331   u32 sid = ~0, session_events;
2332   u64 session_evt_data = ~0;
2333   vcl_session_t *session;
2334   u8 add_event = 0;
2335
2336   switch (e->event_type)
2337     {
2338     case SESSION_IO_EVT_RX:
2339       sid = e->session_index;
2340       if (!(session = vcl_session_get (wrk, sid)))
2341         break;
2342       vcl_fifo_rx_evt_valid_or_break (session);
2343       session_events = session->vep.ev.events;
2344       if (!(EPOLLIN & session->vep.ev.events) || session->has_rx_evt)
2345         break;
2346       add_event = 1;
2347       events[*num_ev].events |= EPOLLIN;
2348       session_evt_data = session->vep.ev.data.u64;
2349       session->has_rx_evt = 1;
2350       break;
2351     case SESSION_IO_EVT_TX:
2352       sid = e->session_index;
2353       if (!(session = vcl_session_get (wrk, sid)))
2354         break;
2355       session_events = session->vep.ev.events;
2356       if (!(EPOLLOUT & session_events))
2357         break;
2358       add_event = 1;
2359       events[*num_ev].events |= EPOLLOUT;
2360       session_evt_data = session->vep.ev.data.u64;
2361       svm_fifo_reset_tx_ntf (session->tx_fifo);
2362       break;
2363     case SESSION_CTRL_EVT_ACCEPTED:
2364       session = vcl_session_accepted (wrk,
2365                                       (session_accepted_msg_t *) e->data);
2366       if (!session)
2367         break;
2368
2369       session_events = session->vep.ev.events;
2370       if (!(EPOLLIN & session_events))
2371         break;
2372
2373       add_event = 1;
2374       events[*num_ev].events |= EPOLLIN;
2375       session_evt_data = session->vep.ev.data.u64;
2376       break;
2377     case SESSION_CTRL_EVT_CONNECTED:
2378       connected_msg = (session_connected_msg_t *) e->data;
2379       vcl_session_connected_handler (wrk, connected_msg);
2380       /* Generate EPOLLOUT because there's no connected event */
2381       sid = vcl_session_index_from_vpp_handle (wrk, connected_msg->handle);
2382       if (!(session = vcl_session_get (wrk, sid)))
2383         break;
2384       session_events = session->vep.ev.events;
2385       if (!(EPOLLOUT & session_events))
2386         break;
2387       add_event = 1;
2388       events[*num_ev].events |= EPOLLOUT;
2389       session_evt_data = session->vep.ev.data.u64;
2390       break;
2391     case SESSION_CTRL_EVT_DISCONNECTED:
2392       disconnected_msg = (session_disconnected_msg_t *) e->data;
2393       session = vcl_session_disconnected_handler (wrk, disconnected_msg);
2394       if (!session)
2395         break;
2396       session_events = session->vep.ev.events;
2397       if (!((EPOLLHUP | EPOLLRDHUP) & session_events))
2398         break;
2399       add_event = 1;
2400       events[*num_ev].events |= EPOLLHUP | EPOLLRDHUP;
2401       session_evt_data = session->vep.ev.data.u64;
2402       break;
2403     case SESSION_CTRL_EVT_RESET:
2404       sid = vcl_session_reset_handler (wrk, (session_reset_msg_t *) e->data);
2405       if (!(session = vcl_session_get (wrk, sid)))
2406         break;
2407       session_events = session->vep.ev.events;
2408       if (!((EPOLLHUP | EPOLLRDHUP) & session_events))
2409         break;
2410       add_event = 1;
2411       events[*num_ev].events |= EPOLLHUP | EPOLLRDHUP;
2412       session_evt_data = session->vep.ev.data.u64;
2413       break;
2414     case SESSION_CTRL_EVT_UNLISTEN_REPLY:
2415       vcl_session_unlisten_reply_handler (wrk, e->data);
2416       break;
2417     case SESSION_CTRL_EVT_REQ_WORKER_UPDATE:
2418       vcl_session_req_worker_update_handler (wrk, e->data);
2419       break;
2420     case SESSION_CTRL_EVT_WORKER_UPDATE_REPLY:
2421       vcl_session_worker_update_reply_handler (wrk, e->data);
2422       break;
2423     default:
2424       VDBG (0, "unhandled: %u", e->event_type);
2425       break;
2426     }
2427
2428   if (add_event)
2429     {
2430       events[*num_ev].data.u64 = session_evt_data;
2431       if (EPOLLONESHOT & session_events)
2432         {
2433           session = vcl_session_get (wrk, sid);
2434           session->vep.ev.events = 0;
2435         }
2436       *num_ev += 1;
2437     }
2438 }
2439
2440 static int
2441 vcl_epoll_wait_handle_mq (vcl_worker_t * wrk, svm_msg_q_t * mq,
2442                           struct epoll_event *events, u32 maxevents,
2443                           double wait_for_time, u32 * num_ev)
2444 {
2445   svm_msg_q_msg_t *msg;
2446   session_event_t *e;
2447   int i;
2448
2449   if (vec_len (wrk->mq_msg_vector) && svm_msg_q_is_empty (mq))
2450     goto handle_dequeued;
2451
2452   svm_msg_q_lock (mq);
2453   if (svm_msg_q_is_empty (mq))
2454     {
2455       if (!wait_for_time)
2456         {
2457           svm_msg_q_unlock (mq);
2458           return 0;
2459         }
2460       else if (wait_for_time < 0)
2461         {
2462           svm_msg_q_wait (mq);
2463         }
2464       else
2465         {
2466           if (svm_msg_q_timedwait (mq, wait_for_time / 1e3))
2467             {
2468               svm_msg_q_unlock (mq);
2469               return 0;
2470             }
2471         }
2472     }
2473   vcl_mq_dequeue_batch (wrk, mq);
2474   svm_msg_q_unlock (mq);
2475
2476 handle_dequeued:
2477   for (i = 0; i < vec_len (wrk->mq_msg_vector); i++)
2478     {
2479       msg = vec_elt_at_index (wrk->mq_msg_vector, i);
2480       e = svm_msg_q_msg_data (mq, msg);
2481       if (*num_ev < maxevents)
2482         vcl_epoll_wait_handle_mq_event (wrk, e, events, num_ev);
2483       else
2484         vec_add1 (wrk->unhandled_evts_vector, *e);
2485       svm_msg_q_free_msg (mq, msg);
2486     }
2487   vec_reset_length (wrk->mq_msg_vector);
2488   vcl_handle_pending_wrk_updates (wrk);
2489   return *num_ev;
2490 }
2491
2492 static int
2493 vppcom_epoll_wait_condvar (vcl_worker_t * wrk, struct epoll_event *events,
2494                            int maxevents, u32 n_evts, double wait_for_time)
2495 {
2496   double wait = 0, start = 0;
2497
2498   if (!n_evts)
2499     {
2500       wait = wait_for_time;
2501       start = clib_time_now (&wrk->clib_time);
2502     }
2503
2504   do
2505     {
2506       vcl_epoll_wait_handle_mq (wrk, wrk->app_event_queue, events, maxevents,
2507                                 wait, &n_evts);
2508       if (n_evts)
2509         return n_evts;
2510       if (wait == -1)
2511         continue;
2512
2513       wait = wait - (clib_time_now (&wrk->clib_time) - start);
2514     }
2515   while (wait > 0);
2516
2517   return 0;
2518 }
2519
2520 static int
2521 vppcom_epoll_wait_eventfd (vcl_worker_t * wrk, struct epoll_event *events,
2522                            int maxevents, u32 n_evts, double wait_for_time)
2523 {
2524   vcl_mq_evt_conn_t *mqc;
2525   int __clib_unused n_read;
2526   int n_mq_evts, i;
2527   u64 buf;
2528
2529   vec_validate (wrk->mq_events, pool_elts (wrk->mq_evt_conns));
2530 again:
2531   n_mq_evts = epoll_wait (wrk->mqs_epfd, wrk->mq_events,
2532                           vec_len (wrk->mq_events), wait_for_time);
2533   for (i = 0; i < n_mq_evts; i++)
2534     {
2535       mqc = vcl_mq_evt_conn_get (wrk, wrk->mq_events[i].data.u32);
2536       n_read = read (mqc->mq_fd, &buf, sizeof (buf));
2537       vcl_epoll_wait_handle_mq (wrk, mqc->mq, events, maxevents, 0, &n_evts);
2538     }
2539   if (!n_evts && n_mq_evts > 0)
2540     goto again;
2541
2542   return (int) n_evts;
2543 }
2544
2545 int
2546 vppcom_epoll_wait (uint32_t vep_handle, struct epoll_event *events,
2547                    int maxevents, double wait_for_time)
2548 {
2549   vcl_worker_t *wrk = vcl_worker_get_current ();
2550   vcl_session_t *vep_session;
2551   u32 n_evts = 0;
2552   int i;
2553
2554   if (PREDICT_FALSE (maxevents <= 0))
2555     {
2556       VDBG (0, "ERROR: Invalid maxevents (%d)!", maxevents);
2557       return VPPCOM_EINVAL;
2558     }
2559
2560   vep_session = vcl_session_get_w_handle (wrk, vep_handle);
2561   if (!vep_session)
2562     return VPPCOM_EBADFD;
2563
2564   if (PREDICT_FALSE (!vep_session->is_vep))
2565     {
2566       VDBG (0, "ERROR: vep_idx (%u) is not a vep!", vep_handle);
2567       return VPPCOM_EINVAL;
2568     }
2569
2570   memset (events, 0, sizeof (*events) * maxevents);
2571
2572   if (vec_len (wrk->unhandled_evts_vector))
2573     {
2574       for (i = 0; i < vec_len (wrk->unhandled_evts_vector); i++)
2575         {
2576           vcl_epoll_wait_handle_mq_event (wrk, &wrk->unhandled_evts_vector[i],
2577                                           events, &n_evts);
2578           if (n_evts == maxevents)
2579             {
2580               i += 1;
2581               break;
2582             }
2583         }
2584       vec_delete (wrk->unhandled_evts_vector, i, 0);
2585     }
2586
2587   if (vcm->cfg.use_mq_eventfd)
2588     return vppcom_epoll_wait_eventfd (wrk, events, maxevents, n_evts,
2589                                       wait_for_time);
2590
2591   return vppcom_epoll_wait_condvar (wrk, events, maxevents, n_evts,
2592                                     wait_for_time);
2593 }
2594
2595 int
2596 vppcom_session_attr (uint32_t session_handle, uint32_t op,
2597                      void *buffer, uint32_t * buflen)
2598 {
2599   vcl_worker_t *wrk = vcl_worker_get_current ();
2600   vcl_session_t *session;
2601   int rv = VPPCOM_OK;
2602   u32 *flags = buffer, tmp_flags = 0;
2603   vppcom_endpt_t *ep = buffer;
2604
2605   session = vcl_session_get_w_handle (wrk, session_handle);
2606   if (!session)
2607     return VPPCOM_EBADFD;
2608
2609   switch (op)
2610     {
2611     case VPPCOM_ATTR_GET_NREAD:
2612       rv = vcl_session_read_ready (session);
2613       VDBG (2, "VPPCOM_ATTR_GET_NREAD: sh %u, nread = %d", session_handle,
2614             rv);
2615       break;
2616
2617     case VPPCOM_ATTR_GET_NWRITE:
2618       rv = vcl_session_write_ready (session);
2619       VDBG (2, "VPPCOM_ATTR_GET_NWRITE: sh %u, nwrite = %d", session_handle,
2620             rv);
2621       break;
2622
2623     case VPPCOM_ATTR_GET_FLAGS:
2624       if (PREDICT_TRUE (buffer && buflen && (*buflen >= sizeof (*flags))))
2625         {
2626           *flags = O_RDWR | (VCL_SESS_ATTR_TEST (session->attr,
2627                                                  VCL_SESS_ATTR_NONBLOCK));
2628           *buflen = sizeof (*flags);
2629           VDBG (2, "VPPCOM_ATTR_GET_FLAGS: sh %u, flags = 0x%08x, "
2630                 "is_nonblocking = %u", session_handle, *flags,
2631                 VCL_SESS_ATTR_TEST (session->attr, VCL_SESS_ATTR_NONBLOCK));
2632         }
2633       else
2634         rv = VPPCOM_EINVAL;
2635       break;
2636
2637     case VPPCOM_ATTR_SET_FLAGS:
2638       if (PREDICT_TRUE (buffer && buflen && (*buflen == sizeof (*flags))))
2639         {
2640           if (*flags & O_NONBLOCK)
2641             VCL_SESS_ATTR_SET (session->attr, VCL_SESS_ATTR_NONBLOCK);
2642           else
2643             VCL_SESS_ATTR_CLR (session->attr, VCL_SESS_ATTR_NONBLOCK);
2644
2645           VDBG (2, "VPPCOM_ATTR_SET_FLAGS: sh %u, flags = 0x%08x,"
2646                 " is_nonblocking = %u", session_handle, *flags,
2647                 VCL_SESS_ATTR_TEST (session->attr, VCL_SESS_ATTR_NONBLOCK));
2648         }
2649       else
2650         rv = VPPCOM_EINVAL;
2651       break;
2652
2653     case VPPCOM_ATTR_GET_PEER_ADDR:
2654       if (PREDICT_TRUE (buffer && buflen &&
2655                         (*buflen >= sizeof (*ep)) && ep->ip))
2656         {
2657           ep->is_ip4 = session->transport.is_ip4;
2658           ep->port = session->transport.rmt_port;
2659           if (session->transport.is_ip4)
2660             clib_memcpy_fast (ep->ip, &session->transport.rmt_ip.ip4,
2661                               sizeof (ip4_address_t));
2662           else
2663             clib_memcpy_fast (ep->ip, &session->transport.rmt_ip.ip6,
2664                               sizeof (ip6_address_t));
2665           *buflen = sizeof (*ep);
2666           VDBG (1, "VPPCOM_ATTR_GET_PEER_ADDR: sh %u, is_ip4 = %u, "
2667                 "addr = %U, port %u", session_handle, ep->is_ip4,
2668                 format_ip46_address, &session->transport.rmt_ip,
2669                 ep->is_ip4 ? IP46_TYPE_IP4 : IP46_TYPE_IP6,
2670                 clib_net_to_host_u16 (ep->port));
2671         }
2672       else
2673         rv = VPPCOM_EINVAL;
2674       break;
2675
2676     case VPPCOM_ATTR_GET_LCL_ADDR:
2677       if (PREDICT_TRUE (buffer && buflen &&
2678                         (*buflen >= sizeof (*ep)) && ep->ip))
2679         {
2680           ep->is_ip4 = session->transport.is_ip4;
2681           ep->port = session->transport.lcl_port;
2682           if (session->transport.is_ip4)
2683             clib_memcpy_fast (ep->ip, &session->transport.lcl_ip.ip4,
2684                               sizeof (ip4_address_t));
2685           else
2686             clib_memcpy_fast (ep->ip, &session->transport.lcl_ip.ip6,
2687                               sizeof (ip6_address_t));
2688           *buflen = sizeof (*ep);
2689           VDBG (1, "VPPCOM_ATTR_GET_LCL_ADDR: sh %u, is_ip4 = %u, addr = %U"
2690                 " port %d", session_handle, ep->is_ip4, format_ip46_address,
2691                 &session->transport.lcl_ip,
2692                 ep->is_ip4 ? IP46_TYPE_IP4 : IP46_TYPE_IP6,
2693                 clib_net_to_host_u16 (ep->port));
2694         }
2695       else
2696         rv = VPPCOM_EINVAL;
2697       break;
2698
2699     case VPPCOM_ATTR_GET_LIBC_EPFD:
2700       rv = session->libc_epfd;
2701       VDBG (2, "VPPCOM_ATTR_GET_LIBC_EPFD: libc_epfd %d", rv);
2702       break;
2703
2704     case VPPCOM_ATTR_SET_LIBC_EPFD:
2705       if (PREDICT_TRUE (buffer && buflen &&
2706                         (*buflen == sizeof (session->libc_epfd))))
2707         {
2708           session->libc_epfd = *(int *) buffer;
2709           *buflen = sizeof (session->libc_epfd);
2710
2711           VDBG (2, "VPPCOM_ATTR_SET_LIBC_EPFD: libc_epfd %d, buflen %d",
2712                 session->libc_epfd, *buflen);
2713         }
2714       else
2715         rv = VPPCOM_EINVAL;
2716       break;
2717
2718     case VPPCOM_ATTR_GET_PROTOCOL:
2719       if (buffer && buflen && (*buflen >= sizeof (int)))
2720         {
2721           *(int *) buffer = session->session_type;
2722           *buflen = sizeof (int);
2723
2724           VDBG (2, "VPPCOM_ATTR_GET_PROTOCOL: %d (%s), buflen %d",
2725                 *(int *) buffer, *(int *) buffer ? "UDP" : "TCP", *buflen);
2726         }
2727       else
2728         rv = VPPCOM_EINVAL;
2729       break;
2730
2731     case VPPCOM_ATTR_GET_LISTEN:
2732       if (buffer && buflen && (*buflen >= sizeof (int)))
2733         {
2734           *(int *) buffer = VCL_SESS_ATTR_TEST (session->attr,
2735                                                 VCL_SESS_ATTR_LISTEN);
2736           *buflen = sizeof (int);
2737
2738           VDBG (2, "VPPCOM_ATTR_GET_LISTEN: %d, buflen %d", *(int *) buffer,
2739                 *buflen);
2740         }
2741       else
2742         rv = VPPCOM_EINVAL;
2743       break;
2744
2745     case VPPCOM_ATTR_GET_ERROR:
2746       if (buffer && buflen && (*buflen >= sizeof (int)))
2747         {
2748           *(int *) buffer = 0;
2749           *buflen = sizeof (int);
2750
2751           VDBG (2, "VPPCOM_ATTR_GET_ERROR: %d, buflen %d, #VPP-TBD#",
2752                 *(int *) buffer, *buflen);
2753         }
2754       else
2755         rv = VPPCOM_EINVAL;
2756       break;
2757
2758     case VPPCOM_ATTR_GET_TX_FIFO_LEN:
2759       if (buffer && buflen && (*buflen >= sizeof (u32)))
2760         {
2761
2762           /* VPP-TBD */
2763           *(size_t *) buffer = (session->sndbuf_size ? session->sndbuf_size :
2764                                 session->tx_fifo ? session->tx_fifo->nitems :
2765                                 vcm->cfg.tx_fifo_size);
2766           *buflen = sizeof (u32);
2767
2768           VDBG (2, "VPPCOM_ATTR_GET_TX_FIFO_LEN: %u (0x%x), buflen %d,"
2769                 " #VPP-TBD#", *(size_t *) buffer, *(size_t *) buffer,
2770                 *buflen);
2771         }
2772       else
2773         rv = VPPCOM_EINVAL;
2774       break;
2775
2776     case VPPCOM_ATTR_SET_TX_FIFO_LEN:
2777       if (buffer && buflen && (*buflen == sizeof (u32)))
2778         {
2779           /* VPP-TBD */
2780           session->sndbuf_size = *(u32 *) buffer;
2781           VDBG (2, "VPPCOM_ATTR_SET_TX_FIFO_LEN: %u (0x%x), buflen %d,"
2782                 " #VPP-TBD#", session->sndbuf_size, session->sndbuf_size,
2783                 *buflen);
2784         }
2785       else
2786         rv = VPPCOM_EINVAL;
2787       break;
2788
2789     case VPPCOM_ATTR_GET_RX_FIFO_LEN:
2790       if (buffer && buflen && (*buflen >= sizeof (u32)))
2791         {
2792
2793           /* VPP-TBD */
2794           *(size_t *) buffer = (session->rcvbuf_size ? session->rcvbuf_size :
2795                                 session->rx_fifo ? session->rx_fifo->nitems :
2796                                 vcm->cfg.rx_fifo_size);
2797           *buflen = sizeof (u32);
2798
2799           VDBG (2, "VPPCOM_ATTR_GET_RX_FIFO_LEN: %u (0x%x), buflen %d, "
2800                 "#VPP-TBD#", *(size_t *) buffer, *(size_t *) buffer, *buflen);
2801         }
2802       else
2803         rv = VPPCOM_EINVAL;
2804       break;
2805
2806     case VPPCOM_ATTR_SET_RX_FIFO_LEN:
2807       if (buffer && buflen && (*buflen == sizeof (u32)))
2808         {
2809           /* VPP-TBD */
2810           session->rcvbuf_size = *(u32 *) buffer;
2811           VDBG (2, "VPPCOM_ATTR_SET_RX_FIFO_LEN: %u (0x%x), buflen %d,"
2812                 " #VPP-TBD#", session->sndbuf_size, session->sndbuf_size,
2813                 *buflen);
2814         }
2815       else
2816         rv = VPPCOM_EINVAL;
2817       break;
2818
2819     case VPPCOM_ATTR_GET_REUSEADDR:
2820       if (buffer && buflen && (*buflen >= sizeof (int)))
2821         {
2822           /* VPP-TBD */
2823           *(int *) buffer = VCL_SESS_ATTR_TEST (session->attr,
2824                                                 VCL_SESS_ATTR_REUSEADDR);
2825           *buflen = sizeof (int);
2826
2827           VDBG (2, "VPPCOM_ATTR_GET_REUSEADDR: %d, buflen %d, #VPP-TBD#",
2828                 *(int *) buffer, *buflen);
2829         }
2830       else
2831         rv = VPPCOM_EINVAL;
2832       break;
2833
2834     case VPPCOM_ATTR_SET_REUSEADDR:
2835       if (buffer && buflen && (*buflen == sizeof (int)) &&
2836           !VCL_SESS_ATTR_TEST (session->attr, VCL_SESS_ATTR_LISTEN))
2837         {
2838           /* VPP-TBD */
2839           if (*(int *) buffer)
2840             VCL_SESS_ATTR_SET (session->attr, VCL_SESS_ATTR_REUSEADDR);
2841           else
2842             VCL_SESS_ATTR_CLR (session->attr, VCL_SESS_ATTR_REUSEADDR);
2843
2844           VDBG (2, "VPPCOM_ATTR_SET_REUSEADDR: %d, buflen %d, #VPP-TBD#",
2845                 VCL_SESS_ATTR_TEST (session->attr, VCL_SESS_ATTR_REUSEADDR),
2846                 *buflen);
2847         }
2848       else
2849         rv = VPPCOM_EINVAL;
2850       break;
2851
2852     case VPPCOM_ATTR_GET_REUSEPORT:
2853       if (buffer && buflen && (*buflen >= sizeof (int)))
2854         {
2855           /* VPP-TBD */
2856           *(int *) buffer = VCL_SESS_ATTR_TEST (session->attr,
2857                                                 VCL_SESS_ATTR_REUSEPORT);
2858           *buflen = sizeof (int);
2859
2860           VDBG (2, "VPPCOM_ATTR_GET_REUSEPORT: %d, buflen %d, #VPP-TBD#",
2861                 *(int *) buffer, *buflen);
2862         }
2863       else
2864         rv = VPPCOM_EINVAL;
2865       break;
2866
2867     case VPPCOM_ATTR_SET_REUSEPORT:
2868       if (buffer && buflen && (*buflen == sizeof (int)) &&
2869           !VCL_SESS_ATTR_TEST (session->attr, VCL_SESS_ATTR_LISTEN))
2870         {
2871           /* VPP-TBD */
2872           if (*(int *) buffer)
2873             VCL_SESS_ATTR_SET (session->attr, VCL_SESS_ATTR_REUSEPORT);
2874           else
2875             VCL_SESS_ATTR_CLR (session->attr, VCL_SESS_ATTR_REUSEPORT);
2876
2877           VDBG (2, "VPPCOM_ATTR_SET_REUSEPORT: %d, buflen %d, #VPP-TBD#",
2878                 VCL_SESS_ATTR_TEST (session->attr, VCL_SESS_ATTR_REUSEPORT),
2879                 *buflen);
2880         }
2881       else
2882         rv = VPPCOM_EINVAL;
2883       break;
2884
2885     case VPPCOM_ATTR_GET_BROADCAST:
2886       if (buffer && buflen && (*buflen >= sizeof (int)))
2887         {
2888           /* VPP-TBD */
2889           *(int *) buffer = VCL_SESS_ATTR_TEST (session->attr,
2890                                                 VCL_SESS_ATTR_BROADCAST);
2891           *buflen = sizeof (int);
2892
2893           VDBG (2, "VPPCOM_ATTR_GET_BROADCAST: %d, buflen %d, #VPP-TBD#",
2894                 *(int *) buffer, *buflen);
2895         }
2896       else
2897         rv = VPPCOM_EINVAL;
2898       break;
2899
2900     case VPPCOM_ATTR_SET_BROADCAST:
2901       if (buffer && buflen && (*buflen == sizeof (int)))
2902         {
2903           /* VPP-TBD */
2904           if (*(int *) buffer)
2905             VCL_SESS_ATTR_SET (session->attr, VCL_SESS_ATTR_BROADCAST);
2906           else
2907             VCL_SESS_ATTR_CLR (session->attr, VCL_SESS_ATTR_BROADCAST);
2908
2909           VDBG (2, "VPPCOM_ATTR_SET_BROADCAST: %d, buflen %d, #VPP-TBD#",
2910                 VCL_SESS_ATTR_TEST (session->attr, VCL_SESS_ATTR_BROADCAST),
2911                 *buflen);
2912         }
2913       else
2914         rv = VPPCOM_EINVAL;
2915       break;
2916
2917     case VPPCOM_ATTR_GET_V6ONLY:
2918       if (buffer && buflen && (*buflen >= sizeof (int)))
2919         {
2920           /* VPP-TBD */
2921           *(int *) buffer = VCL_SESS_ATTR_TEST (session->attr,
2922                                                 VCL_SESS_ATTR_V6ONLY);
2923           *buflen = sizeof (int);
2924
2925           VDBG (2, "VPPCOM_ATTR_GET_V6ONLY: %d, buflen %d, #VPP-TBD#",
2926                 *(int *) buffer, *buflen);
2927         }
2928       else
2929         rv = VPPCOM_EINVAL;
2930       break;
2931
2932     case VPPCOM_ATTR_SET_V6ONLY:
2933       if (buffer && buflen && (*buflen == sizeof (int)))
2934         {
2935           /* VPP-TBD */
2936           if (*(int *) buffer)
2937             VCL_SESS_ATTR_SET (session->attr, VCL_SESS_ATTR_V6ONLY);
2938           else
2939             VCL_SESS_ATTR_CLR (session->attr, VCL_SESS_ATTR_V6ONLY);
2940
2941           VDBG (2, "VPPCOM_ATTR_SET_V6ONLY: %d, buflen %d, #VPP-TBD#",
2942                 VCL_SESS_ATTR_TEST (session->attr, VCL_SESS_ATTR_V6ONLY),
2943                 *buflen);
2944         }
2945       else
2946         rv = VPPCOM_EINVAL;
2947       break;
2948
2949     case VPPCOM_ATTR_GET_KEEPALIVE:
2950       if (buffer && buflen && (*buflen >= sizeof (int)))
2951         {
2952           /* VPP-TBD */
2953           *(int *) buffer = VCL_SESS_ATTR_TEST (session->attr,
2954                                                 VCL_SESS_ATTR_KEEPALIVE);
2955           *buflen = sizeof (int);
2956
2957           VDBG (2, "VPPCOM_ATTR_GET_KEEPALIVE: %d, buflen %d, #VPP-TBD#",
2958                 *(int *) buffer, *buflen);
2959         }
2960       else
2961         rv = VPPCOM_EINVAL;
2962       break;
2963
2964     case VPPCOM_ATTR_SET_KEEPALIVE:
2965       if (buffer && buflen && (*buflen == sizeof (int)))
2966         {
2967           /* VPP-TBD */
2968           if (*(int *) buffer)
2969             VCL_SESS_ATTR_SET (session->attr, VCL_SESS_ATTR_KEEPALIVE);
2970           else
2971             VCL_SESS_ATTR_CLR (session->attr, VCL_SESS_ATTR_KEEPALIVE);
2972
2973           VDBG (2, "VPPCOM_ATTR_SET_KEEPALIVE: %d, buflen %d, #VPP-TBD#",
2974                 VCL_SESS_ATTR_TEST (session->attr, VCL_SESS_ATTR_KEEPALIVE),
2975                 *buflen);
2976         }
2977       else
2978         rv = VPPCOM_EINVAL;
2979       break;
2980
2981     case VPPCOM_ATTR_GET_TCP_NODELAY:
2982       if (buffer && buflen && (*buflen >= sizeof (int)))
2983         {
2984           /* VPP-TBD */
2985           *(int *) buffer = VCL_SESS_ATTR_TEST (session->attr,
2986                                                 VCL_SESS_ATTR_TCP_NODELAY);
2987           *buflen = sizeof (int);
2988
2989           VDBG (2, "VPPCOM_ATTR_GET_TCP_NODELAY: %d, buflen %d, #VPP-TBD#",
2990                 *(int *) buffer, *buflen);
2991         }
2992       else
2993         rv = VPPCOM_EINVAL;
2994       break;
2995
2996     case VPPCOM_ATTR_SET_TCP_NODELAY:
2997       if (buffer && buflen && (*buflen == sizeof (int)))
2998         {
2999           /* VPP-TBD */
3000           if (*(int *) buffer)
3001             VCL_SESS_ATTR_SET (session->attr, VCL_SESS_ATTR_TCP_NODELAY);
3002           else
3003             VCL_SESS_ATTR_CLR (session->attr, VCL_SESS_ATTR_TCP_NODELAY);
3004
3005           VDBG (2, "VPPCOM_ATTR_SET_TCP_NODELAY: %d, buflen %d, #VPP-TBD#",
3006                 VCL_SESS_ATTR_TEST (session->attr, VCL_SESS_ATTR_TCP_NODELAY),
3007                 *buflen);
3008         }
3009       else
3010         rv = VPPCOM_EINVAL;
3011       break;
3012
3013     case VPPCOM_ATTR_GET_TCP_KEEPIDLE:
3014       if (buffer && buflen && (*buflen >= sizeof (int)))
3015         {
3016           /* VPP-TBD */
3017           *(int *) buffer = VCL_SESS_ATTR_TEST (session->attr,
3018                                                 VCL_SESS_ATTR_TCP_KEEPIDLE);
3019           *buflen = sizeof (int);
3020
3021           VDBG (2, "VPPCOM_ATTR_GET_TCP_KEEPIDLE: %d, buflen %d, #VPP-TBD#",
3022                 *(int *) buffer, *buflen);
3023         }
3024       else
3025         rv = VPPCOM_EINVAL;
3026       break;
3027
3028     case VPPCOM_ATTR_SET_TCP_KEEPIDLE:
3029       if (buffer && buflen && (*buflen == sizeof (int)))
3030         {
3031           /* VPP-TBD */
3032           if (*(int *) buffer)
3033             VCL_SESS_ATTR_SET (session->attr, VCL_SESS_ATTR_TCP_KEEPIDLE);
3034           else
3035             VCL_SESS_ATTR_CLR (session->attr, VCL_SESS_ATTR_TCP_KEEPIDLE);
3036
3037           VDBG (2, "VPPCOM_ATTR_SET_TCP_KEEPIDLE: %d, buflen %d, #VPP-TBD#",
3038                 VCL_SESS_ATTR_TEST (session->attr,
3039                                     VCL_SESS_ATTR_TCP_KEEPIDLE), *buflen);
3040         }
3041       else
3042         rv = VPPCOM_EINVAL;
3043       break;
3044
3045     case VPPCOM_ATTR_GET_TCP_KEEPINTVL:
3046       if (buffer && buflen && (*buflen >= sizeof (int)))
3047         {
3048           /* VPP-TBD */
3049           *(int *) buffer = VCL_SESS_ATTR_TEST (session->attr,
3050                                                 VCL_SESS_ATTR_TCP_KEEPINTVL);
3051           *buflen = sizeof (int);
3052
3053           VDBG (2, "VPPCOM_ATTR_GET_TCP_KEEPINTVL: %d, buflen %d, #VPP-TBD#",
3054                 *(int *) buffer, *buflen);
3055         }
3056       else
3057         rv = VPPCOM_EINVAL;
3058       break;
3059
3060     case VPPCOM_ATTR_SET_TCP_KEEPINTVL:
3061       if (buffer && buflen && (*buflen == sizeof (int)))
3062         {
3063           /* VPP-TBD */
3064           if (*(int *) buffer)
3065             VCL_SESS_ATTR_SET (session->attr, VCL_SESS_ATTR_TCP_KEEPINTVL);
3066           else
3067             VCL_SESS_ATTR_CLR (session->attr, VCL_SESS_ATTR_TCP_KEEPINTVL);
3068
3069           VDBG (2, "VPPCOM_ATTR_SET_TCP_KEEPINTVL: %d, buflen %d, #VPP-TBD#",
3070                 VCL_SESS_ATTR_TEST (session->attr,
3071                                     VCL_SESS_ATTR_TCP_KEEPINTVL), *buflen);
3072         }
3073       else
3074         rv = VPPCOM_EINVAL;
3075       break;
3076
3077     case VPPCOM_ATTR_GET_TCP_USER_MSS:
3078       if (buffer && buflen && (*buflen >= sizeof (u32)))
3079         {
3080           /* VPP-TBD */
3081           *(u32 *) buffer = session->user_mss;
3082           *buflen = sizeof (int);
3083
3084           VDBG (2, "VPPCOM_ATTR_GET_TCP_USER_MSS: %d, buflen %d, #VPP-TBD#",
3085                 *(int *) buffer, *buflen);
3086         }
3087       else
3088         rv = VPPCOM_EINVAL;
3089       break;
3090
3091     case VPPCOM_ATTR_SET_TCP_USER_MSS:
3092       if (buffer && buflen && (*buflen == sizeof (u32)))
3093         {
3094           /* VPP-TBD */
3095           session->user_mss = *(u32 *) buffer;
3096
3097           VDBG (2, "VPPCOM_ATTR_SET_TCP_USER_MSS: %u, buflen %d, #VPP-TBD#",
3098                 session->user_mss, *buflen);
3099         }
3100       else
3101         rv = VPPCOM_EINVAL;
3102       break;
3103
3104     case VPPCOM_ATTR_SET_SHUT:
3105       if (*flags == SHUT_RD || *flags == SHUT_RDWR)
3106         VCL_SESS_ATTR_SET (session->attr, VCL_SESS_ATTR_SHUT_RD);
3107       if (*flags == SHUT_WR || *flags == SHUT_RDWR)
3108         VCL_SESS_ATTR_SET (session->attr, VCL_SESS_ATTR_SHUT_WR);
3109       break;
3110
3111     case VPPCOM_ATTR_GET_SHUT:
3112       if (VCL_SESS_ATTR_TEST (session->attr, VCL_SESS_ATTR_SHUT_RD))
3113         tmp_flags = 1;
3114       if (VCL_SESS_ATTR_TEST (session->attr, VCL_SESS_ATTR_SHUT_WR))
3115         tmp_flags |= 2;
3116       if (tmp_flags == 1)
3117         *(int *) buffer = SHUT_RD;
3118       else if (tmp_flags == 2)
3119         *(int *) buffer = SHUT_WR;
3120       else if (tmp_flags == 3)
3121         *(int *) buffer = SHUT_RDWR;
3122       *buflen = sizeof (int);
3123       break;
3124     default:
3125       rv = VPPCOM_EINVAL;
3126       break;
3127     }
3128
3129   return rv;
3130 }
3131
3132 int
3133 vppcom_session_recvfrom (uint32_t session_handle, void *buffer,
3134                          uint32_t buflen, int flags, vppcom_endpt_t * ep)
3135 {
3136   vcl_worker_t *wrk = vcl_worker_get_current ();
3137   int rv = VPPCOM_OK;
3138   vcl_session_t *session = 0;
3139
3140   if (ep)
3141     {
3142       session = vcl_session_get_w_handle (wrk, session_handle);
3143       if (PREDICT_FALSE (!session))
3144         {
3145           VDBG (0, "sh 0x%llx is closed!", session_handle);
3146           return VPPCOM_EBADFD;
3147         }
3148       ep->is_ip4 = session->transport.is_ip4;
3149       ep->port = session->transport.rmt_port;
3150     }
3151
3152   if (flags == 0)
3153     rv = vppcom_session_read (session_handle, buffer, buflen);
3154   else if (flags & MSG_PEEK)
3155     rv = vppcom_session_peek (session_handle, buffer, buflen);
3156   else
3157     {
3158       VDBG (0, "Unsupport flags for recvfrom %d", flags);
3159       return VPPCOM_EAFNOSUPPORT;
3160     }
3161
3162   if (ep)
3163     {
3164       if (session->transport.is_ip4)
3165         clib_memcpy_fast (ep->ip, &session->transport.rmt_ip.ip4,
3166                           sizeof (ip4_address_t));
3167       else
3168         clib_memcpy_fast (ep->ip, &session->transport.rmt_ip.ip6,
3169                           sizeof (ip6_address_t));
3170     }
3171
3172   return rv;
3173 }
3174
3175 int
3176 vppcom_session_sendto (uint32_t session_handle, void *buffer,
3177                        uint32_t buflen, int flags, vppcom_endpt_t * ep)
3178 {
3179   if (!buffer)
3180     return VPPCOM_EINVAL;
3181
3182   if (ep)
3183     {
3184       // TBD
3185       return VPPCOM_EINVAL;
3186     }
3187
3188   if (flags)
3189     {
3190       // TBD check the flags and do the right thing
3191       VDBG (2, "handling flags 0x%u (%d) not implemented yet.", flags, flags);
3192     }
3193
3194   return (vppcom_session_write_inline (session_handle, buffer, buflen, 1));
3195 }
3196
3197 int
3198 vppcom_poll (vcl_poll_t * vp, uint32_t n_sids, double wait_for_time)
3199 {
3200   vcl_worker_t *wrk = vcl_worker_get_current ();
3201   f64 timeout = clib_time_now (&wrk->clib_time) + wait_for_time;
3202   u32 i, keep_trying = 1;
3203   svm_msg_q_msg_t msg;
3204   session_event_t *e;
3205   int rv, num_ev = 0;
3206
3207   VDBG (3, "vp %p, nsids %u, wait_for_time %f", vp, n_sids, wait_for_time);
3208
3209   if (!vp)
3210     return VPPCOM_EFAULT;
3211
3212   do
3213     {
3214       vcl_session_t *session;
3215
3216       /* Dequeue all events and drop all unhandled io events */
3217       while (svm_msg_q_sub (wrk->app_event_queue, &msg, SVM_Q_NOWAIT, 0) == 0)
3218         {
3219           e = svm_msg_q_msg_data (wrk->app_event_queue, &msg);
3220           vcl_handle_mq_event (wrk, e);
3221           svm_msg_q_free_msg (wrk->app_event_queue, &msg);
3222         }
3223       vec_reset_length (wrk->unhandled_evts_vector);
3224
3225       for (i = 0; i < n_sids; i++)
3226         {
3227           session = vcl_session_get (wrk, vp[i].sh);
3228           if (!session)
3229             {
3230               vp[i].revents = POLLHUP;
3231               num_ev++;
3232               continue;
3233             }
3234
3235           vp[i].revents = 0;
3236
3237           if (POLLIN & vp[i].events)
3238             {
3239               rv = vcl_session_read_ready (session);
3240               if (rv > 0)
3241                 {
3242                   vp[i].revents |= POLLIN;
3243                   num_ev++;
3244                 }
3245               else if (rv < 0)
3246                 {
3247                   switch (rv)
3248                     {
3249                     case VPPCOM_ECONNRESET:
3250                       vp[i].revents = POLLHUP;
3251                       break;
3252
3253                     default:
3254                       vp[i].revents = POLLERR;
3255                       break;
3256                     }
3257                   num_ev++;
3258                 }
3259             }
3260
3261           if (POLLOUT & vp[i].events)
3262             {
3263               rv = vcl_session_write_ready (session);
3264               if (rv > 0)
3265                 {
3266                   vp[i].revents |= POLLOUT;
3267                   num_ev++;
3268                 }
3269               else if (rv < 0)
3270                 {
3271                   switch (rv)
3272                     {
3273                     case VPPCOM_ECONNRESET:
3274                       vp[i].revents = POLLHUP;
3275                       break;
3276
3277                     default:
3278                       vp[i].revents = POLLERR;
3279                       break;
3280                     }
3281                   num_ev++;
3282                 }
3283             }
3284
3285           if (0)                // Note "done:" label used by VCL_SESSION_LOCK_AND_GET()
3286             {
3287               vp[i].revents = POLLNVAL;
3288               num_ev++;
3289             }
3290         }
3291       if (wait_for_time != -1)
3292         keep_trying = (clib_time_now (&wrk->clib_time) <= timeout) ? 1 : 0;
3293     }
3294   while ((num_ev == 0) && keep_trying);
3295
3296   return num_ev;
3297 }
3298
3299 int
3300 vppcom_mq_epoll_fd (void)
3301 {
3302   vcl_worker_t *wrk = vcl_worker_get_current ();
3303   return wrk->mqs_epfd;
3304 }
3305
3306 int
3307 vppcom_session_index (vcl_session_handle_t session_handle)
3308 {
3309   return session_handle & 0xFFFFFF;
3310 }
3311
3312 int
3313 vppcom_session_worker (vcl_session_handle_t session_handle)
3314 {
3315   return session_handle >> 24;
3316 }
3317
3318 int
3319 vppcom_worker_register (void)
3320 {
3321   if (!vcl_worker_alloc_and_init ())
3322     return VPPCOM_EEXIST;
3323
3324   if (vcl_worker_set_bapi ())
3325     return VPPCOM_EEXIST;
3326
3327   if (vcl_worker_register_with_vpp ())
3328     return VPPCOM_EEXIST;
3329
3330   return VPPCOM_OK;
3331 }
3332
3333 int
3334 vppcom_worker_index (void)
3335 {
3336   return vcl_get_worker_index ();
3337 }
3338
3339 int
3340 vppcom_worker_mqs_epfd (void)
3341 {
3342   vcl_worker_t *wrk = vcl_worker_get_current ();
3343   if (!vcm->cfg.use_mq_eventfd)
3344     return -1;
3345   return wrk->mqs_epfd;
3346 }
3347
3348 /*
3349  * fd.io coding-style-patch-verification: ON
3350  *
3351  * Local Variables:
3352  * eval: (c-set-style "gnu")
3353  * End:
3354  */