src/vnet/session/session_input.c

   1 /* SPDX-License-Identifier: Apache-2.0
   2  * Copyright(c) 2023 Cisco Systems, Inc.
   3  */
   4
   5 #include <vnet/session/session.h>
   6 #include <vnet/session/application.h>
   7
   8 static inline int
   9 mq_try_lock (svm_msg_q_t *mq)
  10 {
  11   int rv, n_try = 0;
  12
  13   while (n_try < 100)
  14     {
  15       rv = svm_msg_q_try_lock (mq);
  16       if (!rv)
  17         return 0;
  18       n_try += 1;
  19       usleep (1);
  20     }
  21
  22   return -1;
  23 }
  24
  25 always_inline u8
  26 mq_event_ring_index (session_evt_type_t et)
  27 {
  28   return (et >= SESSION_CTRL_EVT_RPC ? SESSION_MQ_CTRL_EVT_RING :
  29                                              SESSION_MQ_IO_EVT_RING);
  30 }
  31
  32 void
  33 app_worker_del_all_events (app_worker_t *app_wrk)
  34 {
  35   session_worker_t *wrk;
  36   session_event_t *evt;
  37   u32 thread_index;
  38   session_t *s;
  39
  40   for (thread_index = 0; thread_index < vec_len (app_wrk->wrk_evts);
  41        thread_index++)
  42     {
  43       while (clib_fifo_elts (app_wrk->wrk_evts[thread_index]))
  44         {
  45           clib_fifo_sub2 (app_wrk->wrk_evts[thread_index], evt);
  46           switch (evt->event_type)
  47             {
  48             case SESSION_CTRL_EVT_MIGRATED:
  49               s = session_get (evt->session_index, thread_index);
  50               transport_cleanup (session_get_transport_proto (s),
  51                                  s->connection_index, s->thread_index);
  52               session_free (s);
  53               break;
  54             case SESSION_CTRL_EVT_CLEANUP:
  55               s = session_get (evt->as_u64[0] & 0xffffffff, thread_index);
  56               if (evt->as_u64[0] >> 32 != SESSION_CLEANUP_SESSION)
  57                 break;
  58               uword_to_pointer (evt->as_u64[1], void (*) (session_t * s)) (s);
  59               break;
  60             case SESSION_CTRL_EVT_HALF_CLEANUP:
  61               s = ho_session_get (evt->session_index);
  62               pool_put_index (app_wrk->half_open_table, s->ho_index);
  63               session_free (s);
  64               break;
  65             default:
  66               break;
  67             }
  68         }
  69       wrk = session_main_get_worker (thread_index);
  70       clib_bitmap_set (wrk->app_wrks_pending_ntf, app_wrk->wrk_index, 0);
  71     }
  72 }
  73
  74 always_inline int
  75 app_worker_flush_events_inline (app_worker_t *app_wrk, u32 thread_index,
  76                                 u8 is_builtin)
  77 {
  78   application_t *app = application_get (app_wrk->app_index);
  79   svm_msg_q_t *mq = app_wrk->event_queue;
  80   u8 ring_index, mq_is_cong;
  81   session_state_t old_state;
  82   session_event_t *evt;
  83   u32 n_evts = 128, i;
  84   session_t *s;
  85   int rv;
  86
  87   n_evts = clib_min (n_evts, clib_fifo_elts (app_wrk->wrk_evts[thread_index]));
  88
  89   if (!is_builtin)
  90     {
  91       mq_is_cong = app_worker_mq_is_congested (app_wrk);
  92       if (mq_try_lock (mq))
  93         {
  94           app_worker_set_mq_wrk_congested (app_wrk, thread_index);
  95           return 0;
  96         }
  97     }
  98
  99   for (i = 0; i < n_evts; i++)
 100     {
 101       evt = clib_fifo_head (app_wrk->wrk_evts[thread_index]);
 102       if (!is_builtin)
 103         {
 104           ring_index = mq_event_ring_index (evt->event_type);
 105           if (svm_msg_q_or_ring_is_full (mq, ring_index))
 106             {
 107               app_worker_set_mq_wrk_congested (app_wrk, thread_index);
 108               break;
 109             }
 110         }
 111
 112       switch (evt->event_type)
 113         {
 114         case SESSION_IO_EVT_RX:
 115           s = session_get (evt->session_index, thread_index);
 116           s->flags &= ~SESSION_F_RX_EVT;
 117           /* Application didn't confirm accept yet */
 118           if (PREDICT_FALSE (s->session_state == SESSION_STATE_ACCEPTING ||
 119                              s->session_state == SESSION_STATE_CONNECTING))
 120             break;
 121           app->cb_fns.builtin_app_rx_callback (s);
 122           break;
 123         /* Handle sessions that might not be on current thread */
 124         case SESSION_IO_EVT_BUILTIN_RX:
 125           s = session_get_from_handle_if_valid (evt->session_handle);
 126           if (!s)
 127             break;
 128           s->flags &= ~SESSION_F_RX_EVT;
 129           if (PREDICT_FALSE (s->session_state == SESSION_STATE_ACCEPTING ||
 130                              s->session_state == SESSION_STATE_CONNECTING))
 131             break;
 132           app->cb_fns.builtin_app_rx_callback (s);
 133           break;
 134         case SESSION_IO_EVT_TX:
 135           s = session_get (evt->session_index, thread_index);
 136           app->cb_fns.builtin_app_tx_callback (s);
 137           break;
 138         case SESSION_IO_EVT_TX_MAIN:
 139           s = session_get_from_handle_if_valid (evt->session_handle);
 140           if (!s)
 141             break;
 142           app->cb_fns.builtin_app_tx_callback (s);
 143           break;
 144         case SESSION_CTRL_EVT_BOUND:
 145           /* No app cb function currently */
 146           if (is_builtin)
 147             break;
 148           app->cb_fns.session_listened_callback (
 149             app_wrk->wrk_index, evt->as_u64[1] >> 32, evt->session_handle,
 150             evt->as_u64[1] & 0xffffffff);
 151           break;
 152         case SESSION_CTRL_EVT_ACCEPTED:
 153           s = session_get (evt->session_index, thread_index);
 154           old_state = s->session_state;
 155           if (app->cb_fns.session_accept_callback (s))
 156             {
 157               session_detach_app (s);
 158               break;
 159             }
 160           if (is_builtin)
 161             {
 162               if (old_state >= SESSION_STATE_TRANSPORT_CLOSING)
 163                 {
 164                   session_set_state (s,
 165                                      clib_max (old_state, s->session_state));
 166                   if (!(s->flags & SESSION_F_APP_CLOSED))
 167                     app->cb_fns.session_disconnect_callback (s);
 168                 }
 169             }
 170           break;
 171         case SESSION_CTRL_EVT_CONNECTED:
 172           if (!(evt->as_u64[1] & 0xffffffff))
 173             {
 174               s = session_get (evt->session_index, thread_index);
 175               old_state = s->session_state;
 176             }
 177           else
 178             s = 0;
 179           rv = app->cb_fns.session_connected_callback (
 180             app_wrk->wrk_index, evt->as_u64[1] >> 32, s,
 181             evt->as_u64[1] & 0xffffffff);
 182           if (!s)
 183             break;
 184           if (rv)
 185             {
 186               session_detach_app (s);
 187               break;
 188             }
 189           if (old_state >= SESSION_STATE_TRANSPORT_CLOSING)
 190             {
 191               session_set_state (s, clib_max (old_state, s->session_state));
 192               if (!(s->flags & SESSION_F_APP_CLOSED))
 193                 app->cb_fns.session_disconnect_callback (s);
 194             }
 195           break;
 196         case SESSION_CTRL_EVT_DISCONNECTED:
 197           s = session_get (evt->session_index, thread_index);
 198           if (!(s->flags & SESSION_F_APP_CLOSED))
 199             app->cb_fns.session_disconnect_callback (s);
 200           break;
 201         case SESSION_CTRL_EVT_RESET:
 202           s = session_get (evt->session_index, thread_index);
 203           if (!(s->flags & SESSION_F_APP_CLOSED))
 204             app->cb_fns.session_reset_callback (s);
 205           break;
 206         case SESSION_CTRL_EVT_UNLISTEN_REPLY:
 207           if (is_builtin)
 208             break;
 209           app->cb_fns.session_unlistened_callback (
 210             app_wrk->wrk_index, evt->session_handle, evt->as_u64[1] >> 32,
 211             evt->as_u64[1] & 0xffffffff);
 212           break;
 213         case SESSION_CTRL_EVT_MIGRATED:
 214           s = session_get (evt->session_index, thread_index);
 215           app->cb_fns.session_migrate_callback (s, evt->as_u64[1]);
 216           transport_cleanup (session_get_transport_proto (s),
 217                              s->connection_index, s->thread_index);
 218           session_free (s);
 219           /* Notify app that it has data on the new session */
 220           s = session_get_from_handle (evt->as_u64[1]);
 221           session_send_io_evt_to_thread (s->rx_fifo,
 222                                          SESSION_IO_EVT_BUILTIN_RX);
 223           break;
 224         case SESSION_CTRL_EVT_TRANSPORT_CLOSED:
 225           s = session_get (evt->session_index, thread_index);
 226           /* Notification enqueued before session was refused by app */
 227           if (PREDICT_FALSE (s->app_wrk_index == APP_INVALID_INDEX))
 228             break;
 229           if (app->cb_fns.session_transport_closed_callback)
 230             app->cb_fns.session_transport_closed_callback (s);
 231           break;
 232         case SESSION_CTRL_EVT_CLEANUP:
 233           s = session_get (evt->as_u64[0] & 0xffffffff, thread_index);
 234           /* Notification enqueued before session was refused by app */
 235           if (PREDICT_TRUE (s->app_wrk_index != APP_INVALID_INDEX))
 236             {
 237               if (app->cb_fns.session_cleanup_callback)
 238                 app->cb_fns.session_cleanup_callback (s, evt->as_u64[0] >> 32);
 239             }
 240           if (evt->as_u64[0] >> 32 != SESSION_CLEANUP_SESSION)
 241             break;
 242           uword_to_pointer (evt->as_u64[1], void (*) (session_t * s)) (s);
 243           break;
 244         case SESSION_CTRL_EVT_HALF_CLEANUP:
 245           s = ho_session_get (evt->session_index);
 246           ASSERT (session_vlib_thread_is_cl_thread ());
 247           if (app->cb_fns.half_open_cleanup_callback)
 248             app->cb_fns.half_open_cleanup_callback (s);
 249           pool_put_index (app_wrk->half_open_table, s->ho_index);
 250           session_free (s);
 251           break;
 252         case SESSION_CTRL_EVT_APP_ADD_SEGMENT:
 253           app->cb_fns.add_segment_callback (app_wrk->wrk_index,
 254                                             evt->as_u64[1]);
 255           break;
 256         case SESSION_CTRL_EVT_APP_DEL_SEGMENT:
 257           app->cb_fns.del_segment_callback (app_wrk->wrk_index,
 258                                             evt->as_u64[1]);
 259           break;
 260         default:
 261           clib_warning ("unexpected event: %u", evt->event_type);
 262           ASSERT (0);
 263           break;
 264         }
 265       clib_fifo_advance_head (app_wrk->wrk_evts[thread_index], 1);
 266     }
 267
 268   if (!is_builtin)
 269     {
 270       svm_msg_q_unlock (mq);
 271       if (mq_is_cong && i == n_evts)
 272         app_worker_unset_wrk_mq_congested (app_wrk, thread_index);
 273     }
 274
 275   return 0;
 276 }
 277
 278 int
 279 app_wrk_flush_wrk_events (app_worker_t *app_wrk, u32 thread_index)
 280 {
 281   if (app_worker_application_is_builtin (app_wrk))
 282     return app_worker_flush_events_inline (app_wrk, thread_index,
 283                                            1 /* is_builtin */);
 284   else
 285     return app_worker_flush_events_inline (app_wrk, thread_index,
 286                                            0 /* is_builtin */);
 287 }
 288
 289 static inline int
 290 session_wrk_flush_events (session_worker_t *wrk)
 291 {
 292   app_worker_t *app_wrk;
 293   uword app_wrk_index;
 294   u32 thread_index;
 295
 296   thread_index = wrk->vm->thread_index;
 297   app_wrk_index = clib_bitmap_first_set (wrk->app_wrks_pending_ntf);
 298
 299   while (app_wrk_index != ~0)
 300     {
 301       app_wrk = app_worker_get_if_valid (app_wrk_index);
 302       /* app_wrk events are flushed on free, so should be valid here */
 303       ASSERT (app_wrk != 0);
 304       app_wrk_flush_wrk_events (app_wrk, thread_index);
 305
 306       if (!clib_fifo_elts (app_wrk->wrk_evts[thread_index]))
 307         clib_bitmap_set (wrk->app_wrks_pending_ntf, app_wrk->wrk_index, 0);
 308
 309       app_wrk_index =
 310         clib_bitmap_next_set (wrk->app_wrks_pending_ntf, app_wrk_index + 1);
 311     }
 312
 313   if (!clib_bitmap_is_zero (wrk->app_wrks_pending_ntf))
 314     vlib_node_set_interrupt_pending (wrk->vm, session_input_node.index);
 315
 316   return 0;
 317 }
 318
 319 VLIB_NODE_FN (session_input_node)
 320 (vlib_main_t *vm, vlib_node_runtime_t *node, vlib_frame_t *frame)
 321 {
 322   u32 thread_index = vm->thread_index;
 323   session_worker_t *wrk;
 324
 325   wrk = session_main_get_worker (thread_index);
 326   session_wrk_flush_events (wrk);
 327
 328   return 0;
 329 }
 330
 331 VLIB_REGISTER_NODE (session_input_node) = {
 332   .name = "session-input",
 333   .type = VLIB_NODE_TYPE_INPUT,
 334   .state = VLIB_NODE_STATE_DISABLED,
 335 };
 336
 337 /*
 338  * fd.io coding-style-patch-verification: ON
 339  *
 340  * Local Variables:
 341  * eval: (c-set-style "gnu")
 342  * End:
 343  */