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
  28 #define app_check_thread_and_barrier(_fn, _arg)                         \
  29   if (!vlib_thread_is_main_w_barrier ())                                \
  30     {                                                                   \
  31      vlib_rpc_call_main_thread (_fn, (u8 *) _arg, sizeof(*_arg));       \
  32       return;                                                           \
  33    }
  34
  35 static void
  36 session_mq_listen_handler (void *data)
  37 {
  38   session_listen_msg_t *mp = (session_listen_msg_t *) data;
  39   vnet_listen_args_t _a, *a = &_a;
  40   app_worker_t *app_wrk;
  41   application_t *app;
  42   int rv;
  43
  44   app_check_thread_and_barrier (session_mq_listen_handler, mp);
  45
  46   app = application_lookup (mp->client_index);
  47   if (!app)
  48     return;
  49
  50   clib_memset (a, 0, sizeof (*a));
  51   a->sep.is_ip4 = mp->is_ip4;
  52   clib_memcpy_fast (&a->sep.ip, &mp->ip, sizeof (mp->ip));
  53   a->sep.port = mp->port;
  54   a->sep.fib_index = mp->vrf;
  55   a->sep.sw_if_index = ENDPOINT_INVALID_INDEX;
  56   a->sep.transport_proto = mp->proto;
  57   a->sep_ext.ckpair_index = mp->ckpair_index;
  58   a->sep_ext.crypto_engine = mp->crypto_engine;
  59   a->app_index = app->app_index;
  60   a->wrk_map_index = mp->wrk_index;
  61
  62   if ((rv = vnet_listen (a)))
  63     clib_warning ("listen returned: %d", rv);
  64
  65   app_wrk = application_get_worker (app, mp->wrk_index);
  66   mq_send_session_bound_cb (app_wrk->wrk_index, mp->context, a->handle, rv);
  67   return;
  68 }
  69
  70 static void
  71 session_mq_listen_uri_handler (void *data)
  72 {
  73   session_listen_uri_msg_t *mp = (session_listen_uri_msg_t *) data;
  74   vnet_listen_args_t _a, *a = &_a;
  75   app_worker_t *app_wrk;
  76   application_t *app;
  77   int rv;
  78
  79   app_check_thread_and_barrier (session_mq_listen_uri_handler, mp);
  80
  81   app = application_lookup (mp->client_index);
  82   if (!app)
  83     return;
  84
  85   clib_memset (a, 0, sizeof (*a));
  86   a->uri = (char *) mp->uri;
  87   a->app_index = app->app_index;
  88   rv = vnet_bind_uri (a);
  89
  90   app_wrk = application_get_worker (app, 0);
  91   mq_send_session_bound_cb (app_wrk->wrk_index, mp->context, a->handle, rv);
  92 }
  93
  94 static void
  95 session_mq_connect_handler (void *data)
  96 {
  97   session_connect_msg_t *mp = (session_connect_msg_t *) data;
  98   vnet_connect_args_t _a, *a = &_a;
  99   app_worker_t *app_wrk;
 100   application_t *app;
 101   int rv;
 102
 103   app_check_thread_and_barrier (session_mq_connect_handler, mp);
 104
 105   app = application_lookup (mp->client_index);
 106   if (!app)
 107     return;
 108
 109   clib_memset (a, 0, sizeof (*a));
 110   a->sep.is_ip4 = mp->is_ip4;
 111   clib_memcpy_fast (&a->sep.ip, &mp->ip, sizeof (mp->ip));
 112   a->sep.port = mp->port;
 113   a->sep.transport_proto = mp->proto;
 114   a->sep.peer.fib_index = mp->vrf;
 115   clib_memcpy_fast (&a->sep.peer.ip, &mp->lcl_ip, sizeof (mp->lcl_ip));
 116   a->sep.peer.port = mp->lcl_port;
 117   a->sep.peer.sw_if_index = ENDPOINT_INVALID_INDEX;
 118   a->sep_ext.parent_handle = mp->parent_handle;
 119   a->sep_ext.ckpair_index = mp->ckpair_index;
 120   a->sep_ext.crypto_engine = mp->crypto_engine;
 121   a->sep_ext.flags = mp->flags;
 122   if (mp->hostname_len)
 123     {
 124       vec_validate (a->sep_ext.hostname, mp->hostname_len - 1);
 125       clib_memcpy_fast (a->sep_ext.hostname, mp->hostname, mp->hostname_len);
 126     }
 127   a->api_context = mp->context;
 128   a->app_index = app->app_index;
 129   a->wrk_map_index = mp->wrk_index;
 130
 131   if ((rv = vnet_connect (a)))
 132     {
 133       clib_warning ("connect returned: %U", format_vnet_api_errno, rv);
 134       app_wrk = application_get_worker (app, mp->wrk_index);
 135       mq_send_session_connected_cb (app_wrk->wrk_index, mp->context, 0,
 136                                     /* is_fail */ 1);
 137     }
 138
 139   vec_free (a->sep_ext.hostname);
 140 }
 141
 142 static void
 143 session_mq_connect_uri_handler (void *data)
 144 {
 145   session_connect_uri_msg_t *mp = (session_connect_uri_msg_t *) data;
 146   vnet_connect_args_t _a, *a = &_a;
 147   app_worker_t *app_wrk;
 148   application_t *app;
 149   int rv;
 150
 151   app_check_thread_and_barrier (session_mq_connect_uri_handler, mp);
 152
 153   app = application_lookup (mp->client_index);
 154   if (!app)
 155     return;
 156
 157   clib_memset (a, 0, sizeof (*a));
 158   a->uri = (char *) mp->uri;
 159   a->api_context = mp->context;
 160   a->app_index = app->app_index;
 161   if ((rv = vnet_connect_uri (a)))
 162     {
 163       clib_warning ("connect_uri returned: %d", rv);
 164       app_wrk = application_get_worker (app, 0 /* default wrk only */ );
 165       mq_send_session_connected_cb (app_wrk->wrk_index, mp->context, 0,
 166                                     /* is_fail */ 1);
 167     }
 168 }
 169
 170 static void
 171 session_mq_disconnect_handler (void *data)
 172 {
 173   session_disconnect_msg_t *mp = (session_disconnect_msg_t *) data;
 174   vnet_disconnect_args_t _a, *a = &_a;
 175   application_t *app;
 176
 177   app = application_lookup (mp->client_index);
 178   if (!app)
 179     return;
 180
 181   a->app_index = app->app_index;
 182   a->handle = mp->handle;
 183   vnet_disconnect_session (a);
 184 }
 185
 186 static void
 187 app_mq_detach_handler (void *data)
 188 {
 189   session_app_detach_msg_t *mp = (session_app_detach_msg_t *) data;
 190   vnet_app_detach_args_t _a, *a = &_a;
 191   application_t *app;
 192
 193   app_check_thread_and_barrier (app_mq_detach_handler, mp);
 194
 195   app = application_lookup (mp->client_index);
 196   if (!app)
 197     return;
 198
 199   a->app_index = app->app_index;
 200   a->api_client_index = mp->client_index;
 201   vnet_application_detach (a);
 202 }
 203
 204 static void
 205 session_mq_unlisten_handler (void *data)
 206 {
 207   session_unlisten_msg_t *mp = (session_unlisten_msg_t *) data;
 208   vnet_unlisten_args_t _a, *a = &_a;
 209   app_worker_t *app_wrk;
 210   application_t *app;
 211   int rv;
 212
 213   app_check_thread_and_barrier (session_mq_unlisten_handler, mp);
 214
 215   app = application_lookup (mp->client_index);
 216   if (!app)
 217     return;
 218
 219   clib_memset (a, 0, sizeof (*a));
 220   a->app_index = app->app_index;
 221   a->handle = mp->handle;
 222   a->wrk_map_index = mp->wrk_index;
 223   if ((rv = vnet_unlisten (a)))
 224     clib_warning ("unlisten returned: %d", rv);
 225
 226   app_wrk = application_get_worker (app, a->wrk_map_index);
 227   if (!app_wrk)
 228     return;
 229
 230   mq_send_unlisten_reply (app_wrk, mp->handle, mp->context, rv);
 231 }
 232
 233 static void
 234 session_mq_accepted_reply_handler (void *data)
 235 {
 236   session_accepted_reply_msg_t *mp = (session_accepted_reply_msg_t *) data;
 237   vnet_disconnect_args_t _a = { 0 }, *a = &_a;
 238   session_state_t old_state;
 239   app_worker_t *app_wrk;
 240   session_t *s;
 241
 242   /* Server isn't interested, kill the session */
 243   if (mp->retval)
 244     {
 245       a->app_index = mp->context;
 246       a->handle = mp->handle;
 247       vnet_disconnect_session (a);
 248       return;
 249     }
 250
 251   /* Mail this back from the main thread. We're not polling in main
 252    * thread so we're using other workers for notifications. */
 253   if (vlib_num_workers () && vlib_get_thread_index () != 0
 254       && session_thread_from_handle (mp->handle) == 0)
 255     {
 256       vlib_rpc_call_main_thread (session_mq_accepted_reply_handler,
 257                                  (u8 *) mp, sizeof (*mp));
 258       return;
 259     }
 260
 261   s = session_get_from_handle_if_valid (mp->handle);
 262   if (!s)
 263     return;
 264
 265   app_wrk = app_worker_get (s->app_wrk_index);
 266   if (app_wrk->app_index != mp->context)
 267     {
 268       clib_warning ("app doesn't own session");
 269       return;
 270     }
 271
 272   if (!session_has_transport (s))
 273     {
 274       s->session_state = SESSION_STATE_READY;
 275       if (ct_session_connect_notify (s))
 276         return;
 277     }
 278   else
 279     {
 280       old_state = s->session_state;
 281       s->session_state = SESSION_STATE_READY;
 282
 283       if (!svm_fifo_is_empty_prod (s->rx_fifo))
 284         app_worker_lock_and_send_event (app_wrk, s, SESSION_IO_EVT_RX);
 285
 286       /* Closed while waiting for app to reply. Resend disconnect */
 287       if (old_state >= SESSION_STATE_TRANSPORT_CLOSING)
 288         {
 289           app_worker_close_notify (app_wrk, s);
 290           s->session_state = old_state;
 291           return;
 292         }
 293     }
 294 }
 295
 296 static void
 297 session_mq_reset_reply_handler (void *data)
 298 {
 299   vnet_disconnect_args_t _a = { 0 }, *a = &_a;
 300   session_reset_reply_msg_t *mp;
 301   app_worker_t *app_wrk;
 302   session_t *s;
 303   application_t *app;
 304   u32 index, thread_index;
 305
 306   mp = (session_reset_reply_msg_t *) data;
 307   app = application_lookup (mp->context);
 308   if (!app)
 309     return;
 310
 311   session_parse_handle (mp->handle, &index, &thread_index);
 312   s = session_get_if_valid (index, thread_index);
 313
 314   /* No session or not the right session */
 315   if (!s || s->session_state < SESSION_STATE_TRANSPORT_CLOSING)
 316     return;
 317
 318   app_wrk = app_worker_get (s->app_wrk_index);
 319   if (!app_wrk || app_wrk->app_index != app->app_index)
 320     {
 321       clib_warning ("App %u does not own handle 0x%lx!", app->app_index,
 322                     mp->handle);
 323       return;
 324     }
 325
 326   /* Client objected to resetting the session, log and continue */
 327   if (mp->retval)
 328     {
 329       clib_warning ("client retval %d", mp->retval);
 330       return;
 331     }
 332
 333   /* This comes as a response to a reset, transport only waiting for
 334    * confirmation to remove connection state, no need to disconnect */
 335   a->handle = mp->handle;
 336   a->app_index = app->app_index;
 337   vnet_disconnect_session (a);
 338 }
 339
 340 static void
 341 session_mq_disconnected_handler (void *data)
 342 {
 343   session_disconnected_reply_msg_t *rmp;
 344   vnet_disconnect_args_t _a, *a = &_a;
 345   svm_msg_q_msg_t _msg, *msg = &_msg;
 346   session_disconnected_msg_t *mp;
 347   app_worker_t *app_wrk;
 348   session_event_t *evt;
 349   session_t *s;
 350   application_t *app;
 351   int rv = 0;
 352
 353   mp = (session_disconnected_msg_t *) data;
 354   if (!(s = session_get_from_handle_if_valid (mp->handle)))
 355     {
 356       clib_warning ("could not disconnect handle %llu", mp->handle);
 357       return;
 358     }
 359   app_wrk = app_worker_get (s->app_wrk_index);
 360   app = application_lookup (mp->client_index);
 361   if (!(app_wrk && app && app->app_index == app_wrk->app_index))
 362     {
 363       clib_warning ("could not disconnect session: %llu app: %u",
 364                     mp->handle, mp->client_index);
 365       return;
 366     }
 367
 368   a->handle = mp->handle;
 369   a->app_index = app_wrk->wrk_index;
 370   rv = vnet_disconnect_session (a);
 371
 372   svm_msg_q_lock_and_alloc_msg_w_ring (app_wrk->event_queue,
 373                                        SESSION_MQ_CTRL_EVT_RING,
 374                                        SVM_Q_WAIT, msg);
 375   evt = svm_msg_q_msg_data (app_wrk->event_queue, msg);
 376   clib_memset (evt, 0, sizeof (*evt));
 377   evt->event_type = SESSION_CTRL_EVT_DISCONNECTED_REPLY;
 378   rmp = (session_disconnected_reply_msg_t *) evt->data;
 379   rmp->handle = mp->handle;
 380   rmp->context = mp->context;
 381   rmp->retval = rv;
 382   svm_msg_q_add_and_unlock (app_wrk->event_queue, msg);
 383 }
 384
 385 static void
 386 session_mq_disconnected_reply_handler (void *data)
 387 {
 388   session_disconnected_reply_msg_t *mp;
 389   vnet_disconnect_args_t _a, *a = &_a;
 390   application_t *app;
 391
 392   mp = (session_disconnected_reply_msg_t *) data;
 393
 394   /* Client objected to disconnecting the session, log and continue */
 395   if (mp->retval)
 396     {
 397       clib_warning ("client retval %d", mp->retval);
 398       return;
 399     }
 400
 401   /* Disconnect has been confirmed. Confirm close to transport */
 402   app = application_lookup (mp->context);
 403   if (app)
 404     {
 405       a->handle = mp->handle;
 406       a->app_index = app->app_index;
 407       vnet_disconnect_session (a);
 408     }
 409 }
 410
 411 static void
 412 session_mq_worker_update_handler (void *data)
 413 {
 414   session_worker_update_msg_t *mp = (session_worker_update_msg_t *) data;
 415   session_worker_update_reply_msg_t *rmp;
 416   svm_msg_q_msg_t _msg, *msg = &_msg;
 417   app_worker_t *app_wrk;
 418   u32 owner_app_wrk_map;
 419   session_event_t *evt;
 420   session_t *s;
 421   application_t *app;
 422
 423   app = application_lookup (mp->client_index);
 424   if (!app)
 425     return;
 426   if (!(s = session_get_from_handle_if_valid (mp->handle)))
 427     {
 428       clib_warning ("invalid handle %llu", mp->handle);
 429       return;
 430     }
 431   app_wrk = app_worker_get (s->app_wrk_index);
 432   if (app_wrk->app_index != app->app_index)
 433     {
 434       clib_warning ("app %u does not own session %llu", app->app_index,
 435                     mp->handle);
 436       return;
 437     }
 438   owner_app_wrk_map = app_wrk->wrk_map_index;
 439   app_wrk = application_get_worker (app, mp->wrk_index);
 440
 441   /* This needs to come from the new owner */
 442   if (mp->req_wrk_index == owner_app_wrk_map)
 443     {
 444       session_req_worker_update_msg_t *wump;
 445
 446       svm_msg_q_lock_and_alloc_msg_w_ring (app_wrk->event_queue,
 447                                            SESSION_MQ_CTRL_EVT_RING,
 448                                            SVM_Q_WAIT, msg);
 449       evt = svm_msg_q_msg_data (app_wrk->event_queue, msg);
 450       clib_memset (evt, 0, sizeof (*evt));
 451       evt->event_type = SESSION_CTRL_EVT_REQ_WORKER_UPDATE;
 452       wump = (session_req_worker_update_msg_t *) evt->data;
 453       wump->session_handle = mp->handle;
 454       svm_msg_q_add_and_unlock (app_wrk->event_queue, msg);
 455       return;
 456     }
 457
 458   app_worker_own_session (app_wrk, s);
 459
 460   /*
 461    * Send reply
 462    */
 463   svm_msg_q_lock_and_alloc_msg_w_ring (app_wrk->event_queue,
 464                                        SESSION_MQ_CTRL_EVT_RING,
 465                                        SVM_Q_WAIT, msg);
 466   evt = svm_msg_q_msg_data (app_wrk->event_queue, msg);
 467   clib_memset (evt, 0, sizeof (*evt));
 468   evt->event_type = SESSION_CTRL_EVT_WORKER_UPDATE_REPLY;
 469   rmp = (session_worker_update_reply_msg_t *) evt->data;
 470   rmp->handle = mp->handle;
 471   rmp->rx_fifo = pointer_to_uword (s->rx_fifo);
 472   rmp->tx_fifo = pointer_to_uword (s->tx_fifo);
 473   rmp->segment_handle = session_segment_handle (s);
 474   svm_msg_q_add_and_unlock (app_wrk->event_queue, msg);
 475
 476   /*
 477    * Retransmit messages that may have been lost
 478    */
 479   if (s->tx_fifo && !svm_fifo_is_empty (s->tx_fifo))
 480     session_send_io_evt_to_thread (s->tx_fifo, SESSION_IO_EVT_TX);
 481
 482   if (s->rx_fifo && !svm_fifo_is_empty (s->rx_fifo))
 483     app_worker_lock_and_send_event (app_wrk, s, SESSION_IO_EVT_RX);
 484
 485   if (s->session_state >= SESSION_STATE_TRANSPORT_CLOSING)
 486     app_worker_close_notify (app_wrk, s);
 487 }
 488
 489 vlib_node_registration_t session_queue_node;
 490
 491 typedef struct
 492 {
 493   u32 session_index;
 494   u32 server_thread_index;
 495 } session_queue_trace_t;
 496
 497 /* packet trace format function */
 498 static u8 *
 499 format_session_queue_trace (u8 * s, va_list * args)
 500 {
 501   CLIB_UNUSED (vlib_main_t * vm) = va_arg (*args, vlib_main_t *);
 502   CLIB_UNUSED (vlib_node_t * node) = va_arg (*args, vlib_node_t *);
 503   session_queue_trace_t *t = va_arg (*args, session_queue_trace_t *);
 504
 505   s = format (s, "session index %d thread index %d",
 506               t->session_index, t->server_thread_index);
 507   return s;
 508 }
 509
 510 #define foreach_session_queue_error             \
 511 _(TX, "Packets transmitted")                    \
 512 _(TIMER, "Timer events")                        \
 513 _(NO_BUFFER, "Out of buffers")
 514
 515 typedef enum
 516 {
 517 #define _(sym,str) SESSION_QUEUE_ERROR_##sym,
 518   foreach_session_queue_error
 519 #undef _
 520     SESSION_QUEUE_N_ERROR,
 521 } session_queue_error_t;
 522
 523 static char *session_queue_error_strings[] = {
 524 #define _(sym,string) string,
 525   foreach_session_queue_error
 526 #undef _
 527 };
 528
 529 enum
 530 {
 531   SESSION_TX_NO_BUFFERS = -2,
 532   SESSION_TX_NO_DATA,
 533   SESSION_TX_OK
 534 };
 535
 536 static void
 537 session_tx_trace_frame (vlib_main_t * vm, vlib_node_runtime_t * node,
 538                         u32 next_index, u32 * to_next, u16 n_segs,
 539                         session_t * s, u32 n_trace)
 540 {
 541   session_queue_trace_t *t;
 542   vlib_buffer_t *b;
 543   int i;
 544
 545   for (i = 0; i < clib_min (n_trace, n_segs); i++)
 546     {
 547       b = vlib_get_buffer (vm, to_next[i]);
 548       vlib_trace_buffer (vm, node, next_index, b, 1 /* follow_chain */ );
 549       t = vlib_add_trace (vm, node, b, sizeof (*t));
 550       t->session_index = s->session_index;
 551       t->server_thread_index = s->thread_index;
 552     }
 553   vlib_set_trace_count (vm, node, n_trace - i);
 554 }
 555
 556 always_inline void
 557 session_tx_fifo_chain_tail (vlib_main_t * vm, session_tx_context_t * ctx,
 558                             vlib_buffer_t * b, u16 * n_bufs, u8 peek_data)
 559 {
 560   vlib_buffer_t *chain_b, *prev_b;
 561   u32 chain_bi0, to_deq, left_from_seg;
 562   session_worker_t *wrk;
 563   u16 len_to_deq, n_bytes_read;
 564   u8 *data, j;
 565
 566   wrk = session_main_get_worker (ctx->s->thread_index);
 567   b->flags |= VLIB_BUFFER_TOTAL_LENGTH_VALID;
 568   b->total_length_not_including_first_buffer = 0;
 569
 570   chain_b = b;
 571   left_from_seg = clib_min (ctx->sp.snd_mss - b->current_length,
 572                             ctx->left_to_snd);
 573   to_deq = left_from_seg;
 574   for (j = 1; j < ctx->n_bufs_per_seg; j++)
 575     {
 576       prev_b = chain_b;
 577       len_to_deq = clib_min (to_deq, ctx->deq_per_buf);
 578
 579       *n_bufs -= 1;
 580       chain_bi0 = wrk->tx_buffers[*n_bufs];
 581       chain_b = vlib_get_buffer (vm, chain_bi0);
 582       chain_b->current_data = 0;
 583       data = vlib_buffer_get_current (chain_b);
 584       if (peek_data)
 585         {
 586           n_bytes_read = svm_fifo_peek (ctx->s->tx_fifo,
 587                                         ctx->sp.tx_offset, len_to_deq, data);
 588           ctx->sp.tx_offset += n_bytes_read;
 589         }
 590       else
 591         {
 592           if (ctx->transport_vft->transport_options.tx_type ==
 593               TRANSPORT_TX_DGRAM)
 594             {
 595               svm_fifo_t *f = ctx->s->tx_fifo;
 596               session_dgram_hdr_t *hdr = &ctx->hdr;
 597               u16 deq_now;
 598               deq_now = clib_min (hdr->data_length - hdr->data_offset,
 599                                   len_to_deq);
 600               n_bytes_read = svm_fifo_peek (f, hdr->data_offset, deq_now,
 601                                             data);
 602               ASSERT (n_bytes_read > 0);
 603
 604               hdr->data_offset += n_bytes_read;
 605               if (hdr->data_offset == hdr->data_length)
 606                 {
 607                   u32 offset = hdr->data_length + SESSION_CONN_HDR_LEN;
 608                   svm_fifo_dequeue_drop (f, offset);
 609                 }
 610             }
 611           else
 612             n_bytes_read = svm_fifo_dequeue (ctx->s->tx_fifo,
 613                                              len_to_deq, data);
 614         }
 615       ASSERT (n_bytes_read == len_to_deq);
 616       chain_b->current_length = n_bytes_read;
 617       b->total_length_not_including_first_buffer += chain_b->current_length;
 618
 619       /* update previous buffer */
 620       prev_b->next_buffer = chain_bi0;
 621       prev_b->flags |= VLIB_BUFFER_NEXT_PRESENT;
 622
 623       /* update current buffer */
 624       chain_b->next_buffer = 0;
 625
 626       to_deq -= n_bytes_read;
 627       if (to_deq == 0)
 628         break;
 629     }
 630   ASSERT (to_deq == 0
 631           && b->total_length_not_including_first_buffer == left_from_seg);
 632   ctx->left_to_snd -= left_from_seg;
 633 }
 634
 635 always_inline void
 636 session_tx_fill_buffer (vlib_main_t * vm, session_tx_context_t * ctx,
 637                         vlib_buffer_t * b, u16 * n_bufs, u8 peek_data)
 638 {
 639   u32 len_to_deq;
 640   u8 *data0;
 641   int n_bytes_read;
 642
 643   /*
 644    * Start with the first buffer in chain
 645    */
 646   b->error = 0;
 647   b->flags = VNET_BUFFER_F_LOCALLY_ORIGINATED;
 648   b->current_data = 0;
 649
 650   data0 = vlib_buffer_make_headroom (b, TRANSPORT_MAX_HDRS_LEN);
 651   len_to_deq = clib_min (ctx->left_to_snd, ctx->deq_per_first_buf);
 652
 653   if (peek_data)
 654     {
 655       n_bytes_read = svm_fifo_peek (ctx->s->tx_fifo, ctx->sp.tx_offset,
 656                                     len_to_deq, data0);
 657       ASSERT (n_bytes_read > 0);
 658       /* Keep track of progress locally, transport is also supposed to
 659        * increment it independently when pushing the header */
 660       ctx->sp.tx_offset += n_bytes_read;
 661     }
 662   else
 663     {
 664       if (ctx->transport_vft->transport_options.tx_type == TRANSPORT_TX_DGRAM)
 665         {
 666           session_dgram_hdr_t *hdr = &ctx->hdr;
 667           svm_fifo_t *f = ctx->s->tx_fifo;
 668           u16 deq_now;
 669           u32 offset;
 670
 671           ASSERT (hdr->data_length > hdr->data_offset);
 672           deq_now = clib_min (hdr->data_length - hdr->data_offset,
 673                               len_to_deq);
 674           offset = hdr->data_offset + SESSION_CONN_HDR_LEN;
 675           n_bytes_read = svm_fifo_peek (f, offset, deq_now, data0);
 676           ASSERT (n_bytes_read > 0);
 677
 678           if (ctx->s->session_state == SESSION_STATE_LISTENING)
 679             {
 680               ip_copy (&ctx->tc->rmt_ip, &hdr->rmt_ip, ctx->tc->is_ip4);
 681               ctx->tc->rmt_port = hdr->rmt_port;
 682             }
 683           hdr->data_offset += n_bytes_read;
 684           if (hdr->data_offset == hdr->data_length)
 685             {
 686               offset = hdr->data_length + SESSION_CONN_HDR_LEN;
 687               svm_fifo_dequeue_drop (f, offset);
 688             }
 689         }
 690       else
 691         {
 692           n_bytes_read = svm_fifo_dequeue (ctx->s->tx_fifo,
 693                                            len_to_deq, data0);
 694           ASSERT (n_bytes_read > 0);
 695         }
 696     }
 697   b->current_length = n_bytes_read;
 698   ctx->left_to_snd -= n_bytes_read;
 699
 700   /*
 701    * Fill in the remaining buffers in the chain, if any
 702    */
 703   if (PREDICT_FALSE (ctx->n_bufs_per_seg > 1 && ctx->left_to_snd))
 704     session_tx_fifo_chain_tail (vm, ctx, b, n_bufs, peek_data);
 705 }
 706
 707 always_inline u8
 708 session_tx_not_ready (session_t * s, u8 peek_data)
 709 {
 710   if (peek_data)
 711     {
 712       if (PREDICT_TRUE (s->session_state == SESSION_STATE_READY))
 713         return 0;
 714       /* Can retransmit for closed sessions but can't send new data if
 715        * session is not ready or closed */
 716       else if (s->session_state < SESSION_STATE_READY)
 717         return 1;
 718       else if (s->session_state >= SESSION_STATE_TRANSPORT_CLOSED)
 719         {
 720           /* Allow closed transports to still send custom packets.
 721            * For instance, tcp may want to send acks in time-wait. */
 722           if (s->session_state != SESSION_STATE_TRANSPORT_DELETED
 723               && (s->flags & SESSION_F_CUSTOM_TX))
 724             return 0;
 725           return 2;
 726         }
 727     }
 728   return 0;
 729 }
 730
 731 always_inline transport_connection_t *
 732 session_tx_get_transport (session_tx_context_t * ctx, u8 peek_data)
 733 {
 734   if (peek_data)
 735     {
 736       return ctx->transport_vft->get_connection (ctx->s->connection_index,
 737                                                  ctx->s->thread_index);
 738     }
 739   else
 740     {
 741       if (ctx->s->session_state == SESSION_STATE_LISTENING)
 742         return ctx->transport_vft->get_listener (ctx->s->connection_index);
 743       else
 744         {
 745           return ctx->transport_vft->get_connection (ctx->s->connection_index,
 746                                                      ctx->s->thread_index);
 747         }
 748     }
 749 }
 750
 751 always_inline void
 752 session_tx_set_dequeue_params (vlib_main_t * vm, session_tx_context_t * ctx,
 753                                u32 max_segs, u8 peek_data)
 754 {
 755   u32 n_bytes_per_buf, n_bytes_per_seg;
 756   ctx->max_dequeue = svm_fifo_max_dequeue_cons (ctx->s->tx_fifo);
 757   if (peek_data)
 758     {
 759       /* Offset in rx fifo from where to peek data */
 760       if (PREDICT_FALSE (ctx->sp.tx_offset >= ctx->max_dequeue))
 761         {
 762           ctx->max_len_to_snd = 0;
 763           return;
 764         }
 765       ctx->max_dequeue -= ctx->sp.tx_offset;
 766     }
 767   else
 768     {
 769       if (ctx->transport_vft->transport_options.tx_type == TRANSPORT_TX_DGRAM)
 770         {
 771           if (ctx->max_dequeue <= sizeof (ctx->hdr))
 772             {
 773               ctx->max_len_to_snd = 0;
 774               return;
 775             }
 776           svm_fifo_peek (ctx->s->tx_fifo, 0, sizeof (ctx->hdr),
 777                          (u8 *) & ctx->hdr);
 778           ASSERT (ctx->hdr.data_length > ctx->hdr.data_offset);
 779           ctx->max_dequeue = ctx->hdr.data_length - ctx->hdr.data_offset;
 780         }
 781     }
 782   ASSERT (ctx->max_dequeue > 0);
 783
 784   /* Ensure we're not writing more than transport window allows */
 785   if (ctx->max_dequeue < ctx->sp.snd_space)
 786     {
 787       /* Constrained by tx queue. Try to send only fully formed segments */
 788       ctx->max_len_to_snd = (ctx->max_dequeue > ctx->sp.snd_mss) ?
 789         (ctx->max_dequeue - (ctx->max_dequeue % ctx->sp.snd_mss)) :
 790         ctx->max_dequeue;
 791       /* TODO Nagle ? */
 792     }
 793   else
 794     {
 795       /* Expectation is that snd_space0 is already a multiple of snd_mss */
 796       ctx->max_len_to_snd = ctx->sp.snd_space;
 797     }
 798
 799   /* Check if we're tx constrained by the node */
 800   ctx->n_segs_per_evt = ceil ((f64) ctx->max_len_to_snd / ctx->sp.snd_mss);
 801   if (ctx->n_segs_per_evt > max_segs)
 802     {
 803       ctx->n_segs_per_evt = max_segs;
 804       ctx->max_len_to_snd = max_segs * ctx->sp.snd_mss;
 805     }
 806
 807   n_bytes_per_buf = vlib_buffer_get_default_data_size (vm);
 808   ASSERT (n_bytes_per_buf > TRANSPORT_MAX_HDRS_LEN);
 809   n_bytes_per_seg = TRANSPORT_MAX_HDRS_LEN + ctx->sp.snd_mss;
 810   ctx->n_bufs_per_seg = ceil ((f64) n_bytes_per_seg / n_bytes_per_buf);
 811   ctx->deq_per_buf = clib_min (ctx->sp.snd_mss, n_bytes_per_buf);
 812   ctx->deq_per_first_buf = clib_min (ctx->sp.snd_mss,
 813                                      n_bytes_per_buf -
 814                                      TRANSPORT_MAX_HDRS_LEN);
 815 }
 816
 817 always_inline void
 818 session_tx_maybe_reschedule (session_worker_t * wrk,
 819                              session_tx_context_t * ctx,
 820                              session_evt_elt_t * elt)
 821 {
 822   session_t *s = ctx->s;
 823
 824   svm_fifo_unset_event (s->tx_fifo);
 825   if (svm_fifo_max_dequeue_cons (s->tx_fifo) > ctx->sp.tx_offset)
 826     if (svm_fifo_set_event (s->tx_fifo))
 827       session_evt_add_head_old (wrk, elt);
 828 }
 829
 830 always_inline int
 831 session_tx_fifo_read_and_snd_i (session_worker_t * wrk,
 832                                 vlib_node_runtime_t * node,
 833                                 session_evt_elt_t * elt,
 834                                 int *n_tx_packets, u8 peek_data)
 835 {
 836   u32 n_trace, n_bufs_needed = 0, n_left, pbi, next_index, max_burst;
 837   session_tx_context_t *ctx = &wrk->ctx;
 838   session_main_t *smm = &session_main;
 839   session_event_t *e = &elt->evt;
 840   vlib_main_t *vm = wrk->vm;
 841   transport_proto_t tp;
 842   vlib_buffer_t *pb;
 843   u16 n_bufs, rv;
 844
 845   if (PREDICT_FALSE ((rv = session_tx_not_ready (ctx->s, peek_data))))
 846     {
 847       if (rv < 2)
 848         session_evt_add_old (wrk, elt);
 849       return SESSION_TX_NO_DATA;
 850     }
 851
 852   next_index = smm->session_type_to_next[ctx->s->session_type];
 853   max_burst = VLIB_FRAME_SIZE - *n_tx_packets;
 854
 855   tp = session_get_transport_proto (ctx->s);
 856   ctx->transport_vft = transport_protocol_get_vft (tp);
 857   ctx->tc = session_tx_get_transport (ctx, peek_data);
 858
 859   if (PREDICT_FALSE (e->event_type == SESSION_IO_EVT_TX_FLUSH))
 860     {
 861       if (ctx->transport_vft->flush_data)
 862         ctx->transport_vft->flush_data (ctx->tc);
 863       e->event_type = SESSION_IO_EVT_TX;
 864     }
 865
 866   if (ctx->s->flags & SESSION_F_CUSTOM_TX)
 867     {
 868       u32 n_custom_tx;
 869       ctx->s->flags &= ~SESSION_F_CUSTOM_TX;
 870       n_custom_tx = ctx->transport_vft->custom_tx (ctx->tc, max_burst);
 871       *n_tx_packets += n_custom_tx;
 872       if (PREDICT_FALSE
 873           (ctx->s->session_state >= SESSION_STATE_TRANSPORT_CLOSED))
 874         return SESSION_TX_OK;
 875       max_burst -= n_custom_tx;
 876       if (!max_burst || (ctx->s->flags & SESSION_F_CUSTOM_TX))
 877         {
 878           session_evt_add_old (wrk, elt);
 879           return SESSION_TX_OK;
 880         }
 881     }
 882
 883   transport_connection_snd_params (ctx->tc, &ctx->sp);
 884
 885   if (!ctx->sp.snd_space)
 886     {
 887       /* If the deschedule flag was set, remove session from scheduler.
 888        * Transport is responsible for rescheduling this session. */
 889       if (ctx->sp.flags & TRANSPORT_SND_F_DESCHED)
 890         transport_connection_deschedule (ctx->tc);
 891       /* Request to postpone the session, e.g., zero-wnd and transport
 892        * is not currently probing */
 893       else if (ctx->sp.flags & TRANSPORT_SND_F_POSTPONE)
 894         session_evt_add_old (wrk, elt);
 895       /* This flow queue is "empty" so it should be re-evaluated before
 896        * the ones that have data to send. */
 897       else
 898         session_evt_add_head_old (wrk, elt);
 899
 900       return SESSION_TX_NO_DATA;
 901     }
 902
 903   if (transport_connection_is_tx_paced (ctx->tc))
 904     {
 905       u32 snd_space = transport_connection_tx_pacer_burst (ctx->tc);
 906       if (snd_space < TRANSPORT_PACER_MIN_BURST)
 907         {
 908           session_evt_add_head_old (wrk, elt);
 909           return SESSION_TX_NO_DATA;
 910         }
 911       snd_space = clib_min (ctx->sp.snd_space, snd_space);
 912       ctx->sp.snd_space = snd_space >= ctx->sp.snd_mss ?
 913         snd_space - snd_space % ctx->sp.snd_mss : snd_space;
 914     }
 915
 916   /* Check how much we can pull. */
 917   session_tx_set_dequeue_params (vm, ctx, max_burst, peek_data);
 918
 919   if (PREDICT_FALSE (!ctx->max_len_to_snd))
 920     {
 921       transport_connection_tx_pacer_reset_bucket (ctx->tc, 0);
 922       session_tx_maybe_reschedule (wrk, ctx, elt);
 923       return SESSION_TX_NO_DATA;
 924     }
 925
 926   n_bufs_needed = ctx->n_segs_per_evt * ctx->n_bufs_per_seg;
 927   vec_validate_aligned (wrk->tx_buffers, n_bufs_needed - 1,
 928                         CLIB_CACHE_LINE_BYTES);
 929   n_bufs = vlib_buffer_alloc (vm, wrk->tx_buffers, n_bufs_needed);
 930   if (PREDICT_FALSE (n_bufs < n_bufs_needed))
 931     {
 932       if (n_bufs)
 933         vlib_buffer_free (vm, wrk->tx_buffers, n_bufs);
 934       session_evt_add_head_old (wrk, elt);
 935       vlib_node_increment_counter (wrk->vm, node->node_index,
 936                                    SESSION_QUEUE_ERROR_NO_BUFFER, 1);
 937       return SESSION_TX_NO_BUFFERS;
 938     }
 939
 940   ctx->left_to_snd = ctx->max_len_to_snd;
 941   n_left = ctx->n_segs_per_evt;
 942
 943   while (n_left >= 4)
 944     {
 945       vlib_buffer_t *b0, *b1;
 946       u32 bi0, bi1;
 947
 948       pbi = wrk->tx_buffers[n_bufs - 3];
 949       pb = vlib_get_buffer (vm, pbi);
 950       vlib_prefetch_buffer_header (pb, STORE);
 951       pbi = wrk->tx_buffers[n_bufs - 4];
 952       pb = vlib_get_buffer (vm, pbi);
 953       vlib_prefetch_buffer_header (pb, STORE);
 954
 955       bi0 = wrk->tx_buffers[--n_bufs];
 956       bi1 = wrk->tx_buffers[--n_bufs];
 957
 958       b0 = vlib_get_buffer (vm, bi0);
 959       b1 = vlib_get_buffer (vm, bi1);
 960
 961       session_tx_fill_buffer (vm, ctx, b0, &n_bufs, peek_data);
 962       session_tx_fill_buffer (vm, ctx, b1, &n_bufs, peek_data);
 963
 964       ctx->transport_vft->push_header (ctx->tc, b0);
 965       ctx->transport_vft->push_header (ctx->tc, b1);
 966
 967       n_left -= 2;
 968
 969       VLIB_BUFFER_TRACE_TRAJECTORY_INIT (b0);
 970       VLIB_BUFFER_TRACE_TRAJECTORY_INIT (b1);
 971
 972       vec_add1 (wrk->pending_tx_buffers, bi0);
 973       vec_add1 (wrk->pending_tx_buffers, bi1);
 974       vec_add1 (wrk->pending_tx_nexts, next_index);
 975       vec_add1 (wrk->pending_tx_nexts, next_index);
 976     }
 977   while (n_left)
 978     {
 979       vlib_buffer_t *b0;
 980       u32 bi0;
 981
 982       if (n_left > 1)
 983         {
 984           pbi = wrk->tx_buffers[n_bufs - 2];
 985           pb = vlib_get_buffer (vm, pbi);
 986           vlib_prefetch_buffer_header (pb, STORE);
 987         }
 988
 989       bi0 = wrk->tx_buffers[--n_bufs];
 990       b0 = vlib_get_buffer (vm, bi0);
 991       session_tx_fill_buffer (vm, ctx, b0, &n_bufs, peek_data);
 992
 993       /* Ask transport to push header after current_length and
 994        * total_length_not_including_first_buffer are updated */
 995       ctx->transport_vft->push_header (ctx->tc, b0);
 996
 997       n_left -= 1;
 998
 999       VLIB_BUFFER_TRACE_TRAJECTORY_INIT (b0);
1000
1001       vec_add1 (wrk->pending_tx_buffers, bi0);
1002       vec_add1 (wrk->pending_tx_nexts, next_index);
1003     }
1004
1005   if (PREDICT_FALSE ((n_trace = vlib_get_trace_count (vm, node)) > 0))
1006     session_tx_trace_frame (vm, node, next_index, wrk->pending_tx_buffers,
1007                             ctx->n_segs_per_evt, ctx->s, n_trace);
1008
1009   if (PREDICT_FALSE (n_bufs))
1010     vlib_buffer_free (vm, wrk->tx_buffers, n_bufs);
1011
1012   *n_tx_packets += ctx->n_segs_per_evt;
1013   transport_connection_update_tx_bytes (ctx->tc, ctx->max_len_to_snd);
1014
1015   SESSION_EVT (SESSION_EVT_DEQ, ctx->s, ctx->max_len_to_snd, ctx->max_dequeue,
1016                ctx->s->tx_fifo->has_event, wrk->last_vlib_time);
1017
1018   ASSERT (ctx->left_to_snd == 0);
1019
1020   /* If we couldn't dequeue all bytes reschedule as old flow. Otherwise,
1021    * check if application enqueued more data and reschedule accordingly */
1022   if (ctx->max_len_to_snd < ctx->max_dequeue)
1023     session_evt_add_old (wrk, elt);
1024   else
1025     session_tx_maybe_reschedule (wrk, ctx, elt);
1026
1027   if (!peek_data
1028       && ctx->transport_vft->transport_options.tx_type == TRANSPORT_TX_DGRAM)
1029     {
1030       /* Fix dgram pre header */
1031       if (ctx->max_len_to_snd < ctx->max_dequeue)
1032         svm_fifo_overwrite_head (ctx->s->tx_fifo, (u8 *) & ctx->hdr,
1033                                  sizeof (session_dgram_pre_hdr_t));
1034       /* More data needs to be read */
1035       else if (svm_fifo_max_dequeue_cons (ctx->s->tx_fifo) > 0)
1036         if (svm_fifo_set_event (ctx->s->tx_fifo))
1037           session_evt_add_old (wrk, elt);
1038
1039       if (svm_fifo_needs_deq_ntf (ctx->s->tx_fifo, ctx->max_len_to_snd))
1040         session_dequeue_notify (ctx->s);
1041     }
1042   return SESSION_TX_OK;
1043 }
1044
1045 int
1046 session_tx_fifo_peek_and_snd (session_worker_t * wrk,
1047                               vlib_node_runtime_t * node,
1048                               session_evt_elt_t * e, int *n_tx_packets)
1049 {
1050   return session_tx_fifo_read_and_snd_i (wrk, node, e, n_tx_packets, 1);
1051 }
1052
1053 int
1054 session_tx_fifo_dequeue_and_snd (session_worker_t * wrk,
1055                                  vlib_node_runtime_t * node,
1056                                  session_evt_elt_t * e, int *n_tx_packets)
1057 {
1058   return session_tx_fifo_read_and_snd_i (wrk, node, e, n_tx_packets, 0);
1059 }
1060
1061 int
1062 session_tx_fifo_dequeue_internal (session_worker_t * wrk,
1063                                   vlib_node_runtime_t * node,
1064                                   session_evt_elt_t * e, int *n_tx_packets)
1065 {
1066   session_t *s = wrk->ctx.s;
1067
1068   if (PREDICT_FALSE (s->session_state >= SESSION_STATE_TRANSPORT_CLOSED))
1069     return 0;
1070   svm_fifo_unset_event (s->tx_fifo);
1071   return transport_custom_tx (session_get_transport_proto (s), s,
1072                               VLIB_FRAME_SIZE - *n_tx_packets);
1073 }
1074
1075 always_inline session_t *
1076 session_event_get_session (session_event_t * e, u8 thread_index)
1077 {
1078   return session_get_if_valid (e->session_index, thread_index);
1079 }
1080
1081 always_inline void
1082 session_event_dispatch_ctrl (session_worker_t * wrk, session_evt_elt_t * elt)
1083 {
1084   clib_llist_index_t ei;
1085   void (*fp) (void *);
1086   session_event_t *e;
1087   session_t *s;
1088
1089   ei = clib_llist_entry_index (wrk->event_elts, elt);
1090   e = &elt->evt;
1091
1092   switch (e->event_type)
1093     {
1094     case SESSION_CTRL_EVT_RPC:
1095       fp = e->rpc_args.fp;
1096       (*fp) (e->rpc_args.arg);
1097       break;
1098     case SESSION_CTRL_EVT_CLOSE:
1099       s = session_get_from_handle_if_valid (e->session_handle);
1100       if (PREDICT_FALSE (!s))
1101         break;
1102       session_transport_close (s);
1103       break;
1104     case SESSION_CTRL_EVT_RESET:
1105       s = session_get_from_handle_if_valid (e->session_handle);
1106       if (PREDICT_FALSE (!s))
1107         break;
1108       session_transport_reset (s);
1109       break;
1110     case SESSION_CTRL_EVT_LISTEN:
1111       session_mq_listen_handler (session_evt_ctrl_data (wrk, elt));
1112       break;
1113     case SESSION_CTRL_EVT_LISTEN_URI:
1114       session_mq_listen_uri_handler (session_evt_ctrl_data (wrk, elt));
1115       break;
1116     case SESSION_CTRL_EVT_UNLISTEN:
1117       session_mq_unlisten_handler (session_evt_ctrl_data (wrk, elt));
1118       break;
1119     case SESSION_CTRL_EVT_CONNECT:
1120       session_mq_connect_handler (session_evt_ctrl_data (wrk, elt));
1121       break;
1122     case SESSION_CTRL_EVT_CONNECT_URI:
1123       session_mq_connect_uri_handler (session_evt_ctrl_data (wrk, elt));
1124       break;
1125     case SESSION_CTRL_EVT_DISCONNECT:
1126       session_mq_disconnect_handler (session_evt_ctrl_data (wrk, elt));
1127       break;
1128     case SESSION_CTRL_EVT_DISCONNECTED:
1129       session_mq_disconnected_handler (session_evt_ctrl_data (wrk, elt));
1130       break;
1131     case SESSION_CTRL_EVT_ACCEPTED_REPLY:
1132       session_mq_accepted_reply_handler (session_evt_ctrl_data (wrk, elt));
1133       break;
1134     case SESSION_CTRL_EVT_DISCONNECTED_REPLY:
1135       session_mq_disconnected_reply_handler (session_evt_ctrl_data (wrk,
1136                                                                     elt));
1137       break;
1138     case SESSION_CTRL_EVT_RESET_REPLY:
1139       session_mq_reset_reply_handler (session_evt_ctrl_data (wrk, elt));
1140       break;
1141     case SESSION_CTRL_EVT_WORKER_UPDATE:
1142       session_mq_worker_update_handler (session_evt_ctrl_data (wrk, elt));
1143       break;
1144     case SESSION_CTRL_EVT_APP_DETACH:
1145       app_mq_detach_handler (session_evt_ctrl_data (wrk, elt));
1146       break;
1147     default:
1148       clib_warning ("unhandled event type %d", e->event_type);
1149     }
1150
1151   /* Regrab elements in case pool moved */
1152   elt = pool_elt_at_index (wrk->event_elts, ei);
1153   if (!clib_llist_elt_is_linked (elt, evt_list))
1154     {
1155       e = &elt->evt;
1156       if (e->event_type >= SESSION_CTRL_EVT_BOUND)
1157         session_evt_ctrl_data_free (wrk, elt);
1158       session_evt_elt_free (wrk, elt);
1159     }
1160 }
1161
1162 always_inline void
1163 session_event_dispatch_io (session_worker_t * wrk, vlib_node_runtime_t * node,
1164                            session_evt_elt_t * elt, u32 thread_index,
1165                            int *n_tx_packets)
1166 {
1167   session_main_t *smm = &session_main;
1168   app_worker_t *app_wrk;
1169   clib_llist_index_t ei;
1170   session_event_t *e;
1171   session_t *s;
1172
1173   ei = clib_llist_entry_index (wrk->event_elts, elt);
1174   e = &elt->evt;
1175
1176   switch (e->event_type)
1177     {
1178     case SESSION_IO_EVT_TX_FLUSH:
1179     case SESSION_IO_EVT_TX:
1180       s = session_event_get_session (e, thread_index);
1181       if (PREDICT_FALSE (!s))
1182         break;
1183       CLIB_PREFETCH (s->tx_fifo, 2 * CLIB_CACHE_LINE_BYTES, LOAD);
1184       wrk->ctx.s = s;
1185       /* Spray packets in per session type frames, since they go to
1186        * different nodes */
1187       (smm->session_tx_fns[s->session_type]) (wrk, node, elt, n_tx_packets);
1188       break;
1189     case SESSION_IO_EVT_RX:
1190       s = session_event_get_session (e, thread_index);
1191       if (!s)
1192         break;
1193       transport_app_rx_evt (session_get_transport_proto (s),
1194                             s->connection_index, s->thread_index);
1195       break;
1196     case SESSION_IO_EVT_BUILTIN_RX:
1197       s = session_event_get_session (e, thread_index);
1198       if (PREDICT_FALSE (!s || s->session_state >= SESSION_STATE_CLOSING))
1199         break;
1200       svm_fifo_unset_event (s->rx_fifo);
1201       app_wrk = app_worker_get (s->app_wrk_index);
1202       app_worker_builtin_rx (app_wrk, s);
1203       break;
1204     case SESSION_IO_EVT_BUILTIN_TX:
1205       s = session_get_from_handle_if_valid (e->session_handle);
1206       wrk->ctx.s = s;
1207       if (PREDICT_TRUE (s != 0))
1208         session_tx_fifo_dequeue_internal (wrk, node, elt, n_tx_packets);
1209       break;
1210     default:
1211       clib_warning ("unhandled event type %d", e->event_type);
1212     }
1213
1214   /* Regrab elements in case pool moved */
1215   elt = pool_elt_at_index (wrk->event_elts, ei);
1216   if (!clib_llist_elt_is_linked (elt, evt_list))
1217     session_evt_elt_free (wrk, elt);
1218 }
1219
1220 /* *INDENT-OFF* */
1221 static const u32 session_evt_msg_sizes[] = {
1222 #define _(symc, sym)                                                    \
1223   [SESSION_CTRL_EVT_ ## symc] = sizeof (session_ ## sym ##_msg_t),
1224   foreach_session_ctrl_evt
1225 #undef _
1226 };
1227 /* *INDENT-ON* */
1228
1229 always_inline void
1230 session_evt_add_to_list (session_worker_t * wrk, session_event_t * evt)
1231 {
1232   session_evt_elt_t *elt;
1233
1234   if (evt->event_type >= SESSION_CTRL_EVT_RPC)
1235     {
1236       elt = session_evt_alloc_ctrl (wrk);
1237       if (evt->event_type >= SESSION_CTRL_EVT_BOUND)
1238         {
1239           elt->evt.ctrl_data_index = session_evt_ctrl_data_alloc (wrk);
1240           elt->evt.event_type = evt->event_type;
1241           clib_memcpy_fast (session_evt_ctrl_data (wrk, elt), evt->data,
1242                             session_evt_msg_sizes[evt->event_type]);
1243         }
1244       else
1245         {
1246           /* Internal control events fit into io events footprint */
1247           clib_memcpy_fast (&elt->evt, evt, sizeof (elt->evt));
1248         }
1249     }
1250   else
1251     {
1252       elt = session_evt_alloc_new (wrk);
1253       clib_memcpy_fast (&elt->evt, evt, sizeof (elt->evt));
1254     }
1255 }
1256
1257 static void
1258 session_flush_pending_tx_buffers (session_worker_t * wrk,
1259                                   vlib_node_runtime_t * node)
1260 {
1261   vlib_buffer_enqueue_to_next (wrk->vm, node, wrk->pending_tx_buffers,
1262                                wrk->pending_tx_nexts,
1263                                vec_len (wrk->pending_tx_nexts));
1264   vec_reset_length (wrk->pending_tx_buffers);
1265   vec_reset_length (wrk->pending_tx_nexts);
1266 }
1267
1268 static uword
1269 session_queue_node_fn (vlib_main_t * vm, vlib_node_runtime_t * node,
1270                        vlib_frame_t * frame)
1271 {
1272   session_main_t *smm = vnet_get_session_main ();
1273   u32 thread_index = vm->thread_index, n_to_dequeue;
1274   session_worker_t *wrk = &smm->wrk[thread_index];
1275   session_evt_elt_t *elt, *ctrl_he, *new_he, *old_he;
1276   clib_llist_index_t ei, next_ei, old_ti;
1277   svm_msg_q_msg_t _msg, *msg = &_msg;
1278   int i, n_tx_packets;
1279   session_event_t *evt;
1280   svm_msg_q_t *mq;
1281
1282   SESSION_EVT (SESSION_EVT_DISPATCH_START, wrk);
1283
1284   wrk->last_vlib_time = vlib_time_now (vm);
1285   wrk->last_vlib_us_time = wrk->last_vlib_time * CLIB_US_TIME_FREQ;
1286
1287   /*
1288    *  Update transport time
1289    */
1290   transport_update_time (wrk->last_vlib_time, thread_index);
1291   n_tx_packets = vec_len (wrk->pending_tx_buffers);
1292
1293   /*
1294    *  Dequeue and handle new events
1295    */
1296
1297   /* Try to dequeue what is available. Don't wait for lock.
1298    * XXX: we may need priorities here */
1299   mq = wrk->vpp_event_queue;
1300   n_to_dequeue = svm_msg_q_size (mq);
1301   if (n_to_dequeue && svm_msg_q_try_lock (mq) == 0)
1302     {
1303       for (i = 0; i < n_to_dequeue; i++)
1304         {
1305           svm_msg_q_sub_w_lock (mq, msg);
1306           evt = svm_msg_q_msg_data (mq, msg);
1307           session_evt_add_to_list (wrk, evt);
1308           svm_msg_q_free_msg (mq, msg);
1309         }
1310       svm_msg_q_unlock (mq);
1311     }
1312
1313   /*
1314    * Handle control events
1315    */
1316
1317   ctrl_he = pool_elt_at_index (wrk->event_elts, wrk->ctrl_head);
1318
1319   /* *INDENT-OFF* */
1320   clib_llist_foreach_safe (wrk->event_elts, evt_list, ctrl_he, elt, ({
1321     clib_llist_remove (wrk->event_elts, evt_list, elt);
1322     session_event_dispatch_ctrl (wrk, elt);
1323   }));
1324   /* *INDENT-ON* */
1325
1326   /*
1327    * Handle the new io events.
1328    */
1329
1330   new_he = pool_elt_at_index (wrk->event_elts, wrk->new_head);
1331   old_he = pool_elt_at_index (wrk->event_elts, wrk->old_head);
1332   old_ti = clib_llist_prev_index (old_he, evt_list);
1333
1334   ei = clib_llist_next_index (new_he, evt_list);
1335   while (ei != wrk->new_head && n_tx_packets < VLIB_FRAME_SIZE)
1336     {
1337       elt = pool_elt_at_index (wrk->event_elts, ei);
1338       ei = clib_llist_next_index (elt, evt_list);
1339       clib_llist_remove (wrk->event_elts, evt_list, elt);
1340       session_event_dispatch_io (wrk, node, elt, thread_index, &n_tx_packets);
1341     }
1342
1343   /*
1344    * Handle the old io events, if we had any prior to processing the new ones
1345    */
1346
1347   if (old_ti != wrk->old_head)
1348     {
1349       old_he = pool_elt_at_index (wrk->event_elts, wrk->old_head);
1350       ei = clib_llist_next_index (old_he, evt_list);
1351
1352       while (n_tx_packets < VLIB_FRAME_SIZE)
1353         {
1354           elt = pool_elt_at_index (wrk->event_elts, ei);
1355           next_ei = clib_llist_next_index (elt, evt_list);
1356           clib_llist_remove (wrk->event_elts, evt_list, elt);
1357
1358           session_event_dispatch_io (wrk, node, elt, thread_index,
1359                                      &n_tx_packets);
1360
1361           if (ei == old_ti)
1362             break;
1363
1364           ei = next_ei;
1365         };
1366     }
1367
1368   if (vec_len (wrk->pending_tx_buffers))
1369     session_flush_pending_tx_buffers (wrk, node);
1370
1371   vlib_node_increment_counter (vm, session_queue_node.index,
1372                                SESSION_QUEUE_ERROR_TX, n_tx_packets);
1373
1374   SESSION_EVT (SESSION_EVT_DISPATCH_END, wrk, n_tx_packets);
1375
1376   return n_tx_packets;
1377 }
1378
1379 /* *INDENT-OFF* */
1380 VLIB_REGISTER_NODE (session_queue_node) =
1381 {
1382   .function = session_queue_node_fn,
1383   .flags = VLIB_NODE_FLAG_TRACE_SUPPORTED,
1384   .name = "session-queue",
1385   .format_trace = format_session_queue_trace,
1386   .type = VLIB_NODE_TYPE_INPUT,
1387   .n_errors = ARRAY_LEN (session_queue_error_strings),
1388   .error_strings = session_queue_error_strings,
1389   .state = VLIB_NODE_STATE_DISABLED,
1390 };
1391 /* *INDENT-ON* */
1392
1393 void
1394 dump_thread_0_event_queue (void)
1395 {
1396   session_main_t *smm = vnet_get_session_main ();
1397   vlib_main_t *vm = &vlib_global_main;
1398   u32 my_thread_index = vm->thread_index;
1399   session_event_t _e, *e = &_e;
1400   svm_msg_q_ring_t *ring;
1401   session_t *s0;
1402   svm_msg_q_msg_t *msg;
1403   svm_msg_q_t *mq;
1404   int i, index;
1405
1406   mq = smm->wrk[my_thread_index].vpp_event_queue;
1407   index = mq->q->head;
1408
1409   for (i = 0; i < mq->q->cursize; i++)
1410     {
1411       msg = (svm_msg_q_msg_t *) (&mq->q->data[0] + mq->q->elsize * index);
1412       ring = svm_msg_q_ring (mq, msg->ring_index);
1413       clib_memcpy_fast (e, svm_msg_q_msg_data (mq, msg), ring->elsize);
1414
1415       switch (e->event_type)
1416         {
1417         case SESSION_IO_EVT_TX:
1418           s0 = session_event_get_session (e, my_thread_index);
1419           fformat (stdout, "[%04d] TX session %d\n", i, s0->session_index);
1420           break;
1421
1422         case SESSION_CTRL_EVT_CLOSE:
1423           s0 = session_get_from_handle (e->session_handle);
1424           fformat (stdout, "[%04d] disconnect session %d\n", i,
1425                    s0->session_index);
1426           break;
1427
1428         case SESSION_IO_EVT_BUILTIN_RX:
1429           s0 = session_event_get_session (e, my_thread_index);
1430           fformat (stdout, "[%04d] builtin_rx %d\n", i, s0->session_index);
1431           break;
1432
1433         case SESSION_CTRL_EVT_RPC:
1434           fformat (stdout, "[%04d] RPC call %llx with %llx\n",
1435                    i, (u64) (uword) (e->rpc_args.fp),
1436                    (u64) (uword) (e->rpc_args.arg));
1437           break;
1438
1439         default:
1440           fformat (stdout, "[%04d] unhandled event type %d\n",
1441                    i, e->event_type);
1442           break;
1443         }
1444
1445       index++;
1446
1447       if (index == mq->q->maxsize)
1448         index = 0;
1449     }
1450 }
1451
1452 static u8
1453 session_node_cmp_event (session_event_t * e, svm_fifo_t * f)
1454 {
1455   session_t *s;
1456   switch (e->event_type)
1457     {
1458     case SESSION_IO_EVT_RX:
1459     case SESSION_IO_EVT_TX:
1460     case SESSION_IO_EVT_BUILTIN_RX:
1461     case SESSION_IO_EVT_BUILTIN_TX:
1462     case SESSION_IO_EVT_TX_FLUSH:
1463       if (e->session_index == f->master_session_index)
1464         return 1;
1465       break;
1466     case SESSION_CTRL_EVT_CLOSE:
1467       break;
1468     case SESSION_CTRL_EVT_RPC:
1469       s = session_get_from_handle (e->session_handle);
1470       if (!s)
1471         {
1472           clib_warning ("session has event but doesn't exist!");
1473           break;
1474         }
1475       if (s->rx_fifo == f || s->tx_fifo == f)
1476         return 1;
1477       break;
1478     default:
1479       break;
1480     }
1481   return 0;
1482 }
1483
1484 u8
1485 session_node_lookup_fifo_event (svm_fifo_t * f, session_event_t * e)
1486 {
1487   session_evt_elt_t *elt;
1488   session_worker_t *wrk;
1489   int i, index, found = 0;
1490   svm_msg_q_msg_t *msg;
1491   svm_msg_q_ring_t *ring;
1492   svm_msg_q_t *mq;
1493   u8 thread_index;
1494
1495   ASSERT (e);
1496   thread_index = f->master_thread_index;
1497   wrk = session_main_get_worker (thread_index);
1498
1499   /*
1500    * Search evt queue
1501    */
1502   mq = wrk->vpp_event_queue;
1503   index = mq->q->head;
1504   for (i = 0; i < mq->q->cursize; i++)
1505     {
1506       msg = (svm_msg_q_msg_t *) (&mq->q->data[0] + mq->q->elsize * index);
1507       ring = svm_msg_q_ring (mq, msg->ring_index);
1508       clib_memcpy_fast (e, svm_msg_q_msg_data (mq, msg), ring->elsize);
1509       found = session_node_cmp_event (e, f);
1510       if (found)
1511         return 1;
1512       index = (index + 1) % mq->q->maxsize;
1513     }
1514   /*
1515    * Search pending events vector
1516    */
1517
1518   /* *INDENT-OFF* */
1519   clib_llist_foreach (wrk->event_elts, evt_list,
1520                       pool_elt_at_index (wrk->event_elts, wrk->new_head),
1521                       elt, ({
1522     found = session_node_cmp_event (&elt->evt, f);
1523     if (found)
1524       {
1525         clib_memcpy_fast (e, &elt->evt, sizeof (*e));
1526         goto done;
1527       }
1528   }));
1529   /* *INDENT-ON* */
1530
1531   /* *INDENT-OFF* */
1532   clib_llist_foreach (wrk->event_elts, evt_list,
1533                       pool_elt_at_index (wrk->event_elts, wrk->old_head),
1534                       elt, ({
1535     found = session_node_cmp_event (&elt->evt, f);
1536     if (found)
1537       {
1538         clib_memcpy_fast (e, &elt->evt, sizeof (*e));
1539         goto done;
1540       }
1541   }));
1542   /* *INDENT-ON* */
1543
1544 done:
1545   return found;
1546 }
1547
1548 static clib_error_t *
1549 session_queue_exit (vlib_main_t * vm)
1550 {
1551   if (vec_len (vlib_mains) < 2)
1552     return 0;
1553
1554   /*
1555    * Shut off (especially) worker-thread session nodes.
1556    * Otherwise, vpp can crash as the main thread unmaps the
1557    * API segment.
1558    */
1559   vlib_worker_thread_barrier_sync (vm);
1560   session_node_enable_disable (0 /* is_enable */ );
1561   vlib_worker_thread_barrier_release (vm);
1562   return 0;
1563 }
1564
1565 VLIB_MAIN_LOOP_EXIT_FUNCTION (session_queue_exit);
1566
1567 static uword
1568 session_queue_run_on_main (vlib_main_t * vm)
1569 {
1570   vlib_node_runtime_t *node;
1571
1572   node = vlib_node_get_runtime (vm, session_queue_node.index);
1573   return session_queue_node_fn (vm, node, 0);
1574 }
1575
1576 static uword
1577 session_queue_process (vlib_main_t * vm, vlib_node_runtime_t * rt,
1578                        vlib_frame_t * f)
1579 {
1580   uword *event_data = 0;
1581   f64 timeout = 1.0;
1582   uword event_type;
1583
1584   while (1)
1585     {
1586       vlib_process_wait_for_event_or_clock (vm, timeout);
1587       event_type = vlib_process_get_events (vm, (uword **) & event_data);
1588
1589       switch (event_type)
1590         {
1591         case SESSION_Q_PROCESS_RUN_ON_MAIN:
1592           /* Run session queue node on main thread */
1593           session_queue_run_on_main (vm);
1594           break;
1595         case SESSION_Q_PROCESS_STOP:
1596           timeout = 100000.0;
1597           break;
1598         case ~0:
1599           /* Timed out. Run on main to ensure all events are handled */
1600           session_queue_run_on_main (vm);
1601           break;
1602         }
1603       vec_reset_length (event_data);
1604     }
1605   return 0;
1606 }
1607
1608 /* *INDENT-OFF* */
1609 VLIB_REGISTER_NODE (session_queue_process_node) =
1610 {
1611   .function = session_queue_process,
1612   .type = VLIB_NODE_TYPE_PROCESS,
1613   .name = "session-queue-process",
1614   .state = VLIB_NODE_STATE_DISABLED,
1615 };
1616 /* *INDENT-ON* */
1617
1618 static_always_inline uword
1619 session_queue_pre_input_inline (vlib_main_t * vm, vlib_node_runtime_t * node,
1620                                 vlib_frame_t * frame)
1621 {
1622   session_main_t *sm = &session_main;
1623   if (!sm->wrk[0].vpp_event_queue)
1624     return 0;
1625   node = vlib_node_get_runtime (vm, session_queue_node.index);
1626   return session_queue_node_fn (vm, node, frame);
1627 }
1628
1629 /* *INDENT-OFF* */
1630 VLIB_REGISTER_NODE (session_queue_pre_input_node) =
1631 {
1632   .function = session_queue_pre_input_inline,
1633   .type = VLIB_NODE_TYPE_PRE_INPUT,
1634   .name = "session-queue-main",
1635   .state = VLIB_NODE_STATE_DISABLED,
1636 };
1637 /* *INDENT-ON* */
1638
1639 /*
1640  * fd.io coding-style-patch-verification: ON
1641  *
1642  * Local Variables:
1643  * eval: (c-set-style "gnu")
1644  * End:
1645  */