src/vcl/vppcom.c

   1 /*
   2  * Copyright (c) 2017-2019 Cisco and/or its affiliates.
   3  * Licensed under the Apache License, Version 2.0 (the "License");
   4  * you may not use this
   5  * You may obtain a copy of the License at:
   6  *
   7  *     http://www.apache.org/licenses/LICENSE-2.0
   8  *
   9  * Unless required by applicable law or agreed to in writing, software
  10  * distributed under the License is distributed on an "AS IS" BASIS,
  11  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  12  * See the License for the specific language governing permissions and
  13  * limitations under the License.
  14  */
  15
  16 #include <stdio.h>
  17 #include <stdlib.h>
  18 #include <vcl/vppcom.h>
  19 #include <vcl/vcl_debug.h>
  20 #include <vcl/vcl_private.h>
  21 #include <svm/fifo_segment.h>
  22
  23 __thread uword __vcl_worker_index = ~0;
  24
  25 static inline int
  26 vcl_mq_dequeue_batch (vcl_worker_t * wrk, svm_msg_q_t * mq, u32 n_max_msg)
  27 {
  28   u32 n_msgs = 0, sz, len;
  29
  30   while ((sz = svm_msg_q_size (mq)))
  31     {
  32       len = vec_len (wrk->mq_msg_vector);
  33       vec_validate (wrk->mq_msg_vector, len + sz - 1);
  34       svm_msg_q_sub_raw_batch (mq, wrk->mq_msg_vector + len, sz);
  35       n_msgs += sz;
  36     }
  37   return n_msgs;
  38 }
  39
  40
  41
  42 static void
  43 vcl_msg_add_ext_config (vcl_session_t *s, uword *offset)
  44 {
  45   svm_fifo_chunk_t *c;
  46
  47   c = vcl_segment_alloc_chunk (vcl_vpp_worker_segment_handle (0),
  48                                0 /* one slice only */, s->ext_config->len,
  49                                offset);
  50   if (c)
  51     clib_memcpy_fast (c->data, s->ext_config, s->ext_config->len);
  52 }
  53
  54 static void
  55 vcl_send_session_listen (vcl_worker_t * wrk, vcl_session_t * s)
  56 {
  57   app_session_evt_t _app_evt, *app_evt = &_app_evt;
  58   session_listen_msg_t *mp;
  59   svm_msg_q_t *mq;
  60
  61   mq = vcl_worker_ctrl_mq (wrk);
  62   app_alloc_ctrl_evt_to_vpp (mq, app_evt, SESSION_CTRL_EVT_LISTEN);
  63   mp = (session_listen_msg_t *) app_evt->evt->data;
  64   memset (mp, 0, sizeof (*mp));
  65   mp->client_index = wrk->api_client_handle;
  66   mp->context = s->session_index;
  67   mp->wrk_index = wrk->vpp_wrk_index;
  68   mp->is_ip4 = s->transport.is_ip4;
  69   clib_memcpy_fast (&mp->ip, &s->transport.lcl_ip, sizeof (mp->ip));
  70   mp->port = s->transport.lcl_port;
  71   mp->proto = s->session_type;
  72   mp->vrf = s->vrf;
  73   if (s->flags & VCL_SESSION_F_CONNECTED)
  74     mp->flags = TRANSPORT_CFG_F_CONNECTED;
  75   if (s->ext_config)
  76     vcl_msg_add_ext_config (s, &mp->ext_config);
  77   app_send_ctrl_evt_to_vpp (mq, app_evt);
  78   if (s->ext_config)
  79     {
  80       clib_mem_free (s->ext_config);
  81       s->ext_config = 0;
  82     }
  83 }
  84
  85 static void
  86 vcl_send_session_connect (vcl_worker_t * wrk, vcl_session_t * s)
  87 {
  88   app_session_evt_t _app_evt, *app_evt = &_app_evt;
  89   session_connect_msg_t *mp;
  90   svm_msg_q_t *mq;
  91
  92   mq = vcl_worker_ctrl_mq (wrk);
  93   app_alloc_ctrl_evt_to_vpp (mq, app_evt, SESSION_CTRL_EVT_CONNECT);
  94   mp = (session_connect_msg_t *) app_evt->evt->data;
  95   memset (mp, 0, sizeof (*mp));
  96   mp->client_index = wrk->api_client_handle;
  97   mp->context = s->session_index;
  98   mp->dscp = s->dscp;
  99   mp->wrk_index = wrk->vpp_wrk_index;
 100   mp->is_ip4 = s->transport.is_ip4;
 101   mp->parent_handle = s->parent_handle;
 102   clib_memcpy_fast (&mp->ip, &s->transport.rmt_ip, sizeof (mp->ip));
 103   clib_memcpy_fast (&mp->lcl_ip, &s->transport.lcl_ip, sizeof (mp->lcl_ip));
 104   mp->port = s->transport.rmt_port;
 105   mp->lcl_port = s->transport.lcl_port;
 106   mp->proto = s->session_type;
 107   mp->vrf = s->vrf;
 108   if (s->flags & VCL_SESSION_F_CONNECTED)
 109     mp->flags |= TRANSPORT_CFG_F_CONNECTED;
 110   if (s->ext_config)
 111     vcl_msg_add_ext_config (s, &mp->ext_config);
 112   app_send_ctrl_evt_to_vpp (mq, app_evt);
 113
 114   if (s->ext_config)
 115     {
 116       clib_mem_free (s->ext_config);
 117       s->ext_config = 0;
 118     }
 119 }
 120
 121 void
 122 vcl_send_session_unlisten (vcl_worker_t * wrk, vcl_session_t * s)
 123 {
 124   app_session_evt_t _app_evt, *app_evt = &_app_evt;
 125   session_unlisten_msg_t *mp;
 126   svm_msg_q_t *mq;
 127
 128   mq = vcl_worker_ctrl_mq (wrk);
 129   app_alloc_ctrl_evt_to_vpp (mq, app_evt, SESSION_CTRL_EVT_UNLISTEN);
 130   mp = (session_unlisten_msg_t *) app_evt->evt->data;
 131   memset (mp, 0, sizeof (*mp));
 132   mp->client_index = wrk->api_client_handle;
 133   mp->wrk_index = wrk->vpp_wrk_index;
 134   mp->handle = s->vpp_handle;
 135   mp->context = wrk->wrk_index;
 136   app_send_ctrl_evt_to_vpp (mq, app_evt);
 137 }
 138
 139 static void
 140 vcl_send_session_shutdown (vcl_worker_t *wrk, vcl_session_t *s)
 141 {
 142   app_session_evt_t _app_evt, *app_evt = &_app_evt;
 143   session_shutdown_msg_t *mp;
 144   svm_msg_q_t *mq;
 145
 146   /* Send to thread that owns the session */
 147   mq = s->vpp_evt_q;
 148   app_alloc_ctrl_evt_to_vpp (mq, app_evt, SESSION_CTRL_EVT_SHUTDOWN);
 149   mp = (session_shutdown_msg_t *) app_evt->evt->data;
 150   memset (mp, 0, sizeof (*mp));
 151   mp->client_index = wrk->api_client_handle;
 152   mp->handle = s->vpp_handle;
 153   app_send_ctrl_evt_to_vpp (mq, app_evt);
 154 }
 155
 156 static void
 157 vcl_send_session_disconnect (vcl_worker_t * wrk, vcl_session_t * s)
 158 {
 159   app_session_evt_t _app_evt, *app_evt = &_app_evt;
 160   session_disconnect_msg_t *mp;
 161   svm_msg_q_t *mq;
 162
 163   /* Send to thread that owns the session */
 164   mq = s->vpp_evt_q;
 165   app_alloc_ctrl_evt_to_vpp (mq, app_evt, SESSION_CTRL_EVT_DISCONNECT);
 166   mp = (session_disconnect_msg_t *) app_evt->evt->data;
 167   memset (mp, 0, sizeof (*mp));
 168   mp->client_index = wrk->api_client_handle;
 169   mp->handle = s->vpp_handle;
 170   app_send_ctrl_evt_to_vpp (mq, app_evt);
 171 }
 172
 173 static void
 174 vcl_send_app_detach (vcl_worker_t * wrk)
 175 {
 176   app_session_evt_t _app_evt, *app_evt = &_app_evt;
 177   session_app_detach_msg_t *mp;
 178   svm_msg_q_t *mq;
 179
 180   mq = vcl_worker_ctrl_mq (wrk);
 181   app_alloc_ctrl_evt_to_vpp (mq, app_evt, SESSION_CTRL_EVT_APP_DETACH);
 182   mp = (session_app_detach_msg_t *) app_evt->evt->data;
 183   memset (mp, 0, sizeof (*mp));
 184   mp->client_index = wrk->api_client_handle;
 185   app_send_ctrl_evt_to_vpp (mq, app_evt);
 186 }
 187
 188 static void
 189 vcl_send_session_accepted_reply (svm_msg_q_t * mq, u32 context,
 190                                  session_handle_t handle, int retval)
 191 {
 192   app_session_evt_t _app_evt, *app_evt = &_app_evt;
 193   session_accepted_reply_msg_t *rmp;
 194   app_alloc_ctrl_evt_to_vpp (mq, app_evt, SESSION_CTRL_EVT_ACCEPTED_REPLY);
 195   rmp = (session_accepted_reply_msg_t *) app_evt->evt->data;
 196   rmp->handle = handle;
 197   rmp->context = context;
 198   rmp->retval = retval;
 199   app_send_ctrl_evt_to_vpp (mq, app_evt);
 200 }
 201
 202 static void
 203 vcl_send_session_disconnected_reply (vcl_worker_t * wrk, vcl_session_t * s,
 204                                      int retval)
 205 {
 206   app_session_evt_t _app_evt, *app_evt = &_app_evt;
 207   session_disconnected_reply_msg_t *rmp;
 208   app_alloc_ctrl_evt_to_vpp (s->vpp_evt_q, app_evt,
 209                              SESSION_CTRL_EVT_DISCONNECTED_REPLY);
 210   rmp = (session_disconnected_reply_msg_t *) app_evt->evt->data;
 211   rmp->handle = s->vpp_handle;
 212   rmp->context = wrk->api_client_handle;
 213   rmp->retval = retval;
 214   app_send_ctrl_evt_to_vpp (s->vpp_evt_q, app_evt);
 215 }
 216
 217 static void
 218 vcl_send_session_reset_reply (vcl_worker_t * wrk, vcl_session_t * s,
 219                               int retval)
 220 {
 221   app_session_evt_t _app_evt, *app_evt = &_app_evt;
 222   session_reset_reply_msg_t *rmp;
 223   app_alloc_ctrl_evt_to_vpp (s->vpp_evt_q, app_evt,
 224                              SESSION_CTRL_EVT_RESET_REPLY);
 225   rmp = (session_reset_reply_msg_t *) app_evt->evt->data;
 226   rmp->handle = s->vpp_handle;
 227   rmp->context = wrk->api_client_handle;
 228   rmp->retval = retval;
 229   app_send_ctrl_evt_to_vpp (s->vpp_evt_q, app_evt);
 230 }
 231
 232 void
 233 vcl_send_session_worker_update (vcl_worker_t * wrk, vcl_session_t * s,
 234                                 u32 wrk_index)
 235 {
 236   app_session_evt_t _app_evt, *app_evt = &_app_evt;
 237   session_worker_update_msg_t *mp;
 238
 239   app_alloc_ctrl_evt_to_vpp (s->vpp_evt_q, app_evt,
 240                              SESSION_CTRL_EVT_WORKER_UPDATE);
 241   mp = (session_worker_update_msg_t *) app_evt->evt->data;
 242   mp->client_index = wrk->api_client_handle;
 243   mp->handle = s->vpp_handle;
 244   mp->req_wrk_index = wrk->vpp_wrk_index;
 245   mp->wrk_index = wrk_index;
 246   app_send_ctrl_evt_to_vpp (s->vpp_evt_q, app_evt);
 247 }
 248
 249 int
 250 vcl_send_worker_rpc (u32 dst_wrk_index, void *data, u32 data_len)
 251 {
 252   app_session_evt_t _app_evt, *app_evt = &_app_evt;
 253   session_app_wrk_rpc_msg_t *mp;
 254   vcl_worker_t *dst_wrk, *wrk;
 255   svm_msg_q_t *mq;
 256   int ret = -1;
 257
 258   if (data_len > sizeof (mp->data))
 259     goto done;
 260
 261   clib_spinlock_lock (&vcm->workers_lock);
 262
 263   dst_wrk = vcl_worker_get_if_valid (dst_wrk_index);
 264   if (!dst_wrk)
 265     goto done;
 266
 267   wrk = vcl_worker_get_current ();
 268   mq = vcl_worker_ctrl_mq (wrk);
 269   app_alloc_ctrl_evt_to_vpp (mq, app_evt, SESSION_CTRL_EVT_APP_WRK_RPC);
 270   mp = (session_app_wrk_rpc_msg_t *) app_evt->evt->data;
 271   mp->client_index = wrk->api_client_handle;
 272   mp->wrk_index = dst_wrk->vpp_wrk_index;
 273   clib_memcpy (mp->data, data, data_len);
 274   app_send_ctrl_evt_to_vpp (mq, app_evt);
 275   ret = 0;
 276
 277 done:
 278   clib_spinlock_unlock (&vcm->workers_lock);
 279   return ret;
 280 }
 281
 282 int
 283 vcl_session_transport_attr (vcl_worker_t *wrk, vcl_session_t *s, u8 is_get,
 284                             transport_endpt_attr_t *attr)
 285 {
 286   app_session_evt_t _app_evt, *app_evt = &_app_evt;
 287   session_transport_attr_msg_t *mp;
 288   svm_msg_q_t *mq;
 289   f64 timeout;
 290
 291   ASSERT (!wrk->session_attr_op);
 292   mq = s->vpp_evt_q;
 293   if (PREDICT_FALSE (!mq))
 294     {
 295       /* FIXME: attribute should be stored and sent once session is
 296        * bound/connected to vpp */
 297       return 0;
 298     }
 299
 300   wrk->session_attr_op = 1;
 301   wrk->session_attr_op_rv = -1;
 302
 303   app_alloc_ctrl_evt_to_vpp (mq, app_evt, SESSION_CTRL_EVT_TRANSPORT_ATTR);
 304   mp = (session_transport_attr_msg_t *) app_evt->evt->data;
 305   memset (mp, 0, sizeof (*mp));
 306   mp->client_index = wrk->api_client_handle;
 307   mp->handle = s->vpp_handle;
 308   mp->is_get = is_get;
 309   mp->attr = *attr;
 310   app_send_ctrl_evt_to_vpp (mq, app_evt);
 311
 312   timeout = clib_time_now (&wrk->clib_time) + 1;
 313
 314   while (wrk->session_attr_op && clib_time_now (&wrk->clib_time) < timeout)
 315     vcl_flush_mq_events ();
 316
 317   if (!wrk->session_attr_op_rv && is_get)
 318     *attr = wrk->session_attr_rv;
 319
 320   wrk->session_attr_op = 0;
 321
 322   return wrk->session_attr_op_rv;
 323 }
 324
 325 static u32
 326 vcl_session_accepted_handler (vcl_worker_t * wrk, session_accepted_msg_t * mp,
 327                               u32 ls_index)
 328 {
 329   vcl_session_t *session, *listen_session;
 330   svm_msg_q_t *evt_q;
 331
 332   session = vcl_session_alloc (wrk);
 333
 334   listen_session = vcl_session_get (wrk, ls_index);
 335   if (listen_session->vpp_handle != mp->listener_handle)
 336     {
 337       VDBG (0, "ERROR: listener handle %lu does not match session %u",
 338             mp->listener_handle, ls_index);
 339       goto error;
 340     }
 341
 342   if (vcl_segment_attach_session (
 343         mp->segment_handle, mp->server_rx_fifo, mp->server_tx_fifo,
 344         mp->vpp_event_queue_address, mp->mq_index, 0, session))
 345     {
 346       VDBG (0, "session %u [0x%llx]: failed to attach fifos",
 347             session->session_index, mp->handle);
 348       goto error;
 349     }
 350
 351   session->vpp_handle = mp->handle;
 352   session->session_state = VCL_STATE_READY;
 353   session->transport.rmt_port = mp->rmt.port;
 354   session->transport.is_ip4 = mp->rmt.is_ip4;
 355   clib_memcpy_fast (&session->transport.rmt_ip, &mp->rmt.ip,
 356                     sizeof (ip46_address_t));
 357
 358   vcl_session_table_add_vpp_handle (wrk, mp->handle, session->session_index);
 359   session->transport.lcl_port = mp->lcl.port;
 360   session->transport.lcl_ip = mp->lcl.ip;
 361   session->session_type = listen_session->session_type;
 362   session->is_dgram = vcl_proto_is_dgram (session->session_type);
 363   session->listener_index = listen_session->session_index;
 364   listen_session->n_accepted_sessions++;
 365
 366   vcl_evt (VCL_EVT_ACCEPT, session, listen_session, session_index);
 367
 368   vcl_send_session_accepted_reply (session->vpp_evt_q, mp->context,
 369                                    session->vpp_handle, 0);
 370
 371   return session->session_index;
 372
 373 error:
 374   vcl_segment_attach_mq (vcl_vpp_worker_segment_handle (0),
 375                          mp->vpp_event_queue_address, mp->mq_index, &evt_q);
 376   vcl_send_session_accepted_reply (evt_q, mp->context, mp->handle,
 377                                    VNET_API_ERROR_INVALID_ARGUMENT);
 378   vcl_session_free (wrk, session);
 379   return VCL_INVALID_SESSION_INDEX;
 380 }
 381
 382 static u32
 383 vcl_session_connected_handler (vcl_worker_t * wrk,
 384                                session_connected_msg_t * mp)
 385 {
 386   vcl_session_t *session = 0;
 387   u32 session_index;
 388
 389   session_index = mp->context;
 390   session = vcl_session_get (wrk, session_index);
 391   if (PREDICT_FALSE (!session))
 392     {
 393       VERR ("vpp handle 0x%llx has no session index (%u)!", mp->handle,
 394             session_index);
 395       /* Should not happen but if it does, force vpp session cleanup */
 396       vcl_session_t tmp_session = {
 397         .vpp_handle = mp->handle,
 398         .vpp_evt_q = 0,
 399       };
 400       vcl_segment_attach_session (
 401         mp->segment_handle, mp->server_rx_fifo, mp->server_tx_fifo,
 402         mp->vpp_event_queue_address, mp->mq_index, 0, session);
 403       if (tmp_session.vpp_evt_q)
 404         vcl_send_session_disconnect (wrk, &tmp_session);
 405       return VCL_INVALID_SESSION_INDEX;
 406     }
 407
 408   if (mp->retval)
 409     {
 410       VDBG (0, "session %u: connect failed! %U", session_index,
 411             format_session_error, mp->retval);
 412       session->session_state = VCL_STATE_DETACHED;
 413       session->vpp_handle = VCL_INVALID_SESSION_HANDLE;
 414       return session_index;
 415     }
 416
 417   session->vpp_handle = mp->handle;
 418
 419   /* Add to lookup table. Even if something fails, session cannot be
 420    * cleaned up prior to notifying vpp and going through the cleanup
 421    * "procedure" see @ref vcl_session_cleanup_handler */
 422   vcl_session_table_add_vpp_handle (wrk, mp->handle, session_index);
 423
 424   if (vcl_segment_attach_session (
 425         mp->segment_handle, mp->server_rx_fifo, mp->server_tx_fifo,
 426         mp->vpp_event_queue_address, mp->mq_index, 0, session))
 427     {
 428       VDBG (0, "session %u [0x%llx]: failed to attach fifos",
 429             session->session_index, session->vpp_handle);
 430       session->session_state = VCL_STATE_UPDATED;
 431       vcl_send_session_disconnect (wrk, session);
 432       return session_index;
 433     }
 434
 435   if (mp->ct_rx_fifo)
 436     {
 437       if (vcl_segment_attach_session (mp->ct_segment_handle, mp->ct_rx_fifo,
 438                                       mp->ct_tx_fifo, (uword) ~0, ~0, 1,
 439                                       session))
 440         {
 441           VDBG (0, "session %u [0x%llx]: failed to attach ct fifos",
 442                 session->session_index, session->vpp_handle);
 443           session->session_state = VCL_STATE_UPDATED;
 444           vcl_send_session_disconnect (wrk, session);
 445           return session_index;
 446         }
 447     }
 448
 449   session->transport.is_ip4 = mp->lcl.is_ip4;
 450   clib_memcpy_fast (&session->transport.lcl_ip, &mp->lcl.ip,
 451                     sizeof (session->transport.lcl_ip));
 452   session->transport.lcl_port = mp->lcl.port;
 453
 454   /* Application closed session before connect reply */
 455   if (vcl_session_has_attr (session, VCL_SESS_ATTR_NONBLOCK)
 456       && session->session_state == VCL_STATE_CLOSED)
 457     vcl_send_session_disconnect (wrk, session);
 458   else
 459     session->session_state = VCL_STATE_READY;
 460
 461   VDBG (0, "session %u [0x%llx] connected local: %U:%u remote %U:%u",
 462         session->session_index, session->vpp_handle, vcl_format_ip46_address,
 463         &session->transport.lcl_ip,
 464         session->transport.is_ip4 ? IP46_TYPE_IP4 : IP46_TYPE_IP6,
 465         clib_net_to_host_u16 (session->transport.lcl_port),
 466         vcl_format_ip46_address, &session->transport.rmt_ip,
 467         session->transport.is_ip4 ? IP46_TYPE_IP4 : IP46_TYPE_IP6,
 468         clib_net_to_host_u16 (session->transport.rmt_port));
 469
 470   return session_index;
 471 }
 472
 473 static int
 474 vcl_flag_accepted_session (vcl_session_t * session, u64 handle, u32 flags)
 475 {
 476   vcl_session_msg_t *accepted_msg;
 477   int i;
 478
 479   for (i = 0; i < vec_len (session->accept_evts_fifo); i++)
 480     {
 481       accepted_msg = &session->accept_evts_fifo[i];
 482       if (accepted_msg->accepted_msg.handle == handle)
 483         {
 484           accepted_msg->flags |= flags;
 485           return 1;
 486         }
 487     }
 488   return 0;
 489 }
 490
 491 static u32
 492 vcl_session_reset_handler (vcl_worker_t * wrk,
 493                            session_reset_msg_t * reset_msg)
 494 {
 495   vcl_session_t *session;
 496   u32 sid;
 497
 498   sid = vcl_session_index_from_vpp_handle (wrk, reset_msg->handle);
 499   session = vcl_session_get (wrk, sid);
 500   if (!session)
 501     {
 502       VDBG (0, "request to reset unknown handle 0x%llx", reset_msg->handle);
 503       return VCL_INVALID_SESSION_INDEX;
 504     }
 505
 506   /* Caught a reset before actually accepting the session */
 507   if (session->session_state == VCL_STATE_LISTEN)
 508     {
 509
 510       if (!vcl_flag_accepted_session (session, reset_msg->handle,
 511                                       VCL_ACCEPTED_F_RESET))
 512         VDBG (0, "session was not accepted!");
 513       return VCL_INVALID_SESSION_INDEX;
 514     }
 515
 516   if (session->session_state != VCL_STATE_CLOSED)
 517     session->session_state = VCL_STATE_DISCONNECT;
 518   VDBG (0, "session %u [0x%llx]: reset", sid, reset_msg->handle);
 519   return sid;
 520 }
 521
 522 static u32
 523 vcl_session_bound_handler (vcl_worker_t * wrk, session_bound_msg_t * mp)
 524 {
 525   vcl_session_t *session;
 526   u32 sid = mp->context;
 527
 528   session = vcl_session_get (wrk, sid);
 529   if (mp->retval)
 530     {
 531       VERR ("session %u [0x%llx]: bind failed: %U", sid, mp->handle,
 532             format_session_error, mp->retval);
 533       if (session)
 534         {
 535           session->session_state = VCL_STATE_DETACHED;
 536           session->vpp_handle = mp->handle;
 537           return sid;
 538         }
 539       else
 540         {
 541           VDBG (0, "session %u [0x%llx]: Invalid session index!", sid,
 542                 mp->handle);
 543           return VCL_INVALID_SESSION_INDEX;
 544         }
 545     }
 546
 547   session->vpp_handle = mp->handle;
 548   session->transport.is_ip4 = mp->lcl_is_ip4;
 549   clib_memcpy_fast (&session->transport.lcl_ip, mp->lcl_ip,
 550                     sizeof (ip46_address_t));
 551   session->transport.lcl_port = mp->lcl_port;
 552   vcl_session_table_add_listener (wrk, mp->handle, sid);
 553   session->session_state = VCL_STATE_LISTEN;
 554
 555   if (vcl_session_is_cl (session))
 556     {
 557       if (vcl_segment_attach_session (mp->segment_handle, mp->rx_fifo,
 558                                       mp->tx_fifo, mp->vpp_evt_q, mp->mq_index,
 559                                       0, session))
 560         {
 561           VDBG (0, "session %u [0x%llx]: failed to attach fifos",
 562                 session->session_index, session->vpp_handle);
 563           session->session_state = VCL_STATE_DETACHED;
 564           return VCL_INVALID_SESSION_INDEX;
 565         }
 566     }
 567
 568   VDBG (0, "session %u [0x%llx]: listen succeeded!", sid, mp->handle);
 569   return sid;
 570 }
 571
 572 static void
 573 vcl_session_unlisten_reply_handler (vcl_worker_t * wrk, void *data)
 574 {
 575   session_unlisten_reply_msg_t *mp = (session_unlisten_reply_msg_t *) data;
 576   vcl_session_t *s;
 577
 578   s = vcl_session_get_w_vpp_handle (wrk, mp->handle);
 579   if (!s)
 580     {
 581       VDBG (0, "Unlisten reply with wrong handle %llx", mp->handle);
 582       return;
 583     }
 584   if (s->session_state != VCL_STATE_DISCONNECT)
 585     {
 586       /* Connected udp listener */
 587       if (s->session_type == VPPCOM_PROTO_UDP
 588           && s->session_state == VCL_STATE_CLOSED)
 589         return;
 590
 591       VDBG (0, "Unlisten session in wrong state %llx", mp->handle);
 592       return;
 593     }
 594
 595   if (mp->retval)
 596     VDBG (0, "ERROR: session %u [0xllx]: unlisten failed: %U",
 597           s->session_index, mp->handle, format_session_error, mp->retval);
 598
 599   if (mp->context != wrk->wrk_index)
 600     VDBG (0, "wrong context");
 601
 602   vcl_session_table_del_vpp_handle (wrk, mp->handle);
 603   vcl_session_free (wrk, s);
 604 }
 605
 606 static void
 607 vcl_session_migrated_handler (vcl_worker_t * wrk, void *data)
 608 {
 609   session_migrated_msg_t *mp = (session_migrated_msg_t *) data;
 610   vcl_session_t *s;
 611   u32 fs_index;
 612
 613   s = vcl_session_get_w_vpp_handle (wrk, mp->handle);
 614   if (!s)
 615     {
 616       VDBG (0, "Migrated notification with wrong handle %llx", mp->handle);
 617       return;
 618     }
 619
 620   /* Only validate if a value is provided */
 621   if (mp->segment_handle != SESSION_INVALID_HANDLE)
 622     {
 623       fs_index = vcl_segment_table_lookup (mp->segment_handle);
 624       if (fs_index == VCL_INVALID_SEGMENT_INDEX)
 625         {
 626           VDBG (0, "segment %lx for session %u is not mounted!",
 627                 mp->segment_handle, s->session_index);
 628           s->session_state = VCL_STATE_DETACHED;
 629           return;
 630         }
 631     }
 632
 633   s->vpp_handle = mp->new_handle;
 634
 635   vcl_segment_attach_mq (vcl_vpp_worker_segment_handle (0), mp->vpp_evt_q,
 636                          mp->vpp_thread_index, &s->vpp_evt_q);
 637
 638   vcl_session_table_del_vpp_handle (wrk, mp->handle);
 639   vcl_session_table_add_vpp_handle (wrk, mp->new_handle, s->session_index);
 640
 641   /* Generate new tx event if we have outstanding data */
 642   if (svm_fifo_has_event (s->tx_fifo))
 643     app_send_io_evt_to_vpp (s->vpp_evt_q,
 644                             s->tx_fifo->shr->master_session_index,
 645                             SESSION_IO_EVT_TX, SVM_Q_WAIT);
 646
 647   VDBG (0, "Migrated 0x%lx to thread %u 0x%lx", mp->handle,
 648         mp->vpp_thread_index, mp->new_handle);
 649 }
 650
 651 static vcl_session_t *
 652 vcl_session_accepted (vcl_worker_t * wrk, session_accepted_msg_t * msg)
 653 {
 654   vcl_session_msg_t *vcl_msg;
 655   vcl_session_t *session;
 656
 657   session = vcl_session_get_w_vpp_handle (wrk, msg->handle);
 658   if (PREDICT_FALSE (session != 0))
 659     VWRN ("session overlap handle %lu state %u!", msg->handle,
 660           session->session_state);
 661
 662   session = vcl_session_table_lookup_listener (wrk, msg->listener_handle);
 663   if (!session)
 664     {
 665       VERR ("couldn't find listen session: listener handle %llx",
 666             msg->listener_handle);
 667       return 0;
 668     }
 669
 670   clib_fifo_add2 (session->accept_evts_fifo, vcl_msg);
 671   vcl_msg->flags = 0;
 672   vcl_msg->accepted_msg = *msg;
 673   /* Session handle points to listener until fully accepted by app */
 674   vcl_session_table_add_vpp_handle (wrk, msg->handle, session->session_index);
 675
 676   return session;
 677 }
 678
 679 static vcl_session_t *
 680 vcl_session_disconnected_handler (vcl_worker_t * wrk,
 681                                   session_disconnected_msg_t * msg)
 682 {
 683   vcl_session_t *session;
 684
 685   session = vcl_session_get_w_vpp_handle (wrk, msg->handle);
 686   if (!session)
 687     {
 688       VWRN ("request to disconnect unknown handle 0x%llx", msg->handle);
 689       return 0;
 690     }
 691
 692   /* Late disconnect notification on a session that has been closed */
 693   if (session->session_state == VCL_STATE_CLOSED)
 694     return 0;
 695
 696   /* Caught a disconnect before actually accepting the session */
 697   if (session->session_state == VCL_STATE_LISTEN)
 698     {
 699       if (!vcl_flag_accepted_session (session, msg->handle,
 700                                       VCL_ACCEPTED_F_CLOSED))
 701         VDBG (0, "session was not accepted!");
 702       return 0;
 703     }
 704
 705   /* If not already reset change state */
 706   if (session->session_state != VCL_STATE_DISCONNECT)
 707     session->session_state = VCL_STATE_VPP_CLOSING;
 708
 709   return session;
 710 }
 711
 712 int
 713 vppcom_session_shutdown (uint32_t session_handle, int how)
 714 {
 715   vcl_worker_t *wrk = vcl_worker_get_current ();
 716   vcl_session_t *session;
 717   vcl_session_state_t state;
 718   u64 vpp_handle;
 719
 720   session = vcl_session_get_w_handle (wrk, session_handle);
 721   if (PREDICT_FALSE (!session))
 722     return VPPCOM_EBADFD;
 723
 724   vpp_handle = session->vpp_handle;
 725   state = session->session_state;
 726
 727   VDBG (1, "session %u [0x%llx] state 0x%x (%s)", session->session_index,
 728         vpp_handle, state, vcl_session_state_str (state));
 729
 730   if (PREDICT_FALSE (state == VCL_STATE_LISTEN))
 731     {
 732       VDBG (0, "ERROR: Cannot shutdown a listen socket!");
 733       return VPPCOM_EBADFD;
 734     }
 735
 736   if (how == SHUT_RD || how == SHUT_RDWR)
 737     {
 738       session->flags |= VCL_SESSION_F_RD_SHUTDOWN;
 739       if (how == SHUT_RD)
 740         return VPPCOM_OK;
 741     }
 742   session->flags |= VCL_SESSION_F_WR_SHUTDOWN;
 743
 744   if (PREDICT_TRUE (state == VCL_STATE_READY))
 745     {
 746       VDBG (1, "session %u [0x%llx]: sending shutdown...",
 747             session->session_index, vpp_handle);
 748
 749       vcl_send_session_shutdown (wrk, session);
 750     }
 751
 752   return VPPCOM_OK;
 753 }
 754
 755 static int
 756 vppcom_session_disconnect (u32 session_handle)
 757 {
 758   vcl_worker_t *wrk = vcl_worker_get_current ();
 759   vcl_session_t *session, *listen_session;
 760   vcl_session_state_t state;
 761   u64 vpp_handle;
 762
 763   session = vcl_session_get_w_handle (wrk, session_handle);
 764   if (!session)
 765     return VPPCOM_EBADFD;
 766
 767   vpp_handle = session->vpp_handle;
 768   state = session->session_state;
 769
 770   VDBG (1, "session %u [0x%llx]: disconnecting state (%s)",
 771         session->session_index, vpp_handle, vcl_session_state_str (state));
 772
 773   if (PREDICT_FALSE (state == VCL_STATE_LISTEN))
 774     {
 775       VDBG (0, "ERROR: Cannot disconnect a listen socket!");
 776       return VPPCOM_EBADFD;
 777     }
 778
 779   if (state == VCL_STATE_VPP_CLOSING)
 780     {
 781       vcl_send_session_disconnected_reply (wrk, session, 0);
 782       VDBG (1, "session %u [0x%llx]: sending disconnect REPLY...",
 783             session->session_index, vpp_handle);
 784     }
 785   else
 786     {
 787       /* Session doesn't have an event queue yet. Probably a non-blocking
 788        * connect. Wait for the reply */
 789       if (PREDICT_FALSE (!session->vpp_evt_q))
 790         return VPPCOM_OK;
 791
 792       VDBG (1, "session %u [0x%llx]: sending disconnect",
 793             session->session_index, vpp_handle);
 794       vcl_send_session_disconnect (wrk, session);
 795     }
 796
 797   if (session->listener_index != VCL_INVALID_SESSION_INDEX)
 798     {
 799       listen_session = vcl_session_get (wrk, session->listener_index);
 800       if (listen_session)
 801         listen_session->n_accepted_sessions--;
 802     }
 803
 804   return VPPCOM_OK;
 805 }
 806
 807 static void
 808 vcl_session_cleanup_handler (vcl_worker_t * wrk, void *data)
 809 {
 810   session_cleanup_msg_t *msg;
 811   vcl_session_t *session;
 812
 813   msg = (session_cleanup_msg_t *) data;
 814   session = vcl_session_get_w_vpp_handle (wrk, msg->handle);
 815   if (!session)
 816     {
 817       VWRN ("disconnect confirmed for unknown handle 0x%llx", msg->handle);
 818       return;
 819     }
 820
 821   if (msg->type == SESSION_CLEANUP_TRANSPORT)
 822     {
 823       /* Transport was cleaned up before we confirmed close. Probably the
 824        * app is still waiting for some data that cannot be delivered.
 825        * Confirm close to make sure everything is cleaned up.
 826        * Move to undetermined state to ensure that the session is not
 827        * removed before both vpp and the app cleanup.
 828        * - If the app closes first, the session is moved to CLOSED state
 829        *   and the session cleanup notification from vpp removes the
 830        *   session.
 831        * - If vpp cleans up the session first, the session is moved to
 832        *   DETACHED state lower and subsequently the close from the app
 833        *   frees the session
 834        */
 835       if (session->session_state == VCL_STATE_VPP_CLOSING)
 836         {
 837           vppcom_session_disconnect (vcl_session_handle (session));
 838           session->session_state = VCL_STATE_UPDATED;
 839         }
 840       else if (session->session_state == VCL_STATE_DISCONNECT)
 841         {
 842           vcl_send_session_reset_reply (wrk, session, 0);
 843           session->session_state = VCL_STATE_UPDATED;
 844         }
 845       return;
 846     }
 847
 848   /* VPP will reuse the handle so clean it up now */
 849   vcl_session_table_del_vpp_handle (wrk, msg->handle);
 850
 851   /* App did not close the connection yet so don't free it. */
 852   if (session->session_state != VCL_STATE_CLOSED)
 853     {
 854       VDBG (0, "session %u: app did not close", session->session_index);
 855       session->session_state = VCL_STATE_DETACHED;
 856       session->vpp_handle = VCL_INVALID_SESSION_HANDLE;
 857       return;
 858     }
 859
 860   /* Session probably tracked with epoll, disconnect not yet handled and
 861    * 1) both transport and session cleanup completed 2) app closed. Wait
 862    * until message is drained to free the session.
 863    * See @ref vcl_handle_mq_event */
 864   if (session->flags & VCL_SESSION_F_PENDING_DISCONNECT)
 865     {
 866       session->flags |= VCL_SESSION_F_PENDING_FREE;
 867       return;
 868     }
 869
 870   vcl_session_free (wrk, session);
 871 }
 872
 873 static void
 874 vcl_session_req_worker_update_handler (vcl_worker_t * wrk, void *data)
 875 {
 876   session_req_worker_update_msg_t *msg;
 877   vcl_session_t *s;
 878
 879   msg = (session_req_worker_update_msg_t *) data;
 880   s = vcl_session_get_w_vpp_handle (wrk, msg->session_handle);
 881   if (!s)
 882     return;
 883
 884   vec_add1 (wrk->pending_session_wrk_updates, s->session_index);
 885 }
 886
 887 static void
 888 vcl_session_worker_update_reply_handler (vcl_worker_t * wrk, void *data)
 889 {
 890   session_worker_update_reply_msg_t *msg;
 891   vcl_session_t *s;
 892
 893   msg = (session_worker_update_reply_msg_t *) data;
 894   s = vcl_session_get_w_vpp_handle (wrk, msg->handle);
 895   if (!s)
 896     {
 897       VDBG (0, "unknown handle 0x%llx", msg->handle);
 898       return;
 899     }
 900
 901   if (s->rx_fifo)
 902     {
 903       if (vcl_segment_attach_session (msg->segment_handle, msg->rx_fifo,
 904                                       msg->tx_fifo, (uword) ~0, ~0, 0, s))
 905         {
 906           VDBG (0, "failed to attach fifos for %u", s->session_index);
 907           return;
 908         }
 909     }
 910   s->session_state = VCL_STATE_UPDATED;
 911
 912   VDBG (0, "session %u[0x%llx] moved to worker %u", s->session_index,
 913         s->vpp_handle, wrk->wrk_index);
 914 }
 915
 916 static int
 917 vcl_api_recv_fd (vcl_worker_t * wrk, int *fds, int n_fds)
 918 {
 919
 920   if (vcm->cfg.vpp_app_socket_api)
 921     return vcl_sapi_recv_fds (wrk, fds, n_fds);
 922
 923   return vcl_bapi_recv_fds (wrk, fds, n_fds);
 924 }
 925
 926 static void
 927 vcl_session_app_add_segment_handler (vcl_worker_t * wrk, void *data)
 928 {
 929   ssvm_segment_type_t seg_type = SSVM_SEGMENT_SHM;
 930   session_app_add_segment_msg_t *msg;
 931   u64 segment_handle;
 932   int fd = -1;
 933
 934   msg = (session_app_add_segment_msg_t *) data;
 935
 936   if (msg->fd_flags)
 937     {
 938       vcl_api_recv_fd (wrk, &fd, 1);
 939       seg_type = SSVM_SEGMENT_MEMFD;
 940     }
 941
 942   segment_handle = msg->segment_handle;
 943   if (segment_handle == VCL_INVALID_SEGMENT_HANDLE)
 944     {
 945       clib_warning ("invalid segment handle");
 946       return;
 947     }
 948
 949   if (vcl_segment_attach (segment_handle, (char *) msg->segment_name,
 950                           seg_type, fd))
 951     {
 952       VDBG (0, "vcl_segment_attach ('%s') failed", msg->segment_name);
 953       return;
 954     }
 955
 956   VDBG (1, "mapped new segment '%s' size %d", msg->segment_name,
 957         msg->segment_size);
 958 }
 959
 960 static void
 961 vcl_session_app_del_segment_handler (vcl_worker_t * wrk, void *data)
 962 {
 963   session_app_del_segment_msg_t *msg = (session_app_del_segment_msg_t *) data;
 964   vcl_segment_detach (msg->segment_handle);
 965   VDBG (1, "Unmapped segment: %d", msg->segment_handle);
 966 }
 967
 968 static void
 969 vcl_worker_rpc_handler (vcl_worker_t * wrk, void *data)
 970 {
 971   if (!vcm->wrk_rpc_fn)
 972     return;
 973
 974   (vcm->wrk_rpc_fn) (((session_app_wrk_rpc_msg_t *) data)->data);
 975 }
 976
 977 static void
 978 vcl_session_transport_attr_reply_handler (vcl_worker_t *wrk, void *data)
 979 {
 980   session_transport_attr_reply_msg_t *mp;
 981
 982   if (!wrk->session_attr_op)
 983     return;
 984
 985   mp = (session_transport_attr_reply_msg_t *) data;
 986
 987   wrk->session_attr_op_rv = mp->retval;
 988   wrk->session_attr_op = 0;
 989   wrk->session_attr_rv = mp->attr;
 990 }
 991
 992 static int
 993 vcl_handle_mq_event (vcl_worker_t * wrk, session_event_t * e)
 994 {
 995   session_disconnected_msg_t *disconnected_msg;
 996   session_connected_msg_t *connected_msg;
 997   session_reset_msg_t *reset_msg;
 998   session_event_t *ecpy;
 999   vcl_session_t *s;
1000   u32 sid;
1001
1002   switch (e->event_type)
1003     {
1004     case SESSION_IO_EVT_RX:
1005     case SESSION_IO_EVT_TX:
1006       s = vcl_session_get (wrk, e->session_index);
1007       if (!s || !vcl_session_is_open (s))
1008         break;
1009       vec_add1 (wrk->unhandled_evts_vector, *e);
1010       break;
1011     case SESSION_CTRL_EVT_BOUND:
1012       /* We can only wait for only one listen so not postponed */
1013       vcl_session_bound_handler (wrk, (session_bound_msg_t *) e->data);
1014       break;
1015     case SESSION_CTRL_EVT_ACCEPTED:
1016       s = vcl_session_accepted (wrk, (session_accepted_msg_t *) e->data);
1017       if (vcl_session_has_attr (s, VCL_SESS_ATTR_NONBLOCK))
1018         {
1019           vec_add2 (wrk->unhandled_evts_vector, ecpy, 1);
1020           *ecpy = *e;
1021           ecpy->postponed = 1;
1022           ecpy->session_index = s->session_index;
1023         }
1024       break;
1025     case SESSION_CTRL_EVT_CONNECTED:
1026       connected_msg = (session_connected_msg_t *) e->data;
1027       sid = vcl_session_connected_handler (wrk, connected_msg);
1028       if (!(s = vcl_session_get (wrk, sid)))
1029         break;
1030       if (vcl_session_has_attr (s, VCL_SESS_ATTR_NONBLOCK))
1031         {
1032           vec_add2 (wrk->unhandled_evts_vector, ecpy, 1);
1033           *ecpy = *e;
1034           ecpy->postponed = 1;
1035           ecpy->session_index = s->session_index;
1036         }
1037       break;
1038     case SESSION_CTRL_EVT_DISCONNECTED:
1039       disconnected_msg = (session_disconnected_msg_t *) e->data;
1040       if (!(s = vcl_session_get_w_vpp_handle (wrk, disconnected_msg->handle)))
1041         break;
1042       if (s->session_state == VCL_STATE_CLOSED)
1043         break;
1044       if (vcl_session_has_attr (s, VCL_SESS_ATTR_NONBLOCK))
1045         {
1046           s->session_state = VCL_STATE_VPP_CLOSING;
1047           s->flags |= VCL_SESSION_F_PENDING_DISCONNECT;
1048           vec_add2 (wrk->unhandled_evts_vector, ecpy, 1);
1049           *ecpy = *e;
1050           ecpy->postponed = 1;
1051           ecpy->session_index = s->session_index;
1052           break;
1053         }
1054       if (!(s = vcl_session_disconnected_handler (wrk, disconnected_msg)))
1055         break;
1056       VDBG (0, "disconnected session %u [0x%llx]", s->session_index,
1057             s->vpp_handle);
1058       break;
1059     case SESSION_CTRL_EVT_RESET:
1060       reset_msg = (session_reset_msg_t *) e->data;
1061       if (!(s = vcl_session_get_w_vpp_handle (wrk, reset_msg->handle)))
1062         break;
1063       if (s->session_state == VCL_STATE_CLOSED)
1064         break;
1065       if (vcl_session_has_attr (s, VCL_SESS_ATTR_NONBLOCK))
1066         {
1067           s->flags |= VCL_SESSION_F_PENDING_DISCONNECT;
1068           s->session_state = VCL_STATE_DISCONNECT;
1069           vec_add2 (wrk->unhandled_evts_vector, ecpy, 1);
1070           *ecpy = *e;
1071           ecpy->postponed = 1;
1072           ecpy->session_index = s->session_index;
1073           break;
1074         }
1075       vcl_session_reset_handler (wrk, (session_reset_msg_t *) e->data);
1076       break;
1077     case SESSION_CTRL_EVT_UNLISTEN_REPLY:
1078       vcl_session_unlisten_reply_handler (wrk, e->data);
1079       break;
1080     case SESSION_CTRL_EVT_MIGRATED:
1081       vcl_session_migrated_handler (wrk, e->data);
1082       break;
1083     case SESSION_CTRL_EVT_CLEANUP:
1084       vcl_session_cleanup_handler (wrk, e->data);
1085       break;
1086     case SESSION_CTRL_EVT_REQ_WORKER_UPDATE:
1087       vcl_session_req_worker_update_handler (wrk, e->data);
1088       break;
1089     case SESSION_CTRL_EVT_WORKER_UPDATE_REPLY:
1090       vcl_session_worker_update_reply_handler (wrk, e->data);
1091       break;
1092     case SESSION_CTRL_EVT_APP_ADD_SEGMENT:
1093       vcl_session_app_add_segment_handler (wrk, e->data);
1094       break;
1095     case SESSION_CTRL_EVT_APP_DEL_SEGMENT:
1096       vcl_session_app_del_segment_handler (wrk, e->data);
1097       break;
1098     case SESSION_CTRL_EVT_APP_WRK_RPC:
1099       vcl_worker_rpc_handler (wrk, e->data);
1100       break;
1101     case SESSION_CTRL_EVT_TRANSPORT_ATTR_REPLY:
1102       vcl_session_transport_attr_reply_handler (wrk, e->data);
1103       break;
1104     default:
1105       clib_warning ("unhandled %u", e->event_type);
1106     }
1107   return VPPCOM_OK;
1108 }
1109
1110 static int
1111 vppcom_wait_for_session_state_change (u32 session_index,
1112                                       vcl_session_state_t state,
1113                                       f64 wait_for_time)
1114 {
1115   vcl_worker_t *wrk = vcl_worker_get_current ();
1116   f64 timeout = clib_time_now (&wrk->clib_time) + wait_for_time;
1117   vcl_session_t *volatile session;
1118   svm_msg_q_msg_t msg;
1119   session_event_t *e;
1120
1121   do
1122     {
1123       session = vcl_session_get (wrk, session_index);
1124       if (PREDICT_FALSE (!session))
1125         {
1126           return VPPCOM_EBADFD;
1127         }
1128       if (session->session_state == state)
1129         {
1130           return VPPCOM_OK;
1131         }
1132       if (session->session_state == VCL_STATE_DETACHED)
1133         {
1134           return VPPCOM_ECONNREFUSED;
1135         }
1136
1137       if (svm_msg_q_sub (wrk->app_event_queue, &msg, SVM_Q_NOWAIT, 0))
1138         {
1139           usleep (100);
1140           continue;
1141         }
1142       e = svm_msg_q_msg_data (wrk->app_event_queue, &msg);
1143       vcl_handle_mq_event (wrk, e);
1144       svm_msg_q_free_msg (wrk->app_event_queue, &msg);
1145     }
1146   while (clib_time_now (&wrk->clib_time) < timeout);
1147
1148   VDBG (0, "timeout waiting for state 0x%x (%s)", state,
1149         vcl_session_state_str (state));
1150   vcl_evt (VCL_EVT_SESSION_TIMEOUT, session, session_state);
1151
1152   return VPPCOM_ETIMEDOUT;
1153 }
1154
1155 static void
1156 vcl_handle_pending_wrk_updates (vcl_worker_t * wrk)
1157 {
1158   vcl_session_state_t state;
1159   vcl_session_t *s;
1160   u32 *sip;
1161
1162   if (PREDICT_TRUE (vec_len (wrk->pending_session_wrk_updates) == 0))
1163     return;
1164
1165   vec_foreach (sip, wrk->pending_session_wrk_updates)
1166   {
1167     s = vcl_session_get (wrk, *sip);
1168     vcl_send_session_worker_update (wrk, s, wrk->wrk_index);
1169     state = s->session_state;
1170     vppcom_wait_for_session_state_change (s->session_index, VCL_STATE_UPDATED,
1171                                           5);
1172     s->session_state = state;
1173   }
1174   vec_reset_length (wrk->pending_session_wrk_updates);
1175 }
1176
1177 void
1178 vcl_worker_flush_mq_events (vcl_worker_t *wrk)
1179 {
1180   svm_msg_q_msg_t *msg;
1181   session_event_t *e;
1182   svm_msg_q_t *mq;
1183   int i;
1184
1185   mq = wrk->app_event_queue;
1186   vcl_mq_dequeue_batch (wrk, mq, ~0);
1187
1188   for (i = 0; i < vec_len (wrk->mq_msg_vector); i++)
1189     {
1190       msg = vec_elt_at_index (wrk->mq_msg_vector, i);
1191       e = svm_msg_q_msg_data (mq, msg);
1192       vcl_handle_mq_event (wrk, e);
1193       svm_msg_q_free_msg (mq, msg);
1194     }
1195   vec_reset_length (wrk->mq_msg_vector);
1196   vcl_handle_pending_wrk_updates (wrk);
1197 }
1198
1199 void
1200 vcl_flush_mq_events (void)
1201 {
1202   vcl_worker_flush_mq_events (vcl_worker_get_current ());
1203 }
1204
1205 static int
1206 vppcom_session_unbind (u32 session_handle)
1207 {
1208   vcl_worker_t *wrk = vcl_worker_get_current ();
1209   session_accepted_msg_t *accepted_msg;
1210   vcl_session_t *session = 0;
1211   vcl_session_msg_t *evt;
1212
1213   session = vcl_session_get_w_handle (wrk, session_handle);
1214   if (!session)
1215     return VPPCOM_EBADFD;
1216
1217   /* Flush pending accept events, if any */
1218   while (clib_fifo_elts (session->accept_evts_fifo))
1219     {
1220       clib_fifo_sub2 (session->accept_evts_fifo, evt);
1221       accepted_msg = &evt->accepted_msg;
1222       vcl_session_table_del_vpp_handle (wrk, accepted_msg->handle);
1223       vcl_send_session_accepted_reply (session->vpp_evt_q,
1224                                        accepted_msg->context,
1225                                        accepted_msg->handle, -1);
1226     }
1227   clib_fifo_free (session->accept_evts_fifo);
1228
1229   vcl_send_session_unlisten (wrk, session);
1230
1231   VDBG (1, "session %u [0x%llx]: sending unbind!", session->session_index,
1232         session->vpp_handle);
1233   vcl_evt (VCL_EVT_UNBIND, session);
1234
1235   session->vpp_handle = ~0;
1236   session->session_state = VCL_STATE_DISCONNECT;
1237
1238   return VPPCOM_OK;
1239 }
1240
1241 /**
1242  * Handle app exit
1243  *
1244  * Notify vpp of the disconnect and mark the worker as free. If we're the
1245  * last worker, do a full cleanup otherwise, since we're probably a forked
1246  * child, avoid syscalls as much as possible. We might've lost privileges.
1247  */
1248 void
1249 vppcom_app_exit (void)
1250 {
1251   if (!pool_elts (vcm->workers))
1252     return;
1253   vcl_worker_cleanup (vcl_worker_get_current (), 1 /* notify vpp */ );
1254   vcl_set_worker_index (~0);
1255   vcl_elog_stop (vcm);
1256 }
1257
1258 static int
1259 vcl_api_attach (void)
1260 {
1261   if (vcm->cfg.vpp_app_socket_api)
1262     return vcl_sapi_attach ();
1263
1264   return vcl_bapi_attach ();
1265 }
1266
1267 static void
1268 vcl_api_retry_attach (vcl_worker_t *wrk)
1269 {
1270   vcl_session_t *s;
1271
1272   if (vcl_api_attach ())
1273     return;
1274
1275   /* Treat listeners as configuration that needs to be re-added to vpp */
1276   pool_foreach (s, wrk->sessions)
1277     {
1278       if (s->flags & VCL_SESSION_F_IS_VEP)
1279         continue;
1280       if (s->session_state == VCL_STATE_LISTEN_NO_MQ)
1281         vppcom_session_listen (vcl_session_handle (s), 10);
1282       else
1283         VDBG (0, "internal error: unexpected state %d", s->session_state);
1284     }
1285 }
1286
1287 static void
1288 vcl_api_handle_disconnect (vcl_worker_t *wrk)
1289 {
1290   wrk->api_client_handle = ~0;
1291   vcl_worker_detach_sessions (wrk);
1292 }
1293
1294 static void
1295 vcl_api_detach (vcl_worker_t * wrk)
1296 {
1297   if (wrk->api_client_handle == ~0)
1298     return;
1299
1300   vcl_send_app_detach (wrk);
1301
1302   if (vcm->cfg.vpp_app_socket_api)
1303     return vcl_sapi_detach (wrk);
1304
1305   return vcl_bapi_disconnect_from_vpp ();
1306 }
1307
1308 /*
1309  * VPPCOM Public API functions
1310  */
1311 int
1312 vppcom_app_create (const char *app_name)
1313 {
1314   vppcom_cfg_t *vcl_cfg = &vcm->cfg;
1315   int rv;
1316
1317   if (vcm->is_init)
1318     {
1319       VDBG (1, "already initialized");
1320       return VPPCOM_EEXIST;
1321     }
1322
1323   vcm->is_init = 1;
1324   vppcom_cfg (&vcm->cfg);
1325   vcl_cfg = &vcm->cfg;
1326
1327   vcm->main_cpu = pthread_self ();
1328   vcm->main_pid = getpid ();
1329   vcm->app_name = format (0, "%s", app_name);
1330   fifo_segment_main_init (&vcm->segment_main, (uword) ~0,
1331                           20 /* timeout in secs */);
1332   pool_alloc (vcm->workers, vcl_cfg->max_workers);
1333   clib_spinlock_init (&vcm->workers_lock);
1334   clib_rwlock_init (&vcm->segment_table_lock);
1335   atexit (vppcom_app_exit);
1336   vcl_elog_init (vcm);
1337
1338   /* Allocate default worker */
1339   vcl_worker_alloc_and_init ();
1340
1341   if ((rv = vcl_api_attach ()))
1342     {
1343       vppcom_app_destroy ();
1344       return rv;
1345     }
1346
1347   VDBG (0, "app_name '%s', my_client_index %d (0x%x)", app_name,
1348         vcm->workers[0].api_client_handle, vcm->workers[0].api_client_handle);
1349
1350   return VPPCOM_OK;
1351 }
1352
1353 void
1354 vppcom_app_destroy (void)
1355 {
1356   vcl_worker_t *wrk, *current_wrk;
1357   void *heap;
1358
1359   if (!pool_elts (vcm->workers))
1360     return;
1361
1362   vcl_evt (VCL_EVT_DETACH, vcm);
1363
1364   current_wrk = vcl_worker_get_current ();
1365
1366   /* *INDENT-OFF* */
1367   pool_foreach (wrk, vcm->workers)  {
1368     if (current_wrk != wrk)
1369       vcl_worker_cleanup (wrk, 0 /* notify vpp */ );
1370   }
1371   /* *INDENT-ON* */
1372
1373   vcl_api_detach (current_wrk);
1374   vcl_worker_cleanup (current_wrk, 0 /* notify vpp */ );
1375   vcl_set_worker_index (~0);
1376
1377   vcl_elog_stop (vcm);
1378
1379   /*
1380    * Free the heap and fix vcm
1381    */
1382   heap = clib_mem_get_heap ();
1383   munmap (clib_mem_get_heap_base (heap), clib_mem_get_heap_size (heap));
1384
1385   vcm = &_vppcom_main;
1386   vcm->is_init = 0;
1387 }
1388
1389 int
1390 vppcom_session_create (u8 proto, u8 is_nonblocking)
1391 {
1392   vcl_worker_t *wrk = vcl_worker_get_current ();
1393   vcl_session_t *session;
1394
1395   session = vcl_session_alloc (wrk);
1396
1397   session->session_type = proto;
1398   session->session_state = VCL_STATE_CLOSED;
1399   session->vpp_handle = ~0;
1400   session->is_dgram = vcl_proto_is_dgram (proto);
1401
1402   if (is_nonblocking)
1403     vcl_session_set_attr (session, VCL_SESS_ATTR_NONBLOCK);
1404
1405   vcl_evt (VCL_EVT_CREATE, session, session_type, session->session_state,
1406            is_nonblocking, session_index);
1407
1408   VDBG (0, "created session %u", session->session_index);
1409
1410   return vcl_session_handle (session);
1411 }
1412
1413 static void
1414 vcl_epoll_lt_add (vcl_worker_t *wrk, vcl_session_t *s)
1415 {
1416   vcl_session_t *cur, *prev;
1417
1418   if (wrk->ep_lt_current == VCL_INVALID_SESSION_INDEX)
1419     {
1420       wrk->ep_lt_current = s->session_index;
1421       s->vep.lt_next = s->session_index;
1422       s->vep.lt_prev = s->session_index;
1423       return;
1424     }
1425
1426   cur = vcl_session_get (wrk, wrk->ep_lt_current);
1427   prev = vcl_session_get (wrk, cur->vep.lt_prev);
1428
1429   prev->vep.lt_next = s->session_index;
1430   s->vep.lt_prev = prev->session_index;
1431
1432   s->vep.lt_next = cur->session_index;
1433   cur->vep.lt_prev = s->session_index;
1434 }
1435
1436 static void
1437 vcl_epoll_lt_del (vcl_worker_t *wrk, vcl_session_t *s)
1438 {
1439   vcl_session_t *prev, *next;
1440
1441   if (s->vep.lt_next == s->session_index)
1442     {
1443       wrk->ep_lt_current = VCL_INVALID_SESSION_INDEX;
1444       s->vep.lt_next = VCL_INVALID_SESSION_INDEX;
1445       return;
1446     }
1447
1448   prev = vcl_session_get (wrk, s->vep.lt_prev);
1449   next = vcl_session_get (wrk, s->vep.lt_next);
1450
1451   prev->vep.lt_next = next->session_index;
1452   next->vep.lt_prev = prev->session_index;
1453
1454   if (s->session_index == wrk->ep_lt_current)
1455     wrk->ep_lt_current = s->vep.lt_next;
1456
1457   s->vep.lt_next = VCL_INVALID_SESSION_INDEX;
1458 }
1459
1460 int
1461 vcl_session_cleanup (vcl_worker_t * wrk, vcl_session_t * s,
1462                      vcl_session_handle_t sh, u8 do_disconnect)
1463 {
1464   int rv = VPPCOM_OK;
1465
1466   VDBG (1, "session %u [0x%llx] closing", s->session_index, s->vpp_handle);
1467
1468   if (s->flags & VCL_SESSION_F_IS_VEP)
1469     {
1470       u32 next_sh = s->vep.next_sh;
1471       while (next_sh != ~0)
1472         {
1473           rv = vppcom_epoll_ctl (sh, EPOLL_CTL_DEL, next_sh, 0);
1474           if (PREDICT_FALSE (rv < 0))
1475             VDBG (0, "vpp handle 0x%llx, sh %u: EPOLL_CTL_DEL vep_idx %u"
1476                   " failed! rv %d (%s)", s->vpp_handle, next_sh,
1477                   s->vep.vep_sh, rv, vppcom_retval_str (rv));
1478           next_sh = s->vep.next_sh;
1479         }
1480       goto free_session;
1481     }
1482
1483   if (s->flags & VCL_SESSION_F_IS_VEP_SESSION)
1484     {
1485       rv = vppcom_epoll_ctl (s->vep.vep_sh, EPOLL_CTL_DEL, sh, 0);
1486       if (rv < 0)
1487         VDBG (0, "session %u [0x%llx]: EPOLL_CTL_DEL vep_idx %u "
1488               "failed! rv %d (%s)", s->session_index, s->vpp_handle,
1489               s->vep.vep_sh, rv, vppcom_retval_str (rv));
1490     }
1491
1492   if (!do_disconnect)
1493     {
1494       VDBG (1, "session %u [0x%llx] disconnect skipped",
1495             s->session_index, s->vpp_handle);
1496       goto cleanup;
1497     }
1498
1499   if (s->session_state == VCL_STATE_LISTEN)
1500     {
1501       rv = vppcom_session_unbind (sh);
1502       if (PREDICT_FALSE (rv < 0))
1503         VDBG (0, "session %u [0x%llx]: listener unbind failed! "
1504               "rv %d (%s)", s->session_index, s->vpp_handle, rv,
1505               vppcom_retval_str (rv));
1506       return rv;
1507     }
1508   else if (vcl_session_is_ready (s)
1509            || (vcl_session_is_connectable_listener (wrk, s)))
1510     {
1511       rv = vppcom_session_disconnect (sh);
1512       if (PREDICT_FALSE (rv < 0))
1513         VDBG (0, "ERROR: session %u [0x%llx]: disconnect failed!"
1514               " rv %d (%s)", s->session_index, s->vpp_handle,
1515               rv, vppcom_retval_str (rv));
1516     }
1517   else if (s->session_state == VCL_STATE_DISCONNECT)
1518     {
1519       vcl_send_session_reset_reply (wrk, s, 0);
1520     }
1521   else if (s->session_state == VCL_STATE_DETACHED)
1522     {
1523       VDBG (0, "vpp freed session %d before close", s->session_index);
1524
1525       if (!(s->flags & VCL_SESSION_F_PENDING_DISCONNECT))
1526         goto free_session;
1527
1528       /* Disconnect/reset messages pending but vpp transport and session
1529        * cleanups already done. Free only after messages drained. */
1530       s->flags |= VCL_SESSION_F_PENDING_FREE;
1531     }
1532
1533   s->session_state = VCL_STATE_CLOSED;
1534
1535   /* Session is removed only after vpp confirms the disconnect */
1536   return rv;
1537
1538 cleanup:
1539   vcl_session_table_del_vpp_handle (wrk, s->vpp_handle);
1540 free_session:
1541   vcl_session_free (wrk, s);
1542   vcl_evt (VCL_EVT_CLOSE, s, rv);
1543
1544   return rv;
1545 }
1546
1547 int
1548 vppcom_session_close (uint32_t session_handle)
1549 {
1550   vcl_worker_t *wrk = vcl_worker_get_current ();
1551   vcl_session_t *session;
1552
1553   session = vcl_session_get_w_handle (wrk, session_handle);
1554   if (!session)
1555     return VPPCOM_EBADFD;
1556   return vcl_session_cleanup (wrk, session, session_handle,
1557                               1 /* do_disconnect */ );
1558 }
1559
1560 int
1561 vppcom_session_bind (uint32_t session_handle, vppcom_endpt_t * ep)
1562 {
1563   vcl_worker_t *wrk = vcl_worker_get_current ();
1564   vcl_session_t *session = 0;
1565
1566   if (!ep || !ep->ip)
1567     return VPPCOM_EINVAL;
1568
1569   session = vcl_session_get_w_handle (wrk, session_handle);
1570   if (!session)
1571     return VPPCOM_EBADFD;
1572
1573   if (session->flags & VCL_SESSION_F_IS_VEP)
1574     {
1575       VDBG (0, "ERROR: cannot bind to epoll session %u!",
1576             session->session_index);
1577       return VPPCOM_EBADFD;
1578     }
1579
1580   session->transport.is_ip4 = ep->is_ip4;
1581   if (ep->is_ip4)
1582     clib_memcpy_fast (&session->transport.lcl_ip.ip4, ep->ip,
1583                       sizeof (ip4_address_t));
1584   else
1585     clib_memcpy_fast (&session->transport.lcl_ip.ip6, ep->ip,
1586                       sizeof (ip6_address_t));
1587   session->transport.lcl_port = ep->port;
1588
1589   VDBG (0,
1590         "session %u handle %u: binding to local %s address %U port %u, "
1591         "proto %s",
1592         session->session_index, session_handle,
1593         session->transport.is_ip4 ? "IPv4" : "IPv6", vcl_format_ip46_address,
1594         &session->transport.lcl_ip,
1595         session->transport.is_ip4 ? IP46_TYPE_IP4 : IP46_TYPE_IP6,
1596         clib_net_to_host_u16 (session->transport.lcl_port),
1597         vppcom_proto_str (session->session_type));
1598   vcl_evt (VCL_EVT_BIND, session);
1599
1600   if (session->session_type == VPPCOM_PROTO_UDP)
1601     vppcom_session_listen (session_handle, 10);
1602
1603   return VPPCOM_OK;
1604 }
1605
1606 int
1607 vppcom_session_listen (uint32_t listen_sh, uint32_t q_len)
1608 {
1609   vcl_worker_t *wrk = vcl_worker_get_current ();
1610   vcl_session_t *listen_session = 0;
1611   u64 listen_vpp_handle;
1612   int rv;
1613
1614   listen_session = vcl_session_get_w_handle (wrk, listen_sh);
1615   if (!listen_session || (listen_session->flags & VCL_SESSION_F_IS_VEP))
1616     return VPPCOM_EBADFD;
1617
1618   listen_vpp_handle = listen_session->vpp_handle;
1619   if (listen_session->session_state == VCL_STATE_LISTEN)
1620     {
1621       VDBG (0, "session %u [0x%llx]: already in listen state!",
1622             listen_sh, listen_vpp_handle);
1623       return VPPCOM_OK;
1624     }
1625
1626   VDBG (0, "session %u: sending vpp listen request...", listen_sh);
1627
1628   /*
1629    * Send listen request to vpp and wait for reply
1630    */
1631   vcl_send_session_listen (wrk, listen_session);
1632   rv = vppcom_wait_for_session_state_change (listen_session->session_index,
1633                                              VCL_STATE_LISTEN,
1634                                              vcm->cfg.session_timeout);
1635
1636   if (PREDICT_FALSE (rv))
1637     {
1638       listen_session = vcl_session_get_w_handle (wrk, listen_sh);
1639       VDBG (0, "session %u [0x%llx]: listen failed! returning %d (%s)",
1640             listen_sh, listen_session->vpp_handle, rv,
1641             vppcom_retval_str (rv));
1642       return rv;
1643     }
1644
1645   return VPPCOM_OK;
1646 }
1647
1648 static int
1649 validate_args_session_accept_ (vcl_worker_t * wrk, vcl_session_t * ls)
1650 {
1651   if (ls->flags & VCL_SESSION_F_IS_VEP)
1652     {
1653       VDBG (0, "ERROR: cannot accept on epoll session %u!",
1654             ls->session_index);
1655       return VPPCOM_EBADFD;
1656     }
1657
1658   if ((ls->session_state != VCL_STATE_LISTEN)
1659       && (!vcl_session_is_connectable_listener (wrk, ls)))
1660     {
1661       VDBG (0,
1662             "ERROR: session [0x%llx]: not in listen state! state 0x%x"
1663             " (%s)",
1664             ls->vpp_handle, ls->session_state,
1665             vcl_session_state_str (ls->session_state));
1666       return VPPCOM_EBADFD;
1667     }
1668   return VPPCOM_OK;
1669 }
1670
1671 int
1672 vppcom_unformat_proto (uint8_t * proto, char *proto_str)
1673 {
1674   if (!strcmp (proto_str, "TCP"))
1675     *proto = VPPCOM_PROTO_TCP;
1676   else if (!strcmp (proto_str, "tcp"))
1677     *proto = VPPCOM_PROTO_TCP;
1678   else if (!strcmp (proto_str, "UDP"))
1679     *proto = VPPCOM_PROTO_UDP;
1680   else if (!strcmp (proto_str, "udp"))
1681     *proto = VPPCOM_PROTO_UDP;
1682   else if (!strcmp (proto_str, "TLS"))
1683     *proto = VPPCOM_PROTO_TLS;
1684   else if (!strcmp (proto_str, "tls"))
1685     *proto = VPPCOM_PROTO_TLS;
1686   else if (!strcmp (proto_str, "QUIC"))
1687     *proto = VPPCOM_PROTO_QUIC;
1688   else if (!strcmp (proto_str, "quic"))
1689     *proto = VPPCOM_PROTO_QUIC;
1690   else if (!strcmp (proto_str, "DTLS"))
1691     *proto = VPPCOM_PROTO_DTLS;
1692   else if (!strcmp (proto_str, "dtls"))
1693     *proto = VPPCOM_PROTO_DTLS;
1694   else if (!strcmp (proto_str, "SRTP"))
1695     *proto = VPPCOM_PROTO_SRTP;
1696   else if (!strcmp (proto_str, "srtp"))
1697     *proto = VPPCOM_PROTO_SRTP;
1698   else
1699     return 1;
1700   return 0;
1701 }
1702
1703 int
1704 vppcom_session_accept (uint32_t listen_session_handle, vppcom_endpt_t * ep,
1705                        uint32_t flags)
1706 {
1707   u32 client_session_index = ~0, listen_session_index, accept_flags = 0;
1708   vcl_worker_t *wrk = vcl_worker_get_current ();
1709   session_accepted_msg_t accepted_msg;
1710   vcl_session_t *listen_session = 0;
1711   vcl_session_t *client_session = 0;
1712   vcl_session_msg_t *evt;
1713   u8 is_nonblocking;
1714   int rv;
1715
1716 again:
1717
1718   listen_session = vcl_session_get_w_handle (wrk, listen_session_handle);
1719   if (!listen_session)
1720     return VPPCOM_EBADFD;
1721
1722   listen_session_index = listen_session->session_index;
1723   if ((rv = validate_args_session_accept_ (wrk, listen_session)))
1724     return rv;
1725
1726   if (clib_fifo_elts (listen_session->accept_evts_fifo))
1727     {
1728       clib_fifo_sub2 (listen_session->accept_evts_fifo, evt);
1729       accept_flags = evt->flags;
1730       accepted_msg = evt->accepted_msg;
1731       goto handle;
1732     }
1733
1734   is_nonblocking = vcl_session_has_attr (listen_session,
1735                                          VCL_SESS_ATTR_NONBLOCK);
1736   while (1)
1737     {
1738       if (svm_msg_q_is_empty (wrk->app_event_queue) && is_nonblocking)
1739         return VPPCOM_EAGAIN;
1740
1741       svm_msg_q_wait (wrk->app_event_queue, SVM_MQ_WAIT_EMPTY);
1742       vcl_worker_flush_mq_events (wrk);
1743       goto again;
1744     }
1745
1746 handle:
1747
1748   client_session_index = vcl_session_accepted_handler (wrk, &accepted_msg,
1749                                                        listen_session_index);
1750   if (client_session_index == VCL_INVALID_SESSION_INDEX)
1751     return VPPCOM_ECONNABORTED;
1752
1753   listen_session = vcl_session_get (wrk, listen_session_index);
1754   client_session = vcl_session_get (wrk, client_session_index);
1755
1756   if (flags & O_NONBLOCK)
1757     vcl_session_set_attr (client_session, VCL_SESS_ATTR_NONBLOCK);
1758
1759   VDBG (1, "listener %u [0x%llx]: Got a connect request! session %u [0x%llx],"
1760         " flags %d, is_nonblocking %u", listen_session->session_index,
1761         listen_session->vpp_handle, client_session_index,
1762         client_session->vpp_handle, flags,
1763         vcl_session_has_attr (client_session, VCL_SESS_ATTR_NONBLOCK));
1764
1765   if (ep)
1766     {
1767       ep->is_ip4 = client_session->transport.is_ip4;
1768       ep->port = client_session->transport.rmt_port;
1769       if (client_session->transport.is_ip4)
1770         clib_memcpy_fast (ep->ip, &client_session->transport.rmt_ip.ip4,
1771                           sizeof (ip4_address_t));
1772       else
1773         clib_memcpy_fast (ep->ip, &client_session->transport.rmt_ip.ip6,
1774                           sizeof (ip6_address_t));
1775     }
1776
1777   VDBG (0,
1778         "listener %u [0x%llx] accepted %u [0x%llx] peer: %U:%u "
1779         "local: %U:%u",
1780         listen_session_handle, listen_session->vpp_handle,
1781         client_session_index, client_session->vpp_handle,
1782         vcl_format_ip46_address, &client_session->transport.rmt_ip,
1783         client_session->transport.is_ip4 ? IP46_TYPE_IP4 : IP46_TYPE_IP6,
1784         clib_net_to_host_u16 (client_session->transport.rmt_port),
1785         vcl_format_ip46_address, &client_session->transport.lcl_ip,
1786         client_session->transport.is_ip4 ? IP46_TYPE_IP4 : IP46_TYPE_IP6,
1787         clib_net_to_host_u16 (client_session->transport.lcl_port));
1788   vcl_evt (VCL_EVT_ACCEPT, client_session, listen_session,
1789            client_session_index);
1790
1791   /*
1792    * Session might have been closed already
1793    */
1794   if (accept_flags)
1795     {
1796       if (accept_flags & VCL_ACCEPTED_F_CLOSED)
1797         client_session->session_state = VCL_STATE_VPP_CLOSING;
1798       else if (accept_flags & VCL_ACCEPTED_F_RESET)
1799         client_session->session_state = VCL_STATE_DISCONNECT;
1800     }
1801   return vcl_session_handle (client_session);
1802 }
1803
1804 int
1805 vppcom_session_connect (uint32_t session_handle, vppcom_endpt_t * server_ep)
1806 {
1807   vcl_worker_t *wrk = vcl_worker_get_current ();
1808   vcl_session_t *session = 0;
1809   u32 session_index;
1810   int rv;
1811
1812   session = vcl_session_get_w_handle (wrk, session_handle);
1813   if (!session)
1814     return VPPCOM_EBADFD;
1815   session_index = session->session_index;
1816
1817   if (PREDICT_FALSE (session->flags & VCL_SESSION_F_IS_VEP))
1818     {
1819       VWRN ("cannot connect epoll session %u!", session->session_index);
1820       return VPPCOM_EBADFD;
1821     }
1822
1823   if (PREDICT_FALSE (vcl_session_is_ready (session)))
1824     {
1825       VDBG (0,
1826             "session %u [0x%llx]: already connected to %U:%d proto %s,"
1827             " state (%s)",
1828             session->session_index, session->vpp_handle,
1829             vcl_format_ip46_address, &session->transport.rmt_ip,
1830             session->transport.is_ip4 ? IP46_TYPE_IP4 : IP46_TYPE_IP6,
1831             clib_net_to_host_u16 (session->transport.rmt_port),
1832             vppcom_proto_str (session->session_type),
1833             vcl_session_state_str (session->session_state));
1834       return VPPCOM_OK;
1835     }
1836
1837   /* Attempt to connect a connectionless listener */
1838   if (PREDICT_FALSE (session->session_state == VCL_STATE_LISTEN))
1839     {
1840       if (session->session_type != VPPCOM_PROTO_UDP)
1841         return VPPCOM_EINVAL;
1842       vcl_send_session_unlisten (wrk, session);
1843       session->session_state = VCL_STATE_CLOSED;
1844     }
1845
1846   session->transport.is_ip4 = server_ep->is_ip4;
1847   vcl_ip_copy_from_ep (&session->transport.rmt_ip, server_ep);
1848   session->transport.rmt_port = server_ep->port;
1849   session->parent_handle = VCL_INVALID_SESSION_HANDLE;
1850   session->flags |= VCL_SESSION_F_CONNECTED;
1851
1852   VDBG (0, "session %u: connecting to peer %U:%d proto %s",
1853         session->session_index, vcl_format_ip46_address,
1854         &session->transport.rmt_ip,
1855         session->transport.is_ip4 ? IP46_TYPE_IP4 : IP46_TYPE_IP6,
1856         clib_net_to_host_u16 (session->transport.rmt_port),
1857         vppcom_proto_str (session->session_type));
1858
1859   vcl_send_session_connect (wrk, session);
1860
1861   if (vcl_session_has_attr (session, VCL_SESS_ATTR_NONBLOCK))
1862     {
1863       /* State set to STATE_UPDATED to ensure the session is not assumed
1864        * to be ready and to also allow the app to close it prior to vpp's
1865        * connected reply. */
1866       session->session_state = VCL_STATE_UPDATED;
1867       return VPPCOM_EINPROGRESS;
1868     }
1869
1870   /*
1871    * Wait for reply from vpp if blocking
1872    */
1873   rv = vppcom_wait_for_session_state_change (session_index, VCL_STATE_READY,
1874                                              vcm->cfg.session_timeout);
1875
1876   session = vcl_session_get (wrk, session_index);
1877   VDBG (0, "session %u [0x%llx]: connect %s!", session->session_index,
1878         session->vpp_handle, rv ? "failed" : "succeeded");
1879
1880   return rv;
1881 }
1882
1883 int
1884 vppcom_session_stream_connect (uint32_t session_handle,
1885                                uint32_t parent_session_handle)
1886 {
1887   vcl_worker_t *wrk = vcl_worker_get_current ();
1888   vcl_session_t *session, *parent_session;
1889   u32 session_index, parent_session_index;
1890   int rv;
1891
1892   session = vcl_session_get_w_handle (wrk, session_handle);
1893   if (!session)
1894     return VPPCOM_EBADFD;
1895   parent_session = vcl_session_get_w_handle (wrk, parent_session_handle);
1896   if (!parent_session)
1897     return VPPCOM_EBADFD;
1898
1899   session_index = session->session_index;
1900   parent_session_index = parent_session->session_index;
1901   if (PREDICT_FALSE (session->flags & VCL_SESSION_F_IS_VEP))
1902     {
1903       VDBG (0, "ERROR: cannot connect epoll session %u!",
1904             session->session_index);
1905       return VPPCOM_EBADFD;
1906     }
1907
1908   if (PREDICT_FALSE (vcl_session_is_ready (session)))
1909     {
1910       VDBG (0,
1911             "session handle %u [0x%llx]: session already "
1912             "connected to session %u [0x%llx] proto %s, state 0x%x (%s)",
1913             session_handle, session->vpp_handle, parent_session_handle,
1914             parent_session->vpp_handle,
1915             vppcom_proto_str (session->session_type), session->session_state,
1916             vcl_session_state_str (session->session_state));
1917       return VPPCOM_OK;
1918     }
1919
1920   /* Connect to quic session specifics */
1921   session->transport.is_ip4 = parent_session->transport.is_ip4;
1922   session->transport.rmt_ip.ip4.as_u32 = (uint32_t) 1;
1923   session->transport.rmt_port = 0;
1924   session->parent_handle = parent_session->vpp_handle;
1925
1926   VDBG (0, "session handle %u: connecting to session %u [0x%llx]",
1927         session_handle, parent_session_handle, parent_session->vpp_handle);
1928
1929   /*
1930    * Send connect request and wait for reply from vpp
1931    */
1932   vcl_send_session_connect (wrk, session);
1933   rv = vppcom_wait_for_session_state_change (session_index, VCL_STATE_READY,
1934                                              vcm->cfg.session_timeout);
1935
1936   session->listener_index = parent_session_index;
1937   parent_session = vcl_session_get_w_handle (wrk, parent_session_handle);
1938   if (parent_session)
1939     parent_session->n_accepted_sessions++;
1940
1941   session = vcl_session_get (wrk, session_index);
1942   VDBG (0, "session %u [0x%llx]: connect %s!", session->session_index,
1943         session->vpp_handle, rv ? "failed" : "succeeded");
1944
1945   return rv;
1946 }
1947
1948 static inline int
1949 vppcom_session_read_internal (uint32_t session_handle, void *buf, int n,
1950                               u8 peek)
1951 {
1952   vcl_worker_t *wrk = vcl_worker_get_current ();
1953   int rv, n_read = 0, is_nonblocking;
1954   vcl_session_t *s = 0;
1955   svm_fifo_t *rx_fifo;
1956   session_event_t *e;
1957   svm_msg_q_t *mq;
1958   u8 is_ct;
1959
1960   if (PREDICT_FALSE (!buf))
1961     return VPPCOM_EFAULT;
1962
1963   s = vcl_session_get_w_handle (wrk, session_handle);
1964   if (PREDICT_FALSE (!s || (s->flags & VCL_SESSION_F_IS_VEP)))
1965     return VPPCOM_EBADFD;
1966
1967   if (PREDICT_FALSE (!vcl_session_is_open (s)))
1968     {
1969       VDBG (0, "session %u[0x%llx] is not open! state 0x%x (%s)",
1970             s->session_index, s->vpp_handle, s->session_state,
1971             vcl_session_state_str (s->session_state));
1972       return vcl_session_closed_error (s);
1973     }
1974
1975   if (PREDICT_FALSE (s->flags & VCL_SESSION_F_RD_SHUTDOWN))
1976     {
1977       /* Vpp would ack the incoming data and enqueue it for reading.
1978        * So even if SHUT_RD is set, we can still read() the data if
1979        * the session is ready.
1980        */
1981       if (!vcl_session_read_ready (s))
1982         {
1983           return 0;
1984         }
1985     }
1986
1987   is_nonblocking = vcl_session_has_attr (s, VCL_SESS_ATTR_NONBLOCK);
1988   is_ct = vcl_session_is_ct (s);
1989   mq = wrk->app_event_queue;
1990   rx_fifo = is_ct ? s->ct_rx_fifo : s->rx_fifo;
1991   s->flags &= ~VCL_SESSION_F_HAS_RX_EVT;
1992
1993   if (svm_fifo_is_empty_cons (rx_fifo))
1994     {
1995       if (is_nonblocking)
1996         {
1997           if (vcl_session_is_closing (s))
1998             return vcl_session_closing_error (s);
1999           if (is_ct)
2000             svm_fifo_unset_event (s->rx_fifo);
2001           svm_fifo_unset_event (rx_fifo);
2002           return VPPCOM_EWOULDBLOCK;
2003         }
2004       while (svm_fifo_is_empty_cons (rx_fifo))
2005         {
2006           if (vcl_session_is_closing (s))
2007             return vcl_session_closing_error (s);
2008
2009           if (is_ct)
2010             svm_fifo_unset_event (s->rx_fifo);
2011           svm_fifo_unset_event (rx_fifo);
2012
2013           svm_msg_q_wait (mq, SVM_MQ_WAIT_EMPTY);
2014           vcl_worker_flush_mq_events (wrk);
2015         }
2016     }
2017
2018 read_again:
2019
2020   if (s->is_dgram)
2021     rv = app_recv_dgram_raw (rx_fifo, buf, n, &s->transport, 0, peek);
2022   else
2023     rv = app_recv_stream_raw (rx_fifo, buf, n, 0, peek);
2024
2025   ASSERT (rv >= 0);
2026
2027   if (peek)
2028     return rv;
2029
2030   n_read += rv;
2031
2032   if (svm_fifo_is_empty_cons (rx_fifo))
2033     {
2034       if (is_ct)
2035         svm_fifo_unset_event (s->rx_fifo);
2036       svm_fifo_unset_event (rx_fifo);
2037       if (!svm_fifo_is_empty_cons (rx_fifo)
2038           && svm_fifo_set_event (rx_fifo) && is_nonblocking)
2039         {
2040           vec_add2 (wrk->unhandled_evts_vector, e, 1);
2041           e->event_type = SESSION_IO_EVT_RX;
2042           e->session_index = s->session_index;
2043         }
2044     }
2045   else if (PREDICT_FALSE (rv < n && !s->is_dgram))
2046     {
2047       /* More data enqueued while reading. Try to drain it
2048        * or fill the buffer. Avoid doing that for dgrams */
2049       buf += rv;
2050       n -= rv;
2051       goto read_again;
2052     }
2053
2054   if (PREDICT_FALSE (svm_fifo_needs_deq_ntf (rx_fifo, n_read)))
2055     {
2056       svm_fifo_clear_deq_ntf (rx_fifo);
2057       app_send_io_evt_to_vpp (s->vpp_evt_q,
2058                               s->rx_fifo->shr->master_session_index,
2059                               SESSION_IO_EVT_RX, SVM_Q_WAIT);
2060     }
2061
2062   VDBG (2, "session %u[0x%llx]: read %d bytes from (%p)", s->session_index,
2063         s->vpp_handle, n_read, rx_fifo);
2064
2065   return n_read;
2066 }
2067
2068 int
2069 vppcom_session_read (uint32_t session_handle, void *buf, size_t n)
2070 {
2071   return (vppcom_session_read_internal (session_handle, buf, n, 0));
2072 }
2073
2074 static int
2075 vppcom_session_peek (uint32_t session_handle, void *buf, int n)
2076 {
2077   return (vppcom_session_read_internal (session_handle, buf, n, 1));
2078 }
2079
2080 int
2081 vppcom_session_read_segments (uint32_t session_handle,
2082                               vppcom_data_segment_t * ds, uint32_t n_segments,
2083                               uint32_t max_bytes)
2084 {
2085   vcl_worker_t *wrk = vcl_worker_get_current ();
2086   int n_read = 0, is_nonblocking;
2087   vcl_session_t *s = 0;
2088   svm_fifo_t *rx_fifo;
2089   svm_msg_q_t *mq;
2090   u8 is_ct;
2091
2092   s = vcl_session_get_w_handle (wrk, session_handle);
2093   if (PREDICT_FALSE (!s || (s->flags & VCL_SESSION_F_IS_VEP)))
2094     return VPPCOM_EBADFD;
2095
2096   if (PREDICT_FALSE (!vcl_session_is_open (s)))
2097     return vcl_session_closed_error (s);
2098
2099   is_nonblocking = vcl_session_has_attr (s, VCL_SESS_ATTR_NONBLOCK);
2100   is_ct = vcl_session_is_ct (s);
2101   mq = wrk->app_event_queue;
2102   rx_fifo = is_ct ? s->ct_rx_fifo : s->rx_fifo;
2103   s->flags &= ~VCL_SESSION_F_HAS_RX_EVT;
2104
2105   if (svm_fifo_is_empty_cons (rx_fifo))
2106     {
2107       if (is_nonblocking)
2108         {
2109           if (is_ct)
2110             svm_fifo_unset_event (s->rx_fifo);
2111           svm_fifo_unset_event (rx_fifo);
2112           return VPPCOM_EWOULDBLOCK;
2113         }
2114       while (svm_fifo_is_empty_cons (rx_fifo))
2115         {
2116           if (vcl_session_is_closing (s))
2117             return vcl_session_closing_error (s);
2118
2119           if (is_ct)
2120             svm_fifo_unset_event (s->rx_fifo);
2121           svm_fifo_unset_event (rx_fifo);
2122
2123           svm_msg_q_wait (mq, SVM_MQ_WAIT_EMPTY);
2124           vcl_worker_flush_mq_events (wrk);
2125         }
2126     }
2127
2128   n_read = svm_fifo_segments (rx_fifo, s->rx_bytes_pending,
2129                               (svm_fifo_seg_t *) ds, &n_segments, max_bytes);
2130   if (n_read < 0)
2131     return VPPCOM_EAGAIN;
2132
2133   if (svm_fifo_max_dequeue_cons (rx_fifo) == n_read)
2134     {
2135       if (is_ct)
2136         svm_fifo_unset_event (s->rx_fifo);
2137       svm_fifo_unset_event (rx_fifo);
2138       if (svm_fifo_max_dequeue_cons (rx_fifo) != n_read
2139           && svm_fifo_set_event (rx_fifo)
2140           && vcl_session_has_attr (s, VCL_SESS_ATTR_NONBLOCK))
2141         {
2142           session_event_t *e;
2143           vec_add2 (wrk->unhandled_evts_vector, e, 1);
2144           e->event_type = SESSION_IO_EVT_RX;
2145           e->session_index = s->session_index;
2146         }
2147     }
2148
2149   s->rx_bytes_pending += n_read;
2150   return n_read;
2151 }
2152
2153 void
2154 vppcom_session_free_segments (uint32_t session_handle, uint32_t n_bytes)
2155 {
2156   vcl_worker_t *wrk = vcl_worker_get_current ();
2157   vcl_session_t *s;
2158   u8 is_ct;
2159
2160   s = vcl_session_get_w_handle (wrk, session_handle);
2161   if (PREDICT_FALSE (!s || (s->flags & VCL_SESSION_F_IS_VEP)))
2162     return;
2163
2164   is_ct = vcl_session_is_ct (s);
2165   svm_fifo_dequeue_drop (is_ct ? s->ct_rx_fifo : s->rx_fifo, n_bytes);
2166
2167   ASSERT (s->rx_bytes_pending >= n_bytes);
2168   s->rx_bytes_pending -= n_bytes;
2169 }
2170
2171 always_inline u8
2172 vcl_fifo_is_writeable (svm_fifo_t * f, u32 len, u8 is_dgram)
2173 {
2174   u32 max_enq = svm_fifo_max_enqueue_prod (f);
2175   if (is_dgram)
2176     return max_enq >= (sizeof (session_dgram_hdr_t) + len);
2177   else
2178     return max_enq > 0;
2179 }
2180
2181 always_inline int
2182 vppcom_session_write_inline (vcl_worker_t * wrk, vcl_session_t * s, void *buf,
2183                              size_t n, u8 is_flush, u8 is_dgram)
2184 {
2185   int n_write, is_nonblocking;
2186   session_evt_type_t et;
2187   svm_fifo_t *tx_fifo;
2188   svm_msg_q_t *mq;
2189   u8 is_ct;
2190
2191   /* Accept zero length writes but just return */
2192   if (PREDICT_FALSE (!n))
2193     return VPPCOM_OK;
2194
2195   if (PREDICT_FALSE (!buf))
2196     return VPPCOM_EFAULT;
2197
2198   if (PREDICT_FALSE (s->flags & VCL_SESSION_F_IS_VEP))
2199     {
2200       VDBG (0, "ERROR: session %u [0x%llx]: cannot write to an epoll"
2201             " session!", s->session_index, s->vpp_handle);
2202       return VPPCOM_EBADFD;
2203     }
2204
2205   if (PREDICT_FALSE (!vcl_session_is_open (s)))
2206     {
2207       VDBG (1, "session %u [0x%llx]: is not open! state 0x%x (%s)",
2208             s->session_index, s->vpp_handle, s->session_state,
2209             vcl_session_state_str (s->session_state));
2210       return vcl_session_closed_error (s);;
2211     }
2212
2213   if (PREDICT_FALSE (s->flags & VCL_SESSION_F_WR_SHUTDOWN))
2214     {
2215       VDBG (1, "session %u [0x%llx]: is shutdown! state 0x%x (%s)",
2216             s->session_index, s->vpp_handle, s->session_state,
2217             vcl_session_state_str (s->session_state));
2218       return VPPCOM_EPIPE;
2219     }
2220
2221   is_ct = vcl_session_is_ct (s);
2222   tx_fifo = is_ct ? s->ct_tx_fifo : s->tx_fifo;
2223   is_nonblocking = vcl_session_has_attr (s, VCL_SESS_ATTR_NONBLOCK);
2224
2225   mq = wrk->app_event_queue;
2226   if (!vcl_fifo_is_writeable (tx_fifo, n, is_dgram))
2227     {
2228       if (is_nonblocking)
2229         {
2230           return VPPCOM_EWOULDBLOCK;
2231         }
2232       while (!vcl_fifo_is_writeable (tx_fifo, n, is_dgram))
2233         {
2234           svm_fifo_add_want_deq_ntf (tx_fifo, SVM_FIFO_WANT_DEQ_NOTIF);
2235           if (vcl_session_is_closing (s))
2236             return vcl_session_closing_error (s);
2237
2238           svm_msg_q_wait (mq, SVM_MQ_WAIT_EMPTY);
2239           vcl_worker_flush_mq_events (wrk);
2240         }
2241     }
2242
2243   et = SESSION_IO_EVT_TX;
2244   if (is_flush && !is_ct)
2245     et = SESSION_IO_EVT_TX_FLUSH;
2246
2247   if (is_dgram)
2248     n_write = app_send_dgram_raw (tx_fifo, &s->transport,
2249                                   s->vpp_evt_q, buf, n, et,
2250                                   0 /* do_evt */ , SVM_Q_WAIT);
2251   else
2252     n_write = app_send_stream_raw (tx_fifo, s->vpp_evt_q, buf, n, et,
2253                                    0 /* do_evt */ , SVM_Q_WAIT);
2254
2255   if (svm_fifo_set_event (s->tx_fifo))
2256     app_send_io_evt_to_vpp (
2257       s->vpp_evt_q, s->tx_fifo->shr->master_session_index, et, SVM_Q_WAIT);
2258
2259   /* The underlying fifo segment can run out of memory */
2260   if (PREDICT_FALSE (n_write < 0))
2261     return VPPCOM_EAGAIN;
2262
2263   VDBG (2, "session %u [0x%llx]: wrote %d bytes", s->session_index,
2264         s->vpp_handle, n_write);
2265
2266   return n_write;
2267 }
2268
2269 int
2270 vppcom_session_write (uint32_t session_handle, void *buf, size_t n)
2271 {
2272   vcl_worker_t *wrk = vcl_worker_get_current ();
2273   vcl_session_t *s;
2274
2275   s = vcl_session_get_w_handle (wrk, session_handle);
2276   if (PREDICT_FALSE (!s))
2277     return VPPCOM_EBADFD;
2278
2279   return vppcom_session_write_inline (wrk, s, buf, n,
2280                                       0 /* is_flush */ , s->is_dgram ? 1 : 0);
2281 }
2282
2283 int
2284 vppcom_session_write_msg (uint32_t session_handle, void *buf, size_t n)
2285 {
2286   vcl_worker_t *wrk = vcl_worker_get_current ();
2287   vcl_session_t *s;
2288
2289   s = vcl_session_get_w_handle (wrk, session_handle);
2290   if (PREDICT_FALSE (!s))
2291     return VPPCOM_EBADFD;
2292
2293   return vppcom_session_write_inline (wrk, s, buf, n,
2294                                       1 /* is_flush */ , s->is_dgram ? 1 : 0);
2295 }
2296
2297 #define vcl_fifo_rx_evt_valid_or_break(_s)                              \
2298 if (PREDICT_FALSE (!_s->rx_fifo))                                       \
2299   break;                                                                \
2300 if (PREDICT_FALSE (svm_fifo_is_empty (_s->rx_fifo)))                    \
2301   {                                                                     \
2302     if (!vcl_session_is_ct (_s))                                        \
2303       {                                                                 \
2304         svm_fifo_unset_event (_s->rx_fifo);                             \
2305         if (svm_fifo_is_empty (_s->rx_fifo))                            \
2306           break;                                                        \
2307       }                                                                 \
2308     else if (svm_fifo_is_empty (_s->ct_rx_fifo))                        \
2309       {                                                                 \
2310         svm_fifo_unset_event (_s->rx_fifo); /* rx evts on actual fifo*/ \
2311         if (svm_fifo_is_empty (_s->ct_rx_fifo))                         \
2312           break;                                                        \
2313       }                                                                 \
2314   }                                                                     \
2315
2316 static void
2317 vcl_select_handle_mq_event (vcl_worker_t * wrk, session_event_t * e,
2318                             unsigned long n_bits, unsigned long *read_map,
2319                             unsigned long *write_map,
2320                             unsigned long *except_map, u32 * bits_set)
2321 {
2322   session_disconnected_msg_t *disconnected_msg;
2323   session_connected_msg_t *connected_msg;
2324   vcl_session_t *s;
2325   u32 sid;
2326
2327   switch (e->event_type)
2328     {
2329     case SESSION_IO_EVT_RX:
2330       sid = e->session_index;
2331       s = vcl_session_get (wrk, sid);
2332       if (!s || !vcl_session_is_open (s))
2333         break;
2334       vcl_fifo_rx_evt_valid_or_break (s);
2335       if (sid < n_bits && read_map)
2336         {
2337           clib_bitmap_set_no_check ((uword *) read_map, sid, 1);
2338           *bits_set += 1;
2339         }
2340       break;
2341     case SESSION_IO_EVT_TX:
2342       sid = e->session_index;
2343       s = vcl_session_get (wrk, sid);
2344       if (!s || !vcl_session_is_open (s))
2345         break;
2346       if (sid < n_bits && write_map)
2347         {
2348           clib_bitmap_set_no_check ((uword *) write_map, sid, 1);
2349           *bits_set += 1;
2350         }
2351       break;
2352     case SESSION_CTRL_EVT_ACCEPTED:
2353       if (!e->postponed)
2354         s = vcl_session_accepted (wrk, (session_accepted_msg_t *) e->data);
2355       else
2356         s = vcl_session_get (wrk, e->session_index);
2357       if (!s)
2358         break;
2359       sid = s->session_index;
2360       if (sid < n_bits && read_map)
2361         {
2362           clib_bitmap_set_no_check ((uword *) read_map, sid, 1);
2363           *bits_set += 1;
2364         }
2365       break;
2366     case SESSION_CTRL_EVT_CONNECTED:
2367       if (!e->postponed)
2368         {
2369           connected_msg = (session_connected_msg_t *) e->data;
2370           sid = vcl_session_connected_handler (wrk, connected_msg);
2371         }
2372       else
2373         sid = e->session_index;
2374       if (sid == VCL_INVALID_SESSION_INDEX)
2375         break;
2376       if (sid < n_bits && write_map)
2377         {
2378           clib_bitmap_set_no_check ((uword *) write_map, sid, 1);
2379           *bits_set += 1;
2380         }
2381       break;
2382     case SESSION_CTRL_EVT_DISCONNECTED:
2383       disconnected_msg = (session_disconnected_msg_t *) e->data;
2384       s = vcl_session_disconnected_handler (wrk, disconnected_msg);
2385       if (!s)
2386         break;
2387       sid = s->session_index;
2388       if (sid < n_bits && except_map)
2389         {
2390           clib_bitmap_set_no_check ((uword *) except_map, sid, 1);
2391           *bits_set += 1;
2392         }
2393       break;
2394     case SESSION_CTRL_EVT_RESET:
2395       sid = vcl_session_reset_handler (wrk, (session_reset_msg_t *) e->data);
2396       if (sid < n_bits && except_map)
2397         {
2398           clib_bitmap_set_no_check ((uword *) except_map, sid, 1);
2399           *bits_set += 1;
2400         }
2401       break;
2402     case SESSION_CTRL_EVT_UNLISTEN_REPLY:
2403       vcl_session_unlisten_reply_handler (wrk, e->data);
2404       break;
2405     case SESSION_CTRL_EVT_MIGRATED:
2406       vcl_session_migrated_handler (wrk, e->data);
2407       break;
2408     case SESSION_CTRL_EVT_CLEANUP:
2409       vcl_session_cleanup_handler (wrk, e->data);
2410       break;
2411     case SESSION_CTRL_EVT_WORKER_UPDATE_REPLY:
2412       vcl_session_worker_update_reply_handler (wrk, e->data);
2413       break;
2414     case SESSION_CTRL_EVT_REQ_WORKER_UPDATE:
2415       vcl_session_req_worker_update_handler (wrk, e->data);
2416       break;
2417     case SESSION_CTRL_EVT_APP_ADD_SEGMENT:
2418       vcl_session_app_add_segment_handler (wrk, e->data);
2419       break;
2420     case SESSION_CTRL_EVT_APP_DEL_SEGMENT:
2421       vcl_session_app_del_segment_handler (wrk, e->data);
2422       break;
2423     case SESSION_CTRL_EVT_APP_WRK_RPC:
2424       vcl_worker_rpc_handler (wrk, e->data);
2425       break;
2426     default:
2427       clib_warning ("unhandled: %u", e->event_type);
2428       break;
2429     }
2430 }
2431
2432 static int
2433 vcl_select_handle_mq (vcl_worker_t * wrk, svm_msg_q_t * mq,
2434                       unsigned long n_bits, unsigned long *read_map,
2435                       unsigned long *write_map, unsigned long *except_map,
2436                       double time_to_wait, u32 * bits_set)
2437 {
2438   svm_msg_q_msg_t *msg;
2439   session_event_t *e;
2440   u32 i;
2441
2442   if (svm_msg_q_is_empty (mq))
2443     {
2444       if (*bits_set)
2445         return 0;
2446
2447       if (!time_to_wait)
2448         return 0;
2449       else if (time_to_wait < 0)
2450         svm_msg_q_wait (mq, SVM_MQ_WAIT_EMPTY);
2451       else
2452         {
2453           if (svm_msg_q_timedwait (mq, time_to_wait))
2454             return 0;
2455         }
2456     }
2457   vcl_mq_dequeue_batch (wrk, mq, ~0);
2458
2459   for (i = 0; i < vec_len (wrk->mq_msg_vector); i++)
2460     {
2461       msg = vec_elt_at_index (wrk->mq_msg_vector, i);
2462       e = svm_msg_q_msg_data (mq, msg);
2463       vcl_select_handle_mq_event (wrk, e, n_bits, read_map, write_map,
2464                                   except_map, bits_set);
2465       svm_msg_q_free_msg (mq, msg);
2466     }
2467   vec_reset_length (wrk->mq_msg_vector);
2468   vcl_handle_pending_wrk_updates (wrk);
2469   return *bits_set;
2470 }
2471
2472 static int
2473 vppcom_select_condvar (vcl_worker_t * wrk, int n_bits,
2474                        vcl_si_set * read_map, vcl_si_set * write_map,
2475                        vcl_si_set * except_map, double time_to_wait,
2476                        u32 * bits_set)
2477 {
2478   double wait = 0, start = 0;
2479
2480   if (!*bits_set)
2481     {
2482       wait = time_to_wait;
2483       start = clib_time_now (&wrk->clib_time);
2484     }
2485
2486   do
2487     {
2488       vcl_select_handle_mq (wrk, wrk->app_event_queue, n_bits, read_map,
2489                             write_map, except_map, wait, bits_set);
2490       if (*bits_set)
2491         return *bits_set;
2492       if (wait == -1)
2493         continue;
2494
2495       wait = wait - (clib_time_now (&wrk->clib_time) - start);
2496     }
2497   while (wait > 0);
2498
2499   return 0;
2500 }
2501
2502 static int
2503 vppcom_select_eventfd (vcl_worker_t * wrk, int n_bits,
2504                        vcl_si_set * read_map, vcl_si_set * write_map,
2505                        vcl_si_set * except_map, double time_to_wait,
2506                        u32 * bits_set)
2507 {
2508   vcl_mq_evt_conn_t *mqc;
2509   int __clib_unused n_read;
2510   int n_mq_evts, i;
2511   u64 buf;
2512
2513   if (PREDICT_FALSE (wrk->api_client_handle == ~0))
2514     {
2515       vcl_api_retry_attach (wrk);
2516       return 0;
2517     }
2518
2519   vec_validate (wrk->mq_events, pool_elts (wrk->mq_evt_conns));
2520   n_mq_evts = epoll_wait (wrk->mqs_epfd, wrk->mq_events,
2521                           vec_len (wrk->mq_events), time_to_wait);
2522   for (i = 0; i < n_mq_evts; i++)
2523     {
2524       if (PREDICT_FALSE (wrk->mq_events[i].data.u32 == ~0))
2525         {
2526           vcl_api_handle_disconnect (wrk);
2527           continue;
2528         }
2529
2530       mqc = vcl_mq_evt_conn_get (wrk, wrk->mq_events[i].data.u32);
2531       n_read = read (mqc->mq_fd, &buf, sizeof (buf));
2532       vcl_select_handle_mq (wrk, mqc->mq, n_bits, read_map, write_map,
2533                             except_map, 0, bits_set);
2534     }
2535
2536   return (n_mq_evts > 0 ? (int) *bits_set : 0);
2537 }
2538
2539 int
2540 vppcom_select (int n_bits, vcl_si_set * read_map, vcl_si_set * write_map,
2541                vcl_si_set * except_map, double time_to_wait)
2542 {
2543   u32 sid, minbits = clib_max (n_bits, BITS (uword)), bits_set = 0;
2544   vcl_worker_t *wrk = vcl_worker_get_current ();
2545   vcl_session_t *s = 0;
2546   int i;
2547
2548   if (n_bits && read_map)
2549     {
2550       clib_bitmap_validate (wrk->rd_bitmap, minbits);
2551       clib_memcpy_fast (wrk->rd_bitmap, read_map,
2552                         vec_len (wrk->rd_bitmap) * sizeof (vcl_si_set));
2553       memset (read_map, 0, vec_len (wrk->rd_bitmap) * sizeof (vcl_si_set));
2554     }
2555   if (n_bits && write_map)
2556     {
2557       clib_bitmap_validate (wrk->wr_bitmap, minbits);
2558       clib_memcpy_fast (wrk->wr_bitmap, write_map,
2559                         vec_len (wrk->wr_bitmap) * sizeof (vcl_si_set));
2560       memset (write_map, 0, vec_len (wrk->wr_bitmap) * sizeof (vcl_si_set));
2561     }
2562   if (n_bits && except_map)
2563     {
2564       clib_bitmap_validate (wrk->ex_bitmap, minbits);
2565       clib_memcpy_fast (wrk->ex_bitmap, except_map,
2566                         vec_len (wrk->ex_bitmap) * sizeof (vcl_si_set));
2567       memset (except_map, 0, vec_len (wrk->ex_bitmap) * sizeof (vcl_si_set));
2568     }
2569
2570   if (!n_bits)
2571     return 0;
2572
2573   if (!write_map)
2574     goto check_rd;
2575
2576   clib_bitmap_foreach (sid, wrk->wr_bitmap)
2577     {
2578       if (!(s = vcl_session_get (wrk, sid)))
2579         {
2580           clib_bitmap_set_no_check ((uword *) write_map, sid, 1);
2581           bits_set++;
2582           continue;
2583         }
2584
2585       if (vcl_session_write_ready (s))
2586         {
2587           clib_bitmap_set_no_check ((uword *) write_map, sid, 1);
2588           bits_set++;
2589         }
2590       else
2591         {
2592           svm_fifo_t *txf = vcl_session_is_ct (s) ? s->ct_tx_fifo : s->tx_fifo;
2593           svm_fifo_add_want_deq_ntf (txf, SVM_FIFO_WANT_DEQ_NOTIF);
2594         }
2595     }
2596
2597 check_rd:
2598   if (!read_map)
2599     goto check_mq;
2600
2601   clib_bitmap_foreach (sid, wrk->rd_bitmap)
2602     {
2603       if (!(s = vcl_session_get (wrk, sid)))
2604         {
2605           clib_bitmap_set_no_check ((uword *) read_map, sid, 1);
2606           bits_set++;
2607           continue;
2608         }
2609
2610       if (vcl_session_read_ready (s))
2611         {
2612           clib_bitmap_set_no_check ((uword *) read_map, sid, 1);
2613           bits_set++;
2614         }
2615     }
2616
2617 check_mq:
2618
2619   for (i = 0; i < vec_len (wrk->unhandled_evts_vector); i++)
2620     {
2621       vcl_select_handle_mq_event (wrk, &wrk->unhandled_evts_vector[i], n_bits,
2622                                   read_map, write_map, except_map, &bits_set);
2623     }
2624   vec_reset_length (wrk->unhandled_evts_vector);
2625
2626   if (vcm->cfg.use_mq_eventfd)
2627     vppcom_select_eventfd (wrk, n_bits, read_map, write_map, except_map,
2628                            time_to_wait, &bits_set);
2629   else
2630     vppcom_select_condvar (wrk, n_bits, read_map, write_map, except_map,
2631                            time_to_wait, &bits_set);
2632
2633   return (bits_set);
2634 }
2635
2636 static inline void
2637 vep_verify_epoll_chain (vcl_worker_t * wrk, u32 vep_handle)
2638 {
2639   vppcom_epoll_t *vep;
2640   u32 sh = vep_handle;
2641   vcl_session_t *s;
2642
2643   if (VPPCOM_DEBUG <= 2)
2644     return;
2645
2646   s = vcl_session_get_w_handle (wrk, vep_handle);
2647   if (PREDICT_FALSE (!s))
2648     {
2649       VDBG (0, "ERROR: Invalid vep_sh (%u)!", vep_handle);
2650       goto done;
2651     }
2652   if (PREDICT_FALSE (!(s->flags & VCL_SESSION_F_IS_VEP)))
2653     {
2654       VDBG (0, "ERROR: vep_sh (%u) is not a vep!", vep_handle);
2655       goto done;
2656     }
2657   vep = &s->vep;
2658   VDBG (0, "vep_sh (%u): Dumping epoll chain\n"
2659         "{\n"
2660         "   is_vep         = %u\n"
2661         "   is_vep_session = %u\n"
2662         "   next_sh        = 0x%x (%u)\n"
2663         "}\n", vep_handle, s->flags & VCL_SESSION_F_IS_VEP,
2664         s->flags & VCL_SESSION_F_IS_VEP_SESSION, vep->next_sh, vep->next_sh);
2665
2666   for (sh = vep->next_sh; sh != ~0; sh = vep->next_sh)
2667     {
2668       s = vcl_session_get_w_handle (wrk, sh);
2669       if (PREDICT_FALSE (!s))
2670         {
2671           VDBG (0, "ERROR: Invalid sh (%u)!", sh);
2672           goto done;
2673         }
2674       if (PREDICT_FALSE (s->flags & VCL_SESSION_F_IS_VEP))
2675         {
2676           VDBG (0, "ERROR: sh (%u) is a vep!", vep_handle);
2677         }
2678       else if (PREDICT_FALSE (!(s->flags & VCL_SESSION_F_IS_VEP_SESSION)))
2679         {
2680           VDBG (0, "ERROR: sh (%u) is not a vep session handle!", sh);
2681           goto done;
2682         }
2683       vep = &s->vep;
2684       if (PREDICT_FALSE (vep->vep_sh != vep_handle))
2685         VDBG (0, "ERROR: session (%u) vep_sh (%u) != vep_sh (%u)!",
2686               sh, s->vep.vep_sh, vep_handle);
2687       if (s->flags & VCL_SESSION_F_IS_VEP_SESSION)
2688         {
2689           VDBG (0, "vep_sh[%u]: sh 0x%x (%u)\n"
2690                 "{\n"
2691                 "   next_sh        = 0x%x (%u)\n"
2692                 "   prev_sh        = 0x%x (%u)\n"
2693                 "   vep_sh         = 0x%x (%u)\n"
2694                 "   ev.events      = 0x%x\n"
2695                 "   ev.data.u64    = 0x%llx\n"
2696                 "   et_mask        = 0x%x\n"
2697                 "}\n",
2698                 vep_handle, sh, sh, vep->next_sh, vep->next_sh, vep->prev_sh,
2699                 vep->prev_sh, vep->vep_sh, vep->vep_sh, vep->ev.events,
2700                 vep->ev.data.u64, vep->et_mask);
2701         }
2702     }
2703
2704 done:
2705   VDBG (0, "vep_sh (%u): Dump complete!\n", vep_handle);
2706 }
2707
2708 int
2709 vppcom_epoll_create (void)
2710 {
2711   vcl_worker_t *wrk = vcl_worker_get_current ();
2712   vcl_session_t *vep_session;
2713
2714   vep_session = vcl_session_alloc (wrk);
2715
2716   vep_session->flags |= VCL_SESSION_F_IS_VEP;
2717   vep_session->vep.vep_sh = ~0;
2718   vep_session->vep.next_sh = ~0;
2719   vep_session->vep.prev_sh = ~0;
2720   vep_session->vpp_handle = ~0;
2721
2722   vcl_evt (VCL_EVT_EPOLL_CREATE, vep_session, vep_session->session_index);
2723   VDBG (0, "Created vep_idx %u", vep_session->session_index);
2724
2725   return vcl_session_handle (vep_session);
2726 }
2727
2728 int
2729 vppcom_epoll_ctl (uint32_t vep_handle, int op, uint32_t session_handle,
2730                   struct epoll_event *event)
2731 {
2732   vcl_worker_t *wrk = vcl_worker_get_current ();
2733   int rv = VPPCOM_OK, add_evt = 0;
2734   vcl_session_t *vep_session;
2735   vcl_session_t *s;
2736   svm_fifo_t *txf;
2737
2738   if (vep_handle == session_handle)
2739     {
2740       VDBG (0, "vep_sh == session handle (%u)!", vep_handle);
2741       return VPPCOM_EINVAL;
2742     }
2743
2744   vep_session = vcl_session_get_w_handle (wrk, vep_handle);
2745   if (PREDICT_FALSE (!vep_session))
2746     {
2747       VDBG (0, "Invalid vep_sh (%u)!", vep_handle);
2748       return VPPCOM_EBADFD;
2749     }
2750   if (PREDICT_FALSE (!(vep_session->flags & VCL_SESSION_F_IS_VEP)))
2751     {
2752       VDBG (0, "vep_sh (%u) is not a vep!", vep_handle);
2753       return VPPCOM_EINVAL;
2754     }
2755
2756   ASSERT (vep_session->vep.vep_sh == ~0);
2757   ASSERT (vep_session->vep.prev_sh == ~0);
2758
2759   s = vcl_session_get_w_handle (wrk, session_handle);
2760   if (PREDICT_FALSE (!s))
2761     {
2762       VDBG (0, "Invalid session_handle (%u)!", session_handle);
2763       return VPPCOM_EBADFD;
2764     }
2765   if (PREDICT_FALSE (s->flags & VCL_SESSION_F_IS_VEP))
2766     {
2767       VDBG (0, "session_handle (%u) is a vep!", vep_handle);
2768       return VPPCOM_EINVAL;
2769     }
2770
2771   switch (op)
2772     {
2773     case EPOLL_CTL_ADD:
2774       if (PREDICT_FALSE (!event))
2775         {
2776           VDBG (0, "EPOLL_CTL_ADD: NULL pointer to epoll_event structure!");
2777           return VPPCOM_EINVAL;
2778         }
2779       if (s->flags & VCL_SESSION_F_IS_VEP_SESSION)
2780         {
2781           VDBG (0, "EPOLL_CTL_ADD: %u already epolled!", s->session_index);
2782           rv = VPPCOM_EEXIST;
2783           goto done;
2784         }
2785       if (vep_session->vep.next_sh != ~0)
2786         {
2787           vcl_session_t *next_session;
2788           next_session = vcl_session_get_w_handle (wrk,
2789                                                    vep_session->vep.next_sh);
2790           if (PREDICT_FALSE (!next_session))
2791             {
2792               VDBG (0, "EPOLL_CTL_ADD: Invalid vep.next_sh (%u) on "
2793                     "vep_idx (%u)!", vep_session->vep.next_sh, vep_handle);
2794               return VPPCOM_EBADFD;
2795             }
2796           ASSERT (next_session->vep.prev_sh == vep_handle);
2797           next_session->vep.prev_sh = session_handle;
2798         }
2799       s->vep.next_sh = vep_session->vep.next_sh;
2800       s->vep.prev_sh = vep_handle;
2801       s->vep.vep_sh = vep_handle;
2802       s->vep.et_mask = VEP_DEFAULT_ET_MASK;
2803       s->vep.lt_next = VCL_INVALID_SESSION_INDEX;
2804       s->vep.ev = *event;
2805       s->flags &= ~VCL_SESSION_F_IS_VEP;
2806       s->flags |= VCL_SESSION_F_IS_VEP_SESSION;
2807       vep_session->vep.next_sh = session_handle;
2808
2809       txf = vcl_session_is_ct (s) ? s->ct_tx_fifo : s->tx_fifo;
2810       if (txf && (event->events & EPOLLOUT))
2811         svm_fifo_add_want_deq_ntf (txf, SVM_FIFO_WANT_DEQ_NOTIF_IF_FULL);
2812
2813       /* Generate EPOLLOUT if tx fifo not full */
2814       if ((event->events & EPOLLOUT) && (vcl_session_write_ready (s) > 0))
2815         {
2816           session_event_t e = { 0 };
2817           e.event_type = SESSION_IO_EVT_TX;
2818           e.session_index = s->session_index;
2819           vec_add1 (wrk->unhandled_evts_vector, e);
2820           add_evt = 1;
2821         }
2822       /* Generate EPOLLIN if rx fifo has data */
2823       if ((event->events & EPOLLIN) && (vcl_session_read_ready (s) > 0))
2824         {
2825           session_event_t e = { 0 };
2826           e.event_type = SESSION_IO_EVT_RX;
2827           e.session_index = s->session_index;
2828           vec_add1 (wrk->unhandled_evts_vector, e);
2829           s->flags &= ~VCL_SESSION_F_HAS_RX_EVT;
2830           add_evt = 1;
2831         }
2832       if (!add_evt && vcl_session_is_closing (s))
2833         {
2834           session_event_t e = { 0 };
2835           if (s->session_state == VCL_STATE_VPP_CLOSING)
2836             e.event_type = SESSION_CTRL_EVT_DISCONNECTED;
2837           else
2838             e.event_type = SESSION_CTRL_EVT_RESET;
2839           e.session_index = s->session_index;
2840           e.postponed = 1;
2841           vec_add1 (wrk->unhandled_evts_vector, e);
2842         }
2843       VDBG (1, "EPOLL_CTL_ADD: vep_sh %u, sh %u, events 0x%x, data 0x%llx!",
2844             vep_handle, session_handle, event->events, event->data.u64);
2845       vcl_evt (VCL_EVT_EPOLL_CTLADD, s, event->events, event->data.u64);
2846       break;
2847
2848     case EPOLL_CTL_MOD:
2849       if (PREDICT_FALSE (!event))
2850         {
2851           VDBG (0, "EPOLL_CTL_MOD: NULL pointer to epoll_event structure!");
2852           rv = VPPCOM_EINVAL;
2853           goto done;
2854         }
2855       else if (PREDICT_FALSE (!(s->flags & VCL_SESSION_F_IS_VEP_SESSION)))
2856         {
2857           VDBG (0, "sh %u EPOLL_CTL_MOD: not a vep session!", session_handle);
2858           rv = VPPCOM_ENOENT;
2859           goto done;
2860         }
2861       else if (PREDICT_FALSE (s->vep.vep_sh != vep_handle))
2862         {
2863           VDBG (0, "EPOLL_CTL_MOD: sh %u vep_sh (%u) != vep_sh (%u)!",
2864                 session_handle, s->vep.vep_sh, vep_handle);
2865           rv = VPPCOM_EINVAL;
2866           goto done;
2867         }
2868
2869       /* Generate EPOLLOUT if session write ready nd event was not on */
2870       if ((event->events & EPOLLOUT) && !(s->vep.ev.events & EPOLLOUT) &&
2871           (vcl_session_write_ready (s) > 0))
2872         {
2873           session_event_t e = { 0 };
2874           e.event_type = SESSION_IO_EVT_TX;
2875           e.session_index = s->session_index;
2876           vec_add1 (wrk->unhandled_evts_vector, e);
2877         }
2878       /* Generate EPOLLIN if session read ready and event was not on */
2879       if ((event->events & EPOLLIN) && !(s->vep.ev.events & EPOLLIN) &&
2880           (vcl_session_read_ready (s) > 0))
2881         {
2882           session_event_t e = { 0 };
2883           e.event_type = SESSION_IO_EVT_RX;
2884           e.session_index = s->session_index;
2885           vec_add1 (wrk->unhandled_evts_vector, e);
2886           s->flags &= ~VCL_SESSION_F_HAS_RX_EVT;
2887         }
2888       s->vep.et_mask = VEP_DEFAULT_ET_MASK;
2889       s->vep.ev = *event;
2890       txf = vcl_session_is_ct (s) ? s->ct_tx_fifo : s->tx_fifo;
2891       if (txf)
2892         {
2893           if (event->events & EPOLLOUT)
2894             svm_fifo_add_want_deq_ntf (txf, SVM_FIFO_WANT_DEQ_NOTIF_IF_FULL);
2895           else
2896             svm_fifo_del_want_deq_ntf (txf, SVM_FIFO_WANT_DEQ_NOTIF_IF_FULL);
2897         }
2898       VDBG (1, "EPOLL_CTL_MOD: vep_sh %u, sh %u, events 0x%x, data 0x%llx!",
2899             vep_handle, session_handle, event->events, event->data.u64);
2900       break;
2901
2902     case EPOLL_CTL_DEL:
2903       if (PREDICT_FALSE (!(s->flags & VCL_SESSION_F_IS_VEP_SESSION)))
2904         {
2905           VDBG (0, "EPOLL_CTL_DEL: %u not a vep session!", session_handle);
2906           rv = VPPCOM_ENOENT;
2907           goto done;
2908         }
2909       else if (PREDICT_FALSE (s->vep.vep_sh != vep_handle))
2910         {
2911           VDBG (0, "EPOLL_CTL_DEL: sh %u vep_sh (%u) != vep_sh (%u)!",
2912                 session_handle, s->vep.vep_sh, vep_handle);
2913           rv = VPPCOM_EINVAL;
2914           goto done;
2915         }
2916
2917       if (s->vep.prev_sh == vep_handle)
2918         vep_session->vep.next_sh = s->vep.next_sh;
2919       else
2920         {
2921           vcl_session_t *prev_session;
2922           prev_session = vcl_session_get_w_handle (wrk, s->vep.prev_sh);
2923           if (PREDICT_FALSE (!prev_session))
2924             {
2925               VDBG (0, "EPOLL_CTL_DEL: Invalid prev_sh (%u) on sh (%u)!",
2926                     s->vep.prev_sh, session_handle);
2927               return VPPCOM_EBADFD;
2928             }
2929           ASSERT (prev_session->vep.next_sh == session_handle);
2930           prev_session->vep.next_sh = s->vep.next_sh;
2931         }
2932       if (s->vep.next_sh != ~0)
2933         {
2934           vcl_session_t *next_session;
2935           next_session = vcl_session_get_w_handle (wrk, s->vep.next_sh);
2936           if (PREDICT_FALSE (!next_session))
2937             {
2938               VDBG (0, "EPOLL_CTL_DEL: Invalid next_sh (%u) on sh (%u)!",
2939                     s->vep.next_sh, session_handle);
2940               return VPPCOM_EBADFD;
2941             }
2942           ASSERT (next_session->vep.prev_sh == session_handle);
2943           next_session->vep.prev_sh = s->vep.prev_sh;
2944         }
2945
2946       if (s->vep.lt_next != VCL_INVALID_SESSION_INDEX)
2947         vcl_epoll_lt_del (wrk, s);
2948
2949       memset (&s->vep, 0, sizeof (s->vep));
2950       s->vep.next_sh = ~0;
2951       s->vep.prev_sh = ~0;
2952       s->vep.vep_sh = ~0;
2953       s->vep.lt_next = VCL_INVALID_SESSION_INDEX;
2954       s->flags &= ~VCL_SESSION_F_IS_VEP_SESSION;
2955
2956       if (vcl_session_is_open (s))
2957         {
2958           txf = vcl_session_is_ct (s) ? s->ct_tx_fifo : s->tx_fifo;
2959           if (txf)
2960             svm_fifo_del_want_deq_ntf (txf, SVM_FIFO_WANT_DEQ_NOTIF_IF_FULL);
2961         }
2962
2963       VDBG (1, "EPOLL_CTL_DEL: vep_idx %u, sh %u!", vep_handle,
2964             session_handle);
2965       vcl_evt (VCL_EVT_EPOLL_CTLDEL, s, vep_sh);
2966       break;
2967
2968     default:
2969       VDBG (0, "Invalid operation (%d)!", op);
2970       rv = VPPCOM_EINVAL;
2971     }
2972
2973   vep_verify_epoll_chain (wrk, vep_handle);
2974
2975 done:
2976   return rv;
2977 }
2978
2979 static inline void
2980 vcl_epoll_wait_handle_mq_event (vcl_worker_t * wrk, session_event_t * e,
2981                                 struct epoll_event *events, u32 * num_ev)
2982 {
2983   session_disconnected_msg_t *disconnected_msg;
2984   session_connected_msg_t *connected_msg;
2985   u32 sid = ~0, session_events;
2986   u64 session_evt_data = ~0;
2987   vcl_session_t *s;
2988   u8 add_event = 0;
2989
2990   switch (e->event_type)
2991     {
2992     case SESSION_IO_EVT_RX:
2993       sid = e->session_index;
2994       s = vcl_session_get (wrk, sid);
2995       if (vcl_session_is_closed (s))
2996         break;
2997       vcl_fifo_rx_evt_valid_or_break (s);
2998       session_events = s->vep.ev.events;
2999       if (!(EPOLLIN & s->vep.ev.events) ||
3000           (s->flags & VCL_SESSION_F_HAS_RX_EVT) ||
3001           (s->vep.lt_next != VCL_INVALID_SESSION_INDEX))
3002         break;
3003       add_event = 1;
3004       events[*num_ev].events = EPOLLIN;
3005       session_evt_data = s->vep.ev.data.u64;
3006       s->flags |= VCL_SESSION_F_HAS_RX_EVT;
3007       break;
3008     case SESSION_IO_EVT_TX:
3009       sid = e->session_index;
3010       s = vcl_session_get (wrk, sid);
3011       if (!s || !vcl_session_is_open (s))
3012         break;
3013       session_events = s->vep.ev.events;
3014       if (!(EPOLLOUT & session_events))
3015         break;
3016       add_event = 1;
3017       events[*num_ev].events = EPOLLOUT;
3018       session_evt_data = s->vep.ev.data.u64;
3019       svm_fifo_reset_has_deq_ntf (vcl_session_is_ct (s) ?
3020                                   s->ct_tx_fifo : s->tx_fifo);
3021       break;
3022     case SESSION_CTRL_EVT_ACCEPTED:
3023       if (!e->postponed)
3024         s = vcl_session_accepted (wrk, (session_accepted_msg_t *) e->data);
3025       else
3026         s = vcl_session_get (wrk, e->session_index);
3027       if (!s)
3028         break;
3029       session_events = s->vep.ev.events;
3030       sid = s->session_index;
3031       if (!(EPOLLIN & session_events) ||
3032           (s->vep.lt_next != VCL_INVALID_SESSION_INDEX))
3033         break;
3034       add_event = 1;
3035       events[*num_ev].events = EPOLLIN;
3036       session_evt_data = s->vep.ev.data.u64;
3037       break;
3038     case SESSION_CTRL_EVT_CONNECTED:
3039       if (!e->postponed)
3040         {
3041           connected_msg = (session_connected_msg_t *) e->data;
3042           sid = vcl_session_connected_handler (wrk, connected_msg);
3043         }
3044       else
3045         sid = e->session_index;
3046       s = vcl_session_get (wrk, sid);
3047       if (vcl_session_is_closed (s))
3048         break;
3049       session_events = s->vep.ev.events;
3050       /* Generate EPOLLOUT because there's no connected event */
3051       if (!(EPOLLOUT & session_events))
3052         break;
3053       add_event = 1;
3054       events[*num_ev].events = EPOLLOUT;
3055       session_evt_data = s->vep.ev.data.u64;
3056       if (s->session_state == VCL_STATE_DETACHED)
3057         events[*num_ev].events |= EPOLLHUP;
3058       break;
3059     case SESSION_CTRL_EVT_DISCONNECTED:
3060       if (!e->postponed)
3061         {
3062           disconnected_msg = (session_disconnected_msg_t *) e->data;
3063           s = vcl_session_disconnected_handler (wrk, disconnected_msg);
3064         }
3065       else
3066         {
3067           s = vcl_session_get (wrk, e->session_index);
3068           s->flags &= ~VCL_SESSION_F_PENDING_DISCONNECT;
3069         }
3070       if (vcl_session_is_closed (s) ||
3071           !(s->flags & VCL_SESSION_F_IS_VEP_SESSION))
3072         {
3073           if (s && (s->flags & VCL_SESSION_F_PENDING_FREE))
3074             vcl_session_free (wrk, s);
3075           break;
3076         }
3077       sid = s->session_index;
3078       session_events = s->vep.ev.events;
3079       add_event = 1;
3080       if (EPOLLRDHUP & session_events)
3081         {
3082           /* If app can distinguish between RDHUP and HUP,
3083            * we make finer control */
3084           events[*num_ev].events = EPOLLRDHUP;
3085           if (s->flags & VCL_SESSION_F_WR_SHUTDOWN)
3086             {
3087               events[*num_ev].events |= EPOLLHUP;
3088             }
3089         }
3090       else
3091         {
3092           events[*num_ev].events = EPOLLHUP;
3093         }
3094       session_evt_data = s->vep.ev.data.u64;
3095
3096       break;
3097     case SESSION_CTRL_EVT_RESET:
3098       if (!e->postponed)
3099         {
3100           sid =
3101             vcl_session_reset_handler (wrk, (session_reset_msg_t *) e->data);
3102           s = vcl_session_get (wrk, sid);
3103         }
3104       else
3105         {
3106           sid = e->session_index;
3107           s = vcl_session_get (wrk, sid);
3108           s->flags &= ~VCL_SESSION_F_PENDING_DISCONNECT;
3109         }
3110       if (vcl_session_is_closed (s) ||
3111           !(s->flags & VCL_SESSION_F_IS_VEP_SESSION))
3112         {
3113           if (s && (s->flags & VCL_SESSION_F_PENDING_FREE))
3114             vcl_session_free (wrk, s);
3115           break;
3116         }
3117       session_events = s->vep.ev.events;
3118       add_event = 1;
3119       events[*num_ev].events = EPOLLHUP | EPOLLRDHUP;
3120       session_evt_data = s->vep.ev.data.u64;
3121       break;
3122     case SESSION_CTRL_EVT_UNLISTEN_REPLY:
3123       vcl_session_unlisten_reply_handler (wrk, e->data);
3124       break;
3125     case SESSION_CTRL_EVT_MIGRATED:
3126       vcl_session_migrated_handler (wrk, e->data);
3127       break;
3128     case SESSION_CTRL_EVT_CLEANUP:
3129       vcl_session_cleanup_handler (wrk, e->data);
3130       break;
3131     case SESSION_CTRL_EVT_REQ_WORKER_UPDATE:
3132       vcl_session_req_worker_update_handler (wrk, e->data);
3133       break;
3134     case SESSION_CTRL_EVT_WORKER_UPDATE_REPLY:
3135       vcl_session_worker_update_reply_handler (wrk, e->data);
3136       break;
3137     case SESSION_CTRL_EVT_APP_ADD_SEGMENT:
3138       vcl_session_app_add_segment_handler (wrk, e->data);
3139       break;
3140     case SESSION_CTRL_EVT_APP_DEL_SEGMENT:
3141       vcl_session_app_del_segment_handler (wrk, e->data);
3142       break;
3143     case SESSION_CTRL_EVT_APP_WRK_RPC:
3144       vcl_worker_rpc_handler (wrk, e->data);
3145       break;
3146     default:
3147       VDBG (0, "unhandled: %u", e->event_type);
3148       break;
3149     }
3150
3151   if (add_event)
3152     {
3153       events[*num_ev].data.u64 = session_evt_data;
3154       if (EPOLLONESHOT & session_events)
3155         {
3156           s = vcl_session_get (wrk, sid);
3157           s->vep.ev.events = 0;
3158         }
3159       else if (!(EPOLLET & session_events))
3160         {
3161           s = vcl_session_get (wrk, sid);
3162           if (s->vep.lt_next == VCL_INVALID_SESSION_INDEX)
3163             vcl_epoll_lt_add (wrk, s);
3164         }
3165       *num_ev += 1;
3166     }
3167 }
3168
3169 static int
3170 vcl_epoll_wait_handle_mq (vcl_worker_t * wrk, svm_msg_q_t * mq,
3171                           struct epoll_event *events, u32 maxevents,
3172                           double wait_for_time, u32 * num_ev)
3173 {
3174   svm_msg_q_msg_t *msg;
3175   session_event_t *e;
3176   int i;
3177
3178   if (vec_len (wrk->mq_msg_vector) && svm_msg_q_is_empty (mq))
3179     goto handle_dequeued;
3180
3181   if (svm_msg_q_is_empty (mq))
3182     {
3183       if (!wait_for_time)
3184         return 0;
3185       else if (wait_for_time < 0)
3186         svm_msg_q_wait (mq, SVM_MQ_WAIT_EMPTY);
3187       else
3188         {
3189           if (svm_msg_q_timedwait (mq, wait_for_time / 1e3))
3190             return 0;
3191         }
3192     }
3193   ASSERT (maxevents > *num_ev);
3194   vcl_mq_dequeue_batch (wrk, mq, ~0);
3195
3196 handle_dequeued:
3197   for (i = 0; i < vec_len (wrk->mq_msg_vector); i++)
3198     {
3199       msg = vec_elt_at_index (wrk->mq_msg_vector, i);
3200       e = svm_msg_q_msg_data (mq, msg);
3201       if (*num_ev < maxevents)
3202         vcl_epoll_wait_handle_mq_event (wrk, e, events, num_ev);
3203       else
3204         vcl_handle_mq_event (wrk, e);
3205       svm_msg_q_free_msg (mq, msg);
3206     }
3207   vec_reset_length (wrk->mq_msg_vector);
3208   vcl_handle_pending_wrk_updates (wrk);
3209   return *num_ev;
3210 }
3211
3212 static int
3213 vppcom_epoll_wait_condvar (vcl_worker_t *wrk, struct epoll_event *events,
3214                            int maxevents, u32 n_evts, double timeout_ms)
3215 {
3216   double end = -1;
3217
3218   if (!n_evts)
3219     {
3220       if (timeout_ms > 0)
3221         end = clib_time_now (&wrk->clib_time) + (timeout_ms / 1e3);
3222     }
3223
3224   do
3225     {
3226       vcl_epoll_wait_handle_mq (wrk, wrk->app_event_queue, events, maxevents,
3227                                 timeout_ms, &n_evts);
3228       if (n_evts || !timeout_ms)
3229         return n_evts;
3230     }
3231   while (end == -1 || clib_time_now (&wrk->clib_time) < end);
3232
3233   return 0;
3234 }
3235
3236 static int
3237 vppcom_epoll_wait_eventfd (vcl_worker_t *wrk, struct epoll_event *events,
3238                            int maxevents, u32 n_evts, double timeout_ms)
3239 {
3240   int __clib_unused n_read;
3241   vcl_mq_evt_conn_t *mqc;
3242   int n_mq_evts, i;
3243   double end = -1;
3244   u64 buf;
3245
3246   if (PREDICT_FALSE (wrk->api_client_handle == ~0))
3247     {
3248       vcl_api_retry_attach (wrk);
3249       return n_evts;
3250     }
3251
3252   vec_validate (wrk->mq_events, pool_elts (wrk->mq_evt_conns));
3253   if (!n_evts)
3254     {
3255       if (timeout_ms > 0)
3256         end = clib_time_now (&wrk->clib_time) + (timeout_ms / 1e3);
3257     }
3258
3259   do
3260     {
3261       n_mq_evts = epoll_wait (wrk->mqs_epfd, wrk->mq_events,
3262                               vec_len (wrk->mq_events), timeout_ms);
3263       if (n_mq_evts < 0)
3264         {
3265           VDBG (0, "epoll_wait error %u", errno);
3266           return n_evts;
3267         }
3268
3269       for (i = 0; i < n_mq_evts; i++)
3270         {
3271           if (PREDICT_FALSE (wrk->mq_events[i].data.u32 == ~0))
3272             {
3273               /* api socket was closed */
3274               vcl_api_handle_disconnect (wrk);
3275               continue;
3276             }
3277
3278           mqc = vcl_mq_evt_conn_get (wrk, wrk->mq_events[i].data.u32);
3279           n_read = read (mqc->mq_fd, &buf, sizeof (buf));
3280           vcl_epoll_wait_handle_mq (wrk, mqc->mq, events, maxevents, 0,
3281                                     &n_evts);
3282         }
3283
3284       if (n_evts || !timeout_ms)
3285         return n_evts;
3286     }
3287   while (end == -1 || clib_time_now (&wrk->clib_time) < end);
3288
3289   return 0;
3290 }
3291
3292 static void
3293 vcl_epoll_wait_handle_lt (vcl_worker_t *wrk, struct epoll_event *events,
3294                           int maxevents, u32 *n_evts)
3295 {
3296   u32 add_event = 0, next;
3297   vcl_session_t *s;
3298   u64 evt_data;
3299   int rv;
3300
3301   ASSERT (wrk->ep_lt_current != VCL_INVALID_SESSION_INDEX);
3302   if (*n_evts >= maxevents)
3303     return;
3304
3305   next = wrk->ep_lt_current;
3306   do
3307     {
3308       s = vcl_session_get (wrk, next);
3309       next = s->vep.lt_next;
3310
3311       if ((s->vep.ev.events & EPOLLIN) && (rv = vcl_session_read_ready (s)))
3312         {
3313           add_event = 1;
3314           events[*n_evts].events |= rv > 0 ? EPOLLIN : EPOLLHUP | EPOLLRDHUP;
3315           evt_data = s->vep.ev.data.u64;
3316         }
3317       if ((s->vep.ev.events & EPOLLOUT) && (rv = vcl_session_write_ready (s)))
3318         {
3319           add_event = 1;
3320           events[*n_evts].events |= rv > 0 ? EPOLLOUT : EPOLLHUP | EPOLLRDHUP;
3321           evt_data = s->vep.ev.data.u64;
3322         }
3323       if (!add_event && s->session_state > VCL_STATE_READY)
3324         {
3325           add_event = 1;
3326           events[*n_evts].events |= EPOLLHUP | EPOLLRDHUP;
3327           evt_data = s->vep.ev.data.u64;
3328         }
3329       if (add_event)
3330         {
3331           events[*n_evts].data.u64 = evt_data;
3332           *n_evts += 1;
3333           add_event = 0;
3334           if (EPOLLONESHOT & s->vep.ev.events)
3335             s->vep.ev.events = 0;
3336           if (*n_evts == maxevents)
3337             {
3338               wrk->ep_lt_current = next;
3339               break;
3340             }
3341         }
3342       else
3343         {
3344           vcl_epoll_lt_del (wrk, s);
3345           if (wrk->ep_lt_current == VCL_INVALID_SESSION_INDEX)
3346             break;
3347         }
3348     }
3349   while (next != wrk->ep_lt_current);
3350 }
3351
3352 int
3353 vppcom_epoll_wait (uint32_t vep_handle, struct epoll_event *events,
3354                    int maxevents, double wait_for_time)
3355 {
3356   vcl_worker_t *wrk = vcl_worker_get_current ();
3357   vcl_session_t *vep_session;
3358   u32 n_evts = 0;
3359   int i;
3360
3361   if (PREDICT_FALSE (maxevents <= 0))
3362     {
3363       VDBG (0, "ERROR: Invalid maxevents (%d)!", maxevents);
3364       return VPPCOM_EINVAL;
3365     }
3366
3367   vep_session = vcl_session_get_w_handle (wrk, vep_handle);
3368   if (!vep_session)
3369     return VPPCOM_EBADFD;
3370
3371   if (PREDICT_FALSE (!(vep_session->flags & VCL_SESSION_F_IS_VEP)))
3372     {
3373       VDBG (0, "ERROR: vep_idx (%u) is not a vep!", vep_handle);
3374       return VPPCOM_EINVAL;
3375     }
3376
3377   if (vec_len (wrk->unhandled_evts_vector))
3378     {
3379       for (i = 0; i < vec_len (wrk->unhandled_evts_vector); i++)
3380         {
3381           vcl_epoll_wait_handle_mq_event (wrk, &wrk->unhandled_evts_vector[i],
3382                                           events, &n_evts);
3383           if (n_evts == maxevents)
3384             {
3385               vec_delete (wrk->unhandled_evts_vector, i + 1, 0);
3386               return n_evts;
3387             }
3388         }
3389       vec_reset_length (wrk->unhandled_evts_vector);
3390     }
3391
3392   if (PREDICT_FALSE (wrk->ep_lt_current != VCL_INVALID_SESSION_INDEX))
3393     vcl_epoll_wait_handle_lt (wrk, events, maxevents, &n_evts);
3394
3395   /* Request to only drain unhandled */
3396   if ((int) wait_for_time == -2)
3397     return n_evts;
3398
3399
3400   if (vcm->cfg.use_mq_eventfd)
3401     n_evts = vppcom_epoll_wait_eventfd (wrk, events, maxevents, n_evts,
3402                                         wait_for_time);
3403   else
3404     n_evts = vppcom_epoll_wait_condvar (wrk, events, maxevents, n_evts,
3405                                         wait_for_time);
3406
3407   return n_evts;
3408 }
3409
3410 int
3411 vppcom_session_attr (uint32_t session_handle, uint32_t op,
3412                      void *buffer, uint32_t * buflen)
3413 {
3414   vcl_worker_t *wrk = vcl_worker_get_current ();
3415   u32 *flags = buffer;
3416   vppcom_endpt_t *ep = buffer;
3417   transport_endpt_attr_t tea;
3418   vcl_session_t *session;
3419   int rv = VPPCOM_OK;
3420
3421   session = vcl_session_get_w_handle (wrk, session_handle);
3422   if (!session)
3423     return VPPCOM_EBADFD;
3424
3425   switch (op)
3426     {
3427     case VPPCOM_ATTR_GET_NREAD:
3428       rv = vcl_session_read_ready (session);
3429       VDBG (2, "VPPCOM_ATTR_GET_NREAD: sh %u, nread = %d", session_handle,
3430             rv);
3431       break;
3432
3433     case VPPCOM_ATTR_GET_NWRITE:
3434       rv = vcl_session_write_ready (session);
3435       VDBG (2, "VPPCOM_ATTR_GET_NWRITE: sh %u, nwrite = %d", session_handle,
3436             rv);
3437       break;
3438
3439     case VPPCOM_ATTR_GET_FLAGS:
3440       if (PREDICT_TRUE (buffer && buflen && (*buflen >= sizeof (*flags))))
3441         {
3442           *flags =
3443             O_RDWR |
3444             (vcl_session_has_attr (session, VCL_SESS_ATTR_NONBLOCK) ?
3445              O_NONBLOCK : 0);
3446           *buflen = sizeof (*flags);
3447           VDBG (2, "VPPCOM_ATTR_GET_FLAGS: sh %u, flags = 0x%08x, "
3448                 "is_nonblocking = %u", session_handle, *flags,
3449                 vcl_session_has_attr (session, VCL_SESS_ATTR_NONBLOCK));
3450         }
3451       else
3452         rv = VPPCOM_EINVAL;
3453       break;
3454
3455     case VPPCOM_ATTR_SET_FLAGS:
3456       if (PREDICT_TRUE (buffer && buflen && (*buflen == sizeof (*flags))))
3457         {
3458           if (*flags & O_NONBLOCK)
3459             vcl_session_set_attr (session, VCL_SESS_ATTR_NONBLOCK);
3460           else
3461             vcl_session_clear_attr (session, VCL_SESS_ATTR_NONBLOCK);
3462
3463           VDBG (2, "VPPCOM_ATTR_SET_FLAGS: sh %u, flags = 0x%08x,"
3464                 " is_nonblocking = %u", session_handle, *flags,
3465                 vcl_session_has_attr (session, VCL_SESS_ATTR_NONBLOCK));
3466         }
3467       else
3468         rv = VPPCOM_EINVAL;
3469       break;
3470
3471     case VPPCOM_ATTR_GET_PEER_ADDR:
3472       if (PREDICT_TRUE (buffer && buflen &&
3473                         (*buflen >= sizeof (*ep)) && ep->ip))
3474         {
3475           ep->is_ip4 = session->transport.is_ip4;
3476           ep->port = session->transport.rmt_port;
3477           if (session->transport.is_ip4)
3478             clib_memcpy_fast (ep->ip, &session->transport.rmt_ip.ip4,
3479                               sizeof (ip4_address_t));
3480           else
3481             clib_memcpy_fast (ep->ip, &session->transport.rmt_ip.ip6,
3482                               sizeof (ip6_address_t));
3483           *buflen = sizeof (*ep);
3484           VDBG (1,
3485                 "VPPCOM_ATTR_GET_PEER_ADDR: sh %u, is_ip4 = %u, "
3486                 "addr = %U, port %u",
3487                 session_handle, ep->is_ip4, vcl_format_ip46_address,
3488                 &session->transport.rmt_ip,
3489                 ep->is_ip4 ? IP46_TYPE_IP4 : IP46_TYPE_IP6,
3490                 clib_net_to_host_u16 (ep->port));
3491         }
3492       else
3493         rv = VPPCOM_EINVAL;
3494       break;
3495
3496     case VPPCOM_ATTR_GET_LCL_ADDR:
3497       if (PREDICT_TRUE (buffer && buflen &&
3498                         (*buflen >= sizeof (*ep)) && ep->ip))
3499         {
3500           ep->is_ip4 = session->transport.is_ip4;
3501           ep->port = session->transport.lcl_port;
3502           if (session->transport.is_ip4)
3503             clib_memcpy_fast (ep->ip, &session->transport.lcl_ip.ip4,
3504                               sizeof (ip4_address_t));
3505           else
3506             clib_memcpy_fast (ep->ip, &session->transport.lcl_ip.ip6,
3507                               sizeof (ip6_address_t));
3508           *buflen = sizeof (*ep);
3509           VDBG (1,
3510                 "VPPCOM_ATTR_GET_LCL_ADDR: sh %u, is_ip4 = %u, addr = %U"
3511                 " port %d",
3512                 session_handle, ep->is_ip4, vcl_format_ip46_address,
3513                 &session->transport.lcl_ip,
3514                 ep->is_ip4 ? IP46_TYPE_IP4 : IP46_TYPE_IP6,
3515                 clib_net_to_host_u16 (ep->port));
3516         }
3517       else
3518         rv = VPPCOM_EINVAL;
3519       break;
3520
3521     case VPPCOM_ATTR_SET_LCL_ADDR:
3522       if (PREDICT_TRUE (buffer && buflen &&
3523                         (*buflen >= sizeof (*ep)) && ep->ip))
3524         {
3525           session->transport.is_ip4 = ep->is_ip4;
3526           session->transport.lcl_port = ep->port;
3527           vcl_ip_copy_from_ep (&session->transport.lcl_ip, ep);
3528           *buflen = sizeof (*ep);
3529           VDBG (1,
3530                 "VPPCOM_ATTR_SET_LCL_ADDR: sh %u, is_ip4 = %u, addr = %U"
3531                 " port %d",
3532                 session_handle, ep->is_ip4, vcl_format_ip46_address,
3533                 &session->transport.lcl_ip,
3534                 ep->is_ip4 ? IP46_TYPE_IP4 : IP46_TYPE_IP6,
3535                 clib_net_to_host_u16 (ep->port));
3536         }
3537       else
3538         rv = VPPCOM_EINVAL;
3539       break;
3540
3541     case VPPCOM_ATTR_GET_LIBC_EPFD:
3542       rv = session->libc_epfd;
3543       VDBG (2, "VPPCOM_ATTR_GET_LIBC_EPFD: libc_epfd %d", rv);
3544       break;
3545
3546     case VPPCOM_ATTR_SET_LIBC_EPFD:
3547       if (PREDICT_TRUE (buffer && buflen &&
3548                         (*buflen == sizeof (session->libc_epfd))))
3549         {
3550           session->libc_epfd = *(int *) buffer;
3551           *buflen = sizeof (session->libc_epfd);
3552
3553           VDBG (2, "VPPCOM_ATTR_SET_LIBC_EPFD: libc_epfd %d, buflen %d",
3554                 session->libc_epfd, *buflen);
3555         }
3556       else
3557         rv = VPPCOM_EINVAL;
3558       break;
3559
3560     case VPPCOM_ATTR_GET_PROTOCOL:
3561       if (buffer && buflen && (*buflen >= sizeof (int)))
3562         {
3563           *(int *) buffer = session->session_type;
3564           *buflen = sizeof (int);
3565
3566           VDBG (2, "VPPCOM_ATTR_GET_PROTOCOL: %d (%s), buflen %d",
3567                 *(int *) buffer, *(int *) buffer ? "UDP" : "TCP", *buflen);
3568         }
3569       else
3570         rv = VPPCOM_EINVAL;
3571       break;
3572
3573     case VPPCOM_ATTR_GET_LISTEN:
3574       if (buffer && buflen && (*buflen >= sizeof (int)))
3575         {
3576           *(int *) buffer = vcl_session_has_attr (session,
3577                                                   VCL_SESS_ATTR_LISTEN);
3578           *buflen = sizeof (int);
3579
3580           VDBG (2, "VPPCOM_ATTR_GET_LISTEN: %d, buflen %d", *(int *) buffer,
3581                 *buflen);
3582         }
3583       else
3584         rv = VPPCOM_EINVAL;
3585       break;
3586
3587     case VPPCOM_ATTR_GET_ERROR:
3588       if (buffer && buflen && (*buflen >= sizeof (int)))
3589         {
3590           *(int *) buffer = 0;
3591           *buflen = sizeof (int);
3592
3593           VDBG (2, "VPPCOM_ATTR_GET_ERROR: %d, buflen %d, #VPP-TBD#",
3594                 *(int *) buffer, *buflen);
3595         }
3596       else
3597         rv = VPPCOM_EINVAL;
3598       break;
3599
3600     case VPPCOM_ATTR_GET_TX_FIFO_LEN:
3601       if (buffer && buflen && (*buflen >= sizeof (u32)))
3602         {
3603
3604           /* VPP-TBD */
3605           *(size_t *) buffer = (session->sndbuf_size ? session->sndbuf_size :
3606                                 session->tx_fifo ?
3607                                 svm_fifo_size (session->tx_fifo) :
3608                                 vcm->cfg.tx_fifo_size);
3609           *buflen = sizeof (u32);
3610
3611           VDBG (2, "VPPCOM_ATTR_GET_TX_FIFO_LEN: %u (0x%x), buflen %d,"
3612                 " #VPP-TBD#", *(size_t *) buffer, *(size_t *) buffer,
3613                 *buflen);
3614         }
3615       else
3616         rv = VPPCOM_EINVAL;
3617       break;
3618
3619     case VPPCOM_ATTR_SET_DSCP:
3620       if (buffer && buflen && (*buflen >= sizeof (u8)))
3621         {
3622           session->dscp = *(u8 *) buffer;
3623
3624           VDBG (2, "VPPCOM_ATTR_SET_DSCP: %u (0x%x), buflen %d,",
3625                 *(u8 *) buffer, *(u8 *) buffer, *buflen);
3626         }
3627       else
3628         rv = VPPCOM_EINVAL;
3629       break;
3630
3631     case VPPCOM_ATTR_SET_TX_FIFO_LEN:
3632       if (buffer && buflen && (*buflen == sizeof (u32)))
3633         {
3634           /* VPP-TBD */
3635           session->sndbuf_size = *(u32 *) buffer;
3636           VDBG (2, "VPPCOM_ATTR_SET_TX_FIFO_LEN: %u (0x%x), buflen %d,"
3637                 " #VPP-TBD#", session->sndbuf_size, session->sndbuf_size,
3638                 *buflen);
3639         }
3640       else
3641         rv = VPPCOM_EINVAL;
3642       break;
3643
3644     case VPPCOM_ATTR_GET_RX_FIFO_LEN:
3645       if (buffer && buflen && (*buflen >= sizeof (u32)))
3646         {
3647
3648           /* VPP-TBD */
3649           *(size_t *) buffer = (session->rcvbuf_size ? session->rcvbuf_size :
3650                                 session->rx_fifo ?
3651                                 svm_fifo_size (session->rx_fifo) :
3652                                 vcm->cfg.rx_fifo_size);
3653           *buflen = sizeof (u32);
3654
3655           VDBG (2, "VPPCOM_ATTR_GET_RX_FIFO_LEN: %u (0x%x), buflen %d, "
3656                 "#VPP-TBD#", *(size_t *) buffer, *(size_t *) buffer, *buflen);
3657         }
3658       else
3659         rv = VPPCOM_EINVAL;
3660       break;
3661
3662     case VPPCOM_ATTR_SET_RX_FIFO_LEN:
3663       if (buffer && buflen && (*buflen == sizeof (u32)))
3664         {
3665           /* VPP-TBD */
3666           session->rcvbuf_size = *(u32 *) buffer;
3667           VDBG (2, "VPPCOM_ATTR_SET_RX_FIFO_LEN: %u (0x%x), buflen %d,"
3668                 " #VPP-TBD#", session->sndbuf_size, session->sndbuf_size,
3669                 *buflen);
3670         }
3671       else
3672         rv = VPPCOM_EINVAL;
3673       break;
3674
3675     case VPPCOM_ATTR_GET_REUSEADDR:
3676       if (buffer && buflen && (*buflen >= sizeof (int)))
3677         {
3678           /* VPP-TBD */
3679           *(int *) buffer = vcl_session_has_attr (session,
3680                                                   VCL_SESS_ATTR_REUSEADDR);
3681           *buflen = sizeof (int);
3682
3683           VDBG (2, "VPPCOM_ATTR_GET_REUSEADDR: %d, buflen %d, #VPP-TBD#",
3684                 *(int *) buffer, *buflen);
3685         }
3686       else
3687         rv = VPPCOM_EINVAL;
3688       break;
3689
3690     case VPPCOM_ATTR_SET_REUSEADDR:
3691       if (buffer && buflen && (*buflen == sizeof (int)) &&
3692           !vcl_session_has_attr (session, VCL_SESS_ATTR_LISTEN))
3693         {
3694           /* VPP-TBD */
3695           if (*(int *) buffer)
3696             vcl_session_set_attr (session, VCL_SESS_ATTR_REUSEADDR);
3697           else
3698             vcl_session_clear_attr (session, VCL_SESS_ATTR_REUSEADDR);
3699
3700           VDBG (2, "VPPCOM_ATTR_SET_REUSEADDR: %d, buflen %d, #VPP-TBD#",
3701                 vcl_session_has_attr (session, VCL_SESS_ATTR_REUSEADDR),
3702                 *buflen);
3703         }
3704       else
3705         rv = VPPCOM_EINVAL;
3706       break;
3707
3708     case VPPCOM_ATTR_GET_REUSEPORT:
3709       if (buffer && buflen && (*buflen >= sizeof (int)))
3710         {
3711           /* VPP-TBD */
3712           *(int *) buffer = vcl_session_has_attr (session,
3713                                                   VCL_SESS_ATTR_REUSEPORT);
3714           *buflen = sizeof (int);
3715
3716           VDBG (2, "VPPCOM_ATTR_GET_REUSEPORT: %d, buflen %d, #VPP-TBD#",
3717                 *(int *) buffer, *buflen);
3718         }
3719       else
3720         rv = VPPCOM_EINVAL;
3721       break;
3722
3723     case VPPCOM_ATTR_SET_REUSEPORT:
3724       if (buffer && buflen && (*buflen == sizeof (int)) &&
3725           !vcl_session_has_attr (session, VCL_SESS_ATTR_LISTEN))
3726         {
3727           /* VPP-TBD */
3728           if (*(int *) buffer)
3729             vcl_session_set_attr (session, VCL_SESS_ATTR_REUSEPORT);
3730           else
3731             vcl_session_clear_attr (session, VCL_SESS_ATTR_REUSEPORT);
3732
3733           VDBG (2, "VPPCOM_ATTR_SET_REUSEPORT: %d, buflen %d, #VPP-TBD#",
3734                 vcl_session_has_attr (session, VCL_SESS_ATTR_REUSEPORT),
3735                 *buflen);
3736         }
3737       else
3738         rv = VPPCOM_EINVAL;
3739       break;
3740
3741     case VPPCOM_ATTR_GET_BROADCAST:
3742       if (buffer && buflen && (*buflen >= sizeof (int)))
3743         {
3744           /* VPP-TBD */
3745           *(int *) buffer = vcl_session_has_attr (session,
3746                                                   VCL_SESS_ATTR_BROADCAST);
3747           *buflen = sizeof (int);
3748
3749           VDBG (2, "VPPCOM_ATTR_GET_BROADCAST: %d, buflen %d, #VPP-TBD#",
3750                 *(int *) buffer, *buflen);
3751         }
3752       else
3753         rv = VPPCOM_EINVAL;
3754       break;
3755
3756     case VPPCOM_ATTR_SET_BROADCAST:
3757       if (buffer && buflen && (*buflen == sizeof (int)))
3758         {
3759           /* VPP-TBD */
3760           if (*(int *) buffer)
3761             vcl_session_set_attr (session, VCL_SESS_ATTR_BROADCAST);
3762           else
3763             vcl_session_clear_attr (session, VCL_SESS_ATTR_BROADCAST);
3764
3765           VDBG (2, "VPPCOM_ATTR_SET_BROADCAST: %d, buflen %d, #VPP-TBD#",
3766                 vcl_session_has_attr (session, VCL_SESS_ATTR_BROADCAST),
3767                 *buflen);
3768         }
3769       else
3770         rv = VPPCOM_EINVAL;
3771       break;
3772
3773     case VPPCOM_ATTR_GET_V6ONLY:
3774       if (buffer && buflen && (*buflen >= sizeof (int)))
3775         {
3776           /* VPP-TBD */
3777           *(int *) buffer = vcl_session_has_attr (session,
3778                                                   VCL_SESS_ATTR_V6ONLY);
3779           *buflen = sizeof (int);
3780
3781           VDBG (2, "VPPCOM_ATTR_GET_V6ONLY: %d, buflen %d, #VPP-TBD#",
3782                 *(int *) buffer, *buflen);
3783         }
3784       else
3785         rv = VPPCOM_EINVAL;
3786       break;
3787
3788     case VPPCOM_ATTR_SET_V6ONLY:
3789       if (buffer && buflen && (*buflen == sizeof (int)))
3790         {
3791           /* VPP-TBD */
3792           if (*(int *) buffer)
3793             vcl_session_set_attr (session, VCL_SESS_ATTR_V6ONLY);
3794           else
3795             vcl_session_clear_attr (session, VCL_SESS_ATTR_V6ONLY);
3796
3797           VDBG (2, "VPPCOM_ATTR_SET_V6ONLY: %d, buflen %d, #VPP-TBD#",
3798                 vcl_session_has_attr (session, VCL_SESS_ATTR_V6ONLY),
3799                 *buflen);
3800         }
3801       else
3802         rv = VPPCOM_EINVAL;
3803       break;
3804
3805     case VPPCOM_ATTR_GET_KEEPALIVE:
3806       if (buffer && buflen && (*buflen >= sizeof (int)))
3807         {
3808           /* VPP-TBD */
3809           *(int *) buffer = vcl_session_has_attr (session,
3810                                                   VCL_SESS_ATTR_KEEPALIVE);
3811           *buflen = sizeof (int);
3812
3813           VDBG (2, "VPPCOM_ATTR_GET_KEEPALIVE: %d, buflen %d, #VPP-TBD#",
3814                 *(int *) buffer, *buflen);
3815         }
3816       else
3817         rv = VPPCOM_EINVAL;
3818       break;
3819
3820     case VPPCOM_ATTR_SET_KEEPALIVE:
3821       if (buffer && buflen && (*buflen == sizeof (int)))
3822         {
3823           /* VPP-TBD */
3824           if (*(int *) buffer)
3825             vcl_session_set_attr (session, VCL_SESS_ATTR_KEEPALIVE);
3826           else
3827             vcl_session_clear_attr (session, VCL_SESS_ATTR_KEEPALIVE);
3828
3829           VDBG (2, "VPPCOM_ATTR_SET_KEEPALIVE: %d, buflen %d, #VPP-TBD#",
3830                 vcl_session_has_attr (session, VCL_SESS_ATTR_KEEPALIVE),
3831                 *buflen);
3832         }
3833       else
3834         rv = VPPCOM_EINVAL;
3835       break;
3836
3837     case VPPCOM_ATTR_GET_TCP_NODELAY:
3838       if (buffer && buflen && (*buflen >= sizeof (int)))
3839         {
3840           /* VPP-TBD */
3841           *(int *) buffer = vcl_session_has_attr (session,
3842                                                   VCL_SESS_ATTR_TCP_NODELAY);
3843           *buflen = sizeof (int);
3844
3845           VDBG (2, "VPPCOM_ATTR_GET_TCP_NODELAY: %d, buflen %d, #VPP-TBD#",
3846                 *(int *) buffer, *buflen);
3847         }
3848       else
3849         rv = VPPCOM_EINVAL;
3850       break;
3851
3852     case VPPCOM_ATTR_SET_TCP_NODELAY:
3853       if (buffer && buflen && (*buflen == sizeof (int)))
3854         {
3855           /* VPP-TBD */
3856           if (*(int *) buffer)
3857             vcl_session_set_attr (session, VCL_SESS_ATTR_TCP_NODELAY);
3858           else
3859             vcl_session_clear_attr (session, VCL_SESS_ATTR_TCP_NODELAY);
3860
3861           VDBG (2, "VPPCOM_ATTR_SET_TCP_NODELAY: %d, buflen %d, #VPP-TBD#",
3862                 vcl_session_has_attr (session, VCL_SESS_ATTR_TCP_NODELAY),
3863                 *buflen);
3864         }
3865       else
3866         rv = VPPCOM_EINVAL;
3867       break;
3868
3869     case VPPCOM_ATTR_GET_TCP_KEEPIDLE:
3870       if (buffer && buflen && (*buflen >= sizeof (int)))
3871         {
3872           /* VPP-TBD */
3873           *(int *) buffer = vcl_session_has_attr (session,
3874                                                   VCL_SESS_ATTR_TCP_KEEPIDLE);
3875           *buflen = sizeof (int);
3876
3877           VDBG (2, "VPPCOM_ATTR_GET_TCP_KEEPIDLE: %d, buflen %d, #VPP-TBD#",
3878                 *(int *) buffer, *buflen);
3879         }
3880       else
3881         rv = VPPCOM_EINVAL;
3882       break;
3883
3884     case VPPCOM_ATTR_SET_TCP_KEEPIDLE:
3885       if (buffer && buflen && (*buflen == sizeof (int)))
3886         {
3887           /* VPP-TBD */
3888           if (*(int *) buffer)
3889             vcl_session_set_attr (session, VCL_SESS_ATTR_TCP_KEEPIDLE);
3890           else
3891             vcl_session_clear_attr (session, VCL_SESS_ATTR_TCP_KEEPIDLE);
3892
3893           VDBG (2, "VPPCOM_ATTR_SET_TCP_KEEPIDLE: %d, buflen %d, #VPP-TBD#",
3894                 vcl_session_has_attr (session,
3895                                       VCL_SESS_ATTR_TCP_KEEPIDLE), *buflen);
3896         }
3897       else
3898         rv = VPPCOM_EINVAL;
3899       break;
3900
3901     case VPPCOM_ATTR_GET_TCP_KEEPINTVL:
3902       if (buffer && buflen && (*buflen >= sizeof (int)))
3903         {
3904           /* VPP-TBD */
3905           *(int *) buffer = vcl_session_has_attr (session,
3906                                                   VCL_SESS_ATTR_TCP_KEEPINTVL);
3907           *buflen = sizeof (int);
3908
3909           VDBG (2, "VPPCOM_ATTR_GET_TCP_KEEPINTVL: %d, buflen %d, #VPP-TBD#",
3910                 *(int *) buffer, *buflen);
3911         }
3912       else
3913         rv = VPPCOM_EINVAL;
3914       break;
3915
3916     case VPPCOM_ATTR_SET_TCP_KEEPINTVL:
3917       if (buffer && buflen && (*buflen == sizeof (int)))
3918         {
3919           /* VPP-TBD */
3920           if (*(int *) buffer)
3921             vcl_session_set_attr (session, VCL_SESS_ATTR_TCP_KEEPINTVL);
3922           else
3923             vcl_session_clear_attr (session, VCL_SESS_ATTR_TCP_KEEPINTVL);
3924
3925           VDBG (2, "VPPCOM_ATTR_SET_TCP_KEEPINTVL: %d, buflen %d, #VPP-TBD#",
3926                 vcl_session_has_attr (session,
3927                                       VCL_SESS_ATTR_TCP_KEEPINTVL), *buflen);
3928         }
3929       else
3930         rv = VPPCOM_EINVAL;
3931       break;
3932
3933     case VPPCOM_ATTR_GET_TCP_USER_MSS:
3934       if (!(buffer && buflen && (*buflen >= sizeof (u32))))
3935         {
3936           rv = VPPCOM_EINVAL;
3937           break;
3938         }
3939
3940       tea.type = TRANSPORT_ENDPT_ATTR_MSS;
3941       tea.mss = *(u32 *) buffer;
3942       if (vcl_session_transport_attr (wrk, session, 1 /* is_get */, &tea))
3943         rv = VPPCOM_ENOPROTOOPT;
3944
3945       if (!rv)
3946         {
3947           *(u32 *) buffer = tea.mss;
3948           *buflen = sizeof (int);
3949         }
3950
3951       VDBG (2, "VPPCOM_ATTR_GET_TCP_USER_MSS: %d, buflen %d", *(int *) buffer,
3952             *buflen);
3953       break;
3954
3955     case VPPCOM_ATTR_SET_TCP_USER_MSS:
3956       if (!(buffer && buflen && (*buflen == sizeof (u32))))
3957         {
3958           rv = VPPCOM_EINVAL;
3959           break;
3960         }
3961
3962       tea.type = TRANSPORT_ENDPT_ATTR_MSS;
3963       tea.mss = *(u32 *) buffer;
3964       if (vcl_session_transport_attr (wrk, session, 0 /* is_get */, &tea))
3965         rv = VPPCOM_ENOPROTOOPT;
3966
3967       VDBG (2, "VPPCOM_ATTR_SET_TCP_USER_MSS: %u, buflen %d", tea.mss,
3968             *buflen);
3969       break;
3970
3971     case VPPCOM_ATTR_SET_CONNECTED:
3972       session->flags |= VCL_SESSION_F_CONNECTED;
3973       break;
3974
3975     case VPPCOM_ATTR_SET_CKPAIR:
3976       if (!(buffer && buflen && (*buflen == sizeof (int))) ||
3977           !vcl_session_has_crypto (session))
3978         {
3979           rv = VPPCOM_EINVAL;
3980           break;
3981         }
3982       if (!session->ext_config)
3983         {
3984           vcl_session_alloc_ext_cfg (session, TRANSPORT_ENDPT_EXT_CFG_CRYPTO,
3985                                      sizeof (transport_endpt_ext_cfg_t));
3986         }
3987       else if (session->ext_config->type != TRANSPORT_ENDPT_EXT_CFG_CRYPTO)
3988         {
3989           rv = VPPCOM_EINVAL;
3990           break;
3991         }
3992
3993       session->ext_config->crypto.ckpair_index = *(uint32_t *) buffer;
3994       break;
3995
3996     case VPPCOM_ATTR_SET_VRF:
3997       if (!(buffer && buflen && (*buflen == sizeof (u32))))
3998         {
3999           rv = VPPCOM_EINVAL;
4000           break;
4001         }
4002       session->vrf = *(u32 *) buffer;
4003       break;
4004
4005     case VPPCOM_ATTR_GET_VRF:
4006       if (!(buffer && buflen && (*buflen >= sizeof (u32))))
4007         {
4008           rv = VPPCOM_EINVAL;
4009           break;
4010         }
4011       *(u32 *) buffer = session->vrf;
4012       *buflen = sizeof (u32);
4013       break;
4014
4015     case VPPCOM_ATTR_GET_DOMAIN:
4016       if (!(buffer && buflen && (*buflen >= sizeof (int))))
4017         {
4018           rv = VPPCOM_EINVAL;
4019           break;
4020         }
4021
4022       if (session->transport.is_ip4)
4023         *(int *) buffer = AF_INET;
4024       else
4025         *(int *) buffer = AF_INET6;
4026       *buflen = sizeof (int);
4027
4028       VDBG (2, "VPPCOM_ATTR_GET_DOMAIN: %d, buflen %u", *(int *) buffer,
4029             *buflen);
4030       break;
4031
4032     case VPPCOM_ATTR_SET_ENDPT_EXT_CFG:
4033       if (!(buffer && buflen && (*buflen > 0)))
4034         {
4035           rv = VPPCOM_EINVAL;
4036           break;
4037         }
4038       if (session->ext_config)
4039         {
4040           rv = VPPCOM_EINVAL;
4041           break;
4042         }
4043       vcl_session_alloc_ext_cfg (session, TRANSPORT_ENDPT_EXT_CFG_NONE,
4044                                  *buflen + sizeof (u32));
4045       clib_memcpy (session->ext_config->data, buffer, *buflen);
4046       session->ext_config->len = *buflen;
4047       break;
4048
4049     default:
4050       rv = VPPCOM_EINVAL;
4051       break;
4052     }
4053
4054   return rv;
4055 }
4056
4057 int
4058 vppcom_session_recvfrom (uint32_t session_handle, void *buffer,
4059                          uint32_t buflen, int flags, vppcom_endpt_t * ep)
4060 {
4061   vcl_worker_t *wrk = vcl_worker_get_current ();
4062   vcl_session_t *session;
4063   int rv = VPPCOM_OK;
4064
4065   if (flags == 0)
4066     rv = vppcom_session_read (session_handle, buffer, buflen);
4067   else if (flags & MSG_PEEK)
4068     rv = vppcom_session_peek (session_handle, buffer, buflen);
4069   else
4070     {
4071       VDBG (0, "Unsupport flags for recvfrom %d", flags);
4072       return VPPCOM_EAFNOSUPPORT;
4073     }
4074
4075   if (ep && rv > 0)
4076     {
4077       session = vcl_session_get_w_handle (wrk, session_handle);
4078       if (session->transport.is_ip4)
4079         clib_memcpy_fast (ep->ip, &session->transport.rmt_ip.ip4,
4080                           sizeof (ip4_address_t));
4081       else
4082         clib_memcpy_fast (ep->ip, &session->transport.rmt_ip.ip6,
4083                           sizeof (ip6_address_t));
4084       ep->is_ip4 = session->transport.is_ip4;
4085       ep->port = session->transport.rmt_port;
4086     }
4087
4088   return rv;
4089 }
4090
4091 int
4092 vppcom_session_sendto (uint32_t session_handle, void *buffer,
4093                        uint32_t buflen, int flags, vppcom_endpt_t * ep)
4094 {
4095   vcl_worker_t *wrk = vcl_worker_get_current ();
4096   vcl_session_t *s;
4097
4098   s = vcl_session_get_w_handle (wrk, session_handle);
4099   if (PREDICT_FALSE (!s))
4100     return VPPCOM_EBADFD;
4101
4102   if (ep)
4103     {
4104       if (!vcl_session_is_cl (s))
4105         return VPPCOM_EINVAL;
4106
4107       s->transport.is_ip4 = ep->is_ip4;
4108       s->transport.rmt_port = ep->port;
4109       vcl_ip_copy_from_ep (&s->transport.rmt_ip, ep);
4110
4111       /* Session not connected/bound in vpp. Create it by 'connecting' it */
4112       if (PREDICT_FALSE (s->session_state == VCL_STATE_CLOSED))
4113         {
4114           u32 session_index = s->session_index;
4115           f64 timeout = vcm->cfg.session_timeout;
4116           int rv;
4117
4118           vcl_send_session_connect (wrk, s);
4119           rv = vppcom_wait_for_session_state_change (session_index,
4120                                                      VCL_STATE_READY,
4121                                                      timeout);
4122           if (rv < 0)
4123             return rv;
4124           s = vcl_session_get (wrk, session_index);
4125         }
4126     }
4127
4128   if (flags)
4129     {
4130       // TBD check the flags and do the right thing
4131       VDBG (2, "handling flags 0x%u (%d) not implemented yet.", flags, flags);
4132     }
4133
4134   return (vppcom_session_write_inline (wrk, s, buffer, buflen, 1,
4135                                        s->is_dgram ? 1 : 0));
4136 }
4137
4138 int
4139 vppcom_poll (vcl_poll_t * vp, uint32_t n_sids, double wait_for_time)
4140 {
4141   vcl_worker_t *wrk = vcl_worker_get_current ();
4142   f64 timeout = clib_time_now (&wrk->clib_time) + wait_for_time;
4143   u32 i, keep_trying = 1;
4144   svm_msg_q_msg_t msg;
4145   session_event_t *e;
4146   int rv, num_ev = 0;
4147
4148   VDBG (3, "vp %p, nsids %u, wait_for_time %f", vp, n_sids, wait_for_time);
4149
4150   if (!vp)
4151     return VPPCOM_EFAULT;
4152
4153   do
4154     {
4155       vcl_session_t *session;
4156
4157       /* Dequeue all events and drop all unhandled io events */
4158       while (svm_msg_q_sub (wrk->app_event_queue, &msg, SVM_Q_NOWAIT, 0) == 0)
4159         {
4160           e = svm_msg_q_msg_data (wrk->app_event_queue, &msg);
4161           vcl_handle_mq_event (wrk, e);
4162           svm_msg_q_free_msg (wrk->app_event_queue, &msg);
4163         }
4164       vec_reset_length (wrk->unhandled_evts_vector);
4165
4166       for (i = 0; i < n_sids; i++)
4167         {
4168           session = vcl_session_get (wrk, vp[i].sh);
4169           if (!session)
4170             {
4171               vp[i].revents = POLLHUP;
4172               num_ev++;
4173               continue;
4174             }
4175
4176           vp[i].revents = 0;
4177
4178           if (POLLIN & vp[i].events)
4179             {
4180               rv = vcl_session_read_ready (session);
4181               if (rv > 0)
4182                 {
4183                   vp[i].revents |= POLLIN;
4184                   num_ev++;
4185                 }
4186               else if (rv < 0)
4187                 {
4188                   switch (rv)
4189                     {
4190                     case VPPCOM_ECONNRESET:
4191                       vp[i].revents = POLLHUP;
4192                       break;
4193
4194                     default:
4195                       vp[i].revents = POLLERR;
4196                       break;
4197                     }
4198                   num_ev++;
4199                 }
4200             }
4201
4202           if (POLLOUT & vp[i].events)
4203             {
4204               rv = vcl_session_write_ready (session);
4205               if (rv > 0)
4206                 {
4207                   vp[i].revents |= POLLOUT;
4208                   num_ev++;
4209                 }
4210               else if (rv < 0)
4211                 {
4212                   switch (rv)
4213                     {
4214                     case VPPCOM_ECONNRESET:
4215                       vp[i].revents = POLLHUP;
4216                       break;
4217
4218                     default:
4219                       vp[i].revents = POLLERR;
4220                       break;
4221                     }
4222                   num_ev++;
4223                 }
4224             }
4225
4226           if (0)                // Note "done:" label used by VCL_SESSION_LOCK_AND_GET()
4227             {
4228               vp[i].revents = POLLNVAL;
4229               num_ev++;
4230             }
4231         }
4232       if (wait_for_time != -1)
4233         keep_trying = (clib_time_now (&wrk->clib_time) <= timeout) ? 1 : 0;
4234     }
4235   while ((num_ev == 0) && keep_trying);
4236
4237   return num_ev;
4238 }
4239
4240 int
4241 vppcom_mq_epoll_fd (void)
4242 {
4243   vcl_worker_t *wrk = vcl_worker_get_current ();
4244   return wrk->mqs_epfd;
4245 }
4246
4247 int
4248 vppcom_session_index (vcl_session_handle_t session_handle)
4249 {
4250   return session_handle & 0xFFFFFF;
4251 }
4252
4253 int
4254 vppcom_session_worker (vcl_session_handle_t session_handle)
4255 {
4256   return session_handle >> 24;
4257 }
4258
4259 int
4260 vppcom_worker_register (void)
4261 {
4262   if (!vcl_worker_alloc_and_init ())
4263     return VPPCOM_EEXIST;
4264
4265   if (vcl_worker_register_with_vpp ())
4266     return VPPCOM_EEXIST;
4267
4268   return VPPCOM_OK;
4269 }
4270
4271 void
4272 vppcom_worker_unregister (void)
4273 {
4274   vcl_worker_cleanup (vcl_worker_get_current (), 1 /* notify vpp */ );
4275   vcl_set_worker_index (~0);
4276 }
4277
4278 void
4279 vppcom_worker_index_set (int index)
4280 {
4281   vcl_set_worker_index (index);
4282 }
4283
4284 int
4285 vppcom_worker_index (void)
4286 {
4287   return vcl_get_worker_index ();
4288 }
4289
4290 int
4291 vppcom_worker_mqs_epfd (void)
4292 {
4293   vcl_worker_t *wrk = vcl_worker_get_current ();
4294   if (!vcm->cfg.use_mq_eventfd)
4295     return -1;
4296   return wrk->mqs_epfd;
4297 }
4298
4299 int
4300 vppcom_session_is_connectable_listener (uint32_t session_handle)
4301 {
4302   vcl_session_t *session;
4303   vcl_worker_t *wrk = vcl_worker_get_current ();
4304   session = vcl_session_get_w_handle (wrk, session_handle);
4305   if (!session)
4306     return VPPCOM_EBADFD;
4307   return vcl_session_is_connectable_listener (wrk, session);
4308 }
4309
4310 int
4311 vppcom_session_listener (uint32_t session_handle)
4312 {
4313   vcl_worker_t *wrk = vcl_worker_get_current ();
4314   vcl_session_t *listen_session, *session;
4315   session = vcl_session_get_w_handle (wrk, session_handle);
4316   if (!session)
4317     return VPPCOM_EBADFD;
4318   if (session->listener_index == VCL_INVALID_SESSION_INDEX)
4319     return VPPCOM_EBADFD;
4320   listen_session = vcl_session_get_w_handle (wrk, session->listener_index);
4321   if (!listen_session)
4322     return VPPCOM_EBADFD;
4323   return vcl_session_handle (listen_session);
4324 }
4325
4326 int
4327 vppcom_session_n_accepted (uint32_t session_handle)
4328 {
4329   vcl_worker_t *wrk = vcl_worker_get_current ();
4330   vcl_session_t *session = vcl_session_get_w_handle (wrk, session_handle);
4331   if (!session)
4332     return VPPCOM_EBADFD;
4333   return session->n_accepted_sessions;
4334 }
4335
4336 const char *
4337 vppcom_proto_str (vppcom_proto_t proto)
4338 {
4339   char const *proto_str;
4340
4341   switch (proto)
4342     {
4343     case VPPCOM_PROTO_TCP:
4344       proto_str = "TCP";
4345       break;
4346     case VPPCOM_PROTO_UDP:
4347       proto_str = "UDP";
4348       break;
4349     case VPPCOM_PROTO_TLS:
4350       proto_str = "TLS";
4351       break;
4352     case VPPCOM_PROTO_QUIC:
4353       proto_str = "QUIC";
4354       break;
4355     case VPPCOM_PROTO_DTLS:
4356       proto_str = "DTLS";
4357       break;
4358     case VPPCOM_PROTO_SRTP:
4359       proto_str = "SRTP";
4360       break;
4361     default:
4362       proto_str = "UNKNOWN";
4363       break;
4364     }
4365   return proto_str;
4366 }
4367
4368 const char *
4369 vppcom_retval_str (int retval)
4370 {
4371   char const *st;
4372
4373   switch (retval)
4374     {
4375     case VPPCOM_OK:
4376       st = "VPPCOM_OK";
4377       break;
4378
4379     case VPPCOM_EAGAIN:
4380       st = "VPPCOM_EAGAIN";
4381       break;
4382
4383     case VPPCOM_EFAULT:
4384       st = "VPPCOM_EFAULT";
4385       break;
4386
4387     case VPPCOM_ENOMEM:
4388       st = "VPPCOM_ENOMEM";
4389       break;
4390
4391     case VPPCOM_EINVAL:
4392       st = "VPPCOM_EINVAL";
4393       break;
4394
4395     case VPPCOM_EBADFD:
4396       st = "VPPCOM_EBADFD";
4397       break;
4398
4399     case VPPCOM_EAFNOSUPPORT:
4400       st = "VPPCOM_EAFNOSUPPORT";
4401       break;
4402
4403     case VPPCOM_ECONNABORTED:
4404       st = "VPPCOM_ECONNABORTED";
4405       break;
4406
4407     case VPPCOM_ECONNRESET:
4408       st = "VPPCOM_ECONNRESET";
4409       break;
4410
4411     case VPPCOM_ENOTCONN:
4412       st = "VPPCOM_ENOTCONN";
4413       break;
4414
4415     case VPPCOM_ECONNREFUSED:
4416       st = "VPPCOM_ECONNREFUSED";
4417       break;
4418
4419     case VPPCOM_ETIMEDOUT:
4420       st = "VPPCOM_ETIMEDOUT";
4421       break;
4422
4423     default:
4424       st = "UNKNOWN_STATE";
4425       break;
4426     }
4427
4428   return st;
4429 }
4430
4431 int
4432 vppcom_add_cert_key_pair (vppcom_cert_key_pair_t *ckpair)
4433 {
4434   if (vcm->cfg.vpp_app_socket_api)
4435     return vcl_sapi_add_cert_key_pair (ckpair);
4436   else
4437     return vcl_bapi_add_cert_key_pair (ckpair);
4438 }
4439
4440 int
4441 vppcom_del_cert_key_pair (uint32_t ckpair_index)
4442 {
4443   if (vcm->cfg.vpp_app_socket_api)
4444     return vcl_sapi_del_cert_key_pair (ckpair_index);
4445   else
4446     return vcl_bapi_del_cert_key_pair (ckpair_index);
4447 }
4448
4449 /*
4450  * fd.io coding-style-patch-verification: ON
4451  *
4452  * Local Variables:
4453  * eval: (c-set-style "gnu")
4454  * End:
4455  */