+#include <sys/timerfd.h>
+
+static inline void
+session_wrk_send_evt_to_main (session_worker_t *wrk, session_evt_elt_t *elt)
+{
+ session_evt_elt_t *he;
+ uword thread_index;
+ u8 is_empty;
+
+ thread_index = wrk->vm->thread_index;
+ he = clib_llist_elt (wrk->event_elts, wrk->evts_pending_main);
+ is_empty = clib_llist_is_empty (wrk->event_elts, evt_list, he);
+ clib_llist_add_tail (wrk->event_elts, evt_list, elt, he);
+ if (is_empty)
+ session_send_rpc_evt_to_thread (0, session_wrk_handle_evts_main_rpc,
+ uword_to_pointer (thread_index, void *));
+}
+
+#define app_check_thread_and_barrier(_wrk, _elt) \
+ if (!vlib_thread_is_main_w_barrier ()) \
+ { \
+ session_wrk_send_evt_to_main (wrk, elt); \
+ return; \
+ }
+
+static void
+session_wrk_timerfd_update (session_worker_t *wrk, u64 time_ns)
+{
+ struct itimerspec its;
+
+ its.it_value.tv_sec = 0;
+ its.it_value.tv_nsec = time_ns;
+ its.it_interval.tv_sec = 0;
+ its.it_interval.tv_nsec = its.it_value.tv_nsec;
+
+ if (timerfd_settime (wrk->timerfd, 0, &its, NULL) == -1)
+ clib_warning ("timerfd_settime");
+}
+
+always_inline u64
+session_wrk_tfd_timeout (session_wrk_state_t state, u32 thread_index)
+{
+ if (state == SESSION_WRK_INTERRUPT)
+ return thread_index ? 1e6 : vlib_num_workers () ? 5e8 : 1e6;
+ else if (state == SESSION_WRK_IDLE)
+ return thread_index ? 1e8 : vlib_num_workers () ? 5e8 : 1e8;
+ else
+ return 0;
+}
+
+static inline void
+session_wrk_set_state (session_worker_t *wrk, session_wrk_state_t state)
+{
+ u64 time_ns;
+
+ wrk->state = state;
+ if (wrk->timerfd == -1)
+ return;
+ time_ns = session_wrk_tfd_timeout (state, wrk->vm->thread_index);
+ session_wrk_timerfd_update (wrk, time_ns);
+}
+
+static transport_endpt_ext_cfg_t *
+session_mq_get_ext_config (application_t *app, uword offset)
+{
+ svm_fifo_chunk_t *c;
+ fifo_segment_t *fs;
+
+ fs = application_get_rx_mqs_segment (app);
+ c = fs_chunk_ptr (fs->h, offset);
+ return (transport_endpt_ext_cfg_t *) c->data;
+}
+
+static void
+session_mq_free_ext_config (application_t *app, uword offset)
+{
+ svm_fifo_chunk_t *c;
+ fifo_segment_t *fs;
+
+ fs = application_get_rx_mqs_segment (app);
+ c = fs_chunk_ptr (fs->h, offset);
+ fifo_segment_collect_chunk (fs, 0 /* only one slice */, c);
+}
+
+static void
+session_mq_listen_handler (session_worker_t *wrk, session_evt_elt_t *elt)
+{
+ vnet_listen_args_t _a, *a = &_a;
+ session_listen_msg_t *mp;
+ app_worker_t *app_wrk;
+ application_t *app;
+ int rv;
+
+ app_check_thread_and_barrier (wrk, elt);
+
+ mp = session_evt_ctrl_data (wrk, elt);
+ app = application_lookup (mp->client_index);
+ if (!app)
+ return;
+
+ clib_memset (a, 0, sizeof (*a));
+ a->sep.is_ip4 = mp->is_ip4;
+ ip_copy (&a->sep.ip, &mp->ip, mp->is_ip4);
+ a->sep.port = mp->port;
+ a->sep.fib_index = mp->vrf;
+ a->sep.sw_if_index = ENDPOINT_INVALID_INDEX;
+ a->sep.transport_proto = mp->proto;
+ a->app_index = app->app_index;
+ a->wrk_map_index = mp->wrk_index;
+ a->sep_ext.transport_flags = mp->flags;
+
+ if (mp->ext_config)
+ a->sep_ext.ext_cfg = session_mq_get_ext_config (app, mp->ext_config);
+
+ if ((rv = vnet_listen (a)))
+ session_worker_stat_error_inc (wrk, rv, 1);
+
+ app_wrk = application_get_worker (app, mp->wrk_index);
+ app_worker_listened_notify (app_wrk, a->handle, mp->context, rv);
+
+ if (mp->ext_config)
+ session_mq_free_ext_config (app, mp->ext_config);
+
+ /* Make sure events are flushed before releasing barrier, to avoid
+ * potential race with accept. */
+ app_wrk_flush_wrk_events (app_wrk, 0);
+}
+
+static void
+session_mq_listen_uri_handler (session_worker_t *wrk, session_evt_elt_t *elt)
+{
+ vnet_listen_args_t _a, *a = &_a;
+ session_listen_uri_msg_t *mp;
+ app_worker_t *app_wrk;
+ application_t *app;
+ int rv;
+
+ app_check_thread_and_barrier (wrk, elt);
+
+ mp = session_evt_ctrl_data (wrk, elt);
+ app = application_lookup (mp->client_index);
+ if (!app)
+ return;
+
+ clib_memset (a, 0, sizeof (*a));
+ a->uri = (char *) mp->uri;
+ a->app_index = app->app_index;
+ rv = vnet_bind_uri (a);
+
+ app_wrk = application_get_worker (app, 0);
+ app_worker_listened_notify (app_wrk, a->handle, mp->context, rv);
+ app_wrk_flush_wrk_events (app_wrk, 0);
+}
+
+static void
+session_mq_connect_one (session_connect_msg_t *mp)
+{
+ vnet_connect_args_t _a, *a = &_a;
+ app_worker_t *app_wrk;
+ session_worker_t *wrk;
+ application_t *app;
+ int rv;
+
+ app = application_lookup (mp->client_index);
+ if (!app)
+ return;
+
+ clib_memset (a, 0, sizeof (*a));
+ a->sep.is_ip4 = mp->is_ip4;
+ clib_memcpy_fast (&a->sep.ip, &mp->ip, sizeof (mp->ip));
+ a->sep.port = mp->port;
+ a->sep.transport_proto = mp->proto;
+ a->sep.peer.fib_index = mp->vrf;
+ a->sep.dscp = mp->dscp;
+ clib_memcpy_fast (&a->sep.peer.ip, &mp->lcl_ip, sizeof (mp->lcl_ip));
+ if (mp->is_ip4)
+ {
+ ip46_address_mask_ip4 (&a->sep.ip);
+ ip46_address_mask_ip4 (&a->sep.peer.ip);
+ }
+ a->sep.peer.port = mp->lcl_port;
+ a->sep.peer.sw_if_index = ENDPOINT_INVALID_INDEX;
+ a->sep_ext.parent_handle = mp->parent_handle;
+ a->sep_ext.transport_flags = mp->flags;
+ a->api_context = mp->context;
+ a->app_index = app->app_index;
+ a->wrk_map_index = mp->wrk_index;
+
+ if (mp->ext_config)
+ a->sep_ext.ext_cfg = session_mq_get_ext_config (app, mp->ext_config);
+
+ if ((rv = vnet_connect (a)))
+ {
+ wrk = session_main_get_worker (vlib_get_thread_index ());
+ session_worker_stat_error_inc (wrk, rv, 1);
+ app_wrk = application_get_worker (app, mp->wrk_index);
+ app_worker_connect_notify (app_wrk, 0, rv, mp->context);
+ }
+
+ if (mp->ext_config)
+ session_mq_free_ext_config (app, mp->ext_config);
+}
+
+static void
+session_mq_handle_connects_rpc (void *arg)
+{
+ u32 max_connects = 32, n_connects = 0;
+ session_evt_elt_t *he, *elt, *next;
+ session_worker_t *fwrk;
+
+ ASSERT (session_vlib_thread_is_cl_thread ());
+
+ /* Pending connects on linked list pertaining to first worker */
+ fwrk = session_main_get_worker (transport_cl_thread ());
+ if (!fwrk->n_pending_connects)
+ return;
+
+ he = clib_llist_elt (fwrk->event_elts, fwrk->pending_connects);
+ elt = clib_llist_next (fwrk->event_elts, evt_list, he);
+
+ /* Avoid holding the worker for too long */
+ while (n_connects < max_connects && elt != he)
+ {
+ next = clib_llist_next (fwrk->event_elts, evt_list, elt);
+ clib_llist_remove (fwrk->event_elts, evt_list, elt);
+ session_mq_connect_one (session_evt_ctrl_data (fwrk, elt));
+ session_evt_ctrl_data_free (fwrk, elt);
+ clib_llist_put (fwrk->event_elts, elt);
+ elt = next;
+ n_connects += 1;
+ }
+
+ /* Decrement with worker barrier */
+ fwrk->n_pending_connects -= n_connects;
+ if (fwrk->n_pending_connects > 0)
+ {
+ session_send_rpc_evt_to_thread_force (fwrk->vm->thread_index,
+ session_mq_handle_connects_rpc, 0);
+ }
+}
+
+static void
+session_mq_connect_handler (session_worker_t *wrk, session_evt_elt_t *elt)
+{
+ u32 thread_index = wrk - session_main.wrk;
+ session_evt_elt_t *he;
+
+ if (PREDICT_FALSE (thread_index > transport_cl_thread ()))
+ {
+ clib_warning ("Connect on wrong thread. Dropping");
+ return;
+ }
+
+ /* If on worker, check if main has any pending messages. Avoids reordering
+ * with other control messages that need to be handled by main
+ */
+ if (thread_index)
+ {
+ he = clib_llist_elt (wrk->event_elts, wrk->evts_pending_main);
+
+ /* Events pending on main, postpone to avoid reordering */
+ if (!clib_llist_is_empty (wrk->event_elts, evt_list, he))
+ {
+ clib_llist_add_tail (wrk->event_elts, evt_list, elt, he);
+ return;
+ }
+ }
+
+ /* Add to pending list to be handled by first worker */
+ he = clib_llist_elt (wrk->event_elts, wrk->pending_connects);
+ clib_llist_add_tail (wrk->event_elts, evt_list, elt, he);
+
+ /* Decremented with worker barrier */
+ wrk->n_pending_connects += 1;
+ if (wrk->n_pending_connects == 1)
+ {
+ session_send_rpc_evt_to_thread_force (thread_index,
+ session_mq_handle_connects_rpc, 0);
+ }
+}
+
+static void
+session_mq_connect_uri_handler (session_worker_t *wrk, session_evt_elt_t *elt)
+{
+ vnet_connect_args_t _a, *a = &_a;
+ session_connect_uri_msg_t *mp;
+ app_worker_t *app_wrk;
+ application_t *app;
+ int rv;
+
+ app_check_thread_and_barrier (wrk, elt);
+
+ mp = session_evt_ctrl_data (wrk, elt);
+ app = application_lookup (mp->client_index);
+ if (!app)
+ return;
+
+ clib_memset (a, 0, sizeof (*a));
+ a->uri = (char *) mp->uri;
+ a->api_context = mp->context;
+ a->app_index = app->app_index;
+ if ((rv = vnet_connect_uri (a)))
+ {
+ session_worker_stat_error_inc (wrk, rv, 1);
+ app_wrk = application_get_worker (app, 0 /* default wrk only */ );
+ app_worker_connect_notify (app_wrk, 0, rv, mp->context);
+ }
+}
+
+static void
+session_mq_shutdown_handler (void *data)
+{
+ session_shutdown_msg_t *mp = (session_shutdown_msg_t *) data;
+ vnet_shutdown_args_t _a, *a = &_a;
+ application_t *app;
+
+ app = application_lookup (mp->client_index);
+ if (!app)
+ return;
+
+ a->app_index = app->app_index;
+ a->handle = mp->handle;
+ vnet_shutdown_session (a);
+}
+
+static void
+session_mq_disconnect_handler (void *data)
+{
+ session_disconnect_msg_t *mp = (session_disconnect_msg_t *) data;
+ vnet_disconnect_args_t _a, *a = &_a;
+ application_t *app;
+
+ app = application_lookup (mp->client_index);
+ if (!app)
+ return;
+
+ a->app_index = app->app_index;
+ a->handle = mp->handle;
+ vnet_disconnect_session (a);
+}
+
+static void
+app_mq_detach_handler (session_worker_t *wrk, session_evt_elt_t *elt)
+{
+ vnet_app_detach_args_t _a, *a = &_a;
+ session_app_detach_msg_t *mp;
+ application_t *app;
+
+ app_check_thread_and_barrier (wrk, elt);
+
+ mp = session_evt_ctrl_data (wrk, elt);
+ app = application_lookup (mp->client_index);
+ if (!app)
+ return;
+
+ a->app_index = app->app_index;
+ a->api_client_index = mp->client_index;
+ vnet_application_detach (a);
+}
+
+static void
+session_mq_unlisten_handler (session_worker_t *wrk, session_evt_elt_t *elt)
+{
+ vnet_unlisten_args_t _a, *a = &_a;
+ session_unlisten_msg_t *mp;
+ app_worker_t *app_wrk;
+ session_handle_t sh;
+ application_t *app;
+ int rv;
+
+ app_check_thread_and_barrier (wrk, elt);
+
+ mp = session_evt_ctrl_data (wrk, elt);
+ sh = mp->handle;
+
+ app = application_lookup (mp->client_index);
+ if (!app)
+ return;
+
+ clib_memset (a, 0, sizeof (*a));
+ a->app_index = app->app_index;
+ a->handle = sh;
+ a->wrk_map_index = mp->wrk_index;
+
+ if ((rv = vnet_unlisten (a)))
+ session_worker_stat_error_inc (wrk, rv, 1);
+
+ app_wrk = application_get_worker (app, a->wrk_map_index);
+ if (!app_wrk)
+ return;
+
+ app_worker_unlisten_reply (app_wrk, sh, mp->context, rv);
+}
+
+static void
+session_mq_accepted_reply_handler (session_worker_t *wrk,
+ session_evt_elt_t *elt)
+{
+ vnet_disconnect_args_t _a = { 0 }, *a = &_a;
+ session_accepted_reply_msg_t *mp;
+ session_state_t old_state;
+ app_worker_t *app_wrk;
+ session_t *s;
+
+ mp = session_evt_ctrl_data (wrk, elt);
+
+ /* Mail this back from the main thread. We're not polling in main
+ * thread so we're using other workers for notifications. */
+ if (session_thread_from_handle (mp->handle) == 0 && vlib_num_workers () &&
+ vlib_get_thread_index () != 0)
+ {
+ session_wrk_send_evt_to_main (wrk, elt);
+ return;
+ }
+
+ s = session_get_from_handle_if_valid (mp->handle);
+ if (!s)
+ return;
+
+ app_wrk = app_worker_get (s->app_wrk_index);
+ if (app_wrk->app_index != mp->context)
+ {
+ clib_warning ("app doesn't own session");
+ return;
+ }
+
+ /* Server isn't interested, disconnect the session */
+ if (mp->retval)
+ {
+ a->app_index = mp->context;
+ a->handle = mp->handle;
+ vnet_disconnect_session (a);
+ s->app_wrk_index = SESSION_INVALID_INDEX;
+ return;
+ }
+
+ /* Special handling for cut-through sessions */
+ if (!session_has_transport (s))
+ {
+ session_set_state (s, SESSION_STATE_READY);
+ ct_session_connect_notify (s, SESSION_E_NONE);
+ return;
+ }
+
+ old_state = s->session_state;
+ session_set_state (s, SESSION_STATE_READY);
+
+ if (!svm_fifo_is_empty_prod (s->rx_fifo))
+ app_worker_rx_notify (app_wrk, s);
+
+ /* Closed while waiting for app to reply. Resend disconnect */
+ if (old_state >= SESSION_STATE_TRANSPORT_CLOSING)
+ {
+ app_worker_close_notify (app_wrk, s);
+ session_set_state (s, old_state);
+ return;
+ }
+}
+
+static void
+session_mq_reset_reply_handler (void *data)
+{
+ vnet_disconnect_args_t _a = { 0 }, *a = &_a;
+ session_reset_reply_msg_t *mp;
+ app_worker_t *app_wrk;
+ session_t *s;
+ application_t *app;
+ u32 index, thread_index;
+
+ mp = (session_reset_reply_msg_t *) data;
+ app = application_lookup (mp->context);
+ if (!app)
+ return;
+
+ session_parse_handle (mp->handle, &index, &thread_index);
+ s = session_get_if_valid (index, thread_index);
+
+ /* No session or not the right session */
+ if (!s || s->session_state < SESSION_STATE_TRANSPORT_CLOSING)
+ return;
+
+ app_wrk = app_worker_get (s->app_wrk_index);
+ if (!app_wrk || app_wrk->app_index != app->app_index)
+ {
+ clib_warning ("App %u does not own handle 0x%lx!", app->app_index,
+ mp->handle);
+ return;
+ }
+
+ /* Client objected to resetting the session, log and continue */
+ if (mp->retval)
+ {
+ clib_warning ("client retval %d", mp->retval);
+ return;
+ }
+
+ /* This comes as a response to a reset, transport only waiting for
+ * confirmation to remove connection state, no need to disconnect */
+ a->handle = mp->handle;
+ a->app_index = app->app_index;
+ vnet_disconnect_session (a);
+}
+
+static void
+session_mq_disconnected_handler (void *data)
+{
+ session_disconnected_reply_msg_t *rmp;
+ vnet_disconnect_args_t _a, *a = &_a;
+ svm_msg_q_msg_t _msg, *msg = &_msg;
+ session_disconnected_msg_t *mp;
+ app_worker_t *app_wrk;
+ session_event_t *evt;
+ session_t *s;
+ application_t *app;
+ int rv = 0;
+
+ mp = (session_disconnected_msg_t *) data;
+ if (!(s = session_get_from_handle_if_valid (mp->handle)))
+ {
+ clib_warning ("could not disconnect handle %llu", mp->handle);
+ return;
+ }
+ app_wrk = app_worker_get (s->app_wrk_index);
+ app = application_lookup (mp->client_index);
+ if (!(app_wrk && app && app->app_index == app_wrk->app_index))
+ {
+ clib_warning ("could not disconnect session: %llu app: %u",
+ mp->handle, mp->client_index);
+ return;
+ }
+
+ a->handle = mp->handle;
+ a->app_index = app_wrk->wrk_index;
+ rv = vnet_disconnect_session (a);
+
+ svm_msg_q_lock_and_alloc_msg_w_ring (app_wrk->event_queue,
+ SESSION_MQ_CTRL_EVT_RING,
+ SVM_Q_WAIT, msg);
+ evt = svm_msg_q_msg_data (app_wrk->event_queue, msg);
+ clib_memset (evt, 0, sizeof (*evt));
+ evt->event_type = SESSION_CTRL_EVT_DISCONNECTED_REPLY;
+ rmp = (session_disconnected_reply_msg_t *) evt->data;
+ rmp->handle = mp->handle;
+ rmp->context = mp->context;
+ rmp->retval = rv;
+ svm_msg_q_add_and_unlock (app_wrk->event_queue, msg);
+}
+
+static void
+session_mq_disconnected_reply_handler (void *data)
+{
+ session_disconnected_reply_msg_t *mp;
+ vnet_disconnect_args_t _a, *a = &_a;
+ application_t *app;
+
+ mp = (session_disconnected_reply_msg_t *) data;
+
+ /* Client objected to disconnecting the session, log and continue */
+ if (mp->retval)
+ {
+ clib_warning ("client retval %d", mp->retval);
+ return;
+ }
+
+ /* Disconnect has been confirmed. Confirm close to transport */
+ app = application_lookup (mp->context);
+ if (app)
+ {
+ a->handle = mp->handle;
+ a->app_index = app->app_index;
+ vnet_disconnect_session (a);
+ }
+}
+
+static void
+session_mq_worker_update_handler (void *data)
+{
+ session_worker_update_msg_t *mp = (session_worker_update_msg_t *) data;
+ session_worker_update_reply_msg_t *rmp;
+ svm_msg_q_msg_t _msg, *msg = &_msg;
+ app_worker_t *app_wrk;
+ u32 owner_app_wrk_map;
+ session_event_t *evt;
+ session_t *s;
+ application_t *app;
+ int rv;
+
+ app = application_lookup (mp->client_index);
+ if (!app)
+ return;
+ if (!(s = session_get_from_handle_if_valid (mp->handle)))
+ {
+ clib_warning ("invalid handle %llu", mp->handle);
+ return;
+ }
+ app_wrk = app_worker_get (s->app_wrk_index);
+ if (app_wrk->app_index != app->app_index)
+ {
+ clib_warning ("app %u does not own session %llu", app->app_index,
+ mp->handle);
+ return;
+ }
+ owner_app_wrk_map = app_wrk->wrk_map_index;
+ app_wrk = application_get_worker (app, mp->wrk_index);
+
+ /* This needs to come from the new owner */
+ if (mp->req_wrk_index == owner_app_wrk_map)
+ {
+ session_req_worker_update_msg_t *wump;
+
+ svm_msg_q_lock_and_alloc_msg_w_ring (app_wrk->event_queue,
+ SESSION_MQ_CTRL_EVT_RING,
+ SVM_Q_WAIT, msg);
+ evt = svm_msg_q_msg_data (app_wrk->event_queue, msg);
+ clib_memset (evt, 0, sizeof (*evt));
+ evt->event_type = SESSION_CTRL_EVT_REQ_WORKER_UPDATE;
+ wump = (session_req_worker_update_msg_t *) evt->data;
+ wump->session_handle = mp->handle;
+ svm_msg_q_add_and_unlock (app_wrk->event_queue, msg);
+ return;
+ }
+
+ rv = app_worker_own_session (app_wrk, s);
+ if (rv)
+ session_stat_error_inc (rv, 1);
+
+ /*
+ * Send reply
+ */
+ svm_msg_q_lock_and_alloc_msg_w_ring (app_wrk->event_queue,
+ SESSION_MQ_CTRL_EVT_RING,
+ SVM_Q_WAIT, msg);
+ evt = svm_msg_q_msg_data (app_wrk->event_queue, msg);
+ clib_memset (evt, 0, sizeof (*evt));
+ evt->event_type = SESSION_CTRL_EVT_WORKER_UPDATE_REPLY;
+ rmp = (session_worker_update_reply_msg_t *) evt->data;
+ rmp->handle = mp->handle;
+ if (s->rx_fifo)
+ rmp->rx_fifo = fifo_segment_fifo_offset (s->rx_fifo);
+ if (s->tx_fifo)
+ rmp->tx_fifo = fifo_segment_fifo_offset (s->tx_fifo);
+ rmp->segment_handle = session_segment_handle (s);
+ svm_msg_q_add_and_unlock (app_wrk->event_queue, msg);
+
+ /*
+ * Retransmit messages that may have been lost
+ */
+ if (s->tx_fifo && !svm_fifo_is_empty (s->tx_fifo))
+ session_send_io_evt_to_thread (s->tx_fifo, SESSION_IO_EVT_TX);
+
+ if (s->rx_fifo && !svm_fifo_is_empty (s->rx_fifo))
+ app_worker_rx_notify (app_wrk, s);
+
+ if (s->session_state >= SESSION_STATE_TRANSPORT_CLOSING)
+ app_worker_close_notify (app_wrk, s);
+}
+
+static void
+session_mq_app_wrk_rpc_handler (void *data)
+{
+ session_app_wrk_rpc_msg_t *mp = (session_app_wrk_rpc_msg_t *) data;
+ svm_msg_q_msg_t _msg, *msg = &_msg;
+ session_app_wrk_rpc_msg_t *rmp;
+ app_worker_t *app_wrk;
+ session_event_t *evt;
+ application_t *app;
+
+ app = application_lookup (mp->client_index);
+ if (!app)
+ return;
+
+ app_wrk = application_get_worker (app, mp->wrk_index);
+
+ svm_msg_q_lock_and_alloc_msg_w_ring (app_wrk->event_queue,
+ SESSION_MQ_CTRL_EVT_RING, SVM_Q_WAIT,
+ msg);
+ evt = svm_msg_q_msg_data (app_wrk->event_queue, msg);
+ clib_memset (evt, 0, sizeof (*evt));
+ evt->event_type = SESSION_CTRL_EVT_APP_WRK_RPC;
+ rmp = (session_app_wrk_rpc_msg_t *) evt->data;
+ clib_memcpy (rmp->data, mp->data, sizeof (mp->data));
+ svm_msg_q_add_and_unlock (app_wrk->event_queue, msg);
+}
+
+static void
+session_mq_transport_attr_handler (void *data)
+{
+ session_transport_attr_msg_t *mp = (session_transport_attr_msg_t *) data;
+ session_transport_attr_reply_msg_t *rmp;
+ svm_msg_q_msg_t _msg, *msg = &_msg;
+ app_worker_t *app_wrk;
+ session_event_t *evt;
+ application_t *app;
+ session_t *s;
+ int rv;
+
+ app = application_lookup (mp->client_index);
+ if (!app)
+ return;
+
+ if (!(s = session_get_from_handle_if_valid (mp->handle)))
+ {
+ clib_warning ("invalid handle %llu", mp->handle);
+ return;
+ }
+ app_wrk = app_worker_get (s->app_wrk_index);
+ if (app_wrk->app_index != app->app_index)
+ {
+ clib_warning ("app %u does not own session %llu", app->app_index,
+ mp->handle);
+ return;
+ }
+
+ rv = session_transport_attribute (s, mp->is_get, &mp->attr);
+
+ svm_msg_q_lock_and_alloc_msg_w_ring (
+ app_wrk->event_queue, SESSION_MQ_CTRL_EVT_RING, SVM_Q_WAIT, msg);
+ evt = svm_msg_q_msg_data (app_wrk->event_queue, msg);
+ clib_memset (evt, 0, sizeof (*evt));
+ evt->event_type = SESSION_CTRL_EVT_TRANSPORT_ATTR_REPLY;
+ rmp = (session_transport_attr_reply_msg_t *) evt->data;
+ rmp->handle = mp->handle;
+ rmp->retval = rv;
+ rmp->is_get = mp->is_get;
+ if (!rv && mp->is_get)
+ rmp->attr = mp->attr;
+ svm_msg_q_add_and_unlock (app_wrk->event_queue, msg);
+}
+
+void
+session_wrk_handle_evts_main_rpc (void *args)
+{
+ vlib_main_t *vm = vlib_get_main ();
+ clib_llist_index_t ei, next_ei;
+ session_evt_elt_t *he, *elt;
+ session_worker_t *fwrk;
+ u32 thread_index;
+
+ vlib_worker_thread_barrier_sync (vm);
+
+ thread_index = pointer_to_uword (args);
+ fwrk = session_main_get_worker (thread_index);
+
+ he = clib_llist_elt (fwrk->event_elts, fwrk->evts_pending_main);
+ ei = clib_llist_next_index (he, evt_list);
+
+ while (ei != fwrk->evts_pending_main)
+ {
+ elt = clib_llist_elt (fwrk->event_elts, ei);
+ next_ei = clib_llist_next_index (elt, evt_list);
+ clib_llist_remove (fwrk->event_elts, evt_list, elt);
+ switch (elt->evt.event_type)
+ {
+ case SESSION_CTRL_EVT_LISTEN:
+ session_mq_listen_handler (fwrk, elt);
+ break;
+ case SESSION_CTRL_EVT_UNLISTEN:
+ session_mq_unlisten_handler (fwrk, elt);
+ break;
+ case SESSION_CTRL_EVT_APP_DETACH:
+ app_mq_detach_handler (fwrk, elt);
+ break;
+ case SESSION_CTRL_EVT_CONNECT_URI:
+ session_mq_connect_uri_handler (fwrk, elt);
+ break;
+ case SESSION_CTRL_EVT_ACCEPTED_REPLY:
+ session_mq_accepted_reply_handler (fwrk, elt);
+ break;
+ case SESSION_CTRL_EVT_CONNECT:
+ session_mq_connect_handler (fwrk, elt);
+ break;
+ default:
+ clib_warning ("unhandled %u", elt->evt.event_type);
+ ALWAYS_ASSERT (0);
+ break;
+ }
+
+ /* Regrab element in case pool moved */
+ elt = clib_llist_elt (fwrk->event_elts, ei);
+ if (!clib_llist_elt_is_linked (elt, evt_list))
+ {
+ session_evt_ctrl_data_free (fwrk, elt);
+ clib_llist_put (fwrk->event_elts, elt);
+ }
+ ei = next_ei;
+ }
+
+ vlib_worker_thread_barrier_release (vm);
+}