VCL IOEvent external API callback
Change-Id: I417357b00c43b27872aa3f681335bdc1ef574eca
Signed-off-by: Keith Burns (alagalah) <alagalah@gmail.com>
Signed-off-by: Dave Wallace <dwallacelf@gmail.com>
diff --git a/src/vcl/sock_test.h b/src/vcl/sock_test.h
index f9f5c70..26dc3f1 100644
--- a/src/vcl/sock_test.h
+++ b/src/vcl/sock_test.h
@@ -65,6 +65,7 @@
typedef struct __attribute__ ((packed))
{
uint32_t magic;
+ uint32_t seq_num;
uint32_t test;
uint32_t ctrl_handle;
uint32_t num_test_sockets;
@@ -215,6 +216,7 @@
printf (" test config (%p):\n"
SOCK_TEST_SEPARATOR_STRING
" magic: 0x%08x\n"
+ " seq_num: 0x%08x\n"
"%-5s test: %s (%d)\n"
" ctrl handle: %d (0x%x)\n"
"%-5s num test sockets: %u (0x%08x)\n"
@@ -224,7 +226,7 @@
"%-5s num writes: %lu (0x%08lx)\n"
" client tx bytes: %lu (0x%08lx)\n"
SOCK_TEST_SEPARATOR_STRING,
- (void *) cfg, cfg->magic,
+ (void *) cfg, cfg->magic, cfg->seq_num,
is_client && (cfg->test == SOCK_TEST_TYPE_UNI) ?
"'"SOCK_TEST_TOKEN_RUN_UNI"'" :
is_client && (cfg->test == SOCK_TEST_TYPE_BI) ?
diff --git a/src/vcl/sock_test_client.c b/src/vcl/sock_test_client.c
index 1ed4b89..e88b2b9 100644
--- a/src/vcl/sock_test_client.c
+++ b/src/vcl/sock_test_client.c
@@ -37,6 +37,7 @@
#endif
struct sockaddr_storage server_addr;
uint32_t server_addr_size;
+ uint32_t cfg_seq_num;
sock_test_socket_t ctrl_socket;
sock_test_socket_t *test_socket;
uint32_t num_test_sockets;
@@ -57,12 +58,18 @@
if (socket->cfg.verbose)
sock_test_cfg_dump (&socket->cfg, 1 /* is_client */ );
+ ctrl->cfg.seq_num = ++scm->cfg_seq_num;
+ if (socket->cfg.verbose)
+ {
+ printf ("CLIENT (fd %d): Sending config sent to server.\n", socket->fd);
+ sock_test_cfg_dump (&ctrl->cfg, 1 /* is_client */ );
+ }
tx_bytes = sock_test_write (socket->fd, (uint8_t *) & ctrl->cfg,
sizeof (ctrl->cfg), NULL, ctrl->cfg.verbose);
if (tx_bytes < 0)
{
- fprintf (stderr, "CLIENT: ERROR: write test cfg failed (%d)!\n",
- tx_bytes);
+ fprintf (stderr, "CLIENT (fd %d): ERROR: write test cfg failed (%d)!\n",
+ socket->fd, tx_bytes);
return tx_bytes;
}
@@ -73,22 +80,34 @@
if (rl_cfg->magic != SOCK_TEST_CFG_CTRL_MAGIC)
{
- fprintf (stderr, "CLIENT: ERROR: Bad server reply cfg -- aborting!\n");
+ fprintf (stderr, "CLIENT (fd %d): ERROR: Bad server reply cfg "
+ "-- aborting!\n", socket->fd);
return -1;
}
- if (socket->cfg.verbose)
- {
- printf ("CLIENT (fd %d): Got config back from server.\n", socket->fd);
- sock_test_cfg_dump (rl_cfg, 1 /* is_client */ );
- }
if ((rx_bytes != sizeof (sock_test_cfg_t))
|| !sock_test_cfg_verify (rl_cfg, &ctrl->cfg))
{
- fprintf (stderr, "CLIENT: ERROR: Invalid config received "
- "from server -- aborting!\n");
- sock_test_cfg_dump (rl_cfg, 1 /* is_client */ );
+ fprintf (stderr, "CLIENT (fd %d): ERROR: Invalid config received "
+ "from server!\n", socket->fd);
+ if (rx_bytes != sizeof (sock_test_cfg_t))
+ {
+ fprintf (stderr, "\tRx bytes %d != cfg size %lu\n",
+ rx_bytes, sizeof (sock_test_cfg_t));
+ }
+ else
+ {
+ sock_test_cfg_dump (rl_cfg, 1 /* is_client */ );
+ fprintf (stderr, "CLIENT (fd %d): Valid config sent to server.\n",
+ socket->fd);
+ sock_test_cfg_dump (&ctrl->cfg, 1 /* is_client */ );
+ }
return -1;
}
+ else if (socket->cfg.verbose)
+ {
+ printf ("CLIENT (fd %d): Got config back from server.\n", socket->fd);
+ sock_test_cfg_dump (rl_cfg, 1 /* is_client */ );
+ }
ctrl->cfg.ctrl_handle = ((ctrl->cfg.ctrl_handle == ~0) ?
rl_cfg->ctrl_handle : ctrl->cfg.ctrl_handle);
@@ -118,7 +137,8 @@
tsock = &scm->test_socket[n];
tsock->cfg = ctrl->cfg;
sock_test_socket_buf_alloc (tsock);
- sock_test_cfg_sync (tsock);
+ if (sock_test_cfg_sync (tsock))
+ return;
memcpy (tsock->txbuf, ctrl->txbuf, nbytes);
memset (&tsock->stats, 0, sizeof (tsock->stats));
diff --git a/src/vcl/vcl_event.c b/src/vcl/vcl_event.c
index d8bd8c7..dafa250 100644
--- a/src/vcl/vcl_event.c
+++ b/src/vcl/vcl_event.c
@@ -59,10 +59,10 @@
}
void
-vce_clear_event (vce_event_thread_t *evt, vce_event_t *ev)
+vce_clear_event (vce_event_thread_t *evt, u32 ev_idx)
{
clib_spinlock_lock (&(evt->events_lockp));
- pool_put (evt->vce_events, ev);
+ pool_put_index (evt->vce_events, ev_idx);
clib_spinlock_unlock (&(evt->events_lockp));
}
@@ -70,11 +70,10 @@
vce_get_event_from_index(vce_event_thread_t *evt, u32 ev_idx)
{
vce_event_t *ev = 0;
+ /* Assumes caller has obtained the spinlock (evt->events_lockp) */
- clib_spinlock_lock (&(evt->events_lockp));
if ( ! pool_is_free_index (evt->vce_events, ev_idx))
ev = pool_elt_at_index (evt->vce_events, ev_idx);
- clib_spinlock_unlock (&(evt->events_lockp));
return ev;
}
@@ -96,7 +95,7 @@
vce_event_handler_reg_t *
vce_register_handler (vce_event_thread_t *evt, vce_event_key_t *evk,
- vce_event_callback_t cb, void *cb_args)
+ vce_event_callback_t cb, void *cb_args)
{
vce_event_handler_reg_t *handler;
vce_event_handler_reg_t *old_handler = 0;
@@ -115,25 +114,25 @@
/* If we are just re-registering, ignore and move on
* else store the old handler_fn for unregister to re-instate */
if (old_handler->handler_fn == cb)
- {
+ {
- clib_spinlock_unlock (&evt->handlers_lockp);
+ clib_spinlock_unlock (&evt->handlers_lockp);
- /* Signal event thread that a handler exists in case any
- * recycled events requiring this handler are pending */
- pthread_mutex_lock (&(evt->generator_lock));
- pthread_cond_signal (&(evt->generator_cond));
- pthread_mutex_unlock (&(evt->generator_lock));
- return old_handler;
- }
+ /* Signal event thread that a handler exists in case any
+ * recycled events requiring this handler are pending */
+ pthread_mutex_lock (&(evt->generator_lock));
+ pthread_cond_signal (&(evt->generator_cond));
+ pthread_mutex_unlock (&(evt->generator_lock));
+ return old_handler;
+ }
}
pool_get (evt->vce_event_handlers, handler);
handler_index = (u32) (handler - evt->vce_event_handlers);
handler->handler_fn = cb;
- handler->replaced_handler_idx = (p) ? p[0] : ~0;
- handler->ev_idx = ~0; //This will be set by the event thread if event happens
+ handler->replaced_handler_idx = (u32) ((p) ? p[0] : ~0);
+ handler->ev_idx = (u32) ~0; //This will be set by the event thread if event happens
handler->evk = evk->as_u64;
handler->handler_fn_args = cb_args;
@@ -210,61 +209,64 @@
u32 ev_idx;
vce_event_handler_reg_t *handler;
uword *p;
+ u32 recycle_count = 0;
pthread_mutex_lock (&(evt->generator_lock));
- clib_spinlock_lock (&(evt->events_lockp));
- evt->recycle_event = 1; // Used for recycling events with no handlers
- clib_spinlock_unlock (&(evt->events_lockp));
-
- do
+ while (1)
{
- while ( (clib_fifo_elts (evt->event_index_fifo) == 0) ||
- evt->recycle_event)
- {
- clib_spinlock_lock (&(evt->events_lockp));
- evt->recycle_event = 0;
- clib_spinlock_unlock (&(evt->events_lockp));
- pthread_cond_wait (&(evt->generator_cond),
- &(evt->generator_lock));
- }
+ uword fifo_depth = clib_fifo_elts (evt->event_index_fifo);
+ while ((fifo_depth == 0) || (recycle_count == fifo_depth))
+ {
+ recycle_count = 0;
+ pthread_cond_wait (&(evt->generator_cond), &(evt->generator_lock));
+ fifo_depth = clib_fifo_elts (evt->event_index_fifo);
+ }
/* Remove event */
clib_spinlock_lock (&(evt->events_lockp));
-
clib_fifo_sub1 (evt->event_index_fifo, ev_idx);
- ev = pool_elt_at_index (evt->vce_events, ev_idx);
-
+ ev = vce_get_event_from_index (evt, ev_idx);
ASSERT(ev);
-
+ if (recycle_count && ev->recycle)
+ {
+ clib_fifo_add1 (evt->event_index_fifo, ev_idx);
+ clib_spinlock_unlock (&(evt->events_lockp));
+ continue;
+ }
clib_spinlock_lock (&evt->handlers_lockp);
p = hash_get (evt->handlers_index_by_event_key, ev->evk.as_u64);
if (!p)
- {
- /* If an event falls in the woods, and there is no handler to hear it,
- * does it make any sound?
- * I don't know either, so lets try recycling the event */
- clib_fifo_add1 (evt->event_index_fifo, ev_idx);
- evt->recycle_event = 1;
+ {
+ /* If an event falls in the woods, and there is no handler to hear it,
+ * does it make any sound?
+ * I don't know either, so lets biff the event */
+ pool_put(evt->vce_events, ev);
clib_spinlock_unlock (&(evt->events_lockp));
- clib_spinlock_unlock (&evt->handlers_lockp);
+ clib_spinlock_unlock (&evt->handlers_lockp);
pthread_mutex_unlock (&(evt->generator_lock));
- }
+ }
else
{
+ u32 evt_recycle = ev->recycle;
handler = pool_elt_at_index (evt->vce_event_handlers, p[0]);
handler->ev_idx = ev_idx;
+ ev->recycle = 0;
clib_spinlock_unlock (&(evt->events_lockp));
clib_spinlock_unlock (&evt->handlers_lockp);
pthread_mutex_unlock (&(evt->generator_lock));
(handler->handler_fn)(handler);
+
+ clib_spinlock_lock (&(evt->events_lockp));
+ ev = vce_get_event_from_index (evt, ev_idx);
+ recycle_count += (!evt_recycle && ev && ev->recycle) ? 1 : 0;
+ clib_spinlock_unlock(&(evt->events_lockp));
}
pthread_mutex_lock (&(evt->generator_lock));
}
- while (1);
return NULL;
}
@@ -281,5 +283,6 @@
clib_spinlock_init (&(evt->handlers_lockp));
return pthread_create (&(evt->thread), NULL /* attr */ ,
- vce_event_thread_fn, evt);
+ vce_event_thread_fn, evt);
}
+
diff --git a/src/vcl/vcl_event.h b/src/vcl/vcl_event.h
index f2a85a0..2188466 100644
--- a/src/vcl/vcl_event.h
+++ b/src/vcl/vcl_event.h
@@ -39,8 +39,8 @@
typedef struct vce_event_
{
vce_event_key_t evk;
- u32 refcnt;
- void *data;
+ u32 recycle;
+ u64 data[2]; // Hard code size to avoid allocator thrashing.
} vce_event_t;
typedef void (*vce_event_callback_t) (void *reg /*vce_event_handler_reg_t* */);
@@ -87,9 +87,9 @@
* - removes event from event_pool
*
* @param evt - vce_event_thread_t - event system state
- * @param ev - vce_event_t - event to remove
+ * @param ev_idx - u32 - index of event to remove
*/
-void vce_clear_event (vce_event_thread_t *evt, vce_event_t *ev);
+void vce_clear_event (vce_event_thread_t *evt, u32 ev_idx);
/**
* @brief vce_get_event_from_index()
@@ -102,6 +102,20 @@
vce_event_t * vce_get_event_from_index(vce_event_thread_t *evt, u32 ev_idx);
/**
+ * @brief vce_get_event_data()
+ *
+ * @param ev - vce_event_t * - event
+ * @param data_size - u32 - required size of data
+ *
+ * @return vce_event_t *
+ */
+always_inline void * vce_get_event_data(vce_event_t *ev, u32 data_size)
+{
+ ASSERT(sizeof(ev->data) >= data_size);
+ return (&ev->data);
+}
+
+/**
* @brief vce_get_event_handler()
* - returns handler if exists or 0
* @param evt - vce_event_thread_t - event system state
diff --git a/src/vcl/vppcom.c b/src/vcl/vppcom.c
index cab2f60..a66926d 100644
--- a/src/vcl/vppcom.c
+++ b/src/vcl/vppcom.c
@@ -195,9 +195,12 @@
{
VCL_EVENT_INVALID_EVENT,
VCL_EVENT_CONNECT_REQ_ACCEPTED,
+ VCL_EVENT_IOEVENT_RX_FIFO,
+ VCL_EVENT_IOEVENT_TX_FIFO,
VCL_EVENT_N_EVENTS
} vcl_event_id_t;
+
typedef struct vce_event_connect_request_
{
u32 accepted_session_index;
@@ -210,6 +213,22 @@
void *user_cb_data;
} vppcom_session_listener_t;
+typedef struct vppcom_session_ioevent_
+{
+ vppcom_session_ioevent_cb user_cb;
+ void *user_cb_data;
+} vppcom_session_ioevent_t;
+
+typedef struct vppcom_session_io_thread_
+{
+ pthread_t thread;
+ pthread_mutex_t vce_io_lock;
+ pthread_cond_t vce_io_cond;
+ u32 *active_session_indexes; //pool
+ vppcom_session_ioevent_t *ioevents; //pool
+ clib_spinlock_t io_sessions_lockp;
+} vppcom_session_io_thread_t;
+
typedef struct vppcom_main_t_
{
u8 init;
@@ -254,6 +273,9 @@
/* Event thread */
vce_event_thread_t event_thread;
+ /* IO thread */
+ vppcom_session_io_thread_t session_io_thread;
+
/* VPP Event-logger */
elog_main_t elog_main;
elog_track_t elog_track;
@@ -364,6 +386,9 @@
/*
* VPPCOM Utility Functions
*/
+
+
+
static inline int
vppcom_session_at_index (u32 session_index, session_t * volatile *sess)
{
@@ -379,6 +404,81 @@
return VPPCOM_OK;
}
+void *
+vppcom_session_io_thread_fn (void *arg)
+{
+ vppcom_session_io_thread_t *evt = (vppcom_session_io_thread_t *) arg;
+ u32 *session_indexes = 0, *session_index;
+ int i, rv;
+ u32 bytes = 0;
+ session_t *session;
+
+ while (1)
+ {
+ vec_reset_length (session_indexes);
+ clib_spinlock_lock (&evt->io_sessions_lockp);
+ pool_foreach (session_index, evt->active_session_indexes, (
+ {
+ vec_add1
+ (session_indexes,
+ *session_index);
+ }
+ ));
+ clib_spinlock_unlock (&evt->io_sessions_lockp);
+ if (session_indexes)
+ {
+ for (i = 0; i < vec_len (session_indexes); ++i)
+ {
+ VCL_LOCK_AND_GET_SESSION (session_indexes[i], &session);
+ bytes = svm_fifo_max_dequeue (session->rx_fifo);
+ clib_spinlock_unlock (&vcm->sessions_lockp);
+
+ if (bytes)
+ {
+ vppcom_ioevent_t *eio;
+ vce_event_t *ev;
+ u32 ev_idx;
+
+ clib_spinlock_lock (&vcm->event_thread.events_lockp);
+
+ pool_get (vcm->event_thread.vce_events, ev);
+ ev_idx = (u32) (ev - vcm->event_thread.vce_events);
+ eio = vce_get_event_data (ev, sizeof (*eio));
+ ev->evk.eid = VCL_EVENT_IOEVENT_RX_FIFO;
+ ev->evk.session_index = session_indexes[i];
+ eio->bytes = bytes;
+ eio->session_index = session_indexes[i];
+
+ clib_spinlock_unlock (&vcm->event_thread.events_lockp);
+
+ rv = vce_generate_event (&vcm->event_thread, ev_idx);
+ }
+ }
+ }
+ struct timespec ts;
+ ts.tv_sec = 0;
+ ts.tv_nsec = 1000000; /* 1 millisecond */
+ nanosleep (&ts, NULL);
+ }
+done:
+ clib_spinlock_unlock (&vcm->sessions_lockp);
+ return NULL;
+}
+
+int
+vppcom_start_io_event_thread (vppcom_session_io_thread_t * evt,
+ u8 max_sessions)
+{
+ pthread_cond_init (&(evt->vce_io_cond), NULL);
+ pthread_mutex_init (&(evt->vce_io_lock), NULL);
+
+ clib_spinlock_init (&(evt->io_sessions_lockp));
+
+ return pthread_create (&(evt->thread), NULL /* attr */ ,
+ vppcom_session_io_thread_fn, evt);
+}
+
+
static inline void
vppcom_session_table_add_listener (u64 listener_handle, u32 value)
{
@@ -462,6 +562,32 @@
*/
void
+vce_registered_ioevent_handler_fn (void *arg)
+{
+ vce_event_handler_reg_t *reg = (vce_event_handler_reg_t *) arg;
+ vppcom_ioevent_t *eio;
+ vce_event_t *ev;
+ u32 ioevt_ndx = (u64) (reg->handler_fn_args);
+ vppcom_session_ioevent_t *ioevent, ioevent_;
+
+ clib_spinlock_lock (&(vcm->event_thread.events_lockp));
+ ev = vce_get_event_from_index (&vcm->event_thread, reg->ev_idx);
+ eio = vce_get_event_data (ev, sizeof (*eio));
+ clib_spinlock_unlock (&(vcm->event_thread.events_lockp));
+
+ clib_spinlock_lock (&vcm->session_io_thread.io_sessions_lockp);
+ ioevent = pool_elt_at_index (vcm->session_io_thread.ioevents, ioevt_ndx);
+ ioevent_ = *ioevent;
+ clib_spinlock_unlock (&vcm->session_io_thread.io_sessions_lockp);
+ (ioevent_.user_cb) (eio, ioevent_.user_cb_data);
+ vce_clear_event (&vcm->event_thread, reg->ev_idx);
+ return;
+
+ /*TODO - Unregister check in close for this listener */
+
+}
+
+void
vce_registered_listener_connect_handler_fn (void *arg)
{
vce_event_handler_reg_t *reg = (vce_event_handler_reg_t *) arg;
@@ -475,12 +601,12 @@
vppcom_session_listener_t *session_listener =
(vppcom_session_listener_t *) reg->handler_fn_args;
+ clib_spinlock_lock (&(vcm->event_thread.events_lockp));
ev = vce_get_event_from_index (&vcm->event_thread, reg->ev_idx);
-
- ecr = (vce_event_connect_request_t *) ev->data;
+ ecr = vce_get_event_data (ev, sizeof (*ecr));
+ clib_spinlock_unlock (&(vcm->event_thread.events_lockp));
VCL_LOCK_AND_GET_SESSION (ecr->accepted_session_index, &new_session);
-
ep.is_ip4 = new_session->peer_addr.is_ip4;
ep.port = new_session->peer_port;
if (new_session->peer_addr.is_ip4)
@@ -498,8 +624,18 @@
(session_listener->user_cb) (ecr->accepted_session_index, &ep,
session_listener->user_cb_data);
- /*TODO - Unregister check in close for this listener */
+ if (vcm->session_io_thread.io_sessions_lockp)
+ {
+ /* Throw this new accepted session index into the rx poll thread pool */
+ clib_spinlock_lock (&vcm->session_io_thread.io_sessions_lockp);
+ u32 *active_session_index;
+ pool_get (vcm->session_io_thread.active_session_indexes,
+ active_session_index);
+ *active_session_index = ecr->accepted_session_index;
+ clib_spinlock_unlock (&vcm->session_io_thread.io_sessions_lockp);
+ }
+ /*TODO - Unregister check in close for this listener */
return;
done:
@@ -541,7 +677,7 @@
vce_event_t *ev;
/* Retrieve the VCL_EVENT_CONNECT_REQ_ACCEPTED event */
ev = vce_get_event_from_index (&vcm->event_thread, reg->ev_idx);
- vce_event_connect_request_t *ecr = (vce_event_connect_request_t *) ev->data;
+ vce_event_connect_request_t *ecr = vce_get_event_data (ev, sizeof (*ecr));
/* Add the accepted_session_index to the FIFO */
clib_spinlock_lock (&vcm->session_fifo_lockp);
@@ -551,7 +687,7 @@
/* Recycling the event. */
clib_spinlock_lock (&(vcm->event_thread.events_lockp));
- vcm->event_thread.recycle_event = 1;
+ ev->recycle = 1;
clib_fifo_add1 (vcm->event_thread.event_index_fifo, reg->ev_idx);
clib_spinlock_unlock (&(vcm->event_thread.events_lockp));
}
@@ -811,8 +947,8 @@
(vcm->cfg.app_scope_global ? APP_OPTIONS_FLAGS_USE_GLOBAL_SCOPE : 0) |
(app_is_proxy ? APP_OPTIONS_FLAGS_IS_PROXY : 0);
bmp->options[APP_OPTIONS_PROXY_TRANSPORT] =
- (vcm->cfg.app_proxy_transport_tcp ? 1 << TRANSPORT_PROTO_TCP : 0) |
- (vcm->cfg.app_proxy_transport_udp ? 1 << TRANSPORT_PROTO_UDP : 0);
+ (u64) ((vcm->cfg.app_proxy_transport_tcp ? 1 << TRANSPORT_PROTO_TCP : 0) |
+ (vcm->cfg.app_proxy_transport_udp ? 1 << TRANSPORT_PROTO_UDP : 0));
bmp->options[APP_OPTIONS_SEGMENT_SIZE] = vcm->cfg.segment_size;
bmp->options[APP_OPTIONS_ADD_SEGMENT_SIZE] = vcm->cfg.add_segment_size;
bmp->options[APP_OPTIONS_RX_FIFO_SIZE] = vcm->cfg.rx_fifo_size;
@@ -1085,6 +1221,16 @@
/*
* Setup session
*/
+ if (vcm->session_io_thread.io_sessions_lockp)
+ {
+ // Add this connection to the active io sessions list
+ clib_spinlock_lock (&vcm->session_io_thread.io_sessions_lockp);
+ u32 *active_session_index;
+ pool_get (vcm->session_io_thread.active_session_indexes,
+ active_session_index);
+ *active_session_index = session_index;
+ clib_spinlock_unlock (&vcm->session_io_thread.io_sessions_lockp);
+ }
session->vpp_event_queue = uword_to_pointer (mp->vpp_event_queue_address,
svm_queue_t *);
@@ -1360,7 +1506,7 @@
/* Allocate local session and set it up */
pool_get (vcm->sessions, session);
memset (session, 0, sizeof (*session));
- session_index = session - vcm->sessions;
+ session_index = (u32) (session - vcm->sessions);
rx_fifo = uword_to_pointer (mp->server_rx_fifo, svm_fifo_t *);
rx_fifo->client_session_index = session_index;
@@ -1388,10 +1534,8 @@
clib_spinlock_lock (&vcm->event_thread.events_lockp);
pool_get (vcm->event_thread.vce_events, ev);
- ev->data = clib_mem_alloc (sizeof (vce_event_connect_request_t));
- ev->refcnt = 0;
ev_idx = (u32) (ev - vcm->event_thread.vce_events);
- ecr = ev->data;
+ ecr = vce_get_event_data (ev, sizeof (*ecr));
ev->evk.eid = VCL_EVENT_CONNECT_REQ_ACCEPTED;
listen_session = vppcom_session_table_lookup_listener (mp->listener_handle);
ev->evk.session_index = (u32) (listen_session - vcm->sessions);
@@ -1400,7 +1544,6 @@
clib_spinlock_unlock (&vcm->event_thread.events_lockp);
rv = vce_generate_event (&vcm->event_thread, ev_idx);
-
ASSERT (rv == 0);
if (VPPCOM_DEBUG > 1)
@@ -1454,6 +1597,9 @@
}
+/* VPP combines bind and listen as one operation. VCL manages the separation
+ * of bind and listen locally via vppcom_session_bind() and
+ * vppcom_session_listen() */
static void
vppcom_send_bind_sock (session_t * session, u32 session_index)
{
@@ -2194,6 +2340,7 @@
vcm->cfg.listen_queue_size);
vppcom_cfg_read (conf_fname);
+
env_var_str = getenv (VPPCOM_ENV_API_PREFIX);
if (env_var_str)
{
@@ -2310,7 +2457,6 @@
rv = vce_start_event_thread (&(vcm->event_thread), 20);
-
if (VPPCOM_DEBUG > 0)
clib_warning ("VCL<%d>: sending session enable", getpid ());
@@ -2704,7 +2850,7 @@
if (VPPCOM_DEBUG > 0)
clib_warning ("VCL<%d>: vpp handle 0x%llx, "
- "sid %u: sending bind request...",
+ "sid %u: sending VPP bind+listen request...",
getpid (), listen_vpp_handle, listen_session_index);
vppcom_send_bind_sock (listen_session, listen_session_index);
@@ -2717,10 +2863,10 @@
if (PREDICT_FALSE (retval))
{
if (VPPCOM_DEBUG > 0)
- clib_warning ("VCL<%d>: vpp handle 0x%llx, sid %u: bind failed! "
- "returning %d (%s)", getpid (),
- listen_session->vpp_handle, listen_session_index,
- retval, vppcom_retval_str (retval));
+ clib_warning
+ ("VCL<%d>: vpp handle 0x%llx, sid %u: bind+listen failed! "
+ "returning %d (%s)", getpid (), listen_session->vpp_handle,
+ listen_session_index, retval, vppcom_retval_str (retval));
clib_spinlock_unlock (&vcm->sessions_lockp);
rv = retval;
goto done;
@@ -2746,15 +2892,21 @@
vce_event_key_t evk;
vppcom_session_listener_t *listener_args;
+ if (!vcm->session_io_thread.io_sessions_lockp)
+ rv = vppcom_start_io_event_thread (&vcm->session_io_thread, 100 /* DAW_TODO: ??? hard-coded value */
+ );
+ if (rv)
+ {
+ goto done;
+ }
rv = vppcom_session_listen (session_index, q_len);
if (rv)
{
goto done;
}
-
/* Register handler for connect_request event on listen_session_index */
- listener_args = clib_mem_alloc (sizeof (vppcom_session_listener_t));
+ listener_args = clib_mem_alloc (sizeof (vppcom_session_listener_t)); // DAW_TODO: Use a pool instead of thrashing the memory allocator!
listener_args->user_cb = cb;
listener_args->user_cb_data = ptr;
listener_args->user_errcb = errcb;
@@ -2841,23 +2993,25 @@
evk.eid = VCL_EVENT_CONNECT_REQ_ACCEPTED;
reg = vce_register_handler (&vcm->event_thread, &evk,
vce_connect_request_handler_fn, 0);
+ clib_spinlock_lock (&(vcm->event_thread.events_lockp));
ev = vce_get_event_from_index (&vcm->event_thread, reg->ev_idx);
pthread_mutex_lock (®->handler_lock);
while (!ev)
{
- rv =
- pthread_cond_timedwait (®->handler_cond, ®->handler_lock, &ts);
+ clib_spinlock_unlock (&(vcm->event_thread.events_lockp));
+ rv = pthread_cond_timedwait (®->handler_cond,
+ ®->handler_lock, &ts);
if (rv == ETIMEDOUT)
{
rv = VPPCOM_EAGAIN;
goto cleanup;
}
+ clib_spinlock_lock (&(vcm->event_thread.events_lockp));
ev = vce_get_event_from_index (&vcm->event_thread, reg->ev_idx);
}
- result = (vce_event_connect_request_t *) ev->data;
+ result = vce_get_event_data (ev, sizeof (*result));
client_session_index = result->accepted_session_index;
-
-
+ clib_spinlock_unlock (&(vcm->event_thread.events_lockp));
/* Remove from the FIFO used to service epoll */
clib_spinlock_lock (&vcm->session_fifo_lockp);
@@ -2982,8 +3136,17 @@
clib_spinlock_unlock (&vcm->sessions_lockp);
rv = (int) client_session_index;
- vce_clear_event (&vcm->event_thread, ev);
-
+ vce_clear_event (&vcm->event_thread, reg->ev_idx);
+ if (vcm->session_io_thread.io_sessions_lockp)
+ {
+ /* Throw this new accepted session index into the rx poll thread pool */
+ clib_spinlock_lock (&vcm->session_io_thread.io_sessions_lockp);
+ u32 *active_session_index;
+ pool_get (vcm->session_io_thread.active_session_indexes,
+ active_session_index);
+ *active_session_index = client_session_index;
+ clib_spinlock_unlock (&vcm->session_io_thread.io_sessions_lockp);
+ }
cleanup:
vce_unregister_handler (&vcm->event_thread, reg);
pthread_mutex_unlock (®->handler_lock);
@@ -3285,6 +3448,41 @@
}
int
+vppcom_session_register_ioevent_cb (uint32_t session_index,
+ vppcom_session_ioevent_cb cb,
+ uint8_t rx, void *ptr)
+{
+ int rv = VPPCOM_OK;
+ vce_event_key_t evk;
+ vppcom_session_ioevent_t *ioevent;
+
+ if (!vcm->session_io_thread.io_sessions_lockp)
+ rv = vppcom_start_io_event_thread (&vcm->session_io_thread, 100 /* DAW_TODO: ??? hard-coded value */
+ );
+
+ if (rv == VPPCOM_OK)
+ {
+ void *io_evt_ndx;
+
+ /* Register handler for ioevent on session_index */
+ clib_spinlock_lock (&vcm->session_io_thread.io_sessions_lockp);
+ pool_get (vcm->session_io_thread.ioevents, ioevent);
+ io_evt_ndx = (void *) (ioevent - vcm->session_io_thread.ioevents);
+ ioevent->user_cb = cb;
+ ioevent->user_cb_data = ptr;
+ clib_spinlock_unlock (&vcm->session_io_thread.io_sessions_lockp);
+
+ evk.session_index = session_index;
+ evk.eid = rx ? VCL_EVENT_IOEVENT_RX_FIFO : VCL_EVENT_IOEVENT_TX_FIFO;
+
+ (void) vce_register_handler (&vcm->event_thread, &evk,
+ vce_registered_ioevent_handler_fn,
+ io_evt_ndx);
+ }
+ return rv;
+}
+
+int
vppcom_session_write (uint32_t session_index, void *buf, size_t n)
{
session_t *session = 0;
@@ -3339,7 +3537,12 @@
}
while (!is_nonblocking && (n_write <= 0));
- /* If event wasn't set, add one */
+ /* If event wasn't set, add one
+ *
+ * To reduce context switching, can check if an
+ * event is already there for this event_key, but for now
+ * this will suffice. */
+
if ((n_write > 0) && svm_fifo_set_event (tx_fifo))
{
/* Fabricate TX event, send to vpp */
diff --git a/src/vcl/vppcom.h b/src/vcl/vppcom.h
index 34a69b2..c752e50 100644
--- a/src/vcl/vppcom.h
+++ b/src/vcl/vppcom.h
@@ -144,6 +144,13 @@
short *revents;
} vcl_poll_t;
+typedef struct vppcom_ioevent_
+{
+ uint32_t session_index;
+ size_t bytes;
+} vppcom_ioevent_t;
+
+
/*
* VPPCOM Public API Functions
*/
@@ -221,6 +228,34 @@
void *);
/**
+ * User registered callback for IO events (rx/tx)
+ * @param vppcom_ioevent_t* -
+ * @param void* - user passed arg to pass back
+ */
+typedef void (*vppcom_session_ioevent_cb) (vppcom_ioevent_t *, void *);
+
+/**
+ * @brief vppcom_session_register_listener accepts a bound session_index, and
+ * listens for connections.
+ *
+ * On successful connection, calls registered callback (cb) with new
+ * session_index.
+ *
+ * On error, calls registered error callback (errcb).
+ *
+ * @param session_index - bound session_index to create listener on
+ * @param cb - on new accepted session callback
+ * @param errcb - on failure callback
+ * @param flags - placeholder for future use. Must be ZERO
+ * @param q_len - max listener connection backlog
+ * @param ptr - user data
+ * @return
+ */
+extern int vppcom_session_register_ioevent_cb (uint32_t session_index,
+ vppcom_session_ioevent_cb cb,
+ uint8_t rx, void *ptr);
+
+/**
* User registered ERROR callback for any errors associated with
* handling vppcom_session_register_listener() and connections
* @param void* - user passed arg to pass back