+static void
+ct_accept_rpc_wrk_handler (void *rpc_args)
+{
+ u32 thread_index, n_connects, i, n_pending;
+ const u32 max_connects = 32;
+ ct_worker_t *wrk;
+ u8 need_rpc = 0;
+
+ thread_index = pointer_to_uword (rpc_args);
+ wrk = ct_worker_get (thread_index);
+
+ /* Connects could be handled without worker barrier so grab lock */
+ clib_spinlock_lock (&wrk->pending_connects_lock);
+
+ n_pending = clib_fifo_elts (wrk->pending_connects);
+ n_connects = clib_min (n_pending, max_connects);
+ vec_validate (wrk->new_connects, n_connects);
+
+ for (i = 0; i < n_connects; i++)
+ clib_fifo_sub1 (wrk->pending_connects, wrk->new_connects[i]);
+
+ if (n_pending == n_connects)
+ wrk->have_connects = 0;
+ else
+ need_rpc = 1;
+
+ clib_spinlock_unlock (&wrk->pending_connects_lock);
+
+ for (i = 0; i < n_connects; i++)
+ ct_accept_one (thread_index, wrk->new_connects[i]);
+
+ if (need_rpc)
+ session_send_rpc_evt_to_thread_force (
+ thread_index, ct_accept_rpc_wrk_handler,
+ uword_to_pointer (thread_index, void *));
+}
+
+static void
+ct_fwrk_flush_connects (void *rpc_args)
+{
+ u32 thread_index, fwrk_index, n_workers;
+ ct_main_t *cm = &ct_main;
+ ct_worker_t *wrk;
+ u8 need_rpc;
+
+ fwrk_index = cm->fwrk_thread;
+ n_workers = vec_len (cm->fwrk_pending_connects);
+
+ for (thread_index = fwrk_index; thread_index < n_workers; thread_index++)
+ {
+ if (!vec_len (cm->fwrk_pending_connects[thread_index]))
+ continue;
+
+ wrk = ct_worker_get (thread_index);
+
+ /* Connects can be done without worker barrier, grab dst worker lock */
+ if (thread_index != fwrk_index)
+ clib_spinlock_lock (&wrk->pending_connects_lock);
+
+ clib_fifo_add (wrk->pending_connects,
+ cm->fwrk_pending_connects[thread_index],
+ vec_len (cm->fwrk_pending_connects[thread_index]));
+ if (!wrk->have_connects)
+ {
+ wrk->have_connects = 1;
+ need_rpc = 1;
+ }
+
+ if (thread_index != fwrk_index)
+ clib_spinlock_unlock (&wrk->pending_connects_lock);
+
+ vec_reset_length (cm->fwrk_pending_connects[thread_index]);
+
+ if (need_rpc)
+ session_send_rpc_evt_to_thread_force (
+ thread_index, ct_accept_rpc_wrk_handler,
+ uword_to_pointer (thread_index, void *));
+ }
+
+ cm->fwrk_have_flush = 0;
+}
+
+static void
+ct_program_connect_to_wrk (u32 ho_index)
+{
+ ct_main_t *cm = &ct_main;
+ u32 thread_index;
+
+ /* Simple round-robin policy for spreading sessions over workers. We skip
+ * thread index 0, i.e., offset the index by 1, when we have workers as it
+ * is the one dedicated to main thread. Note that n_workers does not include
+ * main thread */
+ cm->n_sessions += 1;
+ thread_index = cm->n_workers ? (cm->n_sessions % cm->n_workers) + 1 : 0;
+
+ /* Pospone flushing of connect request to dst worker until after session
+ * layer fully initializes the half-open session. */
+ vec_add1 (cm->fwrk_pending_connects[thread_index], ho_index);
+ if (!cm->fwrk_have_flush)
+ {
+ session_send_rpc_evt_to_thread_force (
+ cm->fwrk_thread, ct_fwrk_flush_connects,
+ uword_to_pointer (thread_index, void *));
+ cm->fwrk_have_flush = 1;
+ }
+}
+
+static int
+ct_connect (app_worker_t *client_wrk, session_t *ll,
+ session_endpoint_cfg_t *sep)
+{
+ ct_connection_t *ho;
+ u32 ho_index;
+
+ /*
+ * Alloc and init client half-open transport
+ */
+
+ ho = ct_half_open_alloc ();
+ ho_index = ho->c_c_index;
+ ho->c_rmt_port = sep->port;
+ ho->c_lcl_port = 0;
+ ho->c_is_ip4 = sep->is_ip4;
+ ho->client_opaque = sep->opaque;
+ ho->client_wrk = client_wrk->wrk_index;
+ ho->peer_index = ll->session_index;
+ ho->c_proto = TRANSPORT_PROTO_NONE;
+ ho->c_flags |= TRANSPORT_CONNECTION_F_NO_LOOKUP;
+ clib_memcpy (&ho->c_rmt_ip, &sep->ip, sizeof (sep->ip));
+ ho->flags |= CT_CONN_F_CLIENT;
+ ho->c_s_index = ~0;
+ ho->actual_tp = sep->original_tp;
+
+ /*
+ * Program connect on a worker, connected reply comes
+ * after server accepts the connection.
+ */
+ ct_program_connect_to_wrk (ho_index);
+
+ return ho_index;