src/vnet/session/session_input.c - fdio/vpp - Gitiles

 /* SPDX-License-Identifier: Apache-2.0
  * Copyright(c) 2023 Cisco Systems, Inc.
  */

 #include <vnet/session/session.h>
 #include <vnet/session/application.h>

 static inline int
 mq_try_lock (svm_msg_q_t *mq)
 {
   int rv, n_try = 0;

   while (n_try < 100)
     {
       rv = svm_msg_q_try_lock (mq);
       if (!rv)
 	return 0;
       n_try += 1;
       usleep (1);
     }

   return -1;
 }

 always_inline u8
 mq_event_ring_index (session_evt_type_t et)
 {
   return (et >= SESSION_CTRL_EVT_RPC ? SESSION_MQ_CTRL_EVT_RING :
 					     SESSION_MQ_IO_EVT_RING);
 }

 void
 app_worker_del_all_events (app_worker_t *app_wrk)
 {
   session_worker_t *wrk;
   session_event_t *evt;
   u32 thread_index;
   session_t *s;

   for (thread_index = 0; thread_index < vec_len (app_wrk->wrk_evts);
        thread_index++)
     {
       while (clib_fifo_elts (app_wrk->wrk_evts[thread_index]))
 	{
 	  clib_fifo_sub2 (app_wrk->wrk_evts[thread_index], evt);
 	  switch (evt->event_type)
 	    {
 	    case SESSION_CTRL_EVT_MIGRATED:
 	      s = session_get (evt->session_index, thread_index);
 	      transport_cleanup (session_get_transport_proto (s),
 				 s->connection_index, s->thread_index);
 	      session_free (s);
 	      break;
 	    case SESSION_CTRL_EVT_CLEANUP:
 	      s = session_get (evt->as_u64[0] & 0xffffffff, thread_index);
 	      if (evt->as_u64[0] >> 32 != SESSION_CLEANUP_SESSION)
 		break;
 	      uword_to_pointer (evt->as_u64[1], void (*) (session_t * s)) (s);
 	      break;
 	    case SESSION_CTRL_EVT_HALF_CLEANUP:
 	      s = ho_session_get (evt->session_index);
 	      pool_put_index (app_wrk->half_open_table, s->ho_index);
 	      session_free (s);
 	      break;
 	    default:
 	      break;
 	    }
 	}
       wrk = session_main_get_worker (thread_index);
       clib_bitmap_set (wrk->app_wrks_pending_ntf, app_wrk->wrk_index, 0);
     }
 }

 always_inline int
 app_worker_flush_events_inline (app_worker_t *app_wrk, u32 thread_index,
 				u8 is_builtin)
 {
   application_t *app = application_get (app_wrk->app_index);
   svm_msg_q_t *mq = app_wrk->event_queue;
   u8 ring_index, mq_is_cong;
   session_state_t old_state;
   session_event_t *evt;
   u32 n_evts = 128, i;
   session_t *s;
   int rv;

   n_evts = clib_min (n_evts, clib_fifo_elts (app_wrk->wrk_evts[thread_index]));

   if (!is_builtin)
     {
       mq_is_cong = app_worker_mq_is_congested (app_wrk);
       if (mq_try_lock (mq))
 	{
 	  app_worker_set_mq_wrk_congested (app_wrk, thread_index);
 	  return 0;
 	}
     }

   for (i = 0; i < n_evts; i++)
     {
       evt = clib_fifo_head (app_wrk->wrk_evts[thread_index]);
       if (!is_builtin)
 	{
 	  ring_index = mq_event_ring_index (evt->event_type);
 	  if (svm_msg_q_or_ring_is_full (mq, ring_index))
 	    {
 	      app_worker_set_mq_wrk_congested (app_wrk, thread_index);
 	      break;
 	    }
 	}

       switch (evt->event_type)
 	{
 	case SESSION_IO_EVT_RX:
 	  s = session_get (evt->session_index, thread_index);
 	  /* Application didn't confirm accept yet */
 	  if (PREDICT_FALSE (s->session_state == SESSION_STATE_ACCEPTING ||
 			     s->session_state == SESSION_STATE_CONNECTING))
 	    break;
 	  s->flags &= ~SESSION_F_RX_EVT;
 	  app->cb_fns.builtin_app_rx_callback (s);
 	  break;
 	/* Handle sessions that might not be on current thread */
 	case SESSION_IO_EVT_BUILTIN_RX:
 	  s = session_get_from_handle_if_valid (evt->session_handle);
 	  if (!s || s->session_state == SESSION_STATE_ACCEPTING ||
 	      s->session_state == SESSION_STATE_CONNECTING)
 	    break;
 	  s->flags &= ~SESSION_F_RX_EVT;
 	  app->cb_fns.builtin_app_rx_callback (s);
 	  break;
 	case SESSION_IO_EVT_TX:
 	  s = session_get (evt->session_index, thread_index);
 	  app->cb_fns.builtin_app_tx_callback (s);
 	  break;
 	case SESSION_IO_EVT_TX_MAIN:
 	  s = session_get_from_handle_if_valid (evt->session_handle);
 	  if (!s)
 	    break;
 	  app->cb_fns.builtin_app_tx_callback (s);
 	  break;
 	case SESSION_CTRL_EVT_BOUND:
 	  /* No app cb function currently */
 	  if (is_builtin)
 	    break;
 	  mq_send_session_bound_cb (app_wrk->wrk_index, evt->as_u64[1] >> 32,
 				    evt->session_handle,
 				    evt->as_u64[1] & 0xffffffff);
 	  break;
 	case SESSION_CTRL_EVT_ACCEPTED:
 	  s = session_get (evt->session_index, thread_index);
 	  old_state = s->session_state;
 	  if (app->cb_fns.session_accept_callback (s))
 	    {
 	      session_close (s);
 	      s->app_wrk_index = SESSION_INVALID_INDEX;
 	      break;
 	    }
 	  if (is_builtin)
 	    {
 	      if (s->flags & SESSION_F_RX_EVT)
 		{
 		  s->flags &= ~SESSION_F_RX_EVT;
 		  app->cb_fns.builtin_app_rx_callback (s);
 		}
 	      if (old_state >= SESSION_STATE_TRANSPORT_CLOSING)
 		{
 		  session_set_state (s, old_state);
 		  app_worker_close_notify (app_wrk, s);
 		}
 	    }
 	  break;
 	case SESSION_CTRL_EVT_CONNECTED:
 	  if (!(evt->as_u64[1] & 0xffffffff))
 	    {
 	      s = session_get (evt->session_index, thread_index);
 	      old_state = s->session_state;
 	    }
 	  else
 	    s = 0;
 	  rv = app->cb_fns.session_connected_callback (
 	    app_wrk->wrk_index, evt->as_u64[1] >> 32, s,
 	    evt->as_u64[1] & 0xffffffff);
 	  if (!s)
 	    break;
 	  if (rv)
 	    {
 	      session_close (s);
 	      s->app_wrk_index = SESSION_INVALID_INDEX;
 	      break;
 	    }
 	  if (s->flags & SESSION_F_RX_EVT)
 	    {
 	      s->flags &= ~SESSION_F_RX_EVT;
 	      app->cb_fns.builtin_app_rx_callback (s);
 	    }
 	  if (old_state >= SESSION_STATE_TRANSPORT_CLOSING)
 	    {
 	      session_set_state (s, old_state);
 	      app_worker_close_notify (app_wrk, s);
 	    }
 	  break;
 	case SESSION_CTRL_EVT_DISCONNECTED:
 	  s = session_get (evt->session_index, thread_index);
 	  app->cb_fns.session_disconnect_callback (s);
 	  break;
 	case SESSION_CTRL_EVT_RESET:
 	  s = session_get (evt->session_index, thread_index);
 	  app->cb_fns.session_reset_callback (s);
 	  break;
 	case SESSION_CTRL_EVT_UNLISTEN_REPLY:
 	  if (is_builtin)
 	    break;
 	  mq_send_unlisten_reply (app_wrk, evt->session_handle,
 				  evt->as_u64[1] >> 32,
 				  evt->as_u64[1] & 0xffffffff);
 	  break;
 	case SESSION_CTRL_EVT_MIGRATED:
 	  s = session_get (evt->session_index, thread_index);
 	  app->cb_fns.session_migrate_callback (s, evt->as_u64[1]);
 	  transport_cleanup (session_get_transport_proto (s),
 			     s->connection_index, s->thread_index);
 	  session_free (s);
 	  /* Notify app that it has data on the new session */
 	  s = session_get_from_handle (evt->as_u64[1]);
 	  session_send_io_evt_to_thread (s->rx_fifo,
 					 SESSION_IO_EVT_BUILTIN_RX);
 	  break;
 	case SESSION_CTRL_EVT_TRANSPORT_CLOSED:
 	  s = session_get (evt->session_index, thread_index);
 	  if (app->cb_fns.session_transport_closed_callback)
 	    app->cb_fns.session_transport_closed_callback (s);
 	  break;
 	case SESSION_CTRL_EVT_CLEANUP:
 	  s = session_get (evt->as_u64[0] & 0xffffffff, thread_index);
 	  if (app->cb_fns.session_cleanup_callback)
 	    app->cb_fns.session_cleanup_callback (s, evt->as_u64[0] >> 32);
 	  if (evt->as_u64[0] >> 32 != SESSION_CLEANUP_SESSION)
 	    break;
 	  uword_to_pointer (evt->as_u64[1], void (*) (session_t * s)) (s);
 	  break;
 	case SESSION_CTRL_EVT_HALF_CLEANUP:
 	  s = ho_session_get (evt->session_index);
 	  ASSERT (session_vlib_thread_is_cl_thread ());
 	  if (app->cb_fns.half_open_cleanup_callback)
 	    app->cb_fns.half_open_cleanup_callback (s);
 	  pool_put_index (app_wrk->half_open_table, s->ho_index);
 	  session_free (s);
 	  break;
 	case SESSION_CTRL_EVT_APP_ADD_SEGMENT:
 	  app->cb_fns.add_segment_callback (app_wrk->wrk_index,
 					    evt->as_u64[1]);
 	  break;
 	case SESSION_CTRL_EVT_APP_DEL_SEGMENT:
 	  app->cb_fns.del_segment_callback (app_wrk->wrk_index,
 					    evt->as_u64[1]);
 	  break;
 	default:
 	  clib_warning ("unexpected event: %u", evt->event_type);
 	  ASSERT (0);
 	  break;
 	}
       clib_fifo_advance_head (app_wrk->wrk_evts[thread_index], 1);
     }

   if (!is_builtin)
     {
       svm_msg_q_unlock (mq);
       if (mq_is_cong && i == n_evts)
 	app_worker_unset_wrk_mq_congested (app_wrk, thread_index);
     }

   return 0;
 }

 int
 app_wrk_flush_wrk_events (app_worker_t *app_wrk, u32 thread_index)
 {
   if (app_worker_application_is_builtin (app_wrk))
     return app_worker_flush_events_inline (app_wrk, thread_index,
 					   1 /* is_builtin */);
   else
     return app_worker_flush_events_inline (app_wrk, thread_index,
 					   0 /* is_builtin */);
 }

 static inline int
 session_wrk_flush_events (session_worker_t *wrk)
 {
   app_worker_t *app_wrk;
   uword app_wrk_index;
   u32 thread_index;

   thread_index = wrk->vm->thread_index;
   app_wrk_index = clib_bitmap_first_set (wrk->app_wrks_pending_ntf);

   while (app_wrk_index != ~0)
     {
       app_wrk = app_worker_get_if_valid (app_wrk_index);
       /* app_wrk events are flushed on free, so should be valid here */
       ASSERT (app_wrk != 0);
       app_wrk_flush_wrk_events (app_wrk, thread_index);

       if (!clib_fifo_elts (app_wrk->wrk_evts[thread_index]))
 	clib_bitmap_set (wrk->app_wrks_pending_ntf, app_wrk->wrk_index, 0);

       app_wrk_index =
 	clib_bitmap_next_set (wrk->app_wrks_pending_ntf, app_wrk_index + 1);
     }

   if (!clib_bitmap_is_zero (wrk->app_wrks_pending_ntf))
     vlib_node_set_interrupt_pending (wrk->vm, session_input_node.index);

   return 0;
 }

 VLIB_NODE_FN (session_input_node)
 (vlib_main_t *vm, vlib_node_runtime_t *node, vlib_frame_t *frame)
 {
   u32 thread_index = vm->thread_index;
   session_worker_t *wrk;

   wrk = session_main_get_worker (thread_index);
   session_wrk_flush_events (wrk);

   return 0;
 }

 VLIB_REGISTER_NODE (session_input_node) = {
   .name = "session-input",
   .type = VLIB_NODE_TYPE_INPUT,
   .state = VLIB_NODE_STATE_DISABLED,
 };

 /*
  * fd.io coding-style-patch-verification: ON
  *
  * Local Variables:
  * eval: (c-set-style "gnu")
  * End:
  */
	/* SPDX-License-Identifier: Apache-2.0
	* Copyright(c) 2023 Cisco Systems, Inc.
	*/

	#include <vnet/session/session.h>
	#include <vnet/session/application.h>

	static inline int
	mq_try_lock (svm_msg_q_t *mq)
	{
	int rv, n_try = 0;

	while (n_try < 100)
	{
	rv = svm_msg_q_try_lock (mq);
	if (!rv)
	return 0;
	n_try += 1;
	usleep (1);
	}

	return -1;
	}

	always_inline u8
	mq_event_ring_index (session_evt_type_t et)
	{
	return (et >= SESSION_CTRL_EVT_RPC ? SESSION_MQ_CTRL_EVT_RING :
	SESSION_MQ_IO_EVT_RING);
	}

	void
	app_worker_del_all_events (app_worker_t *app_wrk)
	{
	session_worker_t *wrk;
	session_event_t *evt;
	u32 thread_index;
	session_t *s;

	for (thread_index = 0; thread_index < vec_len (app_wrk->wrk_evts);
	thread_index++)
	{
	while (clib_fifo_elts (app_wrk->wrk_evts[thread_index]))
	{
	clib_fifo_sub2 (app_wrk->wrk_evts[thread_index], evt);
	switch (evt->event_type)
	{
	case SESSION_CTRL_EVT_MIGRATED:
	s = session_get (evt->session_index, thread_index);
	transport_cleanup (session_get_transport_proto (s),
	s->connection_index, s->thread_index);
	session_free (s);
	break;
	case SESSION_CTRL_EVT_CLEANUP:
	s = session_get (evt->as_u64[0] & 0xffffffff, thread_index);
	if (evt->as_u64[0] >> 32 != SESSION_CLEANUP_SESSION)
	break;
	uword_to_pointer (evt->as_u64[1], void () (session_t s)) (s);
	break;
	case SESSION_CTRL_EVT_HALF_CLEANUP:
	s = ho_session_get (evt->session_index);
	pool_put_index (app_wrk->half_open_table, s->ho_index);
	session_free (s);
	break;
	default:
	break;
	}
	}
	wrk = session_main_get_worker (thread_index);
	clib_bitmap_set (wrk->app_wrks_pending_ntf, app_wrk->wrk_index, 0);
	}
	}

	always_inline int
	app_worker_flush_events_inline (app_worker_t *app_wrk, u32 thread_index,
	u8 is_builtin)
	{
	application_t *app = application_get (app_wrk->app_index);
	svm_msg_q_t *mq = app_wrk->event_queue;
	u8 ring_index, mq_is_cong;
	session_state_t old_state;
	session_event_t *evt;
	u32 n_evts = 128, i;
	session_t *s;
	int rv;

	n_evts = clib_min (n_evts, clib_fifo_elts (app_wrk->wrk_evts[thread_index]));

	if (!is_builtin)
	{
	mq_is_cong = app_worker_mq_is_congested (app_wrk);
	if (mq_try_lock (mq))
	{
	app_worker_set_mq_wrk_congested (app_wrk, thread_index);
	return 0;
	}
	}

	for (i = 0; i < n_evts; i++)
	{
	evt = clib_fifo_head (app_wrk->wrk_evts[thread_index]);
	if (!is_builtin)
	{
	ring_index = mq_event_ring_index (evt->event_type);
	if (svm_msg_q_or_ring_is_full (mq, ring_index))
	{
	app_worker_set_mq_wrk_congested (app_wrk, thread_index);
	break;
	}
	}

	switch (evt->event_type)
	{
	case SESSION_IO_EVT_RX:
	s = session_get (evt->session_index, thread_index);
	/* Application didn't confirm accept yet */
	if (PREDICT_FALSE (s->session_state == SESSION_STATE_ACCEPTING \|\|
	s->session_state == SESSION_STATE_CONNECTING))
	break;
	s->flags &= ~SESSION_F_RX_EVT;
	app->cb_fns.builtin_app_rx_callback (s);
	break;
	/* Handle sessions that might not be on current thread */
	case SESSION_IO_EVT_BUILTIN_RX:
	s = session_get_from_handle_if_valid (evt->session_handle);
	if (!s \|\| s->session_state == SESSION_STATE_ACCEPTING \|\|
	s->session_state == SESSION_STATE_CONNECTING)
	break;
	s->flags &= ~SESSION_F_RX_EVT;
	app->cb_fns.builtin_app_rx_callback (s);
	break;
	case SESSION_IO_EVT_TX:
	s = session_get (evt->session_index, thread_index);
	app->cb_fns.builtin_app_tx_callback (s);
	break;
	case SESSION_IO_EVT_TX_MAIN:
	s = session_get_from_handle_if_valid (evt->session_handle);
	if (!s)
	break;
	app->cb_fns.builtin_app_tx_callback (s);
	break;
	case SESSION_CTRL_EVT_BOUND:
	/* No app cb function currently */
	if (is_builtin)
	break;
	mq_send_session_bound_cb (app_wrk->wrk_index, evt->as_u64[1] >> 32,
	evt->session_handle,
	evt->as_u64[1] & 0xffffffff);
	break;
	case SESSION_CTRL_EVT_ACCEPTED:
	s = session_get (evt->session_index, thread_index);
	old_state = s->session_state;
	if (app->cb_fns.session_accept_callback (s))
	{
	session_close (s);
	s->app_wrk_index = SESSION_INVALID_INDEX;
	break;
	}
	if (is_builtin)
	{
	if (s->flags & SESSION_F_RX_EVT)
	{
	s->flags &= ~SESSION_F_RX_EVT;
	app->cb_fns.builtin_app_rx_callback (s);
	}
	if (old_state >= SESSION_STATE_TRANSPORT_CLOSING)
	{
	session_set_state (s, old_state);
	app_worker_close_notify (app_wrk, s);
	}
	}
	break;
	case SESSION_CTRL_EVT_CONNECTED:
	if (!(evt->as_u64[1] & 0xffffffff))
	{
	s = session_get (evt->session_index, thread_index);
	old_state = s->session_state;
	}
	else
	s = 0;
	rv = app->cb_fns.session_connected_callback (
	app_wrk->wrk_index, evt->as_u64[1] >> 32, s,
	evt->as_u64[1] & 0xffffffff);
	if (!s)
	break;
	if (rv)
	{
	session_close (s);
	s->app_wrk_index = SESSION_INVALID_INDEX;
	break;
	}
	if (s->flags & SESSION_F_RX_EVT)
	{
	s->flags &= ~SESSION_F_RX_EVT;
	app->cb_fns.builtin_app_rx_callback (s);
	}
	if (old_state >= SESSION_STATE_TRANSPORT_CLOSING)
	{
	session_set_state (s, old_state);
	app_worker_close_notify (app_wrk, s);
	}
	break;
	case SESSION_CTRL_EVT_DISCONNECTED:
	s = session_get (evt->session_index, thread_index);
	app->cb_fns.session_disconnect_callback (s);
	break;
	case SESSION_CTRL_EVT_RESET:
	s = session_get (evt->session_index, thread_index);
	app->cb_fns.session_reset_callback (s);
	break;
	case SESSION_CTRL_EVT_UNLISTEN_REPLY:
	if (is_builtin)
	break;
	mq_send_unlisten_reply (app_wrk, evt->session_handle,
	evt->as_u64[1] >> 32,
	evt->as_u64[1] & 0xffffffff);
	break;
	case SESSION_CTRL_EVT_MIGRATED:
	s = session_get (evt->session_index, thread_index);
	app->cb_fns.session_migrate_callback (s, evt->as_u64[1]);
	transport_cleanup (session_get_transport_proto (s),
	s->connection_index, s->thread_index);
	session_free (s);
	/* Notify app that it has data on the new session */
	s = session_get_from_handle (evt->as_u64[1]);
	session_send_io_evt_to_thread (s->rx_fifo,
	SESSION_IO_EVT_BUILTIN_RX);
	break;
	case SESSION_CTRL_EVT_TRANSPORT_CLOSED:
	s = session_get (evt->session_index, thread_index);
	if (app->cb_fns.session_transport_closed_callback)
	app->cb_fns.session_transport_closed_callback (s);
	break;
	case SESSION_CTRL_EVT_CLEANUP:
	s = session_get (evt->as_u64[0] & 0xffffffff, thread_index);
	if (app->cb_fns.session_cleanup_callback)
	app->cb_fns.session_cleanup_callback (s, evt->as_u64[0] >> 32);
	if (evt->as_u64[0] >> 32 != SESSION_CLEANUP_SESSION)
	break;
	uword_to_pointer (evt->as_u64[1], void () (session_t s)) (s);
	break;
	case SESSION_CTRL_EVT_HALF_CLEANUP:
	s = ho_session_get (evt->session_index);
	ASSERT (session_vlib_thread_is_cl_thread ());
	if (app->cb_fns.half_open_cleanup_callback)
	app->cb_fns.half_open_cleanup_callback (s);
	pool_put_index (app_wrk->half_open_table, s->ho_index);
	session_free (s);
	break;
	case SESSION_CTRL_EVT_APP_ADD_SEGMENT:
	app->cb_fns.add_segment_callback (app_wrk->wrk_index,
	evt->as_u64[1]);
	break;
	case SESSION_CTRL_EVT_APP_DEL_SEGMENT:
	app->cb_fns.del_segment_callback (app_wrk->wrk_index,
	evt->as_u64[1]);
	break;
	default:
	clib_warning ("unexpected event: %u", evt->event_type);
	ASSERT (0);
	break;
	}
	clib_fifo_advance_head (app_wrk->wrk_evts[thread_index], 1);
	}

	if (!is_builtin)
	{
	svm_msg_q_unlock (mq);
	if (mq_is_cong && i == n_evts)
	app_worker_unset_wrk_mq_congested (app_wrk, thread_index);
	}

	return 0;
	}

	int
	app_wrk_flush_wrk_events (app_worker_t *app_wrk, u32 thread_index)
	{
	if (app_worker_application_is_builtin (app_wrk))
	return app_worker_flush_events_inline (app_wrk, thread_index,
	1 /* is_builtin */);
	else
	return app_worker_flush_events_inline (app_wrk, thread_index,
	0 /* is_builtin */);
	}

	static inline int
	session_wrk_flush_events (session_worker_t *wrk)
	{
	app_worker_t *app_wrk;
	uword app_wrk_index;
	u32 thread_index;

	thread_index = wrk->vm->thread_index;
	app_wrk_index = clib_bitmap_first_set (wrk->app_wrks_pending_ntf);

	while (app_wrk_index != ~0)
	{
	app_wrk = app_worker_get_if_valid (app_wrk_index);
	/* app_wrk events are flushed on free, so should be valid here */
	ASSERT (app_wrk != 0);
	app_wrk_flush_wrk_events (app_wrk, thread_index);

	if (!clib_fifo_elts (app_wrk->wrk_evts[thread_index]))
	clib_bitmap_set (wrk->app_wrks_pending_ntf, app_wrk->wrk_index, 0);

	app_wrk_index =
	clib_bitmap_next_set (wrk->app_wrks_pending_ntf, app_wrk_index + 1);
	}

	if (!clib_bitmap_is_zero (wrk->app_wrks_pending_ntf))
	vlib_node_set_interrupt_pending (wrk->vm, session_input_node.index);

	return 0;
	}

	VLIB_NODE_FN (session_input_node)
	(vlib_main_t vm, vlib_node_runtime_t node, vlib_frame_t *frame)
	{
	u32 thread_index = vm->thread_index;
	session_worker_t *wrk;

	wrk = session_main_get_worker (thread_index);
	session_wrk_flush_events (wrk);

	return 0;
	}

	VLIB_REGISTER_NODE (session_input_node) = {
	.name = "session-input",
	.type = VLIB_NODE_TYPE_INPUT,
	.state = VLIB_NODE_STATE_DISABLED,
	};

	/*
	* fd.io coding-style-patch-verification: ON
	*
	* Local Variables:
	* eval: (c-set-style "gnu")
	* End:
	*/