+ 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 (void *data)
+{
+ session_listen_msg_t *mp = (session_listen_msg_t *) data;
+ vnet_listen_args_t _a, *a = &_a;
+ app_worker_t *app_wrk;
+ application_t *app;
+ int rv;
+
+ app_check_thread_and_barrier (session_mq_listen_handler, mp);
+
+ 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)))
+ clib_warning ("listen returned: %U", format_session_error, rv);
+
+ app_wrk = application_get_worker (app, mp->wrk_index);
+ mq_send_session_bound_cb (app_wrk->wrk_index, mp->context, a->handle, rv);
+
+ if (mp->ext_config)
+ session_mq_free_ext_config (app, mp->ext_config);
+}
+
+static void
+session_mq_listen_uri_handler (void *data)
+{
+ session_listen_uri_msg_t *mp = (session_listen_uri_msg_t *) data;
+ vnet_listen_args_t _a, *a = &_a;
+ app_worker_t *app_wrk;
+ application_t *app;
+ int rv;
+
+ app_check_thread_and_barrier (session_mq_listen_uri_handler, mp);
+
+ 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);
+ mq_send_session_bound_cb (app_wrk->wrk_index, mp->context, a->handle, rv);
+}
+
+static void
+session_mq_connect_one (session_connect_msg_t *mp)
+{
+ vnet_connect_args_t _a, *a = &_a;
+ app_worker_t *app_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;
+ 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)))
+ {
+ clib_warning ("connect returned: %U", format_session_error, rv);
+ app_wrk = application_get_worker (app, mp->wrk_index);
+ mq_send_session_connected_cb (app_wrk->wrk_index, mp->context, 0, rv);
+ }
+
+ 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;
+ vlib_main_t *vm = vlib_get_main ();
+ session_evt_elt_t *he, *elt, *next;
+ session_worker_t *fwrk, *wrk;
+
+ ASSERT (vlib_get_thread_index () == 0);
+
+ /* Pending connects on linked list pertaining to first worker */
+ fwrk = session_main_get_worker (1);
+ if (!fwrk->n_pending_connects)
+ goto update_state;
+
+ vlib_worker_thread_barrier_sync (vm);
+
+ he = clib_llist_elt (fwrk->event_elts, fwrk->pending_connects);
+ elt = clib_llist_next (fwrk->event_elts, evt_list, he);
+
+ /* Avoid holding the barrier 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;
+
+ vlib_worker_thread_barrier_release (vm);
+
+update_state:
+
+ /* Switch worker to poll mode if it was in interrupt mode and had work or
+ * back to interrupt if threshold of loops without a connect is passed.
+ * While in poll mode, reprogram connects rpc */
+ wrk = session_main_get_worker (0);
+ if (wrk->state != SESSION_WRK_POLLING)
+ {
+ if (n_connects)
+ {
+ session_wrk_set_state (wrk, SESSION_WRK_POLLING);
+ vlib_node_set_state (vm, session_queue_node.index,
+ VLIB_NODE_STATE_POLLING);
+ wrk->no_connect_loops = 0;
+ }
+ }
+ else
+ {
+ if (!n_connects)
+ {
+ if (++wrk->no_connect_loops > 1e5)
+ {
+ session_wrk_set_state (wrk, SESSION_WRK_INTERRUPT);
+ vlib_node_set_state (vm, session_queue_node.index,
+ VLIB_NODE_STATE_INTERRUPT);
+ }
+ }
+ else
+ wrk->no_connect_loops = 0;
+ }
+
+ if (wrk->state == SESSION_WRK_POLLING)
+ {
+ elt = session_evt_alloc_ctrl (wrk);
+ elt->evt.event_type = SESSION_CTRL_EVT_RPC;
+ elt->evt.rpc_args.fp = session_mq_handle_connects_rpc;
+ }
+}
+
+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;
+
+ /* No workers, so just deal with the connect now */
+ if (PREDICT_FALSE (!thread_index))
+ {
+ session_mq_connect_one (session_evt_ctrl_data (wrk, elt));
+ return;
+ }
+
+ if (PREDICT_FALSE (thread_index != 1))
+ {
+ clib_warning ("Connect on wrong thread. Dropping");
+ return;
+ }
+
+ /* Add to pending list to be handled by main thread */
+ 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)
+ {
+ vlib_node_set_interrupt_pending (vlib_get_main_by_index (0),
+ session_queue_node.index);
+ session_send_rpc_evt_to_thread (0, session_mq_handle_connects_rpc, 0);
+ }
+}
+
+static void
+session_mq_connect_uri_handler (void *data)
+{
+ session_connect_uri_msg_t *mp = (session_connect_uri_msg_t *) data;
+ vnet_connect_args_t _a, *a = &_a;
+ app_worker_t *app_wrk;
+ application_t *app;
+ int rv;
+
+ app_check_thread_and_barrier (session_mq_connect_uri_handler, mp);
+
+ 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)))
+ {
+ clib_warning ("connect_uri returned: %d", rv);
+ app_wrk = application_get_worker (app, 0 /* default wrk only */ );
+ mq_send_session_connected_cb (app_wrk->wrk_index, mp->context, 0, rv);
+ }
+}
+
+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 (void *data)
+{
+ session_app_detach_msg_t *mp = (session_app_detach_msg_t *) data;
+ vnet_app_detach_args_t _a, *a = &_a;
+ application_t *app;
+
+ app_check_thread_and_barrier (app_mq_detach_handler, mp);
+
+ 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_rpc (session_unlisten_msg_t *mp)
+{
+ vlib_main_t *vm = vlib_get_main ();
+ vnet_unlisten_args_t _a, *a = &_a;
+ app_worker_t *app_wrk;
+ session_handle_t sh;
+ application_t *app;
+ u32 context;
+ int rv;
+
+ sh = mp->handle;
+ context = mp->context;
+
+ vlib_worker_thread_barrier_sync (vm);
+
+ 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)))
+ clib_warning ("unlisten returned: %d", rv);
+
+ app_wrk = application_get_worker (app, a->wrk_map_index);
+ if (!app_wrk)
+ return;
+
+ vlib_worker_thread_barrier_release (vm);
+
+ mq_send_unlisten_reply (app_wrk, sh, context, rv);
+ clib_mem_free (mp);
+}
+
+static void
+session_mq_unlisten_handler (session_worker_t *wrk, session_evt_elt_t *elt)
+{
+ u32 thread_index = wrk - session_main.wrk;
+ session_unlisten_msg_t *mp, *arg;
+
+ mp = session_evt_ctrl_data (wrk, elt);
+ arg = clib_mem_alloc (sizeof (session_unlisten_msg_t));
+ clib_memcpy_fast (arg, mp, sizeof (*arg));
+
+ if (PREDICT_FALSE (!thread_index))
+ {
+ session_mq_unlisten_rpc (arg);
+ return;
+ }
+
+ session_send_rpc_evt_to_thread_force (0, session_mq_unlisten_rpc, arg);