1 /* SPDX-License-Identifier: Apache-2.0
2 * Copyright(c) 2023 Cisco Systems, Inc.
5 #include <vnet/session/session.h>
6 #include <vnet/session/application.h>
9 mq_try_lock (svm_msg_q_t *mq)
15 rv = svm_msg_q_try_lock (mq);
26 mq_event_ring_index (session_evt_type_t et)
28 return (et >= SESSION_CTRL_EVT_RPC ? SESSION_MQ_CTRL_EVT_RING :
29 SESSION_MQ_IO_EVT_RING);
33 app_worker_del_all_events (app_worker_t *app_wrk)
35 session_worker_t *wrk;
40 for (thread_index = 0; thread_index < vec_len (app_wrk->wrk_evts);
43 while (clib_fifo_elts (app_wrk->wrk_evts[thread_index]))
45 clib_fifo_sub2 (app_wrk->wrk_evts[thread_index], evt);
46 switch (evt->event_type)
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);
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)
58 uword_to_pointer (evt->as_u64[1], void (*) (session_t * s)) (s);
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);
69 wrk = session_main_get_worker (thread_index);
70 clib_bitmap_set (wrk->app_wrks_pending_ntf, app_wrk->wrk_index, 0);
75 app_worker_flush_events_inline (app_worker_t *app_wrk, u32 thread_index,
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;
87 n_evts = clib_min (n_evts, clib_fifo_elts (app_wrk->wrk_evts[thread_index]));
91 mq_is_cong = app_worker_mq_is_congested (app_wrk);
94 app_worker_set_mq_wrk_congested (app_wrk, thread_index);
99 for (i = 0; i < n_evts; i++)
101 evt = clib_fifo_head (app_wrk->wrk_evts[thread_index]);
104 ring_index = mq_event_ring_index (evt->event_type);
105 if (svm_msg_q_or_ring_is_full (mq, ring_index))
107 app_worker_set_mq_wrk_congested (app_wrk, thread_index);
112 switch (evt->event_type)
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))
121 app->cb_fns.builtin_app_rx_callback (s);
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);
128 s->flags &= ~SESSION_F_RX_EVT;
129 if (PREDICT_FALSE (s->session_state == SESSION_STATE_ACCEPTING ||
130 s->session_state == SESSION_STATE_CONNECTING))
132 app->cb_fns.builtin_app_rx_callback (s);
134 case SESSION_IO_EVT_TX:
135 s = session_get (evt->session_index, thread_index);
136 app->cb_fns.builtin_app_tx_callback (s);
138 case SESSION_IO_EVT_TX_MAIN:
139 s = session_get_from_handle_if_valid (evt->session_handle);
142 app->cb_fns.builtin_app_tx_callback (s);
144 case SESSION_CTRL_EVT_BOUND:
145 /* No app cb function currently */
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);
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))
158 s->app_wrk_index = APP_INVALID_INDEX;
163 if (old_state >= SESSION_STATE_TRANSPORT_CLOSING)
165 session_set_state (s,
166 clib_max (old_state, s->session_state));
167 if (!(s->flags & SESSION_F_APP_CLOSED))
168 app->cb_fns.session_disconnect_callback (s);
172 case SESSION_CTRL_EVT_CONNECTED:
173 if (!(evt->as_u64[1] & 0xffffffff))
175 s = session_get (evt->session_index, thread_index);
176 old_state = s->session_state;
180 rv = app->cb_fns.session_connected_callback (
181 app_wrk->wrk_index, evt->as_u64[1] >> 32, s,
182 evt->as_u64[1] & 0xffffffff);
188 s->app_wrk_index = APP_INVALID_INDEX;
191 if (old_state >= SESSION_STATE_TRANSPORT_CLOSING)
193 session_set_state (s, clib_max (old_state, s->session_state));
194 if (!(s->flags & SESSION_F_APP_CLOSED))
195 app->cb_fns.session_disconnect_callback (s);
198 case SESSION_CTRL_EVT_DISCONNECTED:
199 s = session_get (evt->session_index, thread_index);
200 if (!(s->flags & SESSION_F_APP_CLOSED))
201 app->cb_fns.session_disconnect_callback (s);
203 case SESSION_CTRL_EVT_RESET:
204 s = session_get (evt->session_index, thread_index);
205 if (!(s->flags & SESSION_F_APP_CLOSED))
206 app->cb_fns.session_reset_callback (s);
208 case SESSION_CTRL_EVT_UNLISTEN_REPLY:
211 app->cb_fns.session_unlistened_callback (
212 app_wrk->wrk_index, evt->session_handle, evt->as_u64[1] >> 32,
213 evt->as_u64[1] & 0xffffffff);
215 case SESSION_CTRL_EVT_MIGRATED:
216 s = session_get (evt->session_index, thread_index);
217 app->cb_fns.session_migrate_callback (s, evt->as_u64[1]);
218 transport_cleanup (session_get_transport_proto (s),
219 s->connection_index, s->thread_index);
221 /* Notify app that it has data on the new session */
222 s = session_get_from_handle (evt->as_u64[1]);
223 session_send_io_evt_to_thread (s->rx_fifo,
224 SESSION_IO_EVT_BUILTIN_RX);
226 case SESSION_CTRL_EVT_TRANSPORT_CLOSED:
227 s = session_get (evt->session_index, thread_index);
228 /* Notification enqueued before session was refused by app */
229 if (PREDICT_FALSE (s->app_wrk_index == APP_INVALID_INDEX))
231 if (app->cb_fns.session_transport_closed_callback)
232 app->cb_fns.session_transport_closed_callback (s);
234 case SESSION_CTRL_EVT_CLEANUP:
235 s = session_get (evt->as_u64[0] & 0xffffffff, thread_index);
236 /* Notification enqueued before session was refused by app */
237 if (PREDICT_TRUE (s->app_wrk_index != APP_INVALID_INDEX))
239 if (app->cb_fns.session_cleanup_callback)
240 app->cb_fns.session_cleanup_callback (s, evt->as_u64[0] >> 32);
242 if (evt->as_u64[0] >> 32 != SESSION_CLEANUP_SESSION)
244 uword_to_pointer (evt->as_u64[1], void (*) (session_t * s)) (s);
246 case SESSION_CTRL_EVT_HALF_CLEANUP:
247 s = ho_session_get (evt->session_index);
248 ASSERT (session_vlib_thread_is_cl_thread ());
249 if (app->cb_fns.half_open_cleanup_callback)
250 app->cb_fns.half_open_cleanup_callback (s);
251 pool_put_index (app_wrk->half_open_table, s->ho_index);
254 case SESSION_CTRL_EVT_APP_ADD_SEGMENT:
255 app->cb_fns.add_segment_callback (app_wrk->wrk_index,
258 case SESSION_CTRL_EVT_APP_DEL_SEGMENT:
259 app->cb_fns.del_segment_callback (app_wrk->wrk_index,
263 clib_warning ("unexpected event: %u", evt->event_type);
267 clib_fifo_advance_head (app_wrk->wrk_evts[thread_index], 1);
272 svm_msg_q_unlock (mq);
273 if (mq_is_cong && i == n_evts)
274 app_worker_unset_wrk_mq_congested (app_wrk, thread_index);
281 app_wrk_flush_wrk_events (app_worker_t *app_wrk, u32 thread_index)
283 if (app_worker_application_is_builtin (app_wrk))
284 return app_worker_flush_events_inline (app_wrk, thread_index,
287 return app_worker_flush_events_inline (app_wrk, thread_index,
292 session_wrk_flush_events (session_worker_t *wrk)
294 app_worker_t *app_wrk;
298 thread_index = wrk->vm->thread_index;
299 app_wrk_index = clib_bitmap_first_set (wrk->app_wrks_pending_ntf);
301 while (app_wrk_index != ~0)
303 app_wrk = app_worker_get_if_valid (app_wrk_index);
304 /* app_wrk events are flushed on free, so should be valid here */
305 ASSERT (app_wrk != 0);
306 app_wrk_flush_wrk_events (app_wrk, thread_index);
308 if (!clib_fifo_elts (app_wrk->wrk_evts[thread_index]))
309 clib_bitmap_set (wrk->app_wrks_pending_ntf, app_wrk->wrk_index, 0);
312 clib_bitmap_next_set (wrk->app_wrks_pending_ntf, app_wrk_index + 1);
315 if (!clib_bitmap_is_zero (wrk->app_wrks_pending_ntf))
316 vlib_node_set_interrupt_pending (wrk->vm, session_input_node.index);
321 VLIB_NODE_FN (session_input_node)
322 (vlib_main_t *vm, vlib_node_runtime_t *node, vlib_frame_t *frame)
324 u32 thread_index = vm->thread_index;
325 session_worker_t *wrk;
327 wrk = session_main_get_worker (thread_index);
328 session_wrk_flush_events (wrk);
333 VLIB_REGISTER_NODE (session_input_node) = {
334 .name = "session-input",
335 .type = VLIB_NODE_TYPE_INPUT,
336 .state = VLIB_NODE_STATE_DISABLED,
340 * fd.io coding-style-patch-verification: ON
343 * eval: (c-set-style "gnu")