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