session: async rx event notifications

Move from synchronous flushing of io and ctrl events from transports to
applications to an async model via a new session_input input node that
runs in interrupt mode. Events are coalesced per application worker.

On the one hand, this helps by minimizing message queue locking churn.
And on the other, it opens the possibility for further optimizations of
event message generation, obviates need for rx rescheduling rpcs and is
a first step towards a fully async data/io rx path.

Type: improvement

Signed-off-by: Florin Coras <fcoras@cisco.com>
Change-Id: Id6bebcb65fc9feef8aa02ddf1af6d9ba6f6745ce

1 files changed, 36 insertions, 11 deletions

diff --git a/src/vnet/session/application.h b/src/vnet/session/application.h
index c3d6180022f..5505d91ea09 100644
--- a/src/vnet/session/application.h
+++ b/src/vnet/session/application.h
@@ -77,17 +77,17 @@ typedef struct app_worker_
   /** Pool of half-open session handles. Tracked in case worker detaches */
   session_handle_t *half_open_table;
 
+  /* Per vpp worker fifos of events for app worker */
+  session_event_t **wrk_evts;
+
+  /* Vector of vpp workers mq congestion flags */
+  u8 *wrk_mq_congested;
+
   /** Protects detached seg managers */
   clib_spinlock_t detached_seg_managers_lock;
 
   /** Vector of detached listener segment managers */
   u32 *detached_seg_managers;
-
-  /** Fifo of messages postponed because of mq congestion */
-  app_wrk_postponed_msg_t *postponed_mq_msgs;
-
-  /** Lock to add/sub message from ref @postponed_mq_msgs */
-  clib_spinlock_t postponed_mq_msgs_lock;
 } app_worker_t;
 
 typedef struct app_worker_map_
@@ -317,6 +317,12 @@ void application_enable_rx_mqs_nodes (u8 is_en);
  * App worker
  */
 
+always_inline u8
+app_worker_mq_is_congested (app_worker_t *app_wrk)
+{
+  return app_wrk->mq_congested > 0;
+}
+
 app_worker_t *app_worker_alloc (application_t * app);
 int application_alloc_worker_and_init (application_t * app,
 				       app_worker_t ** wrk);
@@ -331,6 +337,10 @@ session_error_t app_worker_start_listen (app_worker_t *app_wrk,
 					 app_listener_t *lstnr);
 int app_worker_stop_listen (app_worker_t * app_wrk, app_listener_t * al);
 int app_worker_init_accepted (session_t * s);
+int app_worker_listened_notify (app_worker_t *app_wrk, session_handle_t alsh,
+				u32 opaque, int err);
+int app_worker_unlisten_reply (app_worker_t *app_wrk, session_handle_t sh,
+			       u32 opaque, session_error_t err);
 int app_worker_accept_notify (app_worker_t * app_wrk, session_t * s);
 int app_worker_init_connected (app_worker_t * app_wrk, session_t * s);
 int app_worker_connect_notify (app_worker_t * app_wrk, session_t * s,
@@ -343,13 +353,21 @@ int app_worker_transport_closed_notify (app_worker_t * app_wrk,
 int app_worker_reset_notify (app_worker_t * app_wrk, session_t * s);
 int app_worker_cleanup_notify (app_worker_t * app_wrk, session_t * s,
 			       session_cleanup_ntf_t ntf);
+int app_worker_cleanup_notify_custom (app_worker_t *app_wrk, session_t *s,
+				      session_cleanup_ntf_t ntf,
+				      void (*cleanup_cb) (session_t *s));
 int app_worker_migrate_notify (app_worker_t * app_wrk, session_t * s,
 			       session_handle_t new_sh);
-int app_worker_builtin_rx (app_worker_t * app_wrk, session_t * s);
-int app_worker_builtin_tx (app_worker_t * app_wrk, session_t * s);
+int app_worker_rx_notify (app_worker_t *app_wrk, session_t *s);
 int app_worker_session_fifo_tuning (app_worker_t * app_wrk, session_t * s,
 				    svm_fifo_t * f,
 				    session_ft_action_t act, u32 len);
+void app_worker_add_event (app_worker_t *app_wrk, session_t *s,
+			   session_evt_type_t evt_type);
+void app_worker_add_event_custom (app_worker_t *app_wrk, u32 thread_index,
+				  session_event_t *evt);
+int app_wrk_flush_wrk_events (app_worker_t *app_wrk, u32 thread_index);
+void app_worker_del_all_events (app_worker_t *app_wrk);
 segment_manager_t *app_worker_get_listen_segment_manager (app_worker_t *,
 							  session_t *);
 segment_manager_t *app_worker_get_connect_segment_manager (app_worker_t *);
@@ -364,9 +382,10 @@ void app_wrk_send_ctrl_evt_fd (app_worker_t *app_wrk, u8 evt_type, void *msg,
 			       u32 msg_len, int fd);
 void app_wrk_send_ctrl_evt (app_worker_t *app_wrk, u8 evt_type, void *msg,
 			    u32 msg_len);
-int app_worker_send_event (app_worker_t * app, session_t * s, u8 evt);
-int app_worker_lock_and_send_event (app_worker_t * app, session_t * s,
-				    u8 evt_type);
+u8 app_worker_mq_wrk_is_congested (app_worker_t *app_wrk, u32 thread_index);
+void app_worker_set_mq_wrk_congested (app_worker_t *app_wrk, u32 thread_index);
+void app_worker_unset_wrk_mq_congested (app_worker_t *app_wrk,
+					u32 thread_index);
 session_t *app_worker_proxy_listener (app_worker_t * app, u8 fib_proto,
 				      u8 transport_proto);
 void app_worker_del_detached_sm (app_worker_t * app_wrk, u32 sm_index);
@@ -395,6 +414,12 @@ void sapi_socket_close_w_handle (u32 api_handle);
 crypto_engine_type_t app_crypto_engine_type_add (void);
 u8 app_crypto_engine_n_types (void);
 
+static inline u8
+app_worker_application_is_builtin (app_worker_t *app_wrk)
+{
+  return app_wrk->app_is_builtin;
+}
+
 #endif /* SRC_VNET_SESSION_APPLICATION_H_ */
 
 /*