clib_memcpy_fast (&mp->ip, &s->transport.lcl_ip, sizeof (mp->ip));
mp->port = s->transport.lcl_port;
mp->proto = s->session_type;
+ if (s->flags & VCL_SESSION_F_CONNECTED)
+ mp->flags = TRANSPORT_CFG_F_CONNECTED;
app_send_ctrl_evt_to_vpp (mq, app_evt);
}
app_send_ctrl_evt_to_vpp (mq, app_evt);
}
+int
+vcl_send_worker_rpc (u32 dst_wrk_index, void *data, u32 data_len)
+{
+ app_session_evt_t _app_evt, *app_evt = &_app_evt;
+ session_app_wrk_rpc_msg_t *mp;
+ vcl_worker_t *dst_wrk, *wrk;
+ svm_msg_q_t *mq;
+ int ret = -1;
+
+ if (data_len > sizeof (mp->data))
+ goto done;
+
+ clib_spinlock_lock (&vcm->workers_lock);
+
+ dst_wrk = vcl_worker_get_if_valid (dst_wrk_index);
+ if (!dst_wrk)
+ goto done;
+
+ wrk = vcl_worker_get_current ();
+ mq = vcl_worker_ctrl_mq (wrk);
+ app_alloc_ctrl_evt_to_vpp (mq, app_evt, SESSION_CTRL_EVT_APP_WRK_RPC);
+ mp = (session_app_wrk_rpc_msg_t *) app_evt->evt->data;
+ mp->client_index = wrk->my_client_index;
+ mp->wrk_index = dst_wrk->vpp_wrk_index;
+ clib_memcpy (mp->data, data, data_len);
+ app_send_ctrl_evt_to_vpp (mq, app_evt);
+ ret = 0;
+
+done:
+ clib_spinlock_unlock (&vcm->workers_lock);
+ return ret;
+}
+
static u32
vcl_session_accepted_handler (vcl_worker_t * wrk, session_accepted_msg_t * mp,
u32 ls_index)
return VCL_INVALID_SESSION_INDEX;
}
- session->session_state = STATE_DISCONNECT;
+ if (session->session_state != STATE_CLOSED)
+ session->session_state = STATE_DISCONNECT;
VDBG (0, "reset session %u [0x%llx]", sid, reset_msg->handle);
return sid;
}
return;
}
+ if (msg->type == SESSION_CLEANUP_TRANSPORT)
+ {
+ /* Transport was cleaned up before we confirmed close. Probably the
+ * app is still waiting for some data that cannot be delivered.
+ * Confirm close to make sure everything is cleaned up */
+ if (session->session_state == STATE_VPP_CLOSING)
+ vcl_session_cleanup (wrk, session, vcl_session_handle (session),
+ 1 /* do_disconnect */ );
+ return;
+ }
+
vcl_session_table_del_vpp_handle (wrk, msg->handle);
/* Should not happen. App did not close the connection so don't free it. */
if (session->session_state != STATE_CLOSED)
VDBG (1, "Unmapped segment: %d", msg->segment_handle);
}
+static void
+vcl_worker_rpc_handler (vcl_worker_t * wrk, void *data)
+{
+ if (!vcm->wrk_rpc_fn)
+ return;
+
+ (vcm->wrk_rpc_fn) (((session_app_wrk_rpc_msg_t *) data)->data);
+}
+
static int
vcl_handle_mq_event (vcl_worker_t * wrk, session_event_t * e)
{
case SESSION_CTRL_EVT_APP_DEL_SEGMENT:
vcl_session_app_del_segment_handler (wrk, e->data);
break;
+ case SESSION_CTRL_EVT_APP_WRK_RPC:
+ vcl_worker_rpc_handler (wrk, e->data);
+ break;
default:
clib_warning ("unhandled %u", e->event_type);
}
* VPPCOM Public API functions
*/
int
-vppcom_app_create (char *app_name)
+vppcom_app_create (const char *app_name)
{
vppcom_cfg_t *vcl_cfg = &vcm->cfg;
int rv;
void
vppcom_app_destroy (void)
{
+ vcl_worker_t *wrk, *current_wrk;
struct dlmallinfo mi;
- vcl_worker_t *wrk;
mspace heap;
if (!pool_elts (vcm->workers))
vcl_evt (VCL_EVT_DETACH, vcm);
- vcl_send_app_detach (vcl_worker_get_current ());
+ current_wrk = vcl_worker_get_current ();
/* *INDENT-OFF* */
pool_foreach (wrk, vcm->workers, ({
- vcl_worker_cleanup (wrk, 0 /* notify vpp */ );
+ if (current_wrk != wrk)
+ vcl_worker_cleanup (wrk, 0 /* notify vpp */ );
}));
/* *INDENT-ON* */
+ vcl_send_app_detach (current_wrk);
+ vppcom_disconnect_from_vpp ();
+ vcl_worker_cleanup (current_wrk, 0 /* notify vpp */ );
+
vcl_elog_stop (vcm);
- vl_client_disconnect_from_vlib ();
/*
* Free the heap and fix vcm
munmap (mspace_least_addr (heap), mi.arena);
vcm = &_vppcom_main;
+ vcm->is_init = 0;
}
int
*proto = VPPCOM_PROTO_UDP;
else if (!strcmp (proto_str, "udp"))
*proto = VPPCOM_PROTO_UDP;
- else if (!strcmp (proto_str, "UDPC"))
- *proto = VPPCOM_PROTO_UDPC;
- else if (!strcmp (proto_str, "udpc"))
- *proto = VPPCOM_PROTO_UDPC;
else if (!strcmp (proto_str, "TLS"))
*proto = VPPCOM_PROTO_TLS;
else if (!strcmp (proto_str, "tls"))
n_read = app_recv_stream_raw (rx_fifo, buf, n, 0, peek);
if (svm_fifo_is_empty_cons (rx_fifo))
- svm_fifo_unset_event (s->rx_fifo);
+ {
+ svm_fifo_unset_event (s->rx_fifo);
+ if (!svm_fifo_is_empty_cons (rx_fifo)
+ && svm_fifo_set_event (s->rx_fifo) && is_nonblocking)
+ {
+ session_event_t *e;
+ vec_add2 (wrk->unhandled_evts_vector, e, 1);
+ e->event_type = SESSION_IO_EVT_RX;
+ e->session_index = s->session_index;
+ }
+ }
/* Cut-through sessions might request tx notifications on rx fifos */
if (PREDICT_FALSE (rx_fifo->want_deq_ntf))
} \
else if (svm_fifo_is_empty (_s->ct_rx_fifo)) \
{ \
- svm_fifo_unset_event (_s->ct_rx_fifo); \
+ svm_fifo_unset_event (_s->rx_fifo); /* rx evts on actual fifo*/ \
if (svm_fifo_is_empty (_s->ct_rx_fifo)) \
break; \
} \
case SESSION_CTRL_EVT_APP_DEL_SEGMENT:
vcl_session_app_del_segment_handler (wrk, e->data);
break;
+ case SESSION_CTRL_EVT_APP_WRK_RPC:
+ vcl_worker_rpc_handler (wrk, e->data);
+ break;
default:
clib_warning ("unhandled: %u", e->event_type);
break;
clib_bitmap_foreach (sid, wrk->wr_bitmap, ({
if (!(session = vcl_session_get (wrk, sid)))
{
- if (except_map && sid < minbits)
- clib_bitmap_set_no_check (except_map, sid, 1);
- continue;
+ clib_bitmap_set_no_check ((uword*)write_map, sid, 1);
+ bits_set++;
+ continue;
}
- if (vcl_session_write_ready (session) > 0)
+ if (vcl_session_write_ready (session))
{
clib_bitmap_set_no_check ((uword*)write_map, sid, 1);
bits_set++;
clib_bitmap_foreach (sid, wrk->rd_bitmap, ({
if (!(session = vcl_session_get (wrk, sid)))
{
- if (except_map && sid < minbits)
- clib_bitmap_set_no_check (except_map, sid, 1);
- continue;
+ clib_bitmap_set_no_check ((uword*)read_map, sid, 1);
+ bits_set++;
+ continue;
}
- if (vcl_session_read_ready (session) > 0)
+ if (vcl_session_read_ready (session))
{
clib_bitmap_set_no_check ((uword*)read_map, sid, 1);
bits_set++;
{
case SESSION_IO_EVT_RX:
sid = e->session_index;
- if (!(session = vcl_session_get (wrk, sid)))
+ session = vcl_session_get (wrk, sid);
+ if (vcl_session_is_closed (session))
break;
vcl_fifo_rx_evt_valid_or_break (session);
session_events = session->vep.ev.events;
break;
case SESSION_IO_EVT_TX:
sid = e->session_index;
- if (!(session = vcl_session_get (wrk, sid)))
+ session = vcl_session_get (wrk, sid);
+ if (vcl_session_is_closed (session))
break;
session_events = session->vep.ev.events;
if (!(EPOLLOUT & session_events))
connected_msg = (session_connected_msg_t *) e->data;
sid = vcl_session_connected_handler (wrk, connected_msg);
/* Generate EPOLLOUT because there's no connected event */
- if (!(session = vcl_session_get (wrk, sid)))
+ session = vcl_session_get (wrk, sid);
+ if (vcl_session_is_closed (session))
break;
session_events = session->vep.ev.events;
if (!(EPOLLOUT & session_events))
case SESSION_CTRL_EVT_DISCONNECTED:
disconnected_msg = (session_disconnected_msg_t *) e->data;
session = vcl_session_disconnected_handler (wrk, disconnected_msg);
- if (!session)
+ if (vcl_session_is_closed (session))
break;
session_events = session->vep.ev.events;
add_event = 1;
break;
case SESSION_CTRL_EVT_RESET:
sid = vcl_session_reset_handler (wrk, (session_reset_msg_t *) e->data);
- if (!(session = vcl_session_get (wrk, sid)))
+ session = vcl_session_get (wrk, sid);
+ if (vcl_session_is_closed (session))
break;
session_events = session->vep.ev.events;
add_event = 1;
case SESSION_CTRL_EVT_APP_DEL_SEGMENT:
vcl_session_app_del_segment_handler (wrk, e->data);
break;
+ case SESSION_CTRL_EVT_APP_WRK_RPC:
+ vcl_worker_rpc_handler (wrk, e->data);
+ break;
default:
VDBG (0, "unhandled: %u", e->event_type);
break;
}
}
ASSERT (maxevents > *num_ev);
- vcl_mq_dequeue_batch (wrk, mq, maxevents - *num_ev);
+ vcl_mq_dequeue_batch (wrk, mq, ~0);
svm_msg_q_unlock (mq);
handle_dequeued:
{
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
+ vcl_handle_mq_event (wrk, e);
svm_msg_q_free_msg (mq, msg);
}
vec_reset_length (wrk->mq_msg_vector);
continue;
now = clib_time_now (&wrk->clib_time);
- wait -= now - start;
+ wait -= (now - start) * 1e3;
start = now;
}
while (wait > 0);
case VPPCOM_ATTR_GET_FLAGS:
if (PREDICT_TRUE (buffer && buflen && (*buflen >= sizeof (*flags))))
{
- *flags = O_RDWR | (VCL_SESS_ATTR_TEST (session->attr,
- VCL_SESS_ATTR_NONBLOCK));
+ *flags =
+ O_RDWR |
+ (VCL_SESS_ATTR_TEST (session->attr, VCL_SESS_ATTR_NONBLOCK) ?
+ O_NONBLOCK : 0);
*buflen = sizeof (*flags);
VDBG (2, "VPPCOM_ATTR_GET_FLAGS: sh %u, flags = 0x%08x, "
"is_nonblocking = %u", session_handle, *flags,
*(int *) buffer = SHUT_RDWR;
*buflen = sizeof (int);
break;
+
+ case VPPCOM_ATTR_SET_CONNECTED:
+ session->flags |= VCL_SESSION_F_CONNECTED;
+ break;
+
default:
rv = VPPCOM_EINVAL;
break;
vcl_set_worker_index (~0);
}
+void
+vppcom_worker_index_set (int index)
+{
+ vcl_set_worker_index (index);
+}
+
int
vppcom_worker_index (void)
{
return session->n_accepted_sessions;
}
+const char *
+vppcom_proto_str (vppcom_proto_t proto)
+{
+ char const *proto_str;
+
+ switch (proto)
+ {
+ case VPPCOM_PROTO_TCP:
+ proto_str = "TCP";
+ break;
+ case VPPCOM_PROTO_UDP:
+ proto_str = "UDP";
+ break;
+ case VPPCOM_PROTO_TLS:
+ proto_str = "TLS";
+ break;
+ case VPPCOM_PROTO_QUIC:
+ proto_str = "QUIC";
+ break;
+ default:
+ proto_str = "UNKNOWN";
+ break;
+ }
+ return proto_str;
+}
+
+const char *
+vppcom_retval_str (int retval)
+{
+ char const *st;
+
+ switch (retval)
+ {
+ case VPPCOM_OK:
+ st = "VPPCOM_OK";
+ break;
+
+ case VPPCOM_EAGAIN:
+ st = "VPPCOM_EAGAIN";
+ break;
+
+ case VPPCOM_EFAULT:
+ st = "VPPCOM_EFAULT";
+ break;
+
+ case VPPCOM_ENOMEM:
+ st = "VPPCOM_ENOMEM";
+ break;
+
+ case VPPCOM_EINVAL:
+ st = "VPPCOM_EINVAL";
+ break;
+
+ case VPPCOM_EBADFD:
+ st = "VPPCOM_EBADFD";
+ break;
+
+ case VPPCOM_EAFNOSUPPORT:
+ st = "VPPCOM_EAFNOSUPPORT";
+ break;
+
+ case VPPCOM_ECONNABORTED:
+ st = "VPPCOM_ECONNABORTED";
+ break;
+
+ case VPPCOM_ECONNRESET:
+ st = "VPPCOM_ECONNRESET";
+ break;
+
+ case VPPCOM_ENOTCONN:
+ st = "VPPCOM_ENOTCONN";
+ break;
+
+ case VPPCOM_ECONNREFUSED:
+ st = "VPPCOM_ECONNREFUSED";
+ break;
+
+ case VPPCOM_ETIMEDOUT:
+ st = "VPPCOM_ETIMEDOUT";
+ break;
+
+ default:
+ st = "UNKNOWN_STATE";
+ break;
+ }
+
+ return st;
+}
+
/*
* fd.io coding-style-patch-verification: ON
*
Code Review / vpp.git / blobdiff