src/vnet/session/session_node.c

   1 /*
   2  * Copyright (c) 2017 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/session_debug.h>
  24 #include <svm/queue.h>
  25
  26 vlib_node_registration_t session_queue_node;
  27
  28 typedef struct
  29 {
  30   u32 session_index;
  31   u32 server_thread_index;
  32 } session_queue_trace_t;
  33
  34 /* packet trace format function */
  35 static u8 *
  36 format_session_queue_trace (u8 * s, va_list * args)
  37 {
  38   CLIB_UNUSED (vlib_main_t * vm) = va_arg (*args, vlib_main_t *);
  39   CLIB_UNUSED (vlib_node_t * node) = va_arg (*args, vlib_node_t *);
  40   session_queue_trace_t *t = va_arg (*args, session_queue_trace_t *);
  41
  42   s = format (s, "SESSION_QUEUE: session index %d, server thread index %d",
  43               t->session_index, t->server_thread_index);
  44   return s;
  45 }
  46
  47 vlib_node_registration_t session_queue_node;
  48
  49 #define foreach_session_queue_error             \
  50 _(TX, "Packets transmitted")                    \
  51 _(TIMER, "Timer events")                        \
  52 _(NO_BUFFER, "Out of buffers")
  53
  54 typedef enum
  55 {
  56 #define _(sym,str) SESSION_QUEUE_ERROR_##sym,
  57   foreach_session_queue_error
  58 #undef _
  59     SESSION_QUEUE_N_ERROR,
  60 } session_queue_error_t;
  61
  62 static char *session_queue_error_strings[] = {
  63 #define _(sym,string) string,
  64   foreach_session_queue_error
  65 #undef _
  66 };
  67
  68 static void
  69 session_tx_trace_frame (vlib_main_t * vm, vlib_node_runtime_t * node,
  70                         u32 next_index, u32 * to_next, u16 n_segs,
  71                         stream_session_t * s, u32 n_trace)
  72 {
  73   session_queue_trace_t *t;
  74   vlib_buffer_t *b;
  75   int i;
  76
  77   for (i = 0; i < clib_min (n_trace, n_segs); i++)
  78     {
  79       b = vlib_get_buffer (vm, to_next[i - n_segs]);
  80       vlib_trace_buffer (vm, node, next_index, b, 1 /* follow_chain */ );
  81       t = vlib_add_trace (vm, node, b, sizeof (*t));
  82       t->session_index = s->session_index;
  83       t->server_thread_index = s->thread_index;
  84     }
  85   vlib_set_trace_count (vm, node, n_trace - i);
  86 }
  87
  88 always_inline void
  89 session_tx_fifo_chain_tail (vlib_main_t * vm, session_tx_context_t * ctx,
  90                             vlib_buffer_t * b, u16 * n_bufs, u8 peek_data)
  91 {
  92   session_manager_main_t *smm = &session_manager_main;
  93   vlib_buffer_t *chain_b, *prev_b;
  94   u32 chain_bi0, to_deq, left_from_seg;
  95   u16 len_to_deq, n_bytes_read;
  96   u8 *data, j;
  97
  98   b->flags |= VLIB_BUFFER_TOTAL_LENGTH_VALID;
  99   b->total_length_not_including_first_buffer = 0;
 100
 101   chain_b = b;
 102   left_from_seg = clib_min (ctx->snd_mss - b->current_length,
 103                             ctx->left_to_snd);
 104   to_deq = left_from_seg;
 105   for (j = 1; j < ctx->n_bufs_per_seg; j++)
 106     {
 107       prev_b = chain_b;
 108       len_to_deq = clib_min (to_deq, ctx->deq_per_buf);
 109
 110       *n_bufs -= 1;
 111       chain_bi0 = smm->tx_buffers[ctx->s->thread_index][*n_bufs];
 112       _vec_len (smm->tx_buffers[ctx->s->thread_index]) = *n_bufs;
 113
 114       chain_b = vlib_get_buffer (vm, chain_bi0);
 115       chain_b->current_data = 0;
 116       data = vlib_buffer_get_current (chain_b);
 117       if (peek_data)
 118         {
 119           n_bytes_read = svm_fifo_peek (ctx->s->server_tx_fifo,
 120                                         ctx->tx_offset, len_to_deq, data);
 121           ctx->tx_offset += n_bytes_read;
 122         }
 123       else
 124         {
 125           if (ctx->transport_vft->tx_type == TRANSPORT_TX_DGRAM)
 126             {
 127               svm_fifo_t *f = ctx->s->server_tx_fifo;
 128               session_dgram_hdr_t *hdr = &ctx->hdr;
 129               u16 deq_now;
 130               deq_now = clib_min (hdr->data_length - hdr->data_offset,
 131                                   len_to_deq);
 132               n_bytes_read = svm_fifo_peek (f, hdr->data_offset, deq_now,
 133                                             data);
 134               ASSERT (n_bytes_read > 0);
 135
 136               hdr->data_offset += n_bytes_read;
 137               if (hdr->data_offset == hdr->data_length)
 138                 svm_fifo_dequeue_drop (f, hdr->data_length);
 139             }
 140           else
 141             n_bytes_read = svm_fifo_dequeue_nowait (ctx->s->server_tx_fifo,
 142                                                     len_to_deq, data);
 143         }
 144       ASSERT (n_bytes_read == len_to_deq);
 145       chain_b->current_length = n_bytes_read;
 146       b->total_length_not_including_first_buffer += chain_b->current_length;
 147
 148       /* update previous buffer */
 149       prev_b->next_buffer = chain_bi0;
 150       prev_b->flags |= VLIB_BUFFER_NEXT_PRESENT;
 151
 152       /* update current buffer */
 153       chain_b->next_buffer = 0;
 154
 155       to_deq -= n_bytes_read;
 156       if (to_deq == 0)
 157         break;
 158     }
 159   ASSERT (to_deq == 0
 160           && b->total_length_not_including_first_buffer == left_from_seg);
 161   ctx->left_to_snd -= left_from_seg;
 162 }
 163
 164 always_inline int
 165 session_output_try_get_buffers (vlib_main_t * vm,
 166                                 session_manager_main_t * smm,
 167                                 u32 thread_index, u16 * n_bufs, u32 wanted)
 168 {
 169   u32 n_alloc;
 170   vec_validate_aligned (smm->tx_buffers[thread_index], wanted - 1,
 171                         CLIB_CACHE_LINE_BYTES);
 172   n_alloc = vlib_buffer_alloc (vm, &smm->tx_buffers[thread_index][*n_bufs],
 173                                wanted - *n_bufs);
 174   *n_bufs += n_alloc;
 175   _vec_len (smm->tx_buffers[thread_index]) = *n_bufs;
 176   return n_alloc;
 177 }
 178
 179 always_inline void
 180 session_tx_fill_buffer (vlib_main_t * vm, session_tx_context_t * ctx,
 181                         vlib_buffer_t * b, u16 * n_bufs, u8 peek_data)
 182 {
 183   u32 len_to_deq;
 184   u8 *data0;
 185   int n_bytes_read;
 186
 187   /*
 188    * Start with the first buffer in chain
 189    */
 190   b->error = 0;
 191   b->flags = VNET_BUFFER_F_LOCALLY_ORIGINATED;
 192   b->current_data = 0;
 193
 194   data0 = vlib_buffer_make_headroom (b, MAX_HDRS_LEN);
 195   len_to_deq = clib_min (ctx->left_to_snd, ctx->deq_per_first_buf);
 196
 197   if (peek_data)
 198     {
 199       n_bytes_read = svm_fifo_peek (ctx->s->server_tx_fifo, ctx->tx_offset,
 200                                     len_to_deq, data0);
 201       ASSERT (n_bytes_read > 0);
 202       /* Keep track of progress locally, transport is also supposed to
 203        * increment it independently when pushing the header */
 204       ctx->tx_offset += n_bytes_read;
 205     }
 206   else
 207     {
 208       if (ctx->transport_vft->tx_type == TRANSPORT_TX_DGRAM)
 209         {
 210           session_dgram_hdr_t *hdr = &ctx->hdr;
 211           svm_fifo_t *f = ctx->s->server_tx_fifo;
 212           u16 deq_now;
 213           u32 offset;
 214
 215           ASSERT (hdr->data_length > hdr->data_offset);
 216           deq_now = clib_min (hdr->data_length - hdr->data_offset,
 217                               len_to_deq);
 218           offset = hdr->data_offset + SESSION_CONN_HDR_LEN;
 219           n_bytes_read = svm_fifo_peek (f, offset, deq_now, data0);
 220           ASSERT (n_bytes_read > 0);
 221
 222           if (ctx->s->session_state == SESSION_STATE_LISTENING)
 223             {
 224               ip_copy (&ctx->tc->rmt_ip, &hdr->rmt_ip, ctx->tc->is_ip4);
 225               ctx->tc->rmt_port = hdr->rmt_port;
 226             }
 227           hdr->data_offset += n_bytes_read;
 228           if (hdr->data_offset == hdr->data_length)
 229             {
 230               offset = hdr->data_length + SESSION_CONN_HDR_LEN;
 231               svm_fifo_dequeue_drop (f, offset);
 232             }
 233         }
 234       else
 235         {
 236           n_bytes_read = svm_fifo_dequeue_nowait (ctx->s->server_tx_fifo,
 237                                                   len_to_deq, data0);
 238           ASSERT (n_bytes_read > 0);
 239         }
 240     }
 241   b->current_length = n_bytes_read;
 242   ctx->left_to_snd -= n_bytes_read;
 243
 244   /*
 245    * Fill in the remaining buffers in the chain, if any
 246    */
 247   if (PREDICT_FALSE (ctx->n_bufs_per_seg > 1 && ctx->left_to_snd))
 248     session_tx_fifo_chain_tail (vm, ctx, b, n_bufs, peek_data);
 249
 250   /* *INDENT-OFF* */
 251   SESSION_EVT_DBG(SESSION_EVT_DEQ, ctx->s, ({
 252         ed->data[0] = FIFO_EVENT_APP_TX;
 253         ed->data[1] = ctx->max_dequeue;
 254         ed->data[2] = len_to_deq;
 255         ed->data[3] = ctx->left_to_snd;
 256   }));
 257   /* *INDENT-ON* */
 258 }
 259
 260 always_inline u8
 261 session_tx_not_ready (stream_session_t * s, u8 peek_data)
 262 {
 263   if (peek_data)
 264     {
 265       /* Can retransmit for closed sessions but can't send new data if
 266        * session is not ready or closed */
 267       if (s->session_state < SESSION_STATE_READY)
 268         return 1;
 269       if (s->session_state == SESSION_STATE_CLOSED)
 270         return 2;
 271     }
 272   return 0;
 273 }
 274
 275 always_inline transport_connection_t *
 276 session_tx_get_transport (session_tx_context_t * ctx, u8 peek_data)
 277 {
 278   if (peek_data)
 279     {
 280       return ctx->transport_vft->get_connection (ctx->s->connection_index,
 281                                                  ctx->s->thread_index);
 282     }
 283   else
 284     {
 285       if (ctx->s->session_state == SESSION_STATE_LISTENING)
 286         return ctx->transport_vft->get_listener (ctx->s->connection_index);
 287       else
 288         {
 289           return ctx->transport_vft->get_connection (ctx->s->connection_index,
 290                                                      ctx->s->thread_index);
 291         }
 292     }
 293 }
 294
 295 always_inline void
 296 session_tx_set_dequeue_params (vlib_main_t * vm, session_tx_context_t * ctx,
 297                                u32 max_segs, u8 peek_data)
 298 {
 299   u32 n_bytes_per_buf, n_bytes_per_seg;
 300   ctx->max_dequeue = svm_fifo_max_dequeue (ctx->s->server_tx_fifo);
 301   if (peek_data)
 302     {
 303       /* Offset in rx fifo from where to peek data */
 304       ctx->tx_offset = ctx->transport_vft->tx_fifo_offset (ctx->tc);
 305       if (PREDICT_FALSE (ctx->tx_offset >= ctx->max_dequeue))
 306         {
 307           ctx->max_len_to_snd = 0;
 308           return;
 309         }
 310       ctx->max_dequeue -= ctx->tx_offset;
 311     }
 312   else
 313     {
 314       if (ctx->transport_vft->tx_type == TRANSPORT_TX_DGRAM)
 315         {
 316           if (ctx->max_dequeue <= sizeof (ctx->hdr))
 317             {
 318               ctx->max_len_to_snd = 0;
 319               return;
 320             }
 321           svm_fifo_peek (ctx->s->server_tx_fifo, 0, sizeof (ctx->hdr),
 322                          (u8 *) & ctx->hdr);
 323           ASSERT (ctx->hdr.data_length > ctx->hdr.data_offset);
 324           ctx->max_dequeue = ctx->hdr.data_length - ctx->hdr.data_offset;
 325         }
 326     }
 327   ASSERT (ctx->max_dequeue > 0);
 328
 329   /* Ensure we're not writing more than transport window allows */
 330   if (ctx->max_dequeue < ctx->snd_space)
 331     {
 332       /* Constrained by tx queue. Try to send only fully formed segments */
 333       ctx->max_len_to_snd =
 334         (ctx->max_dequeue > ctx->snd_mss) ?
 335         ctx->max_dequeue - ctx->max_dequeue % ctx->snd_mss : ctx->max_dequeue;
 336       /* TODO Nagle ? */
 337     }
 338   else
 339     {
 340       /* Expectation is that snd_space0 is already a multiple of snd_mss */
 341       ctx->max_len_to_snd = ctx->snd_space;
 342     }
 343
 344   /* Check if we're tx constrained by the node */
 345   ctx->n_segs_per_evt = ceil ((f64) ctx->max_len_to_snd / ctx->snd_mss);
 346   if (ctx->n_segs_per_evt > max_segs)
 347     {
 348       ctx->n_segs_per_evt = max_segs;
 349       ctx->max_len_to_snd = max_segs * ctx->snd_mss;
 350     }
 351
 352   n_bytes_per_buf = vlib_buffer_free_list_buffer_size (vm,
 353                                                        VLIB_BUFFER_DEFAULT_FREE_LIST_INDEX);
 354   ASSERT (n_bytes_per_buf > MAX_HDRS_LEN);
 355   n_bytes_per_seg = MAX_HDRS_LEN + ctx->snd_mss;
 356   ctx->n_bufs_per_seg = ceil ((f64) n_bytes_per_seg / n_bytes_per_buf);
 357   ctx->deq_per_buf = clib_min (ctx->snd_mss, n_bytes_per_buf);
 358   ctx->deq_per_first_buf = clib_min (ctx->snd_mss,
 359                                      n_bytes_per_buf - MAX_HDRS_LEN);
 360 }
 361
 362 always_inline int
 363 session_tx_fifo_read_and_snd_i (vlib_main_t * vm, vlib_node_runtime_t * node,
 364                                 session_fifo_event_t * e,
 365                                 stream_session_t * s, int *n_tx_packets,
 366                                 u8 peek_data)
 367 {
 368   u32 next_index, next0, next1, *to_next, n_left_to_next;
 369   u32 n_trace = vlib_get_trace_count (vm, node), n_bufs_needed = 0;
 370   u32 thread_index = s->thread_index, n_left, pbi;
 371   session_manager_main_t *smm = &session_manager_main;
 372   session_tx_context_t *ctx = &smm->ctx[thread_index];
 373   transport_proto_t tp;
 374   vlib_buffer_t *pb;
 375   u16 n_bufs, rv;
 376
 377   if (PREDICT_FALSE ((rv = session_tx_not_ready (s, peek_data))))
 378     {
 379       if (rv < 2)
 380         vec_add1 (smm->pending_event_vector[thread_index], *e);
 381       return 0;
 382     }
 383
 384   next_index = smm->session_type_to_next[s->session_type];
 385   next0 = next1 = next_index;
 386
 387   tp = session_get_transport_proto (s);
 388   ctx->s = s;
 389   ctx->transport_vft = transport_protocol_get_vft (tp);
 390   ctx->tc = session_tx_get_transport (ctx, peek_data);
 391   ctx->snd_mss = ctx->transport_vft->send_mss (ctx->tc);
 392   ctx->snd_space = ctx->transport_vft->send_space (ctx->tc);
 393   if (ctx->snd_space == 0 || ctx->snd_mss == 0)
 394     {
 395       vec_add1 (smm->pending_event_vector[thread_index], *e);
 396       return 0;
 397     }
 398
 399   /* Allow enqueuing of a new event */
 400   svm_fifo_unset_event (s->server_tx_fifo);
 401
 402   /* Check how much we can pull. */
 403   session_tx_set_dequeue_params (vm, ctx, VLIB_FRAME_SIZE - *n_tx_packets,
 404                                  peek_data);
 405
 406   if (PREDICT_FALSE (!ctx->max_len_to_snd))
 407     return 0;
 408
 409   n_bufs = vec_len (smm->tx_buffers[thread_index]);
 410   n_bufs_needed = ctx->n_segs_per_evt * ctx->n_bufs_per_seg;
 411
 412   /*
 413    * Make sure we have at least one full frame of buffers ready
 414    */
 415   if (n_bufs < n_bufs_needed)
 416     {
 417       session_output_try_get_buffers (vm, smm, thread_index, &n_bufs,
 418                                       ctx->n_bufs_per_seg * VLIB_FRAME_SIZE);
 419       if (PREDICT_FALSE (n_bufs < n_bufs_needed))
 420         {
 421           vec_add1 (smm->pending_event_vector[thread_index], *e);
 422           return -1;
 423         }
 424     }
 425
 426   /*
 427    * Write until we fill up a frame
 428    */
 429   vlib_get_next_frame (vm, node, next_index, to_next, n_left_to_next);
 430   if (PREDICT_FALSE (ctx->n_segs_per_evt > n_left_to_next))
 431     {
 432       ctx->n_segs_per_evt = n_left_to_next;
 433       ctx->max_len_to_snd = ctx->snd_mss * n_left_to_next;
 434     }
 435   ctx->left_to_snd = ctx->max_len_to_snd;
 436   n_left = ctx->n_segs_per_evt;
 437
 438   while (n_left >= 4)
 439     {
 440       vlib_buffer_t *b0, *b1;
 441       u32 bi0, bi1;
 442
 443       pbi = smm->tx_buffers[thread_index][n_bufs - 3];
 444       pb = vlib_get_buffer (vm, pbi);
 445       vlib_prefetch_buffer_header (pb, STORE);
 446       pbi = smm->tx_buffers[thread_index][n_bufs - 4];
 447       pb = vlib_get_buffer (vm, pbi);
 448       vlib_prefetch_buffer_header (pb, STORE);
 449
 450       to_next[0] = bi0 = smm->tx_buffers[thread_index][--n_bufs];
 451       to_next[1] = bi1 = smm->tx_buffers[thread_index][--n_bufs];
 452
 453       b0 = vlib_get_buffer (vm, bi0);
 454       b1 = vlib_get_buffer (vm, bi1);
 455
 456       session_tx_fill_buffer (vm, ctx, b0, &n_bufs, peek_data);
 457       session_tx_fill_buffer (vm, ctx, b1, &n_bufs, peek_data);
 458
 459       ctx->transport_vft->push_header (ctx->tc, b0);
 460       ctx->transport_vft->push_header (ctx->tc, b1);
 461
 462       to_next += 2;
 463       n_left_to_next -= 2;
 464       n_left -= 2;
 465
 466       VLIB_BUFFER_TRACE_TRAJECTORY_INIT (b0);
 467       VLIB_BUFFER_TRACE_TRAJECTORY_INIT (b1);
 468
 469       vlib_validate_buffer_enqueue_x2 (vm, node, next_index, to_next,
 470                                        n_left_to_next, bi0, bi1, next0,
 471                                        next1);
 472     }
 473   while (n_left)
 474     {
 475       vlib_buffer_t *b0;
 476       u32 bi0;
 477
 478       ASSERT (n_bufs >= 1);
 479       to_next[0] = bi0 = smm->tx_buffers[thread_index][--n_bufs];
 480       b0 = vlib_get_buffer (vm, bi0);
 481       session_tx_fill_buffer (vm, ctx, b0, &n_bufs, peek_data);
 482
 483       /* Ask transport to push header after current_length and
 484        * total_length_not_including_first_buffer are updated */
 485       ctx->transport_vft->push_header (ctx->tc, b0);
 486
 487       to_next += 1;
 488       n_left_to_next -= 1;
 489       n_left -= 1;
 490
 491       VLIB_BUFFER_TRACE_TRAJECTORY_INIT (b0);
 492
 493       vlib_validate_buffer_enqueue_x1 (vm, node, next_index, to_next,
 494                                        n_left_to_next, bi0, next0);
 495     }
 496
 497   if (PREDICT_FALSE (n_trace > 0))
 498     session_tx_trace_frame (vm, node, next_index, to_next,
 499                             ctx->n_segs_per_evt, s, n_trace);
 500
 501   _vec_len (smm->tx_buffers[thread_index]) = n_bufs;
 502   *n_tx_packets += ctx->n_segs_per_evt;
 503   vlib_put_next_frame (vm, node, next_index, n_left_to_next);
 504
 505   /* If we couldn't dequeue all bytes mark as partially read */
 506   ASSERT (ctx->left_to_snd == 0);
 507   if (ctx->max_len_to_snd < ctx->max_dequeue)
 508     if (svm_fifo_set_event (s->server_tx_fifo))
 509       vec_add1 (smm->pending_event_vector[thread_index], *e);
 510
 511   if (!peek_data && ctx->transport_vft->tx_type == TRANSPORT_TX_DGRAM)
 512     {
 513       /* Fix dgram pre header */
 514       if (ctx->max_len_to_snd < ctx->max_dequeue)
 515         svm_fifo_overwrite_head (s->server_tx_fifo, (u8 *) & ctx->hdr,
 516                                  sizeof (session_dgram_pre_hdr_t));
 517       /* More data needs to be read */
 518       else if (svm_fifo_max_dequeue (s->server_tx_fifo) > 0)
 519         if (svm_fifo_set_event (s->server_tx_fifo))
 520           vec_add1 (smm->pending_event_vector[thread_index], *e);
 521     }
 522   return 0;
 523 }
 524
 525 int
 526 session_tx_fifo_peek_and_snd (vlib_main_t * vm, vlib_node_runtime_t * node,
 527                               session_fifo_event_t * e0,
 528                               stream_session_t * s0, int *n_tx_pkts)
 529 {
 530   return session_tx_fifo_read_and_snd_i (vm, node, e0, s0, n_tx_pkts, 1);
 531 }
 532
 533 int
 534 session_tx_fifo_dequeue_and_snd (vlib_main_t * vm, vlib_node_runtime_t * node,
 535                                  session_fifo_event_t * e0,
 536                                  stream_session_t * s0, int *n_tx_pkts)
 537 {
 538   return session_tx_fifo_read_and_snd_i (vm, node, e0, s0, n_tx_pkts, 0);
 539 }
 540
 541 int
 542 session_tx_fifo_dequeue_internal (vlib_main_t * vm,
 543                                   vlib_node_runtime_t * node,
 544                                   session_fifo_event_t * e0,
 545                                   stream_session_t * s0, int *n_tx_pkts)
 546 {
 547   application_t *app;
 548   app = application_get (s0->opaque);
 549   svm_fifo_unset_event (s0->server_tx_fifo);
 550   return app->cb_fns.builtin_app_tx_callback (s0);
 551 }
 552
 553 always_inline stream_session_t *
 554 session_event_get_session (session_fifo_event_t * e, u8 thread_index)
 555 {
 556   return session_get_if_valid (e->fifo->master_session_index, thread_index);
 557 }
 558
 559 void
 560 dump_thread_0_event_queue (void)
 561 {
 562   session_manager_main_t *smm = vnet_get_session_manager_main ();
 563   vlib_main_t *vm = &vlib_global_main;
 564   u32 my_thread_index = vm->thread_index;
 565   session_fifo_event_t _e, *e = &_e;
 566   stream_session_t *s0;
 567   int i, index;
 568   i8 *headp;
 569
 570   svm_queue_t *q;
 571   q = smm->vpp_event_queues[my_thread_index];
 572
 573   index = q->head;
 574
 575   for (i = 0; i < q->cursize; i++)
 576     {
 577       headp = (i8 *) (&q->data[0] + q->elsize * index);
 578       clib_memcpy (e, headp, q->elsize);
 579
 580       switch (e->event_type)
 581         {
 582         case FIFO_EVENT_APP_TX:
 583           s0 = session_event_get_session (e, my_thread_index);
 584           fformat (stdout, "[%04d] TX session %d\n", i, s0->session_index);
 585           break;
 586
 587         case FIFO_EVENT_DISCONNECT:
 588           s0 = session_get_from_handle (e->session_handle);
 589           fformat (stdout, "[%04d] disconnect session %d\n", i,
 590                    s0->session_index);
 591           break;
 592
 593         case FIFO_EVENT_BUILTIN_RX:
 594           s0 = session_event_get_session (e, my_thread_index);
 595           fformat (stdout, "[%04d] builtin_rx %d\n", i, s0->session_index);
 596           break;
 597
 598         case FIFO_EVENT_RPC:
 599           fformat (stdout, "[%04d] RPC call %llx with %llx\n",
 600                    i, (u64) (e->rpc_args.fp), (u64) (e->rpc_args.arg));
 601           break;
 602
 603         default:
 604           fformat (stdout, "[%04d] unhandled event type %d\n",
 605                    i, e->event_type);
 606           break;
 607         }
 608
 609       index++;
 610
 611       if (index == q->maxsize)
 612         index = 0;
 613     }
 614 }
 615
 616 static u8
 617 session_node_cmp_event (session_fifo_event_t * e, svm_fifo_t * f)
 618 {
 619   stream_session_t *s;
 620   switch (e->event_type)
 621     {
 622     case FIFO_EVENT_APP_RX:
 623     case FIFO_EVENT_APP_TX:
 624     case FIFO_EVENT_BUILTIN_RX:
 625       if (e->fifo == f)
 626         return 1;
 627       break;
 628     case FIFO_EVENT_DISCONNECT:
 629       break;
 630     case FIFO_EVENT_RPC:
 631       s = session_get_from_handle (e->session_handle);
 632       if (!s)
 633         {
 634           clib_warning ("session has event but doesn't exist!");
 635           break;
 636         }
 637       if (s->server_rx_fifo == f || s->server_tx_fifo == f)
 638         return 1;
 639       break;
 640     default:
 641       break;
 642     }
 643   return 0;
 644 }
 645
 646 u8
 647 session_node_lookup_fifo_event (svm_fifo_t * f, session_fifo_event_t * e)
 648 {
 649   session_manager_main_t *smm = vnet_get_session_manager_main ();
 650   svm_queue_t *q;
 651   session_fifo_event_t *pending_event_vector, *evt;
 652   int i, index, found = 0;
 653   i8 *headp;
 654   u8 thread_index;
 655
 656   ASSERT (e);
 657   thread_index = f->master_thread_index;
 658   /*
 659    * Search evt queue
 660    */
 661   q = smm->vpp_event_queues[thread_index];
 662   index = q->head;
 663   for (i = 0; i < q->cursize; i++)
 664     {
 665       headp = (i8 *) (&q->data[0] + q->elsize * index);
 666       clib_memcpy (e, headp, q->elsize);
 667       found = session_node_cmp_event (e, f);
 668       if (found)
 669         return 1;
 670       if (++index == q->maxsize)
 671         index = 0;
 672     }
 673   /*
 674    * Search pending events vector
 675    */
 676   pending_event_vector = smm->pending_event_vector[thread_index];
 677   vec_foreach (evt, pending_event_vector)
 678   {
 679     found = session_node_cmp_event (evt, f);
 680     if (found)
 681       {
 682         clib_memcpy (e, evt, sizeof (*evt));
 683         break;
 684       }
 685   }
 686   return found;
 687 }
 688
 689 static uword
 690 session_queue_node_fn (vlib_main_t * vm, vlib_node_runtime_t * node,
 691                        vlib_frame_t * frame)
 692 {
 693   session_manager_main_t *smm = vnet_get_session_manager_main ();
 694   session_fifo_event_t *my_pending_event_vector, *e;
 695   session_fifo_event_t *my_fifo_events;
 696   u32 n_to_dequeue, n_events;
 697   svm_queue_t *q;
 698   application_t *app;
 699   int n_tx_packets = 0;
 700   u32 thread_index = vm->thread_index;
 701   int i, rv;
 702   f64 now = vlib_time_now (vm);
 703   void (*fp) (void *);
 704
 705   SESSION_EVT_DBG (SESSION_EVT_POLL_GAP_TRACK, smm, thread_index);
 706
 707   /*
 708    *  Update transport time
 709    */
 710   transport_update_time (now, thread_index);
 711
 712   /*
 713    * Get vpp queue events
 714    */
 715   q = smm->vpp_event_queues[thread_index];
 716   if (PREDICT_FALSE (q == 0))
 717     return 0;
 718
 719   my_fifo_events = smm->free_event_vector[thread_index];
 720
 721   /* min number of events we can dequeue without blocking */
 722   n_to_dequeue = q->cursize;
 723   my_pending_event_vector = smm->pending_event_vector[thread_index];
 724
 725   if (!n_to_dequeue && !vec_len (my_pending_event_vector)
 726       && !vec_len (smm->pending_disconnects[thread_index]))
 727     return 0;
 728
 729   SESSION_EVT_DBG (SESSION_EVT_DEQ_NODE, 0);
 730
 731   /*
 732    * If we didn't manage to process previous events try going
 733    * over them again without dequeuing new ones.
 734    */
 735   /* XXX: Block senders to sessions that can't keep up */
 736   if (0 && vec_len (my_pending_event_vector) >= 100)
 737     {
 738       clib_warning ("too many fifo events unsolved");
 739       goto skip_dequeue;
 740     }
 741
 742   /* See you in the next life, don't be late */
 743   if (pthread_mutex_trylock (&q->mutex))
 744     return 0;
 745
 746   for (i = 0; i < n_to_dequeue; i++)
 747     {
 748       vec_add2 (my_fifo_events, e, 1);
 749       svm_queue_sub_raw (q, (u8 *) e);
 750     }
 751
 752   /* The other side of the connection is not polling */
 753   if (q->cursize < (q->maxsize / 8))
 754     (void) pthread_cond_broadcast (&q->condvar);
 755   pthread_mutex_unlock (&q->mutex);
 756
 757   vec_append (my_fifo_events, my_pending_event_vector);
 758   vec_append (my_fifo_events, smm->pending_disconnects[thread_index]);
 759
 760   _vec_len (my_pending_event_vector) = 0;
 761   smm->pending_event_vector[thread_index] = my_pending_event_vector;
 762   _vec_len (smm->pending_disconnects[thread_index]) = 0;
 763
 764 skip_dequeue:
 765   n_events = vec_len (my_fifo_events);
 766   for (i = 0; i < n_events; i++)
 767     {
 768       stream_session_t *s0;     /* $$$ prefetch 1 ahead maybe */
 769       session_fifo_event_t *e0;
 770
 771       e0 = &my_fifo_events[i];
 772       switch (e0->event_type)
 773         {
 774         case FIFO_EVENT_APP_TX:
 775           if (n_tx_packets == VLIB_FRAME_SIZE)
 776             {
 777               vec_add1 (smm->pending_event_vector[thread_index], *e0);
 778               break;
 779             }
 780
 781           s0 = session_event_get_session (e0, thread_index);
 782           if (PREDICT_FALSE (!s0))
 783             {
 784               clib_warning ("It's dead, Jim!");
 785               continue;
 786             }
 787
 788           /* Spray packets in per session type frames, since they go to
 789            * different nodes */
 790           rv = (smm->session_tx_fns[s0->session_type]) (vm, node, e0, s0,
 791                                                         &n_tx_packets);
 792           /* Out of buffers */
 793           if (PREDICT_FALSE (rv < 0))
 794             {
 795               vlib_node_increment_counter (vm, node->node_index,
 796                                            SESSION_QUEUE_ERROR_NO_BUFFER, 1);
 797               continue;
 798             }
 799           break;
 800         case FIFO_EVENT_DISCONNECT:
 801           /* Make sure stream disconnects run after the pending list is
 802            * drained */
 803           s0 = session_get_from_handle (e0->session_handle);
 804           if (!e0->postponed)
 805             {
 806               e0->postponed = 1;
 807               vec_add1 (smm->pending_disconnects[thread_index], *e0);
 808               continue;
 809             }
 810           /* If tx queue is still not empty, wait */
 811           if (svm_fifo_max_dequeue (s0->server_tx_fifo))
 812             {
 813               vec_add1 (smm->pending_disconnects[thread_index], *e0);
 814               continue;
 815             }
 816
 817           stream_session_disconnect_transport (s0);
 818           break;
 819         case FIFO_EVENT_BUILTIN_RX:
 820           s0 = session_event_get_session (e0, thread_index);
 821           if (PREDICT_FALSE (!s0))
 822             continue;
 823           svm_fifo_unset_event (s0->server_rx_fifo);
 824           app = application_get (s0->app_index);
 825           app->cb_fns.builtin_app_rx_callback (s0);
 826           break;
 827         case FIFO_EVENT_RPC:
 828           fp = e0->rpc_args.fp;
 829           (*fp) (e0->rpc_args.arg);
 830           break;
 831
 832         default:
 833           clib_warning ("unhandled event type %d", e0->event_type);
 834         }
 835     }
 836
 837   _vec_len (my_fifo_events) = 0;
 838   smm->free_event_vector[thread_index] = my_fifo_events;
 839
 840   vlib_node_increment_counter (vm, session_queue_node.index,
 841                                SESSION_QUEUE_ERROR_TX, n_tx_packets);
 842
 843   SESSION_EVT_DBG (SESSION_EVT_DISPATCH_END, smm, thread_index);
 844
 845   return n_tx_packets;
 846 }
 847
 848 /* *INDENT-OFF* */
 849 VLIB_REGISTER_NODE (session_queue_node) =
 850 {
 851   .function = session_queue_node_fn,
 852   .name = "session-queue",
 853   .format_trace = format_session_queue_trace,
 854   .type = VLIB_NODE_TYPE_INPUT,
 855   .n_errors = ARRAY_LEN (session_queue_error_strings),
 856   .error_strings = session_queue_error_strings,
 857   .state = VLIB_NODE_STATE_DISABLED,
 858 };
 859 /* *INDENT-ON* */
 860
 861 static clib_error_t *
 862 session_queue_exit (vlib_main_t * vm)
 863 {
 864   if (vec_len (vlib_mains) < 2)
 865     return 0;
 866
 867   /*
 868    * Shut off (especially) worker-thread session nodes.
 869    * Otherwise, vpp can crash as the main thread unmaps the
 870    * API segment.
 871    */
 872   vlib_worker_thread_barrier_sync (vm);
 873   session_node_enable_disable (0 /* is_enable */ );
 874   vlib_worker_thread_barrier_release (vm);
 875   return 0;
 876 }
 877
 878 VLIB_MAIN_LOOP_EXIT_FUNCTION (session_queue_exit);
 879
 880 static uword
 881 session_queue_process (vlib_main_t * vm, vlib_node_runtime_t * rt,
 882                        vlib_frame_t * f)
 883 {
 884   f64 now, timeout = 1.0;
 885   uword *event_data = 0;
 886   uword event_type;
 887
 888   while (1)
 889     {
 890       vlib_process_wait_for_event_or_clock (vm, timeout);
 891       now = vlib_time_now (vm);
 892       event_type = vlib_process_get_events (vm, (uword **) & event_data);
 893
 894       switch (event_type)
 895         {
 896         case SESSION_Q_PROCESS_FLUSH_FRAMES:
 897           /* Flush the frames by updating all transports times */
 898           transport_update_time (now, 0);
 899           break;
 900         case SESSION_Q_PROCESS_STOP:
 901           timeout = 100000.0;
 902           break;
 903         case ~0:
 904           /* Timed out. Update time for all transports to trigger all
 905            * outstanding retransmits. */
 906           transport_update_time (now, 0);
 907           break;
 908         }
 909       vec_reset_length (event_data);
 910     }
 911   return 0;
 912 }
 913
 914 /* *INDENT-OFF* */
 915 VLIB_REGISTER_NODE (session_queue_process_node) =
 916 {
 917   .function = session_queue_process,
 918   .type = VLIB_NODE_TYPE_PROCESS,
 919   .name = "session-queue-process",
 920   .state = VLIB_NODE_STATE_DISABLED,
 921 };
 922 /* *INDENT-ON* */
 923
 924
 925 /*
 926  * fd.io coding-style-patch-verification: ON
 927  *
 928  * Local Variables:
 929  * eval: (c-set-style "gnu")
 930  * End:
 931  */