#include <stdlib.h>
#include <svm/svm_fifo_segment.h>
#include <vcl/vppcom.h>
-#include <vcl/vcl_event.h>
#include <vcl/vcl_debug.h>
#include <vcl/vcl_private.h>
session->session_state = STATE_ACCEPT;
session->transport.rmt_port = mp->port;
session->transport.is_ip4 = mp->is_ip4;
- clib_memcpy (&session->transport.rmt_ip, mp->ip, sizeof (ip46_address_t));
+ clib_memcpy_fast (&session->transport.rmt_ip, mp->ip,
+ sizeof (ip46_address_t));
vcl_session_table_add_vpp_handle (wrk, mp->handle, session->session_index);
session->transport.lcl_port = listen_session->transport.lcl_port;
session->tx_fifo = tx_fifo;
session->vpp_handle = mp->handle;
session->transport.is_ip4 = mp->is_ip4;
- clib_memcpy (&session->transport.lcl_ip, mp->lcl_ip,
- sizeof (session->transport.lcl_ip));
+ clib_memcpy_fast (&session->transport.lcl_ip, mp->lcl_ip,
+ sizeof (session->transport.lcl_ip));
session->transport.lcl_port = mp->lcl_port;
session->session_state = STATE_CONNECT;
session->vpp_handle = mp->handle;
session->transport.is_ip4 = mp->lcl_is_ip4;
- clib_memcpy (&session->transport.lcl_ip, mp->lcl_ip,
- sizeof (ip46_address_t));
+ clib_memcpy_fast (&session->transport.lcl_ip, mp->lcl_ip,
+ sizeof (ip46_address_t));
session->transport.lcl_port = mp->lcl_port;
vcl_session_table_add_listener (wrk, mp->handle, sid);
session->session_state = STATE_LISTEN;
return VPPCOM_OK;
}
+static void
+vcl_cleanup_bapi (void)
+{
+ api_main_t *am = &api_main;
+
+ am->my_client_index = ~0;
+ am->my_registration = 0;
+ am->vl_input_queue = 0;
+ am->msg_index_by_name_and_crc = 0;
+
+ vl_client_api_unmap ();
+}
+
+void
+vcl_app_fork_child_handler (void)
+{
+ u8 *child_name;
+ int rv;
+
+ vcm->current_pid = getpid ();
+ vcl_set_worker_index (0);
+
+ VDBG (0, "initializing forked child");
+ child_name = format (0, "%v-child-%u%c", vcm->app_name, getpid (), 0);
+
+ vcl_cleanup_bapi ();
+ vppcom_api_hookup ();
+ vcm->app_state = STATE_APP_START;
+ rv = vppcom_connect_to_vpp ((char *) child_name);
+ vec_free (child_name);
+ if (rv)
+ {
+ VERR ("couldn't connect to VPP!");
+ return;
+ }
+
+ vcm->app_state = STATE_APP_ADDING_WORKER;
+ vcl_send_app_worker_add_del (1 /* is_add */ );
+ if (vcl_wait_for_app_state_change (STATE_APP_READY))
+ {
+ VERR ("failed to add worker to vpp");
+ return;
+ }
+ VDBG (0, "forked child main worker initialized");
+}
+
/*
* VPPCOM Public API functions
*/
vcl_cfg = &vcm->cfg;
vcm->main_cpu = pthread_self ();
+ vcm->main_pid = vcm->current_pid = getpid ();
+ vcm->app_name = format (0, "%s", app_name);
vppcom_init_error_string_table ();
svm_fifo_segment_main_init (vcl_cfg->segment_baseva,
20 /* timeout in secs */ );
pool_init_fixed (vcm->workers, vcl_cfg->max_workers);
clib_spinlock_init (&vcm->workers_lock);
vcl_worker_alloc_and_init ();
+ pthread_atfork (NULL, NULL, vcl_app_fork_child_handler);
}
if (vcm->my_client_index == ~0)
rv = vppcom_connect_to_vpp (app_name);
if (rv)
{
- clib_warning ("VCL<%d>: ERROR: couldn't connect to VPP!",
- getpid ());
+ VERR ("couldn't connect to VPP!");
return rv;
}
-
- VDBG (0, "VCL<%d>: sending session enable", getpid ());
+ VDBG (0, "sending session enable");
rv = vppcom_app_session_enable ();
if (rv)
{
- clib_warning ("VCL<%d>: ERROR: vppcom_app_session_enable() "
- "failed!", getpid ());
+ VERR ("vppcom_app_session_enable() failed!");
return rv;
}
- VDBG (0, "VCL<%d>: sending app attach", getpid ());
+ VDBG (0, "sending app attach");
rv = vppcom_app_attach ();
if (rv)
{
- clib_warning ("VCL<%d>: ERROR: vppcom_app_attach() failed!",
- getpid ());
+ VERR ("vppcom_app_attach() failed!");
return rv;
}
- VDBG (0, "VCL<%d>: app_name '%s', my_client_index %d (0x%x)",
- getpid (), app_name, vcm->my_client_index, vcm->my_client_index);
+ VDBG (0, "app_name '%s', my_client_index %d (0x%x)",
+ app_name, vcm->my_client_index, vcm->my_client_index);
}
return VPPCOM_OK;
if (vcm->my_client_index == ~0)
return;
- VDBG (0, "VCL<%d>: detaching from VPP, my_client_index %d (0x%x)",
- getpid (), vcm->my_client_index, vcm->my_client_index);
+ VDBG (0, "detaching from VPP, my_client_index %d (0x%x)",
+ vcm->my_client_index, vcm->my_client_index);
vcl_evt (VCL_EVT_DETACH, vcm);
vppcom_app_send_detach ();
rv = vcl_wait_for_app_state_change (STATE_APP_ENABLED);
vcm->cfg.app_timeout = orig_app_timeout;
if (PREDICT_FALSE (rv))
- VDBG (0, "VCL<%d>: application detach timed out! returning %d (%s)",
- getpid (), rv, vppcom_retval_str (rv));
+ VDBG (0, "application detach timed out! returning %d (%s)",
+ rv, vppcom_retval_str (rv));
vcl_elog_stop (vcm);
vl_client_disconnect_from_vlib ();
+ vec_free (vcm->app_name);
vcm->my_client_index = ~0;
vcm->app_state = STATE_APP_START;
}
vcl_evt (VCL_EVT_CREATE, session, session_type, session->session_state,
is_nonblocking, session_index);
- VDBG (0, "VCL<%d>: sid %u", getpid (), session->session_index);
+ VDBG (0, "created sid %u", session->session_index);
return vcl_session_handle (session);
}
session->transport.is_ip4 = ep->is_ip4;
if (ep->is_ip4)
- clib_memcpy (&session->transport.lcl_ip.ip4, ep->ip,
- sizeof (ip4_address_t));
+ clib_memcpy_fast (&session->transport.lcl_ip.ip4, ep->ip,
+ sizeof (ip4_address_t));
else
- clib_memcpy (&session->transport.lcl_ip.ip6, ep->ip,
- sizeof (ip6_address_t));
+ clib_memcpy_fast (&session->transport.lcl_ip.ip6, ep->ip,
+ sizeof (ip6_address_t));
session->transport.lcl_port = ep->port;
VDBG (0, "VCL<%d>: sid %u: binding to local %s address %U port %u, "
svm_msg_q_free_msg (wrk->app_event_queue, &msg);
continue;
}
- clib_memcpy (&accepted_msg, e->data, sizeof (accepted_msg));
+ clib_memcpy_fast (&accepted_msg, e->data, sizeof (accepted_msg));
svm_msg_q_free_msg (wrk->app_event_queue, &msg);
break;
}
ep->is_ip4 = client_session->transport.is_ip4;
ep->port = client_session->transport.rmt_port;
if (client_session->transport.is_ip4)
- clib_memcpy (ep->ip, &client_session->transport.rmt_ip.ip4,
- sizeof (ip4_address_t));
+ clib_memcpy_fast (ep->ip, &client_session->transport.rmt_ip.ip4,
+ sizeof (ip4_address_t));
else
- clib_memcpy (ep->ip, &client_session->transport.rmt_ip.ip6,
- sizeof (ip6_address_t));
+ clib_memcpy_fast (ep->ip, &client_session->transport.rmt_ip.ip6,
+ sizeof (ip6_address_t));
}
if (accepted_msg.server_event_queue_address)
session->transport.is_ip4 = server_ep->is_ip4;
if (session->transport.is_ip4)
- clib_memcpy (&session->transport.rmt_ip.ip4, server_ep->ip,
- sizeof (ip4_address_t));
+ clib_memcpy_fast (&session->transport.rmt_ip.ip4, server_ep->ip,
+ sizeof (ip4_address_t));
else
- clib_memcpy (&session->transport.rmt_ip.ip6, server_ep->ip,
- sizeof (ip6_address_t));
+ clib_memcpy_fast (&session->transport.rmt_ip.ip6, server_ep->ip,
+ sizeof (ip6_address_t));
session->transport.rmt_port = server_ep->port;
VDBG (0, "VCL<%d>: vpp handle 0x%llx, sid %u: connecting to server %s %U "
is_ct = vcl_session_is_ct (s);
mq = is_ct ? s->our_evt_q : wrk->app_event_queue;
rx_fifo = s->rx_fifo;
+ s->has_rx_evt = 0;
if (svm_fifo_is_empty (rx_fifo))
{
if (is_nonblocking)
{
svm_fifo_unset_event (rx_fifo);
- return VPPCOM_OK;
+ return VPPCOM_EWOULDBLOCK;
}
while (svm_fifo_is_empty (rx_fifo))
{
if (svm_fifo_is_empty (rx_fifo))
svm_fifo_unset_event (rx_fifo);
- if (is_ct && n_read + svm_fifo_max_dequeue (rx_fifo) == rx_fifo->nitems)
+ if (is_ct && svm_fifo_want_tx_evt (rx_fifo))
{
- /* If the peer is not polling send notification */
- if (!svm_fifo_has_event (s->rx_fifo))
- app_send_io_evt_to_vpp (s->vpp_evt_q, s->rx_fifo,
- SESSION_IO_EVT_CT_RX, SVM_Q_WAIT);
+ svm_fifo_set_want_tx_evt (s->rx_fifo, 0);
+ app_send_io_evt_to_vpp (s->vpp_evt_q, s->rx_fifo, SESSION_IO_EVT_CT_RX,
+ SVM_Q_WAIT);
}
VDBG (2, "VCL<%d>: vpp handle 0x%llx, sid %u: read %d bytes from (%p)",
is_ct = vcl_session_is_ct (s);
mq = is_ct ? s->our_evt_q : wrk->app_event_queue;
rx_fifo = s->rx_fifo;
+ s->has_rx_evt = 0;
if (svm_fifo_is_empty (rx_fifo))
{
if (is_nonblocking)
{
svm_fifo_unset_event (rx_fifo);
- return VPPCOM_OK;
+ return VPPCOM_EWOULDBLOCK;
}
while (svm_fifo_is_empty (rx_fifo))
{
vppcom_data_segment_copy (void *buf, vppcom_data_segments_t ds, u32 max_bytes)
{
u32 first_copy = clib_min (ds[0].len, max_bytes);
- clib_memcpy (buf, ds[0].data, first_copy);
+ clib_memcpy_fast (buf, ds[0].data, first_copy);
if (first_copy < max_bytes)
{
- clib_memcpy (buf + first_copy, ds[1].data,
- clib_min (ds[1].len, max_bytes - first_copy));
+ clib_memcpy_fast (buf + first_copy, ds[1].data,
+ clib_min (ds[1].len, max_bytes - first_copy));
}
return 0;
}
svm_msg_q_msg_t msg;
session_event_t *e;
svm_msg_q_t *mq;
+ u8 is_ct;
if (PREDICT_FALSE (!buf))
return VPPCOM_EINVAL;
if (PREDICT_FALSE (!s))
return VPPCOM_EBADFD;
- tx_fifo = s->tx_fifo;
- is_nonblocking = VCL_SESS_ATTR_TEST (s->attr, VCL_SESS_ATTR_NONBLOCK);
-
if (PREDICT_FALSE (s->is_vep))
{
clib_warning ("VCL<%d>: ERROR: vpp handle 0x%llx, sid %u: "
return VPPCOM_EBADFD;
}
- if (!(s->session_state & STATE_OPEN))
+ if (PREDICT_FALSE (!(s->session_state & STATE_OPEN)))
{
session_state_t state = s->session_state;
rv = ((state & STATE_DISCONNECT) ? VPPCOM_ECONNRESET : VPPCOM_ENOTCONN);
VDBG (1, "VCL<%d>: vpp handle 0x%llx, sid %u: session is not open! "
- "state 0x%x (%s)",
- getpid (), s->vpp_handle, session_handle,
+ "state 0x%x (%s)", getpid (), s->vpp_handle, session_handle,
state, vppcom_session_state_str (state));
return rv;
}
- mq = vcl_session_is_ct (s) ? s->our_evt_q : wrk->app_event_queue;
+ tx_fifo = s->tx_fifo;
+ is_ct = vcl_session_is_ct (s);
+ is_nonblocking = VCL_SESS_ATTR_TEST (s->attr, VCL_SESS_ATTR_NONBLOCK);
+ mq = is_ct ? s->our_evt_q : wrk->app_event_queue;
if (svm_fifo_is_full (tx_fifo))
{
if (is_nonblocking)
}
while (svm_fifo_is_full (tx_fifo))
{
+ svm_fifo_set_want_tx_evt (tx_fifo, 1);
svm_msg_q_lock (mq);
- while (svm_msg_q_is_empty (mq) && svm_msg_q_timedwait (mq, 10e-6))
- ;
+ if (svm_msg_q_is_empty (mq))
+ svm_msg_q_wait (mq);
+
svm_msg_q_sub_w_lock (mq, &msg);
e = svm_msg_q_msg_data (mq, &msg);
svm_msg_q_unlock (mq);
- if (!vcl_is_tx_evt_for_session (e, s->session_index,
- s->our_evt_q != 0))
+ if (!vcl_is_tx_evt_for_session (e, s->session_index, is_ct))
vcl_handle_mq_event (wrk, e);
svm_msg_q_free_msg (mq, &msg);
}
ASSERT (n_write > 0);
- if (VPPCOM_DEBUG > 2)
- {
- if (n_write <= 0)
- clib_warning ("VCL<%d>: vpp handle 0x%llx, sid %u: "
- "FIFO-FULL (%p)", getpid (), s->vpp_handle,
- session_handle, tx_fifo);
- else
- clib_warning ("VCL<%d>: vpp handle 0x%llx, sid %u: "
- "wrote %d bytes tx-fifo: (%p)", getpid (),
- s->vpp_handle, session_handle, n_write, tx_fifo);
- }
+ VDBG (2, "VCL<%d>: vpp handle 0x%llx, sid %u: wrote %d bytes", getpid (),
+ s->vpp_handle, session_handle, n_write);
+
return n_write;
}
if (n_bits && read_map)
{
clib_bitmap_validate (wrk->rd_bitmap, minbits);
- clib_memcpy (wrk->rd_bitmap, read_map,
- vec_len (wrk->rd_bitmap) * sizeof (clib_bitmap_t));
+ clib_memcpy_fast (wrk->rd_bitmap, read_map,
+ vec_len (wrk->rd_bitmap) * sizeof (clib_bitmap_t));
memset (read_map, 0, vec_len (wrk->rd_bitmap) * sizeof (clib_bitmap_t));
}
if (n_bits && write_map)
{
clib_bitmap_validate (wrk->wr_bitmap, minbits);
- clib_memcpy (wrk->wr_bitmap, write_map,
- vec_len (wrk->wr_bitmap) * sizeof (clib_bitmap_t));
+ clib_memcpy_fast (wrk->wr_bitmap, write_map,
+ vec_len (wrk->wr_bitmap) * sizeof (clib_bitmap_t));
memset (write_map, 0,
vec_len (wrk->wr_bitmap) * sizeof (clib_bitmap_t));
}
if (n_bits && except_map)
{
clib_bitmap_validate (wrk->ex_bitmap, minbits);
- clib_memcpy (wrk->ex_bitmap, except_map,
- vec_len (wrk->ex_bitmap) * sizeof (clib_bitmap_t));
+ clib_memcpy_fast (wrk->ex_bitmap, except_map,
+ vec_len (wrk->ex_bitmap) * sizeof (clib_bitmap_t));
memset (except_map, 0,
vec_len (wrk->ex_bitmap) * sizeof (clib_bitmap_t));
}
sid = e->fifo->client_session_index;
session = vcl_session_get (wrk, sid);
session_events = session->vep.ev.events;
- if (!(EPOLLIN & session->vep.ev.events))
+ if (!(EPOLLIN & session->vep.ev.events) || session->has_rx_evt)
break;
add_event = 1;
events[*num_ev].events |= EPOLLIN;
session_evt_data = session->vep.ev.data.u64;
+ session->has_rx_evt = 1;
break;
case FIFO_EVENT_APP_TX:
sid = e->fifo->client_session_index;
session = vcl_ct_session_get_from_fifo (wrk, e->fifo, 0);
sid = session->session_index;
session_events = session->vep.ev.events;
- if (!(EPOLLIN & session->vep.ev.events))
+ if (!(EPOLLIN & session->vep.ev.events) || session->has_rx_evt)
break;
add_event = 1;
events[*num_ev].events |= EPOLLIN;
session_evt_data = session->vep.ev.data.u64;
+ session->has_rx_evt = 1;
break;
case SESSION_IO_EVT_CT_RX:
session = vcl_ct_session_get_from_fifo (wrk, e->fifo, 1);
session_event_t *e;
int i;
+ if (vec_len (wrk->mq_msg_vector) && svm_msg_q_is_empty (mq))
+ goto handle_dequeued;
+
svm_msg_q_lock (mq);
if (svm_msg_q_is_empty (mq))
{
vcl_mq_dequeue_batch (wrk, mq);
svm_msg_q_unlock (mq);
+handle_dequeued:
for (i = 0; i < vec_len (wrk->mq_msg_vector); i++)
{
msg = vec_elt_at_index (wrk->mq_msg_vector, i);
e = svm_msg_q_msg_data (mq, msg);
- vcl_epoll_wait_handle_mq_event (wrk, e, events, num_ev);
+ if (*num_ev < maxevents)
+ vcl_epoll_wait_handle_mq_event (wrk, e, events, num_ev);
+ else
+ vec_add1 (wrk->unhandled_evts_vector, *e);
svm_msg_q_free_msg (mq, msg);
- if (*num_ev == maxevents)
- {
- i += 1;
- break;
- }
}
-
- vec_delete (wrk->mq_msg_vector, i, 0);
+ vec_reset_length (wrk->mq_msg_vector);
return *num_ev;
}
u64 buf;
vec_validate (wrk->mq_events, pool_elts (wrk->mq_evt_conns));
+again:
n_mq_evts = epoll_wait (wrk->mqs_epfd, wrk->mq_events,
vec_len (wrk->mq_events), wait_for_time);
for (i = 0; i < n_mq_evts; i++)
n_read = read (mqc->mq_fd, &buf, sizeof (buf));
vcl_epoll_wait_handle_mq (wrk, mqc->mq, events, maxevents, 0, &n_evts);
}
+ if (!n_evts && n_mq_evts > 0)
+ goto again;
return (int) n_evts;
}
ep->is_ip4 = session->transport.is_ip4;
ep->port = session->transport.rmt_port;
if (session->transport.is_ip4)
- clib_memcpy (ep->ip, &session->transport.rmt_ip.ip4,
- sizeof (ip4_address_t));
+ clib_memcpy_fast (ep->ip, &session->transport.rmt_ip.ip4,
+ sizeof (ip4_address_t));
else
- clib_memcpy (ep->ip, &session->transport.rmt_ip.ip6,
- sizeof (ip6_address_t));
+ clib_memcpy_fast (ep->ip, &session->transport.rmt_ip.ip6,
+ sizeof (ip6_address_t));
*buflen = sizeof (*ep);
VDBG (1, "VCL<%d>: VPPCOM_ATTR_GET_PEER_ADDR: sid %u, is_ip4 = %u, "
"addr = %U, port %u", getpid (),
ep->is_ip4 = session->transport.is_ip4;
ep->port = session->transport.lcl_port;
if (session->transport.is_ip4)
- clib_memcpy (ep->ip, &session->transport.lcl_ip.ip4,
- sizeof (ip4_address_t));
+ clib_memcpy_fast (ep->ip, &session->transport.lcl_ip.ip4,
+ sizeof (ip4_address_t));
else
- clib_memcpy (ep->ip, &session->transport.lcl_ip.ip6,
- sizeof (ip6_address_t));
+ clib_memcpy_fast (ep->ip, &session->transport.lcl_ip.ip6,
+ sizeof (ip6_address_t));
*buflen = sizeof (*ep);
VDBG (1, "VCL<%d>: VPPCOM_ATTR_GET_LCL_ADDR: sid %u, is_ip4 = %u,"
" addr = %U port %d", getpid (),
if (ep)
{
if (session->transport.is_ip4)
- clib_memcpy (ep->ip, &session->transport.rmt_ip.ip4,
- sizeof (ip4_address_t));
+ clib_memcpy_fast (ep->ip, &session->transport.rmt_ip.ip4,
+ sizeof (ip4_address_t));
else
- clib_memcpy (ep->ip, &session->transport.rmt_ip.ip6,
- sizeof (ip6_address_t));
+ clib_memcpy_fast (ep->ip, &session->transport.rmt_ip.ip6,
+ sizeof (ip6_address_t));
}
return rv;
vcl_worker_t *wrk = vcl_worker_get_current ();
f64 timeout = clib_time_now (&wrk->clib_time) + wait_for_time;
u32 i, keep_trying = 1;
+ svm_msg_q_msg_t msg;
+ session_event_t *e;
int rv, num_ev = 0;
VDBG (3, "VCL<%d>: vp %p, nsids %u, wait_for_time %f",
{
vcl_session_t *session;
- for (i = 0; i < n_sids; i++)
+ /* Dequeue all events and drop all unhandled io events */
+ while (svm_msg_q_sub (wrk->app_event_queue, &msg, SVM_Q_NOWAIT, 0) == 0)
{
- ASSERT (vp[i].revents);
+ e = svm_msg_q_msg_data (wrk->app_event_queue, &msg);
+ vcl_handle_mq_event (wrk, e);
+ svm_msg_q_free_msg (wrk->app_event_queue, &msg);
+ }
+ vec_reset_length (wrk->unhandled_evts_vector);
+ for (i = 0; i < n_sids; i++)
+ {
session = vcl_session_get (wrk, vp[i].sid);
if (!session)
- continue;
+ {
+ vp[i].revents = POLLHUP;
+ num_ev++;
+ continue;
+ }
- if (*vp[i].revents)
- *vp[i].revents = 0;
+ vp[i].revents = 0;
if (POLLIN & vp[i].events)
{
rv = vppcom_session_read_ready (session);
if (rv > 0)
{
- *vp[i].revents |= POLLIN;
+ vp[i].revents |= POLLIN;
num_ev++;
}
else if (rv < 0)
switch (rv)
{
case VPPCOM_ECONNRESET:
- *vp[i].revents = POLLHUP;
+ vp[i].revents = POLLHUP;
break;
default:
- *vp[i].revents = POLLERR;
+ vp[i].revents = POLLERR;
break;
}
num_ev++;
rv = vppcom_session_write_ready (session);
if (rv > 0)
{
- *vp[i].revents |= POLLOUT;
+ vp[i].revents |= POLLOUT;
num_ev++;
}
else if (rv < 0)
switch (rv)
{
case VPPCOM_ECONNRESET:
- *vp[i].revents = POLLHUP;
+ vp[i].revents = POLLHUP;
break;
default:
- *vp[i].revents = POLLERR;
+ vp[i].revents = POLLERR;
break;
}
num_ev++;
if (0) // Note "done:" label used by VCL_SESSION_LOCK_AND_GET()
{
- *vp[i].revents = POLLNVAL;
+ vp[i].revents = POLLNVAL;
num_ev++;
}
}
{
clib_warning ("VCL<%d>: vp[%d].sid %d (0x%x), .events 0x%x, "
".revents 0x%x", getpid (), i, vp[i].sid, vp[i].sid,
- vp[i].events, *vp[i].revents);
+ vp[i].events, vp[i].revents);
}
}
return num_ev;
Code Review / vpp.git / blobdiff