src/vnet/session/session_node.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 file except in compliance with the License.
   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 <math.h>
  17 #include <vlib/vlib.h>
  18 #include <vnet/vnet.h>
  19 #include <vppinfra/elog.h>
  20 #include <vnet/session/transport.h>
  21 #include <vnet/session/session.h>
  22 #include <vnet/session/application.h>
  23 #include <vnet/session/application_interface.h>
  24 #include <vnet/session/application_local.h>
  25 #include <vnet/session/session_debug.h>
  26 #include <svm/queue.h>
  27 #include <sys/timerfd.h>
  28
  29 static inline void
  30 session_wrk_send_evt_to_main (session_worker_t *wrk, session_evt_elt_t *elt)
  31 {
  32   session_evt_elt_t *he;
  33   uword thread_index;
  34   u8 is_empty;
  35
  36   thread_index = wrk->vm->thread_index;
  37   he = clib_llist_elt (wrk->event_elts, wrk->evts_pending_main);
  38   is_empty = clib_llist_is_empty (wrk->event_elts, evt_list, he);
  39   clib_llist_add_tail (wrk->event_elts, evt_list, elt, he);
  40   if (is_empty)
  41     session_send_rpc_evt_to_thread (0, session_wrk_handle_evts_main_rpc,
  42                                     uword_to_pointer (thread_index, void *));
  43 }
  44
  45 #define app_check_thread_and_barrier(_wrk, _elt)                              \
  46   if (!vlib_thread_is_main_w_barrier ())                                      \
  47     {                                                                         \
  48       session_wrk_send_evt_to_main (wrk, elt);                                \
  49       return;                                                                 \
  50     }
  51
  52 static void
  53 session_wrk_timerfd_update (session_worker_t *wrk, u64 time_ns)
  54 {
  55   struct itimerspec its;
  56
  57   its.it_value.tv_sec = 0;
  58   its.it_value.tv_nsec = time_ns;
  59   its.it_interval.tv_sec = 0;
  60   its.it_interval.tv_nsec = its.it_value.tv_nsec;
  61
  62   if (timerfd_settime (wrk->timerfd, 0, &its, NULL) == -1)
  63     clib_warning ("timerfd_settime");
  64 }
  65
  66 always_inline u64
  67 session_wrk_tfd_timeout (session_wrk_state_t state, u32 thread_index)
  68 {
  69   if (state == SESSION_WRK_INTERRUPT)
  70     return thread_index ? 1e6 : vlib_num_workers () ? 5e8 : 1e6;
  71   else if (state == SESSION_WRK_IDLE)
  72     return thread_index ? 1e8 : vlib_num_workers () ? 5e8 : 1e8;
  73   else
  74     return 0;
  75 }
  76
  77 static inline void
  78 session_wrk_set_state (session_worker_t *wrk, session_wrk_state_t state)
  79 {
  80   u64 time_ns;
  81
  82   wrk->state = state;
  83   if (wrk->timerfd == -1)
  84     return;
  85   time_ns = session_wrk_tfd_timeout (state, wrk->vm->thread_index);
  86   session_wrk_timerfd_update (wrk, time_ns);
  87 }
  88
  89 static transport_endpt_ext_cfg_t *
  90 session_mq_get_ext_config (application_t *app, uword offset)
  91 {
  92   svm_fifo_chunk_t *c;
  93   fifo_segment_t *fs;
  94
  95   fs = application_get_rx_mqs_segment (app);
  96   c = fs_chunk_ptr (fs->h, offset);
  97   return (transport_endpt_ext_cfg_t *) c->data;
  98 }
  99
 100 static void
 101 session_mq_free_ext_config (application_t *app, uword offset)
 102 {
 103   svm_fifo_chunk_t *c;
 104   fifo_segment_t *fs;
 105
 106   fs = application_get_rx_mqs_segment (app);
 107   c = fs_chunk_ptr (fs->h, offset);
 108   fifo_segment_collect_chunk (fs, 0 /* only one slice */, c);
 109 }
 110
 111 static void
 112 session_mq_listen_handler (session_worker_t *wrk, session_evt_elt_t *elt)
 113 {
 114   vnet_listen_args_t _a, *a = &_a;
 115   session_listen_msg_t *mp;
 116   app_worker_t *app_wrk;
 117   application_t *app;
 118   int rv;
 119
 120   app_check_thread_and_barrier (wrk, elt);
 121
 122   mp = session_evt_ctrl_data (wrk, elt);
 123   app = application_lookup (mp->client_index);
 124   if (!app)
 125     return;
 126
 127   clib_memset (a, 0, sizeof (*a));
 128   a->sep.is_ip4 = mp->is_ip4;
 129   ip_copy (&a->sep.ip, &mp->ip, mp->is_ip4);
 130   a->sep.port = mp->port;
 131   a->sep.fib_index = mp->vrf;
 132   a->sep.sw_if_index = ENDPOINT_INVALID_INDEX;
 133   a->sep.transport_proto = mp->proto;
 134   a->app_index = app->app_index;
 135   a->wrk_map_index = mp->wrk_index;
 136   a->sep_ext.transport_flags = mp->flags;
 137
 138   if (mp->ext_config)
 139     a->sep_ext.ext_cfg = session_mq_get_ext_config (app, mp->ext_config);
 140
 141   if ((rv = vnet_listen (a)))
 142     session_worker_stat_error_inc (wrk, rv, 1);
 143
 144   app_wrk = application_get_worker (app, mp->wrk_index);
 145   app_worker_listened_notify (app_wrk, a->handle, mp->context, rv);
 146
 147   if (mp->ext_config)
 148     session_mq_free_ext_config (app, mp->ext_config);
 149
 150   /* Make sure events are flushed before releasing barrier, to avoid
 151    * potential race with accept. */
 152   app_wrk_flush_wrk_events (app_wrk, 0);
 153 }
 154
 155 static void
 156 session_mq_listen_uri_handler (session_worker_t *wrk, session_evt_elt_t *elt)
 157 {
 158   vnet_listen_args_t _a, *a = &_a;
 159   session_listen_uri_msg_t *mp;
 160   app_worker_t *app_wrk;
 161   application_t *app;
 162   int rv;
 163
 164   app_check_thread_and_barrier (wrk, elt);
 165
 166   mp = session_evt_ctrl_data (wrk, elt);
 167   app = application_lookup (mp->client_index);
 168   if (!app)
 169     return;
 170
 171   clib_memset (a, 0, sizeof (*a));
 172   a->uri = (char *) mp->uri;
 173   a->app_index = app->app_index;
 174   rv = vnet_bind_uri (a);
 175
 176   app_wrk = application_get_worker (app, 0);
 177   app_worker_listened_notify (app_wrk, a->handle, mp->context, rv);
 178   app_wrk_flush_wrk_events (app_wrk, 0);
 179 }
 180
 181 static void
 182 session_mq_connect_one (session_connect_msg_t *mp)
 183 {
 184   vnet_connect_args_t _a, *a = &_a;
 185   app_worker_t *app_wrk;
 186   session_worker_t *wrk;
 187   application_t *app;
 188   int rv;
 189
 190   app = application_lookup (mp->client_index);
 191   if (!app)
 192     return;
 193
 194   clib_memset (a, 0, sizeof (*a));
 195   a->sep.is_ip4 = mp->is_ip4;
 196   clib_memcpy_fast (&a->sep.ip, &mp->ip, sizeof (mp->ip));
 197   a->sep.port = mp->port;
 198   a->sep.transport_proto = mp->proto;
 199   a->sep.peer.fib_index = mp->vrf;
 200   a->sep.dscp = mp->dscp;
 201   clib_memcpy_fast (&a->sep.peer.ip, &mp->lcl_ip, sizeof (mp->lcl_ip));
 202   if (mp->is_ip4)
 203     {
 204       ip46_address_mask_ip4 (&a->sep.ip);
 205       ip46_address_mask_ip4 (&a->sep.peer.ip);
 206     }
 207   a->sep.peer.port = mp->lcl_port;
 208   a->sep.peer.sw_if_index = ENDPOINT_INVALID_INDEX;
 209   a->sep_ext.parent_handle = mp->parent_handle;
 210   a->sep_ext.transport_flags = mp->flags;
 211   a->api_context = mp->context;
 212   a->app_index = app->app_index;
 213   a->wrk_map_index = mp->wrk_index;
 214
 215   if (mp->ext_config)
 216     a->sep_ext.ext_cfg = session_mq_get_ext_config (app, mp->ext_config);
 217
 218   if ((rv = vnet_connect (a)))
 219     {
 220       wrk = session_main_get_worker (vlib_get_thread_index ());
 221       session_worker_stat_error_inc (wrk, rv, 1);
 222       app_wrk = application_get_worker (app, mp->wrk_index);
 223       app_worker_connect_notify (app_wrk, 0, rv, mp->context);
 224     }
 225
 226   if (mp->ext_config)
 227     session_mq_free_ext_config (app, mp->ext_config);
 228 }
 229
 230 static void
 231 session_mq_handle_connects_rpc (void *arg)
 232 {
 233   u32 max_connects = 32, n_connects = 0;
 234   session_evt_elt_t *he, *elt, *next;
 235   session_worker_t *fwrk;
 236
 237   ASSERT (session_vlib_thread_is_cl_thread ());
 238
 239   /* Pending connects on linked list pertaining to first worker */
 240   fwrk = session_main_get_worker (transport_cl_thread ());
 241   if (!fwrk->n_pending_connects)
 242     return;
 243
 244   he = clib_llist_elt (fwrk->event_elts, fwrk->pending_connects);
 245   elt = clib_llist_next (fwrk->event_elts, evt_list, he);
 246
 247   /* Avoid holding the worker for too long */
 248   while (n_connects < max_connects && elt != he)
 249     {
 250       next = clib_llist_next (fwrk->event_elts, evt_list, elt);
 251       clib_llist_remove (fwrk->event_elts, evt_list, elt);
 252       session_mq_connect_one (session_evt_ctrl_data (fwrk, elt));
 253       session_evt_ctrl_data_free (fwrk, elt);
 254       clib_llist_put (fwrk->event_elts, elt);
 255       elt = next;
 256       n_connects += 1;
 257     }
 258
 259   /* Decrement with worker barrier */
 260   fwrk->n_pending_connects -= n_connects;
 261   if (fwrk->n_pending_connects > 0)
 262     {
 263       session_send_rpc_evt_to_thread_force (fwrk->vm->thread_index,
 264                                             session_mq_handle_connects_rpc, 0);
 265     }
 266 }
 267
 268 static void
 269 session_mq_connect_handler (session_worker_t *wrk, session_evt_elt_t *elt)
 270 {
 271   u32 thread_index = wrk - session_main.wrk;
 272   session_evt_elt_t *he;
 273
 274   if (PREDICT_FALSE (thread_index > transport_cl_thread ()))
 275     {
 276       clib_warning ("Connect on wrong thread. Dropping");
 277       return;
 278     }
 279
 280   /* If on worker, check if main has any pending messages. Avoids reordering
 281    * with other control messages that need to be handled by main
 282    */
 283   if (thread_index)
 284     {
 285       he = clib_llist_elt (wrk->event_elts, wrk->evts_pending_main);
 286
 287       /* Events pending on main, postpone to avoid reordering */
 288       if (!clib_llist_is_empty (wrk->event_elts, evt_list, he))
 289         {
 290           clib_llist_add_tail (wrk->event_elts, evt_list, elt, he);
 291           return;
 292         }
 293     }
 294
 295   /* Add to pending list to be handled by first worker */
 296   he = clib_llist_elt (wrk->event_elts, wrk->pending_connects);
 297   clib_llist_add_tail (wrk->event_elts, evt_list, elt, he);
 298
 299   /* Decremented with worker barrier */
 300   wrk->n_pending_connects += 1;
 301   if (wrk->n_pending_connects == 1)
 302     {
 303       session_send_rpc_evt_to_thread_force (thread_index,
 304                                             session_mq_handle_connects_rpc, 0);
 305     }
 306 }
 307
 308 static void
 309 session_mq_connect_uri_handler (session_worker_t *wrk, session_evt_elt_t *elt)
 310 {
 311   vnet_connect_args_t _a, *a = &_a;
 312   session_connect_uri_msg_t *mp;
 313   app_worker_t *app_wrk;
 314   application_t *app;
 315   int rv;
 316
 317   app_check_thread_and_barrier (wrk, elt);
 318
 319   mp = session_evt_ctrl_data (wrk, elt);
 320   app = application_lookup (mp->client_index);
 321   if (!app)
 322     return;
 323
 324   clib_memset (a, 0, sizeof (*a));
 325   a->uri = (char *) mp->uri;
 326   a->api_context = mp->context;
 327   a->app_index = app->app_index;
 328   if ((rv = vnet_connect_uri (a)))
 329     {
 330       session_worker_stat_error_inc (wrk, rv, 1);
 331       app_wrk = application_get_worker (app, 0 /* default wrk only */ );
 332       app_worker_connect_notify (app_wrk, 0, rv, mp->context);
 333     }
 334 }
 335
 336 static void
 337 session_mq_shutdown_handler (void *data)
 338 {
 339   session_shutdown_msg_t *mp = (session_shutdown_msg_t *) data;
 340   vnet_shutdown_args_t _a, *a = &_a;
 341   application_t *app;
 342
 343   app = application_lookup (mp->client_index);
 344   if (!app)
 345     return;
 346
 347   a->app_index = app->app_index;
 348   a->handle = mp->handle;
 349   vnet_shutdown_session (a);
 350 }
 351
 352 static void
 353 session_mq_disconnect_handler (void *data)
 354 {
 355   session_disconnect_msg_t *mp = (session_disconnect_msg_t *) data;
 356   vnet_disconnect_args_t _a, *a = &_a;
 357   application_t *app;
 358
 359   app = application_lookup (mp->client_index);
 360   if (!app)
 361     return;
 362
 363   a->app_index = app->app_index;
 364   a->handle = mp->handle;
 365   vnet_disconnect_session (a);
 366 }
 367
 368 static void
 369 app_mq_detach_handler (session_worker_t *wrk, session_evt_elt_t *elt)
 370 {
 371   vnet_app_detach_args_t _a, *a = &_a;
 372   session_app_detach_msg_t *mp;
 373   application_t *app;
 374
 375   app_check_thread_and_barrier (wrk, elt);
 376
 377   mp = session_evt_ctrl_data (wrk, elt);
 378   app = application_lookup (mp->client_index);
 379   if (!app)
 380     return;
 381
 382   a->app_index = app->app_index;
 383   a->api_client_index = mp->client_index;
 384   vnet_application_detach (a);
 385 }
 386
 387 static void
 388 session_mq_unlisten_handler (session_worker_t *wrk, session_evt_elt_t *elt)
 389 {
 390   vnet_unlisten_args_t _a, *a = &_a;
 391   session_unlisten_msg_t *mp;
 392   app_worker_t *app_wrk;
 393   session_handle_t sh;
 394   application_t *app;
 395   int rv;
 396
 397   app_check_thread_and_barrier (wrk, elt);
 398
 399   mp = session_evt_ctrl_data (wrk, elt);
 400   sh = mp->handle;
 401
 402   app = application_lookup (mp->client_index);
 403   if (!app)
 404     return;
 405
 406   clib_memset (a, 0, sizeof (*a));
 407   a->app_index = app->app_index;
 408   a->handle = sh;
 409   a->wrk_map_index = mp->wrk_index;
 410
 411   if ((rv = vnet_unlisten (a)))
 412     session_worker_stat_error_inc (wrk, rv, 1);
 413
 414   app_wrk = application_get_worker (app, a->wrk_map_index);
 415   if (!app_wrk)
 416     return;
 417
 418   app_worker_unlisten_reply (app_wrk, sh, mp->context, rv);
 419 }
 420
 421 static void
 422 session_mq_accepted_reply_handler (session_worker_t *wrk,
 423                                    session_evt_elt_t *elt)
 424 {
 425   vnet_disconnect_args_t _a = { 0 }, *a = &_a;
 426   session_accepted_reply_msg_t *mp;
 427   session_state_t old_state;
 428   app_worker_t *app_wrk;
 429   session_t *s;
 430
 431   mp = session_evt_ctrl_data (wrk, elt);
 432
 433   /* Mail this back from the main thread. We're not polling in main
 434    * thread so we're using other workers for notifications. */
 435   if (session_thread_from_handle (mp->handle) == 0 && vlib_num_workers () &&
 436       vlib_get_thread_index () != 0)
 437     {
 438       session_wrk_send_evt_to_main (wrk, elt);
 439       return;
 440     }
 441
 442   s = session_get_from_handle_if_valid (mp->handle);
 443   if (!s)
 444     return;
 445
 446   app_wrk = app_worker_get (s->app_wrk_index);
 447   if (app_wrk->app_index != mp->context)
 448     {
 449       clib_warning ("app doesn't own session");
 450       return;
 451     }
 452
 453   /* Server isn't interested, disconnect the session */
 454   if (mp->retval)
 455     {
 456       a->app_index = mp->context;
 457       a->handle = mp->handle;
 458       vnet_disconnect_session (a);
 459       return;
 460     }
 461
 462   /* Special handling for cut-through sessions */
 463   if (!session_has_transport (s))
 464     {
 465       session_set_state (s, SESSION_STATE_READY);
 466       ct_session_connect_notify (s, SESSION_E_NONE);
 467       return;
 468     }
 469
 470   old_state = s->session_state;
 471   session_set_state (s, SESSION_STATE_READY);
 472
 473   if (!svm_fifo_is_empty_prod (s->rx_fifo))
 474     app_worker_rx_notify (app_wrk, s);
 475
 476   /* Closed while waiting for app to reply. Resend disconnect */
 477   if (old_state >= SESSION_STATE_TRANSPORT_CLOSING)
 478     {
 479       app_worker_close_notify (app_wrk, s);
 480       session_set_state (s, old_state);
 481       return;
 482     }
 483 }
 484
 485 static void
 486 session_mq_reset_reply_handler (void *data)
 487 {
 488   vnet_disconnect_args_t _a = { 0 }, *a = &_a;
 489   session_reset_reply_msg_t *mp;
 490   app_worker_t *app_wrk;
 491   session_t *s;
 492   application_t *app;
 493   u32 index, thread_index;
 494
 495   mp = (session_reset_reply_msg_t *) data;
 496   app = application_lookup (mp->context);
 497   if (!app)
 498     return;
 499
 500   session_parse_handle (mp->handle, &index, &thread_index);
 501   s = session_get_if_valid (index, thread_index);
 502
 503   /* No session or not the right session */
 504   if (!s || s->session_state < SESSION_STATE_TRANSPORT_CLOSING)
 505     return;
 506
 507   app_wrk = app_worker_get (s->app_wrk_index);
 508   if (!app_wrk || app_wrk->app_index != app->app_index)
 509     {
 510       clib_warning ("App %u does not own handle 0x%lx!", app->app_index,
 511                     mp->handle);
 512       return;
 513     }
 514
 515   /* Client objected to resetting the session, log and continue */
 516   if (mp->retval)
 517     {
 518       clib_warning ("client retval %d", mp->retval);
 519       return;
 520     }
 521
 522   /* This comes as a response to a reset, transport only waiting for
 523    * confirmation to remove connection state, no need to disconnect */
 524   a->handle = mp->handle;
 525   a->app_index = app->app_index;
 526   vnet_disconnect_session (a);
 527 }
 528
 529 static void
 530 session_mq_disconnected_handler (void *data)
 531 {
 532   session_disconnected_reply_msg_t *rmp;
 533   vnet_disconnect_args_t _a, *a = &_a;
 534   svm_msg_q_msg_t _msg, *msg = &_msg;
 535   session_disconnected_msg_t *mp;
 536   app_worker_t *app_wrk;
 537   session_event_t *evt;
 538   session_t *s;
 539   application_t *app;
 540   int rv = 0;
 541
 542   mp = (session_disconnected_msg_t *) data;
 543   if (!(s = session_get_from_handle_if_valid (mp->handle)))
 544     {
 545       clib_warning ("could not disconnect handle %llu", mp->handle);
 546       return;
 547     }
 548   app_wrk = app_worker_get (s->app_wrk_index);
 549   app = application_lookup (mp->client_index);
 550   if (!(app_wrk && app && app->app_index == app_wrk->app_index))
 551     {
 552       clib_warning ("could not disconnect session: %llu app: %u",
 553                     mp->handle, mp->client_index);
 554       return;
 555     }
 556
 557   a->handle = mp->handle;
 558   a->app_index = app_wrk->wrk_index;
 559   rv = vnet_disconnect_session (a);
 560
 561   svm_msg_q_lock_and_alloc_msg_w_ring (app_wrk->event_queue,
 562                                        SESSION_MQ_CTRL_EVT_RING,
 563                                        SVM_Q_WAIT, msg);
 564   evt = svm_msg_q_msg_data (app_wrk->event_queue, msg);
 565   clib_memset (evt, 0, sizeof (*evt));
 566   evt->event_type = SESSION_CTRL_EVT_DISCONNECTED_REPLY;
 567   rmp = (session_disconnected_reply_msg_t *) evt->data;
 568   rmp->handle = mp->handle;
 569   rmp->context = mp->context;
 570   rmp->retval = rv;
 571   svm_msg_q_add_and_unlock (app_wrk->event_queue, msg);
 572 }
 573
 574 static void
 575 session_mq_disconnected_reply_handler (void *data)
 576 {
 577   session_disconnected_reply_msg_t *mp;
 578   vnet_disconnect_args_t _a, *a = &_a;
 579   application_t *app;
 580
 581   mp = (session_disconnected_reply_msg_t *) data;
 582
 583   /* Client objected to disconnecting the session, log and continue */
 584   if (mp->retval)
 585     {
 586       clib_warning ("client retval %d", mp->retval);
 587       return;
 588     }
 589
 590   /* Disconnect has been confirmed. Confirm close to transport */
 591   app = application_lookup (mp->context);
 592   if (app)
 593     {
 594       a->handle = mp->handle;
 595       a->app_index = app->app_index;
 596       vnet_disconnect_session (a);
 597     }
 598 }
 599
 600 static void
 601 session_mq_worker_update_handler (void *data)
 602 {
 603   session_worker_update_msg_t *mp = (session_worker_update_msg_t *) data;
 604   session_worker_update_reply_msg_t *rmp;
 605   svm_msg_q_msg_t _msg, *msg = &_msg;
 606   app_worker_t *app_wrk;
 607   u32 owner_app_wrk_map;
 608   session_event_t *evt;
 609   session_t *s;
 610   application_t *app;
 611   int rv;
 612
 613   app = application_lookup (mp->client_index);
 614   if (!app)
 615     return;
 616   if (!(s = session_get_from_handle_if_valid (mp->handle)))
 617     {
 618       clib_warning ("invalid handle %llu", mp->handle);
 619       return;
 620     }
 621   app_wrk = app_worker_get (s->app_wrk_index);
 622   if (app_wrk->app_index != app->app_index)
 623     {
 624       clib_warning ("app %u does not own session %llu", app->app_index,
 625                     mp->handle);
 626       return;
 627     }
 628   owner_app_wrk_map = app_wrk->wrk_map_index;
 629   app_wrk = application_get_worker (app, mp->wrk_index);
 630
 631   /* This needs to come from the new owner */
 632   if (mp->req_wrk_index == owner_app_wrk_map)
 633     {
 634       session_req_worker_update_msg_t *wump;
 635
 636       svm_msg_q_lock_and_alloc_msg_w_ring (app_wrk->event_queue,
 637                                            SESSION_MQ_CTRL_EVT_RING,
 638                                            SVM_Q_WAIT, msg);
 639       evt = svm_msg_q_msg_data (app_wrk->event_queue, msg);
 640       clib_memset (evt, 0, sizeof (*evt));
 641       evt->event_type = SESSION_CTRL_EVT_REQ_WORKER_UPDATE;
 642       wump = (session_req_worker_update_msg_t *) evt->data;
 643       wump->session_handle = mp->handle;
 644       svm_msg_q_add_and_unlock (app_wrk->event_queue, msg);
 645       return;
 646     }
 647
 648   rv = app_worker_own_session (app_wrk, s);
 649   if (rv)
 650     session_stat_error_inc (rv, 1);
 651
 652   /*
 653    * Send reply
 654    */
 655   svm_msg_q_lock_and_alloc_msg_w_ring (app_wrk->event_queue,
 656                                        SESSION_MQ_CTRL_EVT_RING,
 657                                        SVM_Q_WAIT, msg);
 658   evt = svm_msg_q_msg_data (app_wrk->event_queue, msg);
 659   clib_memset (evt, 0, sizeof (*evt));
 660   evt->event_type = SESSION_CTRL_EVT_WORKER_UPDATE_REPLY;
 661   rmp = (session_worker_update_reply_msg_t *) evt->data;
 662   rmp->handle = mp->handle;
 663   if (s->rx_fifo)
 664     rmp->rx_fifo = fifo_segment_fifo_offset (s->rx_fifo);
 665   if (s->tx_fifo)
 666     rmp->tx_fifo = fifo_segment_fifo_offset (s->tx_fifo);
 667   rmp->segment_handle = session_segment_handle (s);
 668   svm_msg_q_add_and_unlock (app_wrk->event_queue, msg);
 669
 670   /*
 671    * Retransmit messages that may have been lost
 672    */
 673   if (s->tx_fifo && !svm_fifo_is_empty (s->tx_fifo))
 674     session_send_io_evt_to_thread (s->tx_fifo, SESSION_IO_EVT_TX);
 675
 676   if (s->rx_fifo && !svm_fifo_is_empty (s->rx_fifo))
 677     app_worker_rx_notify (app_wrk, s);
 678
 679   if (s->session_state >= SESSION_STATE_TRANSPORT_CLOSING)
 680     app_worker_close_notify (app_wrk, s);
 681 }
 682
 683 static void
 684 session_mq_app_wrk_rpc_handler (void *data)
 685 {
 686   session_app_wrk_rpc_msg_t *mp = (session_app_wrk_rpc_msg_t *) data;
 687   svm_msg_q_msg_t _msg, *msg = &_msg;
 688   session_app_wrk_rpc_msg_t *rmp;
 689   app_worker_t *app_wrk;
 690   session_event_t *evt;
 691   application_t *app;
 692
 693   app = application_lookup (mp->client_index);
 694   if (!app)
 695     return;
 696
 697   app_wrk = application_get_worker (app, mp->wrk_index);
 698
 699   svm_msg_q_lock_and_alloc_msg_w_ring (app_wrk->event_queue,
 700                                        SESSION_MQ_CTRL_EVT_RING, SVM_Q_WAIT,
 701                                        msg);
 702   evt = svm_msg_q_msg_data (app_wrk->event_queue, msg);
 703   clib_memset (evt, 0, sizeof (*evt));
 704   evt->event_type = SESSION_CTRL_EVT_APP_WRK_RPC;
 705   rmp = (session_app_wrk_rpc_msg_t *) evt->data;
 706   clib_memcpy (rmp->data, mp->data, sizeof (mp->data));
 707   svm_msg_q_add_and_unlock (app_wrk->event_queue, msg);
 708 }
 709
 710 static void
 711 session_mq_transport_attr_handler (void *data)
 712 {
 713   session_transport_attr_msg_t *mp = (session_transport_attr_msg_t *) data;
 714   session_transport_attr_reply_msg_t *rmp;
 715   svm_msg_q_msg_t _msg, *msg = &_msg;
 716   app_worker_t *app_wrk;
 717   session_event_t *evt;
 718   application_t *app;
 719   session_t *s;
 720   int rv;
 721
 722   app = application_lookup (mp->client_index);
 723   if (!app)
 724     return;
 725
 726   if (!(s = session_get_from_handle_if_valid (mp->handle)))
 727     {
 728       clib_warning ("invalid handle %llu", mp->handle);
 729       return;
 730     }
 731   app_wrk = app_worker_get (s->app_wrk_index);
 732   if (app_wrk->app_index != app->app_index)
 733     {
 734       clib_warning ("app %u does not own session %llu", app->app_index,
 735                     mp->handle);
 736       return;
 737     }
 738
 739   rv = session_transport_attribute (s, mp->is_get, &mp->attr);
 740
 741   svm_msg_q_lock_and_alloc_msg_w_ring (
 742     app_wrk->event_queue, SESSION_MQ_CTRL_EVT_RING, SVM_Q_WAIT, msg);
 743   evt = svm_msg_q_msg_data (app_wrk->event_queue, msg);
 744   clib_memset (evt, 0, sizeof (*evt));
 745   evt->event_type = SESSION_CTRL_EVT_TRANSPORT_ATTR_REPLY;
 746   rmp = (session_transport_attr_reply_msg_t *) evt->data;
 747   rmp->handle = mp->handle;
 748   rmp->retval = rv;
 749   rmp->is_get = mp->is_get;
 750   if (!rv && mp->is_get)
 751     rmp->attr = mp->attr;
 752   svm_msg_q_add_and_unlock (app_wrk->event_queue, msg);
 753 }
 754
 755 void
 756 session_wrk_handle_evts_main_rpc (void *args)
 757 {
 758   vlib_main_t *vm = vlib_get_main ();
 759   clib_llist_index_t ei, next_ei;
 760   session_evt_elt_t *he, *elt;
 761   session_worker_t *fwrk;
 762   u32 thread_index;
 763
 764   vlib_worker_thread_barrier_sync (vm);
 765
 766   thread_index = pointer_to_uword (args);
 767   fwrk = session_main_get_worker (thread_index);
 768
 769   he = clib_llist_elt (fwrk->event_elts, fwrk->evts_pending_main);
 770   ei = clib_llist_next_index (he, evt_list);
 771
 772   while (ei != fwrk->evts_pending_main)
 773     {
 774       elt = clib_llist_elt (fwrk->event_elts, ei);
 775       next_ei = clib_llist_next_index (elt, evt_list);
 776       clib_llist_remove (fwrk->event_elts, evt_list, elt);
 777       switch (elt->evt.event_type)
 778         {
 779         case SESSION_CTRL_EVT_LISTEN:
 780           session_mq_listen_handler (fwrk, elt);
 781           break;
 782         case SESSION_CTRL_EVT_UNLISTEN:
 783           session_mq_unlisten_handler (fwrk, elt);
 784           break;
 785         case SESSION_CTRL_EVT_APP_DETACH:
 786           app_mq_detach_handler (fwrk, elt);
 787           break;
 788         case SESSION_CTRL_EVT_CONNECT_URI:
 789           session_mq_connect_uri_handler (fwrk, elt);
 790           break;
 791         case SESSION_CTRL_EVT_ACCEPTED_REPLY:
 792           session_mq_accepted_reply_handler (fwrk, elt);
 793           break;
 794         case SESSION_CTRL_EVT_CONNECT:
 795           session_mq_connect_handler (fwrk, elt);
 796           break;
 797         default:
 798           clib_warning ("unhandled %u", elt->evt.event_type);
 799           ALWAYS_ASSERT (0);
 800           break;
 801         }
 802
 803       /* Regrab element in case pool moved */
 804       elt = clib_llist_elt (fwrk->event_elts, ei);
 805       if (!clib_llist_elt_is_linked (elt, evt_list))
 806         {
 807           session_evt_ctrl_data_free (fwrk, elt);
 808           clib_llist_put (fwrk->event_elts, elt);
 809         }
 810       ei = next_ei;
 811     }
 812
 813   vlib_worker_thread_barrier_release (vm);
 814 }
 815
 816 vlib_node_registration_t session_queue_node;
 817
 818 typedef struct
 819 {
 820   u32 session_index;
 821   u32 server_thread_index;
 822 } session_queue_trace_t;
 823
 824 /* packet trace format function */
 825 static u8 *
 826 format_session_queue_trace (u8 * s, va_list * args)
 827 {
 828   CLIB_UNUSED (vlib_main_t * vm) = va_arg (*args, vlib_main_t *);
 829   CLIB_UNUSED (vlib_node_t * node) = va_arg (*args, vlib_node_t *);
 830   session_queue_trace_t *t = va_arg (*args, session_queue_trace_t *);
 831
 832   s = format (s, "session index %d thread index %d",
 833               t->session_index, t->server_thread_index);
 834   return s;
 835 }
 836
 837 #define foreach_session_queue_error                                           \
 838   _ (TX, tx, INFO, "Packets transmitted")                                     \
 839   _ (TIMER, timer, INFO, "Timer events")                                      \
 840   _ (NO_BUFFER, no_buffer, ERROR, "Out of buffers")
 841
 842 typedef enum
 843 {
 844 #define _(f, n, s, d) SESSION_QUEUE_ERROR_##f,
 845   foreach_session_queue_error
 846 #undef _
 847     SESSION_QUEUE_N_ERROR,
 848 } session_queue_error_t;
 849
 850 static vlib_error_desc_t session_error_counters[] = {
 851 #define _(f, n, s, d) { #n, d, VL_COUNTER_SEVERITY_##s },
 852   foreach_session_queue_error
 853 #undef _
 854 };
 855
 856 enum
 857 {
 858   SESSION_TX_NO_BUFFERS = -2,
 859   SESSION_TX_NO_DATA,
 860   SESSION_TX_OK
 861 };
 862
 863 static void
 864 session_tx_trace_frame (vlib_main_t *vm, vlib_node_runtime_t *node,
 865                         u32 next_index, vlib_buffer_t **bufs, u16 n_segs,
 866                         session_t *s, u32 n_trace)
 867 {
 868   vlib_buffer_t **b = bufs;
 869
 870   while (n_trace && n_segs)
 871     {
 872       if (PREDICT_TRUE (vlib_trace_buffer (vm, node, next_index, b[0],
 873                                            1 /* follow_chain */)))
 874         {
 875           session_queue_trace_t *t =
 876             vlib_add_trace (vm, node, b[0], sizeof (*t));
 877           t->session_index = s->session_index;
 878           t->server_thread_index = s->thread_index;
 879           n_trace--;
 880         }
 881       b++;
 882       n_segs--;
 883     }
 884   vlib_set_trace_count (vm, node, n_trace);
 885 }
 886
 887 always_inline int
 888 session_tx_fill_dma_transfers (session_worker_t *wrk,
 889                                session_tx_context_t *ctx, vlib_buffer_t *b)
 890 {
 891   vlib_main_t *vm = wrk->vm;
 892   u32 len_to_deq;
 893   u8 *data0 = NULL;
 894   int n_bytes_read, len_write;
 895   svm_fifo_seg_t data_fs[2];
 896
 897   u32 n_segs = 2;
 898   u16 n_transfers = 0;
 899   /*
 900    * Start with the first buffer in chain
 901    */
 902   b->error = 0;
 903   b->flags = VNET_BUFFER_F_LOCALLY_ORIGINATED;
 904   b->current_data = 0;
 905   data0 = vlib_buffer_make_headroom (b, TRANSPORT_MAX_HDRS_LEN);
 906   len_to_deq = clib_min (ctx->left_to_snd, ctx->deq_per_first_buf);
 907
 908   n_bytes_read = svm_fifo_segments (ctx->s->tx_fifo, ctx->sp.tx_offset,
 909                                     data_fs, &n_segs, len_to_deq);
 910
 911   len_write = n_bytes_read;
 912   ASSERT (n_bytes_read == len_to_deq);
 913
 914   while (n_bytes_read)
 915     {
 916       wrk->batch_num++;
 917       vlib_dma_batch_add (vm, wrk->batch, data0, data_fs[n_transfers].data,
 918                           data_fs[n_transfers].len);
 919       data0 += data_fs[n_transfers].len;
 920       n_bytes_read -= data_fs[n_transfers].len;
 921       n_transfers++;
 922     }
 923   return len_write;
 924 }
 925
 926 always_inline int
 927 session_tx_fill_dma_transfers_tail (session_worker_t *wrk,
 928                                     session_tx_context_t *ctx,
 929                                     vlib_buffer_t *b, u32 len_to_deq, u8 *data)
 930 {
 931   vlib_main_t *vm = wrk->vm;
 932   int n_bytes_read, len_write;
 933   svm_fifo_seg_t data_fs[2];
 934   u32 n_segs = 2;
 935   u16 n_transfers = 0;
 936
 937   n_bytes_read = svm_fifo_segments (ctx->s->tx_fifo, ctx->sp.tx_offset,
 938                                     data_fs, &n_segs, len_to_deq);
 939
 940   len_write = n_bytes_read;
 941
 942   ASSERT (n_bytes_read == len_to_deq);
 943
 944   while (n_bytes_read)
 945     {
 946       wrk->batch_num++;
 947       vlib_dma_batch_add (vm, wrk->batch, data, data_fs[n_transfers].data,
 948                           data_fs[n_transfers].len);
 949       data += data_fs[n_transfers].len;
 950       n_bytes_read -= data_fs[n_transfers].len;
 951       n_transfers++;
 952     }
 953
 954   return len_write;
 955 }
 956
 957 always_inline int
 958 session_tx_copy_data (session_worker_t *wrk, session_tx_context_t *ctx,
 959                       vlib_buffer_t *b, u32 len_to_deq, u8 *data0)
 960 {
 961   int n_bytes_read;
 962   if (PREDICT_TRUE (!wrk->dma_enabled))
 963     n_bytes_read =
 964       svm_fifo_peek (ctx->s->tx_fifo, ctx->sp.tx_offset, len_to_deq, data0);
 965   else
 966     n_bytes_read = session_tx_fill_dma_transfers (wrk, ctx, b);
 967   return n_bytes_read;
 968 }
 969
 970 always_inline int
 971 session_tx_copy_data_tail (session_worker_t *wrk, session_tx_context_t *ctx,
 972                            vlib_buffer_t *b, u32 len_to_deq, u8 *data)
 973 {
 974   int n_bytes_read;
 975   if (PREDICT_TRUE (!wrk->dma_enabled))
 976     n_bytes_read =
 977       svm_fifo_peek (ctx->s->tx_fifo, ctx->sp.tx_offset, len_to_deq, data);
 978   else
 979     n_bytes_read =
 980       session_tx_fill_dma_transfers_tail (wrk, ctx, b, len_to_deq, data);
 981   return n_bytes_read;
 982 }
 983
 984 always_inline void
 985 session_tx_fifo_chain_tail (session_worker_t *wrk, session_tx_context_t *ctx,
 986                             vlib_buffer_t *b, u16 *n_bufs, u8 peek_data)
 987 {
 988   vlib_main_t *vm = wrk->vm;
 989   vlib_buffer_t *chain_b, *prev_b;
 990   u32 chain_bi0, to_deq, left_from_seg;
 991   int len_to_deq, n_bytes_read;
 992   u8 *data, j;
 993
 994   b->flags |= VLIB_BUFFER_TOTAL_LENGTH_VALID;
 995   b->total_length_not_including_first_buffer = 0;
 996
 997   chain_b = b;
 998   left_from_seg = clib_min (ctx->sp.snd_mss - b->current_length,
 999                             ctx->left_to_snd);
1000   to_deq = left_from_seg;
1001   for (j = 1; j < ctx->n_bufs_per_seg; j++)
1002     {
1003       prev_b = chain_b;
1004       len_to_deq = clib_min (to_deq, ctx->deq_per_buf);
1005
1006       *n_bufs -= 1;
1007       chain_bi0 = ctx->tx_buffers[*n_bufs];
1008       chain_b = vlib_get_buffer (vm, chain_bi0);
1009       chain_b->current_data = 0;
1010       data = vlib_buffer_get_current (chain_b);
1011       if (peek_data)
1012         {
1013           n_bytes_read =
1014             session_tx_copy_data_tail (wrk, ctx, b, len_to_deq, data);
1015           ctx->sp.tx_offset += n_bytes_read;
1016         }
1017       else
1018         {
1019           if (ctx->transport_vft->transport_options.tx_type ==
1020               TRANSPORT_TX_DGRAM)
1021             {
1022               svm_fifo_t *f = ctx->s->tx_fifo;
1023               session_dgram_hdr_t *hdr = &ctx->hdr;
1024               u16 deq_now;
1025               u32 offset;
1026
1027               deq_now = clib_min (hdr->data_length - hdr->data_offset,
1028                                   len_to_deq);
1029               offset = hdr->data_offset + SESSION_CONN_HDR_LEN;
1030               n_bytes_read = svm_fifo_peek (f, offset, deq_now, data);
1031               ASSERT (n_bytes_read > 0);
1032
1033               hdr->data_offset += n_bytes_read;
1034               if (hdr->data_offset == hdr->data_length)
1035                 {
1036                   offset = hdr->data_length + SESSION_CONN_HDR_LEN;
1037                   svm_fifo_dequeue_drop (f, offset);
1038                   if (ctx->left_to_snd > n_bytes_read)
1039                     svm_fifo_peek (ctx->s->tx_fifo, 0, sizeof (ctx->hdr),
1040                                    (u8 *) & ctx->hdr);
1041                 }
1042               else if (ctx->left_to_snd == n_bytes_read)
1043                 svm_fifo_overwrite_head (ctx->s->tx_fifo, (u8 *) & ctx->hdr,
1044                                          sizeof (session_dgram_pre_hdr_t));
1045             }
1046           else
1047             n_bytes_read = svm_fifo_dequeue (ctx->s->tx_fifo,
1048                                              len_to_deq, data);
1049         }
1050       ASSERT (n_bytes_read == len_to_deq);
1051       chain_b->current_length = n_bytes_read;
1052       b->total_length_not_including_first_buffer += chain_b->current_length;
1053
1054       /* update previous buffer */
1055       prev_b->next_buffer = chain_bi0;
1056       prev_b->flags |= VLIB_BUFFER_NEXT_PRESENT;
1057
1058       /* update current buffer */
1059       chain_b->next_buffer = 0;
1060
1061       to_deq -= n_bytes_read;
1062       if (to_deq == 0)
1063         break;
1064     }
1065   ASSERT (to_deq == 0
1066           && b->total_length_not_including_first_buffer == left_from_seg);
1067   ctx->left_to_snd -= left_from_seg;
1068 }
1069
1070 always_inline void
1071 session_tx_fill_buffer (session_worker_t *wrk, session_tx_context_t *ctx,
1072                         vlib_buffer_t *b, u16 *n_bufs, u8 peek_data)
1073 {
1074   u32 len_to_deq;
1075   u8 *data0;
1076   int n_bytes_read;
1077   /*
1078    * Start with the first buffer in chain
1079    */
1080   b->error = 0;
1081   b->flags = VNET_BUFFER_F_LOCALLY_ORIGINATED;
1082   b->current_data = 0;
1083
1084   data0 = vlib_buffer_make_headroom (b, TRANSPORT_MAX_HDRS_LEN);
1085   len_to_deq = clib_min (ctx->left_to_snd, ctx->deq_per_first_buf);
1086
1087   if (peek_data)
1088     {
1089       n_bytes_read = session_tx_copy_data (wrk, ctx, b, len_to_deq, data0);
1090       ASSERT (n_bytes_read > 0);
1091       /* Keep track of progress locally, transport is also supposed to
1092        * increment it independently when pushing the header */
1093       ctx->sp.tx_offset += n_bytes_read;
1094     }
1095   else
1096     {
1097       if (ctx->transport_vft->transport_options.tx_type == TRANSPORT_TX_DGRAM)
1098         {
1099           session_dgram_hdr_t *hdr = &ctx->hdr;
1100           svm_fifo_t *f = ctx->s->tx_fifo;
1101           u16 deq_now;
1102           u32 offset;
1103
1104           ASSERT (hdr->data_length > hdr->data_offset);
1105           deq_now = clib_min (hdr->data_length - hdr->data_offset,
1106                               len_to_deq);
1107           offset = hdr->data_offset + SESSION_CONN_HDR_LEN;
1108           n_bytes_read = svm_fifo_peek (f, offset, deq_now, data0);
1109           ASSERT (n_bytes_read > 0);
1110
1111           if (transport_connection_is_cless (ctx->tc))
1112             {
1113               clib_memcpy_fast (data0 - sizeof (session_dgram_hdr_t), hdr,
1114                                 sizeof (*hdr));
1115             }
1116           hdr->data_offset += n_bytes_read;
1117           if (hdr->data_offset == hdr->data_length)
1118             {
1119               offset = hdr->data_length + SESSION_CONN_HDR_LEN;
1120               svm_fifo_dequeue_drop (f, offset);
1121               if (ctx->left_to_snd > n_bytes_read)
1122                 svm_fifo_peek (ctx->s->tx_fifo, 0, sizeof (ctx->hdr),
1123                                (u8 *) & ctx->hdr);
1124             }
1125           else if (ctx->left_to_snd == n_bytes_read)
1126             svm_fifo_overwrite_head (ctx->s->tx_fifo, (u8 *) & ctx->hdr,
1127                                      sizeof (session_dgram_pre_hdr_t));
1128         }
1129       else
1130         {
1131           n_bytes_read = svm_fifo_dequeue (ctx->s->tx_fifo,
1132                                            len_to_deq, data0);
1133           ASSERT (n_bytes_read > 0);
1134         }
1135     }
1136
1137   b->current_length = n_bytes_read;
1138   ctx->left_to_snd -= n_bytes_read;
1139
1140   /*
1141    * Fill in the remaining buffers in the chain, if any
1142    */
1143   if (PREDICT_FALSE (ctx->n_bufs_per_seg > 1 && ctx->left_to_snd))
1144     session_tx_fifo_chain_tail (wrk, ctx, b, n_bufs, peek_data);
1145 }
1146
1147 always_inline u8
1148 session_tx_not_ready (session_t * s, u8 peek_data)
1149 {
1150   if (peek_data)
1151     {
1152       if (PREDICT_TRUE (s->session_state == SESSION_STATE_READY))
1153         return 0;
1154       /* Can retransmit for closed sessions but can't send new data if
1155        * session is not ready or closed */
1156       else if (s->session_state < SESSION_STATE_READY)
1157         {
1158           /* Allow accepting session to send custom packets.
1159            * For instance, tcp want to send acks in established, but
1160            * the app has not called accept() yet */
1161           if (s->session_state == SESSION_STATE_ACCEPTING &&
1162               (s->flags & SESSION_F_CUSTOM_TX))
1163             return 0;
1164           return 1;
1165         }
1166       else if (s->session_state >= SESSION_STATE_TRANSPORT_CLOSED)
1167         {
1168           /* Allow closed transports to still send custom packets.
1169            * For instance, tcp may want to send acks in time-wait. */
1170           if (s->session_state != SESSION_STATE_TRANSPORT_DELETED
1171               && (s->flags & SESSION_F_CUSTOM_TX))
1172             return 0;
1173           return 2;
1174         }
1175     }
1176   else
1177     {
1178       if (s->session_state == SESSION_STATE_TRANSPORT_DELETED)
1179         return 2;
1180     }
1181   return 0;
1182 }
1183
1184 always_inline transport_connection_t *
1185 session_tx_get_transport (session_tx_context_t * ctx, u8 peek_data)
1186 {
1187   if (peek_data)
1188     {
1189       return ctx->transport_vft->get_connection (ctx->s->connection_index,
1190                                                  ctx->s->thread_index);
1191     }
1192   else
1193     {
1194       if (ctx->s->session_state == SESSION_STATE_LISTENING)
1195         return ctx->transport_vft->get_listener (ctx->s->connection_index);
1196       else
1197         {
1198           return ctx->transport_vft->get_connection (ctx->s->connection_index,
1199                                                      ctx->s->thread_index);
1200         }
1201     }
1202 }
1203
1204 always_inline void
1205 session_tx_set_dequeue_params (vlib_main_t * vm, session_tx_context_t * ctx,
1206                                u32 max_segs, u8 peek_data)
1207 {
1208   u32 n_bytes_per_buf, n_bytes_per_seg;
1209
1210   n_bytes_per_buf = vlib_buffer_get_default_data_size (vm);
1211   ctx->max_dequeue = svm_fifo_max_dequeue_cons (ctx->s->tx_fifo);
1212
1213   if (peek_data)
1214     {
1215       /* Offset in rx fifo from where to peek data */
1216       if (PREDICT_FALSE (ctx->sp.tx_offset >= ctx->max_dequeue))
1217         {
1218           ctx->max_len_to_snd = 0;
1219           return;
1220         }
1221       ctx->max_dequeue -= ctx->sp.tx_offset;
1222     }
1223   else
1224     {
1225       if (ctx->transport_vft->transport_options.tx_type == TRANSPORT_TX_DGRAM)
1226         {
1227           u32 len, chain_limit;
1228
1229           if (ctx->max_dequeue <= sizeof (ctx->hdr))
1230             {
1231               ctx->max_len_to_snd = 0;
1232               return;
1233             }
1234
1235           svm_fifo_peek (ctx->s->tx_fifo, 0, sizeof (ctx->hdr),
1236                          (u8 *) & ctx->hdr);
1237           /* Zero length dgrams not supported */
1238           if (PREDICT_FALSE (ctx->hdr.data_length == 0))
1239             {
1240               svm_fifo_dequeue_drop (ctx->s->tx_fifo, sizeof (ctx->hdr));
1241               ctx->max_len_to_snd = 0;
1242               return;
1243             }
1244           /* We cannot be sure apps have not enqueued incomplete dgrams */
1245           if (PREDICT_FALSE (ctx->max_dequeue <
1246                              ctx->hdr.data_length + sizeof (ctx->hdr)))
1247             {
1248               ctx->max_len_to_snd = 0;
1249               return;
1250             }
1251           ASSERT (ctx->hdr.data_length > ctx->hdr.data_offset);
1252           len = ctx->hdr.data_length - ctx->hdr.data_offset;
1253
1254           if (ctx->hdr.gso_size)
1255             {
1256               ctx->sp.snd_mss = clib_min (ctx->sp.snd_mss, ctx->hdr.gso_size);
1257             }
1258
1259           /* Process multiple dgrams if smaller than min (buf_space, mss).
1260            * This avoids handling multiple dgrams if they require buffer
1261            * chains */
1262           chain_limit = clib_min (n_bytes_per_buf - TRANSPORT_MAX_HDRS_LEN,
1263                                   ctx->sp.snd_mss);
1264           if (ctx->hdr.data_length <= chain_limit)
1265             {
1266               u32 first_dgram_len, dgram_len, offset, max_offset;
1267               session_dgram_hdr_t hdr;
1268
1269               ctx->sp.snd_mss = clib_min (ctx->sp.snd_mss, len);
1270               offset = ctx->hdr.data_length + sizeof (session_dgram_hdr_t);
1271               first_dgram_len = len;
1272               max_offset = clib_min (ctx->max_dequeue, 16 << 10);
1273
1274               while (offset < max_offset)
1275                 {
1276                   svm_fifo_peek (ctx->s->tx_fifo, offset, sizeof (ctx->hdr),
1277                                  (u8 *) & hdr);
1278                   dgram_len = hdr.data_length - hdr.data_offset;
1279                   if (offset + sizeof (hdr) + hdr.data_length >
1280                         ctx->max_dequeue ||
1281                       first_dgram_len != dgram_len)
1282                     break;
1283                   /* Assert here to allow test above with zero length dgrams */
1284                   ASSERT (hdr.data_length > hdr.data_offset);
1285                   len += dgram_len;
1286                   offset += sizeof (hdr) + hdr.data_length;
1287                 }
1288             }
1289
1290           ctx->max_dequeue = len;
1291         }
1292     }
1293   ASSERT (ctx->max_dequeue > 0);
1294
1295   /* Ensure we're not writing more than transport window allows */
1296   if (ctx->max_dequeue < ctx->sp.snd_space)
1297     {
1298       /* Constrained by tx queue. Try to send only fully formed segments */
1299       ctx->max_len_to_snd = (ctx->max_dequeue > ctx->sp.snd_mss) ?
1300         (ctx->max_dequeue - (ctx->max_dequeue % ctx->sp.snd_mss)) :
1301         ctx->max_dequeue;
1302       /* TODO Nagle ? */
1303     }
1304   else
1305     {
1306       /* Expectation is that snd_space0 is already a multiple of snd_mss */
1307       ctx->max_len_to_snd = ctx->sp.snd_space;
1308     }
1309
1310   /* Check if we're tx constrained by the node */
1311   ctx->n_segs_per_evt = ceil ((f64) ctx->max_len_to_snd / ctx->sp.snd_mss);
1312   if (ctx->n_segs_per_evt > max_segs)
1313     {
1314       ctx->n_segs_per_evt = max_segs;
1315       ctx->max_len_to_snd = max_segs * ctx->sp.snd_mss;
1316     }
1317
1318   ASSERT (n_bytes_per_buf > TRANSPORT_MAX_HDRS_LEN);
1319   if (ctx->n_segs_per_evt > 1)
1320     {
1321       u32 n_bytes_last_seg, n_bufs_last_seg;
1322
1323       n_bytes_per_seg = TRANSPORT_MAX_HDRS_LEN + ctx->sp.snd_mss;
1324       n_bytes_last_seg = TRANSPORT_MAX_HDRS_LEN + ctx->max_len_to_snd
1325         - ((ctx->n_segs_per_evt - 1) * ctx->sp.snd_mss);
1326       ctx->n_bufs_per_seg = ceil ((f64) n_bytes_per_seg / n_bytes_per_buf);
1327       n_bufs_last_seg = ceil ((f64) n_bytes_last_seg / n_bytes_per_buf);
1328       ctx->n_bufs_needed = ((ctx->n_segs_per_evt - 1) * ctx->n_bufs_per_seg)
1329         + n_bufs_last_seg;
1330     }
1331   else
1332     {
1333       n_bytes_per_seg = TRANSPORT_MAX_HDRS_LEN + ctx->max_len_to_snd;
1334       ctx->n_bufs_per_seg = ceil ((f64) n_bytes_per_seg / n_bytes_per_buf);
1335       ctx->n_bufs_needed = ctx->n_bufs_per_seg;
1336     }
1337
1338   ctx->deq_per_buf = clib_min (ctx->sp.snd_mss, n_bytes_per_buf);
1339   ctx->deq_per_first_buf = clib_min (ctx->sp.snd_mss,
1340                                      n_bytes_per_buf -
1341                                      TRANSPORT_MAX_HDRS_LEN);
1342 }
1343
1344 always_inline void
1345 session_tx_maybe_reschedule (session_worker_t * wrk,
1346                              session_tx_context_t * ctx,
1347                              session_evt_elt_t * elt)
1348 {
1349   session_t *s = ctx->s;
1350
1351   svm_fifo_unset_event (s->tx_fifo);
1352   if (svm_fifo_max_dequeue_cons (s->tx_fifo) > ctx->sp.tx_offset)
1353     {
1354       if (svm_fifo_set_event (s->tx_fifo))
1355         session_evt_add_head_old (wrk, elt);
1356     }
1357   else
1358     {
1359       transport_connection_deschedule (ctx->tc);
1360     }
1361 }
1362
1363 always_inline void
1364 session_tx_add_pending_buffer (session_worker_t *wrk, u32 bi, u32 next_index)
1365 {
1366   if (PREDICT_TRUE (!wrk->dma_enabled))
1367     {
1368       vec_add1 (wrk->pending_tx_buffers, bi);
1369       vec_add1 (wrk->pending_tx_nexts, next_index);
1370     }
1371   else
1372     {
1373       session_dma_transfer *dma_transfer = &wrk->dma_trans[wrk->trans_tail];
1374       vec_add1 (dma_transfer->pending_tx_buffers, bi);
1375       vec_add1 (dma_transfer->pending_tx_nexts, next_index);
1376     }
1377 }
1378
1379 always_inline int
1380 session_tx_fifo_read_and_snd_i (session_worker_t * wrk,
1381                                 vlib_node_runtime_t * node,
1382                                 session_evt_elt_t * elt,
1383                                 int *n_tx_packets, u8 peek_data)
1384 {
1385   u32 n_trace, n_left, pbi, next_index, max_burst;
1386   session_tx_context_t *ctx = &wrk->ctx;
1387   session_main_t *smm = &session_main;
1388   session_event_t *e = &elt->evt;
1389   vlib_main_t *vm = wrk->vm;
1390   transport_proto_t tp;
1391   vlib_buffer_t *pb;
1392   u16 n_bufs, rv;
1393
1394   if (PREDICT_FALSE ((rv = session_tx_not_ready (ctx->s, peek_data))))
1395     {
1396       if (rv < 2)
1397         session_evt_add_old (wrk, elt);
1398       return SESSION_TX_NO_DATA;
1399     }
1400
1401   next_index = smm->session_type_to_next[ctx->s->session_type];
1402   max_burst = SESSION_NODE_FRAME_SIZE - *n_tx_packets;
1403
1404   tp = session_get_transport_proto (ctx->s);
1405   ctx->transport_vft = transport_protocol_get_vft (tp);
1406   ctx->tc = session_tx_get_transport (ctx, peek_data);
1407
1408   if (PREDICT_FALSE (e->event_type == SESSION_IO_EVT_TX_FLUSH))
1409     {
1410       if (ctx->transport_vft->flush_data)
1411         ctx->transport_vft->flush_data (ctx->tc);
1412       e->event_type = SESSION_IO_EVT_TX;
1413     }
1414
1415   if (ctx->s->flags & SESSION_F_CUSTOM_TX)
1416     {
1417       u32 n_custom_tx;
1418       ctx->s->flags &= ~SESSION_F_CUSTOM_TX;
1419       ctx->sp.max_burst_size = max_burst;
1420       n_custom_tx = ctx->transport_vft->custom_tx (ctx->tc, &ctx->sp);
1421       *n_tx_packets += n_custom_tx;
1422       if (PREDICT_FALSE
1423           (ctx->s->session_state >= SESSION_STATE_TRANSPORT_CLOSED))
1424         return SESSION_TX_OK;
1425       max_burst -= n_custom_tx;
1426       if (!max_burst || (ctx->s->flags & SESSION_F_CUSTOM_TX))
1427         {
1428           session_evt_add_old (wrk, elt);
1429           return SESSION_TX_OK;
1430         }
1431     }
1432
1433   /* Connection previously descheduled because it had no data to send.
1434    * Clear descheduled flag and reset pacer if in use */
1435   if (transport_connection_is_descheduled (ctx->tc))
1436     transport_connection_clear_descheduled (ctx->tc);
1437
1438   transport_connection_snd_params (ctx->tc, &ctx->sp);
1439
1440   if (!ctx->sp.snd_space)
1441     {
1442       /* If the deschedule flag was set, remove session from scheduler.
1443        * Transport is responsible for rescheduling this session. */
1444       if (ctx->sp.flags & TRANSPORT_SND_F_DESCHED)
1445         transport_connection_deschedule (ctx->tc);
1446       /* Request to postpone the session, e.g., zero-wnd and transport
1447        * is not currently probing */
1448       else if (ctx->sp.flags & TRANSPORT_SND_F_POSTPONE)
1449         session_evt_add_old (wrk, elt);
1450       /* This flow queue is "empty" so it should be re-evaluated before
1451        * the ones that have data to send. */
1452       else
1453         session_evt_add_head_old (wrk, elt);
1454
1455       return SESSION_TX_NO_DATA;
1456     }
1457
1458   if (transport_connection_is_tx_paced (ctx->tc))
1459     {
1460       u32 snd_space = transport_connection_tx_pacer_burst (ctx->tc);
1461       if (snd_space < TRANSPORT_PACER_MIN_BURST)
1462         {
1463           session_evt_add_head_old (wrk, elt);
1464           return SESSION_TX_NO_DATA;
1465         }
1466       snd_space = clib_min (ctx->sp.snd_space, snd_space);
1467       ctx->sp.snd_space = snd_space >= ctx->sp.snd_mss ?
1468         snd_space - snd_space % ctx->sp.snd_mss : snd_space;
1469     }
1470
1471   /* Check how much we can pull. */
1472   session_tx_set_dequeue_params (vm, ctx, max_burst, peek_data);
1473
1474   if (PREDICT_FALSE (!ctx->max_len_to_snd))
1475     {
1476       transport_connection_tx_pacer_reset_bucket (ctx->tc, 0);
1477       session_tx_maybe_reschedule (wrk, ctx, elt);
1478       return SESSION_TX_NO_DATA;
1479     }
1480
1481   vec_validate_aligned (ctx->tx_buffers, ctx->n_bufs_needed - 1,
1482                         CLIB_CACHE_LINE_BYTES);
1483   n_bufs = vlib_buffer_alloc (vm, ctx->tx_buffers, ctx->n_bufs_needed);
1484   if (PREDICT_FALSE (n_bufs < ctx->n_bufs_needed))
1485     {
1486       if (n_bufs)
1487         vlib_buffer_free (vm, ctx->tx_buffers, n_bufs);
1488       session_evt_add_head_old (wrk, elt);
1489       vlib_node_increment_counter (wrk->vm, node->node_index,
1490                                    SESSION_QUEUE_ERROR_NO_BUFFER, 1);
1491       return SESSION_TX_NO_BUFFERS;
1492     }
1493
1494   if (transport_connection_is_tx_paced (ctx->tc))
1495     transport_connection_tx_pacer_update_bytes (ctx->tc, ctx->max_len_to_snd);
1496
1497   ctx->left_to_snd = ctx->max_len_to_snd;
1498   n_left = ctx->n_segs_per_evt;
1499
1500   vec_validate (ctx->transport_pending_bufs, n_left);
1501
1502   while (n_left >= 4)
1503     {
1504       vlib_buffer_t *b0, *b1;
1505       u32 bi0, bi1;
1506
1507       pbi = ctx->tx_buffers[n_bufs - 3];
1508       pb = vlib_get_buffer (vm, pbi);
1509       vlib_prefetch_buffer_header (pb, STORE);
1510       pbi = ctx->tx_buffers[n_bufs - 4];
1511       pb = vlib_get_buffer (vm, pbi);
1512       vlib_prefetch_buffer_header (pb, STORE);
1513
1514       bi0 = ctx->tx_buffers[--n_bufs];
1515       bi1 = ctx->tx_buffers[--n_bufs];
1516
1517       b0 = vlib_get_buffer (vm, bi0);
1518       b1 = vlib_get_buffer (vm, bi1);
1519
1520       session_tx_fill_buffer (wrk, ctx, b0, &n_bufs, peek_data);
1521       session_tx_fill_buffer (wrk, ctx, b1, &n_bufs, peek_data);
1522
1523       ctx->transport_pending_bufs[ctx->n_segs_per_evt - n_left] = b0;
1524       ctx->transport_pending_bufs[ctx->n_segs_per_evt - n_left + 1] = b1;
1525       n_left -= 2;
1526
1527       session_tx_add_pending_buffer (wrk, bi0, next_index);
1528       session_tx_add_pending_buffer (wrk, bi1, next_index);
1529     }
1530   while (n_left)
1531     {
1532       vlib_buffer_t *b0;
1533       u32 bi0;
1534
1535       if (n_left > 1)
1536         {
1537           pbi = ctx->tx_buffers[n_bufs - 2];
1538           pb = vlib_get_buffer (vm, pbi);
1539           vlib_prefetch_buffer_header (pb, STORE);
1540         }
1541
1542       bi0 = ctx->tx_buffers[--n_bufs];
1543       b0 = vlib_get_buffer (vm, bi0);
1544       session_tx_fill_buffer (wrk, ctx, b0, &n_bufs, peek_data);
1545
1546       ctx->transport_pending_bufs[ctx->n_segs_per_evt - n_left] = b0;
1547       n_left -= 1;
1548
1549       session_tx_add_pending_buffer (wrk, bi0, next_index);
1550     }
1551
1552   /* Ask transport to push headers */
1553   ctx->transport_vft->push_header (ctx->tc, ctx->transport_pending_bufs,
1554                                    ctx->n_segs_per_evt);
1555
1556   if (PREDICT_FALSE ((n_trace = vlib_get_trace_count (vm, node)) > 0))
1557     session_tx_trace_frame (vm, node, next_index, ctx->transport_pending_bufs,
1558                             ctx->n_segs_per_evt, ctx->s, n_trace);
1559
1560   if (PREDICT_FALSE (n_bufs))
1561     vlib_buffer_free (vm, ctx->tx_buffers, n_bufs);
1562
1563   *n_tx_packets += ctx->n_segs_per_evt;
1564
1565   SESSION_EVT (SESSION_EVT_DEQ, ctx->s, ctx->max_len_to_snd, ctx->max_dequeue,
1566                ctx->s->tx_fifo->shr->has_event, wrk->last_vlib_time);
1567
1568   ASSERT (ctx->left_to_snd == 0);
1569
1570   /* If we couldn't dequeue all bytes reschedule as old flow. Otherwise,
1571    * check if application enqueued more data and reschedule accordingly */
1572   if (ctx->max_len_to_snd < ctx->max_dequeue)
1573     session_evt_add_old (wrk, elt);
1574   else
1575     session_tx_maybe_reschedule (wrk, ctx, elt);
1576
1577   if (!peek_data)
1578     {
1579       u32 n_dequeued = ctx->max_len_to_snd;
1580       if (ctx->transport_vft->transport_options.tx_type == TRANSPORT_TX_DGRAM)
1581         n_dequeued += ctx->n_segs_per_evt * SESSION_CONN_HDR_LEN;
1582       if (svm_fifo_needs_deq_ntf (ctx->s->tx_fifo, n_dequeued))
1583         session_dequeue_notify (ctx->s);
1584     }
1585   return SESSION_TX_OK;
1586 }
1587
1588 int
1589 session_tx_fifo_peek_and_snd (session_worker_t * wrk,
1590                               vlib_node_runtime_t * node,
1591                               session_evt_elt_t * e, int *n_tx_packets)
1592 {
1593   return session_tx_fifo_read_and_snd_i (wrk, node, e, n_tx_packets, 1);
1594 }
1595
1596 int
1597 session_tx_fifo_dequeue_and_snd (session_worker_t * wrk,
1598                                  vlib_node_runtime_t * node,
1599                                  session_evt_elt_t * e, int *n_tx_packets)
1600 {
1601   return session_tx_fifo_read_and_snd_i (wrk, node, e, n_tx_packets, 0);
1602 }
1603
1604 int
1605 session_tx_fifo_dequeue_internal (session_worker_t * wrk,
1606                                   vlib_node_runtime_t * node,
1607                                   session_evt_elt_t * elt, int *n_tx_packets)
1608 {
1609   transport_send_params_t *sp = &wrk->ctx.sp;
1610   session_t *s = wrk->ctx.s;
1611   clib_llist_index_t ei;
1612   u32 n_packets;
1613
1614   if (PREDICT_FALSE (s->session_state >= SESSION_STATE_TRANSPORT_CLOSED))
1615     return 0;
1616
1617   /* Clear custom-tx flag used to request reschedule for tx */
1618   s->flags &= ~SESSION_F_CUSTOM_TX;
1619
1620   sp->flags = 0;
1621   sp->bytes_dequeued = 0;
1622   sp->max_burst_size = clib_min (SESSION_NODE_FRAME_SIZE - *n_tx_packets,
1623                                  TRANSPORT_PACER_MAX_BURST_PKTS);
1624
1625   /* Grab elt index since app transports can enqueue events on tx */
1626   ei = clib_llist_entry_index (wrk->event_elts, elt);
1627
1628   n_packets = transport_custom_tx (session_get_transport_proto (s), s, sp);
1629   *n_tx_packets += n_packets;
1630
1631   elt = clib_llist_elt (wrk->event_elts, ei);
1632
1633   if (s->flags & SESSION_F_CUSTOM_TX)
1634     {
1635       session_evt_add_old (wrk, elt);
1636     }
1637   else if (!(sp->flags & TRANSPORT_SND_F_DESCHED))
1638     {
1639       svm_fifo_unset_event (s->tx_fifo);
1640       if (svm_fifo_max_dequeue_cons (s->tx_fifo))
1641         if (svm_fifo_set_event (s->tx_fifo))
1642           session_evt_add_head_old (wrk, elt);
1643     }
1644
1645   if (sp->bytes_dequeued &&
1646       svm_fifo_needs_deq_ntf (s->tx_fifo, sp->bytes_dequeued))
1647     session_dequeue_notify (s);
1648
1649   return n_packets;
1650 }
1651
1652 always_inline session_t *
1653 session_event_get_session (session_worker_t * wrk, session_event_t * e)
1654 {
1655   if (PREDICT_FALSE (pool_is_free_index (wrk->sessions, e->session_index)))
1656     return 0;
1657
1658   ASSERT (session_is_valid (e->session_index, wrk->vm->thread_index));
1659   return pool_elt_at_index (wrk->sessions, e->session_index);
1660 }
1661
1662 always_inline void
1663 session_event_dispatch_ctrl (session_worker_t * wrk, session_evt_elt_t * elt)
1664 {
1665   clib_llist_index_t ei;
1666   void (*fp) (void *);
1667   session_event_t *e;
1668   session_t *s;
1669
1670   ei = clib_llist_entry_index (wrk->event_elts, elt);
1671   e = &elt->evt;
1672
1673   switch (e->event_type)
1674     {
1675     case SESSION_CTRL_EVT_RPC:
1676       fp = e->rpc_args.fp;
1677       (*fp) (e->rpc_args.arg);
1678       break;
1679     case SESSION_CTRL_EVT_HALF_CLOSE:
1680       s = session_get_from_handle_if_valid (e->session_handle);
1681       if (PREDICT_FALSE (!s))
1682         break;
1683       session_transport_half_close (s);
1684       break;
1685     case SESSION_CTRL_EVT_CLOSE:
1686       s = session_get_from_handle_if_valid (e->session_handle);
1687       if (PREDICT_FALSE (!s))
1688         break;
1689       session_transport_close (s);
1690       break;
1691     case SESSION_CTRL_EVT_RESET:
1692       s = session_get_from_handle_if_valid (e->session_handle);
1693       if (PREDICT_FALSE (!s))
1694         break;
1695       session_transport_reset (s);
1696       break;
1697     case SESSION_CTRL_EVT_LISTEN:
1698       session_mq_listen_handler (wrk, elt);
1699       break;
1700     case SESSION_CTRL_EVT_LISTEN_URI:
1701       session_mq_listen_uri_handler (wrk, elt);
1702       break;
1703     case SESSION_CTRL_EVT_UNLISTEN:
1704       session_mq_unlisten_handler (wrk, elt);
1705       break;
1706     case SESSION_CTRL_EVT_CONNECT:
1707       session_mq_connect_handler (wrk, elt);
1708       break;
1709     case SESSION_CTRL_EVT_CONNECT_URI:
1710       session_mq_connect_uri_handler (wrk, elt);
1711       break;
1712     case SESSION_CTRL_EVT_SHUTDOWN:
1713       session_mq_shutdown_handler (session_evt_ctrl_data (wrk, elt));
1714       break;
1715     case SESSION_CTRL_EVT_DISCONNECT:
1716       session_mq_disconnect_handler (session_evt_ctrl_data (wrk, elt));
1717       break;
1718     case SESSION_CTRL_EVT_DISCONNECTED:
1719       session_mq_disconnected_handler (session_evt_ctrl_data (wrk, elt));
1720       break;
1721     case SESSION_CTRL_EVT_ACCEPTED_REPLY:
1722       session_mq_accepted_reply_handler (wrk, elt);
1723       break;
1724     case SESSION_CTRL_EVT_DISCONNECTED_REPLY:
1725       session_mq_disconnected_reply_handler (session_evt_ctrl_data (wrk,
1726                                                                     elt));
1727       break;
1728     case SESSION_CTRL_EVT_RESET_REPLY:
1729       session_mq_reset_reply_handler (session_evt_ctrl_data (wrk, elt));
1730       break;
1731     case SESSION_CTRL_EVT_WORKER_UPDATE:
1732       session_mq_worker_update_handler (session_evt_ctrl_data (wrk, elt));
1733       break;
1734     case SESSION_CTRL_EVT_APP_DETACH:
1735       app_mq_detach_handler (wrk, elt);
1736       break;
1737     case SESSION_CTRL_EVT_APP_WRK_RPC:
1738       session_mq_app_wrk_rpc_handler (session_evt_ctrl_data (wrk, elt));
1739       break;
1740     case SESSION_CTRL_EVT_TRANSPORT_ATTR:
1741       session_mq_transport_attr_handler (session_evt_ctrl_data (wrk, elt));
1742       break;
1743     default:
1744       clib_warning ("unhandled event type %d", e->event_type);
1745     }
1746
1747   /* Regrab elements in case pool moved */
1748   elt = clib_llist_elt (wrk->event_elts, ei);
1749   if (!clib_llist_elt_is_linked (elt, evt_list))
1750     {
1751       e = &elt->evt;
1752       if (e->event_type >= SESSION_CTRL_EVT_BOUND)
1753         session_evt_ctrl_data_free (wrk, elt);
1754       clib_llist_put (wrk->event_elts, elt);
1755     }
1756   SESSION_EVT (SESSION_EVT_COUNTS, CNT_CTRL_EVTS, 1, wrk);
1757 }
1758
1759 always_inline void
1760 session_event_dispatch_io (session_worker_t * wrk, vlib_node_runtime_t * node,
1761                            session_evt_elt_t * elt, int *n_tx_packets)
1762 {
1763   session_main_t *smm = &session_main;
1764   app_worker_t *app_wrk;
1765   clib_llist_index_t ei;
1766   session_event_t *e;
1767   session_t *s;
1768
1769   ei = clib_llist_entry_index (wrk->event_elts, elt);
1770   e = &elt->evt;
1771
1772   switch (e->event_type)
1773     {
1774     case SESSION_IO_EVT_TX_FLUSH:
1775     case SESSION_IO_EVT_TX:
1776       s = session_event_get_session (wrk, e);
1777       if (PREDICT_FALSE (!s))
1778         break;
1779       CLIB_PREFETCH (s->tx_fifo, sizeof (*(s->tx_fifo)), LOAD);
1780       wrk->ctx.s = s;
1781       /* Spray packets in per session type frames, since they go to
1782        * different nodes */
1783       (smm->session_tx_fns[s->session_type]) (wrk, node, elt, n_tx_packets);
1784       break;
1785     case SESSION_IO_EVT_RX:
1786       s = session_event_get_session (wrk, e);
1787       if (!s)
1788         break;
1789       transport_app_rx_evt (session_get_transport_proto (s),
1790                             s->connection_index, s->thread_index);
1791       break;
1792     case SESSION_IO_EVT_BUILTIN_RX:
1793       s = session_event_get_session (wrk, e);
1794       if (PREDICT_FALSE (!s || s->session_state >= SESSION_STATE_CLOSING))
1795         break;
1796       svm_fifo_unset_event (s->rx_fifo);
1797       app_wrk = app_worker_get (s->app_wrk_index);
1798       app_worker_rx_notify (app_wrk, s);
1799       break;
1800     case SESSION_IO_EVT_TX_MAIN:
1801       s = session_get_if_valid (e->session_index, 0 /* main thread */);
1802       if (PREDICT_FALSE (!s))
1803         break;
1804       wrk->ctx.s = s;
1805       if (PREDICT_TRUE (s != 0))
1806         (smm->session_tx_fns[s->session_type]) (wrk, node, elt, n_tx_packets);
1807       break;
1808     default:
1809       clib_warning ("unhandled event type %d", e->event_type);
1810     }
1811
1812   SESSION_EVT (SESSION_EVT_IO_EVT_COUNTS, e->event_type, 1, wrk);
1813
1814   /* Regrab elements in case pool moved */
1815   elt = clib_llist_elt (wrk->event_elts, ei);
1816   if (!clib_llist_elt_is_linked (elt, evt_list))
1817     clib_llist_put (wrk->event_elts, elt);
1818 }
1819
1820 /* *INDENT-OFF* */
1821 static const u32 session_evt_msg_sizes[] = {
1822 #define _(symc, sym)                                                    \
1823   [SESSION_CTRL_EVT_ ## symc] = sizeof (session_ ## sym ##_msg_t),
1824   foreach_session_ctrl_evt
1825 #undef _
1826 };
1827 /* *INDENT-ON* */
1828
1829 always_inline void
1830 session_update_time_subscribers (session_main_t *smm, clib_time_type_t now,
1831                                  u32 thread_index)
1832 {
1833   session_update_time_fn *fn;
1834
1835   vec_foreach (fn, smm->update_time_fns)
1836     (*fn) (now, thread_index);
1837 }
1838
1839 always_inline void
1840 session_evt_add_to_list (session_worker_t * wrk, session_event_t * evt)
1841 {
1842   session_evt_elt_t *elt;
1843
1844   if (evt->event_type >= SESSION_CTRL_EVT_RPC)
1845     {
1846       elt = session_evt_alloc_ctrl (wrk);
1847       if (evt->event_type >= SESSION_CTRL_EVT_BOUND)
1848         {
1849           elt->evt.ctrl_data_index = session_evt_ctrl_data_alloc (wrk);
1850           elt->evt.event_type = evt->event_type;
1851           clib_memcpy_fast (session_evt_ctrl_data (wrk, elt), evt->data,
1852                             session_evt_msg_sizes[evt->event_type]);
1853         }
1854       else
1855         {
1856           /* Internal control events fit into io events footprint */
1857           clib_memcpy_fast (&elt->evt, evt, sizeof (elt->evt));
1858         }
1859     }
1860   else
1861     {
1862       elt = session_evt_alloc_new (wrk);
1863       clib_memcpy_fast (&elt->evt, evt, sizeof (elt->evt));
1864     }
1865 }
1866
1867 static void
1868 session_flush_pending_tx_buffers (session_worker_t * wrk,
1869                                   vlib_node_runtime_t * node)
1870 {
1871   vlib_buffer_enqueue_to_next_vec (wrk->vm, node, &wrk->pending_tx_buffers,
1872                                    &wrk->pending_tx_nexts,
1873                                    vec_len (wrk->pending_tx_nexts));
1874   vec_reset_length (wrk->pending_tx_buffers);
1875   vec_reset_length (wrk->pending_tx_nexts);
1876 }
1877
1878 int
1879 session_wrk_handle_mq (session_worker_t *wrk, svm_msg_q_t *mq)
1880 {
1881   svm_msg_q_msg_t _msg, *msg = &_msg;
1882   u32 i, n_to_dequeue = 0;
1883   session_event_t *evt;
1884
1885   n_to_dequeue = svm_msg_q_size (mq);
1886   for (i = 0; i < n_to_dequeue; i++)
1887     {
1888       svm_msg_q_sub_raw (mq, msg);
1889       evt = svm_msg_q_msg_data (mq, msg);
1890       session_evt_add_to_list (wrk, evt);
1891       svm_msg_q_free_msg (mq, msg);
1892     }
1893
1894   return n_to_dequeue;
1895 }
1896
1897 static void
1898 session_wrk_update_state (session_worker_t *wrk)
1899 {
1900   vlib_main_t *vm = wrk->vm;
1901
1902   if (wrk->state == SESSION_WRK_POLLING)
1903     {
1904       if (clib_llist_elts (wrk->event_elts) == 5 &&
1905           vlib_last_vectors_per_main_loop (vm) < 1)
1906         {
1907           session_wrk_set_state (wrk, SESSION_WRK_INTERRUPT);
1908           vlib_node_set_state (vm, session_queue_node.index,
1909                                VLIB_NODE_STATE_INTERRUPT);
1910         }
1911     }
1912   else if (wrk->state == SESSION_WRK_INTERRUPT)
1913     {
1914       if (clib_llist_elts (wrk->event_elts) > 5 ||
1915           vlib_last_vectors_per_main_loop (vm) > 1)
1916         {
1917           session_wrk_set_state (wrk, SESSION_WRK_POLLING);
1918           vlib_node_set_state (vm, session_queue_node.index,
1919                                VLIB_NODE_STATE_POLLING);
1920         }
1921       else if (PREDICT_FALSE (!pool_elts (wrk->sessions)))
1922         {
1923           session_wrk_set_state (wrk, SESSION_WRK_IDLE);
1924         }
1925     }
1926   else
1927     {
1928       if (clib_llist_elts (wrk->event_elts))
1929         {
1930           session_wrk_set_state (wrk, SESSION_WRK_INTERRUPT);
1931         }
1932     }
1933 }
1934
1935 static uword
1936 session_queue_node_fn (vlib_main_t * vm, vlib_node_runtime_t * node,
1937                        vlib_frame_t * frame)
1938 {
1939   u32 thread_index = vm->thread_index, __clib_unused n_evts;
1940   session_evt_elt_t *elt, *ctrl_he, *new_he, *old_he;
1941   session_main_t *smm = vnet_get_session_main ();
1942   session_worker_t *wrk = &smm->wrk[thread_index];
1943   clib_llist_index_t ei, next_ei, old_ti;
1944   int n_tx_packets;
1945
1946   SESSION_EVT (SESSION_EVT_DISPATCH_START, wrk);
1947
1948   session_wrk_update_time (wrk, vlib_time_now (vm));
1949
1950   /*
1951    *  Update transport time
1952    */
1953   session_update_time_subscribers (smm, wrk->last_vlib_time, thread_index);
1954   n_tx_packets = vec_len (wrk->pending_tx_buffers);
1955   SESSION_EVT (SESSION_EVT_DSP_CNTRS, UPDATE_TIME, wrk);
1956
1957   if (PREDICT_FALSE (wrk->dma_enabled))
1958     {
1959       if (wrk->trans_head == ((wrk->trans_tail + 1) & (wrk->trans_size - 1)))
1960         return 0;
1961       wrk->batch = vlib_dma_batch_new (vm, wrk->config_index);
1962       if (!wrk->batch)
1963         return 0;
1964     }
1965
1966   /*
1967    *  Dequeue new internal mq events
1968    */
1969
1970   n_evts = session_wrk_handle_mq (wrk, wrk->vpp_event_queue);
1971   SESSION_EVT (SESSION_EVT_DSP_CNTRS, MQ_DEQ, wrk, n_evts);
1972
1973   /*
1974    * Handle control events
1975    */
1976
1977   ei = wrk->ctrl_head;
1978   ctrl_he = clib_llist_elt (wrk->event_elts, ei);
1979   next_ei = clib_llist_next_index (ctrl_he, evt_list);
1980   old_ti = clib_llist_prev_index (ctrl_he, evt_list);
1981   while (ei != old_ti)
1982     {
1983       ei = next_ei;
1984       elt = clib_llist_elt (wrk->event_elts, next_ei);
1985       next_ei = clib_llist_next_index (elt, evt_list);
1986       clib_llist_remove (wrk->event_elts, evt_list, elt);
1987       session_event_dispatch_ctrl (wrk, elt);
1988     }
1989
1990   SESSION_EVT (SESSION_EVT_DSP_CNTRS, CTRL_EVTS, wrk);
1991
1992   /*
1993    * Handle the new io events.
1994    */
1995
1996   new_he = clib_llist_elt (wrk->event_elts, wrk->new_head);
1997   old_he = clib_llist_elt (wrk->event_elts, wrk->old_head);
1998   old_ti = clib_llist_prev_index (old_he, evt_list);
1999
2000   ei = clib_llist_next_index (new_he, evt_list);
2001   while (ei != wrk->new_head && n_tx_packets < SESSION_NODE_FRAME_SIZE)
2002     {
2003       elt = clib_llist_elt (wrk->event_elts, ei);
2004       ei = clib_llist_next_index (elt, evt_list);
2005       clib_llist_remove (wrk->event_elts, evt_list, elt);
2006       session_event_dispatch_io (wrk, node, elt, &n_tx_packets);
2007     }
2008
2009   SESSION_EVT (SESSION_EVT_DSP_CNTRS, NEW_IO_EVTS, wrk);
2010
2011   /*
2012    * Handle the old io events, if we had any prior to processing the new ones
2013    */
2014
2015   if (old_ti != wrk->old_head)
2016     {
2017       old_he = clib_llist_elt (wrk->event_elts, wrk->old_head);
2018       ei = clib_llist_next_index (old_he, evt_list);
2019
2020       while (n_tx_packets < SESSION_NODE_FRAME_SIZE)
2021         {
2022           elt = clib_llist_elt (wrk->event_elts, ei);
2023           next_ei = clib_llist_next_index (elt, evt_list);
2024           clib_llist_remove (wrk->event_elts, evt_list, elt);
2025
2026           session_event_dispatch_io (wrk, node, elt, &n_tx_packets);
2027
2028           if (ei == old_ti)
2029             break;
2030
2031           ei = next_ei;
2032         };
2033     }
2034
2035   if (PREDICT_FALSE (wrk->dma_enabled))
2036     {
2037       if (wrk->batch_num)
2038         {
2039           vlib_dma_batch_set_cookie (vm, wrk->batch, wrk->trans_tail);
2040           wrk->batch_num = 0;
2041           wrk->trans_tail++;
2042           if (wrk->trans_tail == wrk->trans_size)
2043             wrk->trans_tail = 0;
2044         }
2045
2046       vlib_dma_batch_submit (vm, wrk->batch);
2047     }
2048
2049   SESSION_EVT (SESSION_EVT_DSP_CNTRS, OLD_IO_EVTS, wrk);
2050
2051   if (vec_len (wrk->pending_tx_buffers))
2052     session_flush_pending_tx_buffers (wrk, node);
2053
2054   vlib_node_increment_counter (vm, session_queue_node.index,
2055                                SESSION_QUEUE_ERROR_TX, n_tx_packets);
2056
2057   SESSION_EVT (SESSION_EVT_DISPATCH_END, wrk, n_tx_packets);
2058
2059   if (wrk->flags & SESSION_WRK_F_ADAPTIVE)
2060     session_wrk_update_state (wrk);
2061
2062   return n_tx_packets;
2063 }
2064
2065 /* *INDENT-OFF* */
2066 VLIB_REGISTER_NODE (session_queue_node) = {
2067   .function = session_queue_node_fn,
2068   .flags = VLIB_NODE_FLAG_TRACE_SUPPORTED,
2069   .name = "session-queue",
2070   .format_trace = format_session_queue_trace,
2071   .type = VLIB_NODE_TYPE_INPUT,
2072   .n_errors = SESSION_QUEUE_N_ERROR,
2073   .error_counters = session_error_counters,
2074   .state = VLIB_NODE_STATE_DISABLED,
2075 };
2076 /* *INDENT-ON* */
2077
2078 static clib_error_t *
2079 session_wrk_tfd_read_ready (clib_file_t *cf)
2080 {
2081   session_worker_t *wrk = session_main_get_worker (cf->private_data);
2082   u64 buf;
2083   int rv;
2084
2085   vlib_node_set_interrupt_pending (wrk->vm, session_queue_node.index);
2086   rv = read (wrk->timerfd, &buf, sizeof (buf));
2087   if (rv < 0 && errno != EAGAIN)
2088     clib_unix_warning ("failed");
2089   return 0;
2090 }
2091
2092 static clib_error_t *
2093 session_wrk_tfd_write_ready (clib_file_t *cf)
2094 {
2095   return 0;
2096 }
2097
2098 void
2099 session_wrk_enable_adaptive_mode (session_worker_t *wrk)
2100 {
2101   u32 thread_index = wrk->vm->thread_index;
2102   clib_file_t template = { 0 };
2103
2104   if ((wrk->timerfd = timerfd_create (CLOCK_MONOTONIC, TFD_NONBLOCK)) < 0)
2105     clib_warning ("timerfd_create");
2106
2107   template.read_function = session_wrk_tfd_read_ready;
2108   template.write_function = session_wrk_tfd_write_ready;
2109   template.file_descriptor = wrk->timerfd;
2110   template.private_data = thread_index;
2111   template.polling_thread_index = thread_index;
2112   template.description = format (0, "session-wrk-tfd-%u", thread_index);
2113
2114   wrk->timerfd_file = clib_file_add (&file_main, &template);
2115   wrk->flags |= SESSION_WRK_F_ADAPTIVE;
2116 }
2117
2118 static clib_error_t *
2119 session_queue_exit (vlib_main_t * vm)
2120 {
2121   if (vlib_get_n_threads () < 2)
2122     return 0;
2123
2124   /*
2125    * Shut off (especially) worker-thread session nodes.
2126    * Otherwise, vpp can crash as the main thread unmaps the
2127    * API segment.
2128    */
2129   vlib_worker_thread_barrier_sync (vm);
2130   session_node_enable_disable (0 /* is_enable */ );
2131   vlib_worker_thread_barrier_release (vm);
2132   return 0;
2133 }
2134
2135 VLIB_MAIN_LOOP_EXIT_FUNCTION (session_queue_exit);
2136
2137 static uword
2138 session_queue_run_on_main (vlib_main_t * vm)
2139 {
2140   vlib_node_runtime_t *node;
2141
2142   node = vlib_node_get_runtime (vm, session_queue_node.index);
2143   return session_queue_node_fn (vm, node, 0);
2144 }
2145
2146 static uword
2147 session_queue_process (vlib_main_t * vm, vlib_node_runtime_t * rt,
2148                        vlib_frame_t * f)
2149 {
2150   uword *event_data = 0;
2151   f64 timeout = 1.0;
2152   uword event_type;
2153
2154   while (1)
2155     {
2156       vlib_process_wait_for_event_or_clock (vm, timeout);
2157       event_type = vlib_process_get_events (vm, (uword **) & event_data);
2158
2159       switch (event_type)
2160         {
2161         case SESSION_Q_PROCESS_RUN_ON_MAIN:
2162           /* Run session queue node on main thread */
2163           session_queue_run_on_main (vm);
2164           break;
2165         case SESSION_Q_PROCESS_STOP:
2166           vlib_node_set_state (vm, session_queue_process_node.index,
2167                                VLIB_NODE_STATE_DISABLED);
2168           timeout = 100000.0;
2169           break;
2170         case ~0:
2171           /* Timed out. Run on main to ensure all events are handled */
2172           session_queue_run_on_main (vm);
2173           break;
2174         }
2175       vec_reset_length (event_data);
2176     }
2177   return 0;
2178 }
2179
2180 /* *INDENT-OFF* */
2181 VLIB_REGISTER_NODE (session_queue_process_node) =
2182 {
2183   .function = session_queue_process,
2184   .type = VLIB_NODE_TYPE_PROCESS,
2185   .name = "session-queue-process",
2186   .state = VLIB_NODE_STATE_DISABLED,
2187 };
2188 /* *INDENT-ON* */
2189
2190 static_always_inline uword
2191 session_queue_pre_input_inline (vlib_main_t * vm, vlib_node_runtime_t * node,
2192                                 vlib_frame_t * frame)
2193 {
2194   session_main_t *sm = &session_main;
2195   if (!sm->wrk[0].vpp_event_queue)
2196     return 0;
2197   node = vlib_node_get_runtime (vm, session_queue_node.index);
2198   return session_queue_node_fn (vm, node, frame);
2199 }
2200
2201 /* *INDENT-OFF* */
2202 VLIB_REGISTER_NODE (session_queue_pre_input_node) =
2203 {
2204   .function = session_queue_pre_input_inline,
2205   .type = VLIB_NODE_TYPE_PRE_INPUT,
2206   .name = "session-queue-main",
2207   .state = VLIB_NODE_STATE_DISABLED,
2208 };
2209 /* *INDENT-ON* */
2210
2211 /*
2212  * fd.io coding-style-patch-verification: ON
2213  *
2214  * Local Variables:
2215  * eval: (c-set-style "gnu")
2216  * End:
2217  */