From 460dce6e2d017cc7b2151fd0fa61d464570489d7 Mon Sep 17 00:00:00 2001 From: Florin Coras Date: Fri, 27 Jul 2018 05:45:06 -0700 Subject: [PATCH] vcl: add read/write udp support Change-Id: Ie6171c12055cde6915856de340839f5da1b1b1da Signed-off-by: Florin Coras --- src/vcl/vcl_bapi.c | 14 +++ src/vcl/vcl_cfg.c | 1 + src/vcl/vcl_private.h | 9 ++ src/vcl/vcl_test_server.c | 17 ++-- src/vcl/vppcom.c | 142 +++++++++++++++++-------------- src/vnet/session-apps/echo_server.c | 10 ++- src/vnet/session/application.c | 6 +- src/vnet/session/application_interface.h | 39 ++++++--- src/vnet/session/session.h | 4 +- 9 files changed, 149 insertions(+), 93 deletions(-) diff --git a/src/vcl/vcl_bapi.c b/src/vcl/vcl_bapi.c index 0201cd82966..311df64528c 100644 --- a/src/vcl/vcl_bapi.c +++ b/src/vcl/vcl_bapi.c @@ -132,6 +132,7 @@ vl_api_map_another_segment_t_handler (vl_api_map_another_segment_t * mp) svm_fifo_segment_create_args_t *a = &_a; int rv; + vcm->mounting_segment = 1; memset (a, 0, sizeof (*a)); a->segment_name = (char *) mp->segment_name; a->segment_size = mp->segment_size; @@ -147,6 +148,7 @@ vl_api_map_another_segment_t_handler (vl_api_map_another_segment_t * mp) VDBG (1, "VCL<%d>: mapped new segment '%s' size %d", getpid (), mp->segment_name, mp->segment_size); + vcm->mounting_segment = 0; } static void @@ -356,6 +358,18 @@ done: vppcom_session_table_add_listener (mp->handle, session_index); session->session_state = STATE_LISTEN; + if (session->is_dgram) + { + svm_fifo_t *rx_fifo, *tx_fifo; + session->vpp_evt_q = uword_to_pointer (mp->vpp_evt_q, svm_msg_q_t *); + rx_fifo = uword_to_pointer (mp->rx_fifo, svm_fifo_t *); + rx_fifo->client_session_index = session_index; + tx_fifo = uword_to_pointer (mp->tx_fifo, svm_fifo_t *); + tx_fifo->client_session_index = session_index; + session->rx_fifo = rx_fifo; + session->tx_fifo = tx_fifo; + } + VDBG (1, "VCL<%d>: vpp handle 0x%llx, sid %u: bind succeeded!", getpid (), mp->handle, mp->context); done_unlock: diff --git a/src/vcl/vcl_cfg.c b/src/vcl/vcl_cfg.c index 279a975719e..6e6a0ac0da4 100644 --- a/src/vcl/vcl_cfg.c +++ b/src/vcl/vcl_cfg.c @@ -21,6 +21,7 @@ */ static vppcom_main_t _vppcom_main = { .debug = VPPCOM_DEBUG_INIT, + .init = 0, .my_client_index = ~0 }; diff --git a/src/vcl/vcl_private.h b/src/vcl/vcl_private.h index 327a7fc02fe..af58cdc3ac9 100644 --- a/src/vcl/vcl_private.h +++ b/src/vcl/vcl_private.h @@ -237,6 +237,9 @@ typedef struct vppcom_main_t_ /** Pool of cut through registrations */ vcl_cut_through_registration_t *cut_through_registrations; + /** Flag indicating that a new segment is being mounted */ + volatile u32 mounting_segment; + #ifdef VCL_ELOG /* VPP Event-logger */ elog_main_t elog_main; @@ -313,6 +316,12 @@ vcl_session_get_index_from_handle (u64 handle) return VCL_INVALID_SESSION_INDEX; } +static inline u8 +vcl_session_is_ct (vcl_session_t * s) +{ + return (s->our_evt_q != 0); +} + static inline int vppcom_session_at_index (u32 session_index, vcl_session_t * volatile *sess) { diff --git a/src/vcl/vcl_test_server.c b/src/vcl/vcl_test_server.c index 6a2fda0be57..b49383eb73a 100644 --- a/src/vcl/vcl_test_server.c +++ b/src/vcl/vcl_test_server.c @@ -425,14 +425,17 @@ main (int argc, char **argv) return -1; } - rv = vppcom_session_listen (ssm->listen_fd, 10); - if (rv < 0) + if (!ssm->cfg.transport_udp) { - errno_val = errno = -rv; - perror ("ERROR in main()"); - fprintf (stderr, "SERVER: ERROR: listen failed " - "(errno = %d)!\n", errno_val); - return -1; + rv = vppcom_session_listen (ssm->listen_fd, 10); + if (rv < 0) + { + errno_val = errno = -rv; + perror ("ERROR in main()"); + fprintf (stderr, "SERVER: ERROR: listen failed " + "(errno = %d)!\n", errno_val); + return -1; + } } ssm->epfd = vppcom_epoll_create (); diff --git a/src/vcl/vppcom.c b/src/vcl/vppcom.c index d1c4413b2a8..a74e55af634 100644 --- a/src/vcl/vppcom.c +++ b/src/vcl/vppcom.c @@ -33,9 +33,15 @@ static void vcl_wait_for_memory (void *mem) { u8 __clib_unused test; + if (vcm->mounting_segment) + { + while (vcm->mounting_segment) + ; + return; + } if (1 || vcm->debug) { - sleep (1); + usleep (1e5); return; } if (signal (SIGSEGV, sigsegv_signal)) @@ -359,6 +365,8 @@ vcl_session_accepted_handler (session_accepted_msg_t * mp) hash_set (vcm->session_index_by_vpp_handles, mp->handle, session_index); session->transport.lcl_port = listen_session->transport.lcl_port; session->transport.lcl_ip = listen_session->transport.lcl_ip; + session->session_type = listen_session->session_type; + session->is_dgram = session->session_type == VPPCOM_PROTO_UDP; VDBG (1, "VCL<%d>: vpp handle 0x%llx, sid %u: client accept request from %s" " address %U port %d queue %p!", getpid (), mp->handle, session_index, @@ -405,13 +413,18 @@ done: if (rv) goto done_unlock; + rx_fifo = uword_to_pointer (mp->server_rx_fifo, svm_fifo_t *); + tx_fifo = uword_to_pointer (mp->server_tx_fifo, svm_fifo_t *); + vcl_wait_for_memory (rx_fifo); + rx_fifo->client_session_index = session_index; + tx_fifo->client_session_index = session_index; + if (mp->client_event_queue_address) { session->vpp_evt_q = uword_to_pointer (mp->server_event_queue_address, svm_msg_q_t *); session->our_evt_q = uword_to_pointer (mp->client_event_queue_address, svm_msg_q_t *); - vcl_wait_for_memory (session->vpp_evt_q); session->ct_registration = vcl_ct_registration_add (session->our_evt_q, session_index); } @@ -419,11 +432,6 @@ done: session->vpp_evt_q = uword_to_pointer (mp->vpp_event_queue_address, svm_msg_q_t *); - rx_fifo = uword_to_pointer (mp->server_rx_fifo, svm_fifo_t *); - rx_fifo->client_session_index = session_index; - tx_fifo = uword_to_pointer (mp->server_tx_fifo, svm_fifo_t *); - tx_fifo->client_session_index = session_index; - session->rx_fifo = rx_fifo; session->tx_fifo = tx_fifo; session->vpp_handle = mp->handle; @@ -762,11 +770,10 @@ vppcom_session_create (u8 proto, u8 is_nonblocking) session->session_type = proto; session->session_state = STATE_START; session->vpp_handle = ~0; + session->is_dgram = proto == VPPCOM_PROTO_UDP; if (is_nonblocking) VCL_SESS_ATTR_SET (session->attr, VCL_SESS_ATTR_NONBLOCK); - else - VCL_SESS_ATTR_CLR (session->attr, VCL_SESS_ATTR_NONBLOCK); vcl_evt (VCL_EVT_CREATE, session, session_type, session->session_state, is_nonblocking, session_index); @@ -929,6 +936,10 @@ vppcom_session_bind (uint32_t session_index, vppcom_endpt_t * ep) session->session_type ? "UDP" : "TCP"); vcl_evt (VCL_EVT_BIND, session); VCL_SESSION_UNLOCK (); + + if (session->session_type == VPPCOM_PROTO_UDP) + vppcom_session_listen (session_index, 10); + done: return rv; } @@ -1225,12 +1236,20 @@ vcl_is_rx_evt_for_session (session_event_t * e, u32 sid, u8 is_ct) return (e->event_type == SESSION_IO_EVT_CT_TX); } +static inline u8 +vcl_session_is_readable (vcl_session_t * s) +{ + return ((s->session_state & STATE_OPEN) + || (s->session_state == STATE_LISTEN + && s->session_type == VPPCOM_PROTO_UDP)); +} + static inline int vppcom_session_read_internal (uint32_t session_index, void *buf, int n, u8 peek) { int n_read = 0, rv, is_nonblocking; - vcl_session_t *session = 0; + vcl_session_t *s = 0; svm_fifo_t *rx_fifo; svm_msg_q_msg_t msg; session_event_t *e; @@ -1239,9 +1258,9 @@ vppcom_session_read_internal (uint32_t session_index, void *buf, int n, ASSERT (buf); - VCL_SESSION_LOCK_AND_GET (session_index, &session); + VCL_SESSION_LOCK_AND_GET (session_index, &s); - if (PREDICT_FALSE (session->is_vep)) + if (PREDICT_FALSE (s->is_vep)) { VCL_SESSION_UNLOCK (); clib_warning ("VCL<%d>: ERROR: sid %u: cannot " @@ -1250,24 +1269,24 @@ vppcom_session_read_internal (uint32_t session_index, void *buf, int n, goto done; } - is_nonblocking = VCL_SESS_ATTR_TEST (session->attr, VCL_SESS_ATTR_NONBLOCK); - rx_fifo = session->rx_fifo; + is_nonblocking = VCL_SESS_ATTR_TEST (s->attr, VCL_SESS_ATTR_NONBLOCK); + rx_fifo = s->rx_fifo; - if (PREDICT_FALSE (!(session->session_state & STATE_OPEN))) + if (PREDICT_FALSE (!vcl_session_is_readable (s))) { - session_state_t state = session->session_state; + session_state_t state = s->session_state; VCL_SESSION_UNLOCK (); rv = ((state & STATE_DISCONNECT) ? VPPCOM_ECONNRESET : VPPCOM_ENOTCONN); VDBG (0, "VCL<%d>: vpp handle 0x%llx, sid %u: %s session is not open! " "state 0x%x (%s), returning %d (%s)", - getpid (), session->vpp_handle, session_index, state, + getpid (), s->vpp_handle, session_index, state, vppcom_session_state_str (state), rv, vppcom_retval_str (rv)); goto done; } VCL_SESSION_UNLOCK (); - mq = session->our_evt_q ? session->our_evt_q : vcm->app_event_queue; + mq = vcl_session_is_ct (s) ? s->our_evt_q : vcm->app_event_queue; is_full = svm_fifo_is_full (rx_fifo); if (svm_fifo_is_empty (rx_fifo)) @@ -1286,7 +1305,7 @@ vppcom_session_read_internal (uint32_t session_index, void *buf, int n, svm_msg_q_sub_w_lock (mq, &msg); e = svm_msg_q_msg_data (mq, &msg); if (!vcl_is_rx_evt_for_session (e, session_index, - session->our_evt_q != 0)) + s->our_evt_q != 0)) { vcl_handle_mq_ctrl_event (e); svm_msg_q_free_msg (mq, &msg); @@ -1303,27 +1322,24 @@ vppcom_session_read_internal (uint32_t session_index, void *buf, int n, } } - if (peek) - n_read = svm_fifo_peek (rx_fifo, 0, n, buf); + if (s->is_dgram) + n_read = app_recv_dgram_raw (rx_fifo, buf, n, &s->transport, 1, peek); else - n_read = svm_fifo_dequeue_nowait (rx_fifo, n, buf); - ASSERT (n_read > 0); - svm_fifo_unset_event (rx_fifo); + n_read = app_recv_stream_raw (rx_fifo, buf, n, 1, peek); - if (session->our_evt_q && is_full) - app_send_io_evt_to_vpp (session->vpp_evt_q, rx_fifo, SESSION_IO_EVT_CT_RX, + if (vcl_session_is_ct (s) && is_full) + app_send_io_evt_to_vpp (s->vpp_evt_q, rx_fifo, SESSION_IO_EVT_CT_RX, SVM_Q_WAIT); - if (VPPCOM_DEBUG > 2) { if (n_read > 0) clib_warning ("VCL<%d>: vpp handle 0x%llx, sid %u: read %d bytes " - "from (%p)", getpid (), session->vpp_handle, + "from (%p)", getpid (), s->vpp_handle, session_index, n_read, rx_fifo); else clib_warning ("VCL<%d>: vpp handle 0x%llx, sid %u: nothing read! " - "returning %d (%s)", getpid (), session->vpp_handle, + "returning %d (%s)", getpid (), s->vpp_handle, session_index, rv, vppcom_retval_str (rv)); } return n_read; @@ -1389,45 +1405,46 @@ int vppcom_session_write (uint32_t session_index, void *buf, size_t n) { int rv, n_write, is_nonblocking; - vcl_session_t *session = 0; + vcl_session_t *s = 0; svm_fifo_t *tx_fifo = 0; + session_evt_type_t et; svm_msg_q_msg_t msg; session_event_t *e; svm_msg_q_t *mq; ASSERT (buf); - VCL_SESSION_LOCK_AND_GET (session_index, &session); + VCL_SESSION_LOCK_AND_GET (session_index, &s); - tx_fifo = session->tx_fifo; - is_nonblocking = VCL_SESS_ATTR_TEST (session->attr, VCL_SESS_ATTR_NONBLOCK); + tx_fifo = s->tx_fifo; + is_nonblocking = VCL_SESS_ATTR_TEST (s->attr, VCL_SESS_ATTR_NONBLOCK); - if (PREDICT_FALSE (session->is_vep)) + if (PREDICT_FALSE (s->is_vep)) { VCL_SESSION_UNLOCK (); clib_warning ("VCL<%d>: ERROR: vpp handle 0x%llx, sid %u: " "cannot write to an epoll session!", - getpid (), session->vpp_handle, session_index); + getpid (), s->vpp_handle, session_index); rv = VPPCOM_EBADFD; goto done; } - if (!(session->session_state & STATE_OPEN)) + if (!(s->session_state & STATE_OPEN)) { - session_state_t state = session->session_state; + session_state_t state = s->session_state; rv = ((state & STATE_DISCONNECT) ? VPPCOM_ECONNRESET : VPPCOM_ENOTCONN); VCL_SESSION_UNLOCK (); VDBG (1, "VCL<%d>: vpp handle 0x%llx, sid %u: session is not open! " "state 0x%x (%s)", - getpid (), session->vpp_handle, session_index, + getpid (), s->vpp_handle, session_index, state, vppcom_session_state_str (state)); goto done; } VCL_SESSION_UNLOCK (); - mq = session->our_evt_q ? session->our_evt_q : vcm->app_event_queue; + mq = vcl_session_is_ct (s) ? s->our_evt_q : vcm->app_event_queue; if (svm_fifo_is_full (tx_fifo)) { if (is_nonblocking) @@ -1448,7 +1465,7 @@ vppcom_session_write (uint32_t session_index, void *buf, size_t n) svm_msg_q_sub_w_lock (mq, &msg); e = svm_msg_q_msg_data (mq, &msg); if (!vcl_is_tx_evt_for_session (e, session_index, - session->our_evt_q != 0)) + s->our_evt_q != 0)) { vcl_handle_mq_ctrl_event (e); svm_msg_q_free_msg (mq, &msg); @@ -1465,31 +1482,27 @@ vppcom_session_write (uint32_t session_index, void *buf, size_t n) } } - n_write = svm_fifo_enqueue_nowait (tx_fifo, n, (void *) buf); - ASSERT (n_write > 0); + ASSERT (FIFO_EVENT_APP_TX + 1 == SESSION_IO_EVT_CT_TX); + et = FIFO_EVENT_APP_TX + vcl_session_is_ct (s); + if (s->is_dgram) + n_write = app_send_dgram_raw (tx_fifo, &s->transport, + s->vpp_evt_q, buf, n, et, SVM_Q_WAIT); + else + n_write = app_send_stream_raw (tx_fifo, s->vpp_evt_q, buf, n, et, + SVM_Q_WAIT); - if (svm_fifo_set_event (tx_fifo)) - { - session_evt_type_t et; - VCL_SESSION_LOCK_AND_GET (session_index, &session); - et = session->our_evt_q ? SESSION_IO_EVT_CT_TX : FIFO_EVENT_APP_TX; - app_send_io_evt_to_vpp (session->vpp_evt_q, tx_fifo, et, SVM_Q_WAIT); - VCL_SESSION_UNLOCK (); - VDBG (1, "VCL<%d>: vpp handle 0x%llx, sid %u: added FIFO_EVENT_APP_TX " - "to vpp_event_q %p, n_write %d", getpid (), - session->vpp_handle, session_index, session->vpp_evt_q, n_write); - } + 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 (), session->vpp_handle, + "FIFO-FULL (%p)", getpid (), s->vpp_handle, session_index, tx_fifo); else clib_warning ("VCL<%d>: vpp handle 0x%llx, sid %u: " "wrote %d bytes tx-fifo: (%p)", getpid (), - session->vpp_handle, session_index, n_write, tx_fifo); + s->vpp_handle, session_index, n_write, tx_fifo); } return n_write; @@ -2885,18 +2898,11 @@ vppcom_session_recvfrom (uint32_t session_index, void *buffer, VCL_SESSION_UNLOCK (); VDBG (0, "VCL<%d>: invalid session, sid (%u) has been closed!", getpid (), session_index); - rv = VPPCOM_EBADFD; VCL_SESSION_UNLOCK (); - goto done; + return VPPCOM_EBADFD; } 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)); - else - clib_memcpy (ep->ip, &session->transport.rmt_ip.ip6, - sizeof (ip6_address_t)); VCL_SESSION_UNLOCK (); } @@ -2908,10 +2914,16 @@ vppcom_session_recvfrom (uint32_t session_index, void *buffer, { clib_warning ("VCL<%d>: Unsupport flags for recvfrom %d", getpid (), flags); - rv = VPPCOM_EAFNOSUPPORT; + return VPPCOM_EAFNOSUPPORT; } -done: + if (session->transport.is_ip4) + clib_memcpy (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)); + return rv; } diff --git a/src/vnet/session-apps/echo_server.c b/src/vnet/session-apps/echo_server.c index 770f4ba7337..14ab36d796c 100644 --- a/src/vnet/session-apps/echo_server.c +++ b/src/vnet/session-apps/echo_server.c @@ -211,14 +211,16 @@ echo_server_rx_callback (stream_session_t * s) actual_transfer = app_recv_stream_raw (rx_fifo, esm->rx_buf[thread_index], max_transfer, - 0 /* don't clear event */ ); + 0 /* don't clear event */ , + 0 /* peek */ ); } else { actual_transfer = app_recv_dgram_raw (rx_fifo, esm->rx_buf[thread_index], max_transfer, &at, - 0 /* don't clear event */ ); + 0 /* don't clear event */ , + 0 /* peek */ ); } ASSERT (actual_transfer == max_transfer); /* test_bytes (esm, actual_transfer); */ @@ -232,14 +234,14 @@ echo_server_rx_callback (stream_session_t * s) n_written = app_send_stream_raw (tx_fifo, esm->vpp_queue[thread_index], esm->rx_buf[thread_index], - actual_transfer, 0); + actual_transfer, FIFO_EVENT_APP_TX, 0); } else { n_written = app_send_dgram_raw (tx_fifo, &at, esm->vpp_queue[s->thread_index], esm->rx_buf[thread_index], - actual_transfer, 0); + actual_transfer, FIFO_EVENT_APP_TX, 0); } if (n_written != max_transfer) diff --git a/src/vnet/session/application.c b/src/vnet/session/application.c index 99f0dff7fab..757e12e1b58 100644 --- a/src/vnet/session/application.c +++ b/src/vnet/session/application.c @@ -841,7 +841,8 @@ app_send_io_evt_rx (application_t * app, stream_session_t * s, u8 lock) svm_msg_q_msg_t msg; svm_msg_q_t *mq; - if (PREDICT_FALSE (s->session_state != SESSION_STATE_READY)) + if (PREDICT_FALSE (s->session_state != SESSION_STATE_READY + && s->session_state != SESSION_STATE_LISTENING)) { /* Session is closed so app will never clean up. Flush rx fifo */ if (s->session_state == SESSION_STATE_CLOSED) @@ -917,8 +918,9 @@ app_send_io_evt_tx (application_t * app, stream_session_t * s, u8 lock) typedef int (app_send_evt_handler_fn) (application_t *app, stream_session_t *s, u8 lock); -static app_send_evt_handler_fn * const app_send_evt_handler_fns[2] = { +static app_send_evt_handler_fn * const app_send_evt_handler_fns[3] = { app_send_io_evt_rx, + 0, app_send_io_evt_tx, }; /* *INDENT-ON* */ diff --git a/src/vnet/session/application_interface.h b/src/vnet/session/application_interface.h index ffe2a64c245..daba169a297 100644 --- a/src/vnet/session/application_interface.h +++ b/src/vnet/session/application_interface.h @@ -354,7 +354,8 @@ app_send_io_evt_to_vpp (svm_msg_q_t * mq, svm_fifo_t * f, u8 evt_type, always_inline int app_send_dgram_raw (svm_fifo_t * f, app_session_transport_t * at, - svm_msg_q_t * vpp_evt_q, u8 * data, u32 len, u8 noblock) + svm_msg_q_t * vpp_evt_q, u8 * data, u32 len, u8 evt_type, + u8 noblock) { u32 max_enqueue, actual_write; session_dgram_hdr_t hdr; @@ -379,7 +380,7 @@ app_send_dgram_raw (svm_fifo_t * f, app_session_transport_t * at, if ((rv = svm_fifo_enqueue_nowait (f, actual_write, data)) > 0) { if (svm_fifo_set_event (f)) - app_send_io_evt_to_vpp (vpp_evt_q, f, FIFO_EVENT_APP_TX, noblock); + app_send_io_evt_to_vpp (vpp_evt_q, f, evt_type, noblock); } ASSERT (rv); return rv; @@ -389,19 +390,19 @@ always_inline int app_send_dgram (app_session_t * s, u8 * data, u32 len, u8 noblock) { return app_send_dgram_raw (s->tx_fifo, &s->transport, s->vpp_evt_q, data, - len, noblock); + len, FIFO_EVENT_APP_TX, noblock); } always_inline int app_send_stream_raw (svm_fifo_t * f, svm_msg_q_t * vpp_evt_q, u8 * data, - u32 len, u8 noblock) + u32 len, u8 evt_type, u8 noblock) { int rv; if ((rv = svm_fifo_enqueue_nowait (f, len, data)) > 0) { if (svm_fifo_set_event (f)) - app_send_io_evt_to_vpp (vpp_evt_q, f, FIFO_EVENT_APP_TX, noblock); + app_send_io_evt_to_vpp (vpp_evt_q, f, evt_type, noblock); } return rv; } @@ -409,7 +410,8 @@ app_send_stream_raw (svm_fifo_t * f, svm_msg_q_t * vpp_evt_q, u8 * data, always_inline int app_send_stream (app_session_t * s, u8 * data, u32 len, u8 noblock) { - return app_send_stream_raw (s->tx_fifo, s->vpp_evt_q, data, len, noblock); + return app_send_stream_raw (s->tx_fifo, s->vpp_evt_q, data, len, + FIFO_EVENT_APP_TX, noblock); } always_inline int @@ -422,17 +424,22 @@ app_send (app_session_t * s, u8 * data, u32 len, u8 noblock) always_inline int app_recv_dgram_raw (svm_fifo_t * f, u8 * buf, u32 len, - app_session_transport_t * at, u8 clear_evt) + app_session_transport_t * at, u8 clear_evt, u8 peek) { session_dgram_pre_hdr_t ph; u32 max_deq; int rv; - if (clear_evt) - svm_fifo_unset_event (f); max_deq = svm_fifo_max_dequeue (f); if (max_deq < sizeof (session_dgram_hdr_t)) - return 0; + { + if (clear_evt) + svm_fifo_unset_event (f); + return 0; + } + + if (clear_evt) + svm_fifo_unset_event (f); svm_fifo_peek (f, 0, sizeof (ph), (u8 *) & ph); ASSERT (ph.data_length >= ph.data_offset); @@ -440,6 +447,8 @@ app_recv_dgram_raw (svm_fifo_t * f, u8 * buf, u32 len, svm_fifo_peek (f, sizeof (ph), sizeof (*at), (u8 *) at); len = clib_min (len, ph.data_length - ph.data_offset); rv = svm_fifo_peek (f, ph.data_offset + SESSION_CONN_HDR_LEN, len, buf); + if (peek) + return rv; ph.data_offset += rv; if (ph.data_offset == ph.data_length) svm_fifo_dequeue_drop (f, ph.data_length + SESSION_CONN_HDR_LEN); @@ -451,21 +460,25 @@ app_recv_dgram_raw (svm_fifo_t * f, u8 * buf, u32 len, always_inline int app_recv_dgram (app_session_t * s, u8 * buf, u32 len) { - return app_recv_dgram_raw (s->rx_fifo, buf, len, &s->transport, 1); + return app_recv_dgram_raw (s->rx_fifo, buf, len, &s->transport, 1, 0); } always_inline int -app_recv_stream_raw (svm_fifo_t * f, u8 * buf, u32 len, u8 clear_evt) +app_recv_stream_raw (svm_fifo_t * f, u8 * buf, u32 len, u8 clear_evt, u8 peek) { if (clear_evt) svm_fifo_unset_event (f); + + if (peek) + return svm_fifo_peek (f, 0, len, buf); + return svm_fifo_dequeue_nowait (f, len, buf); } always_inline int app_recv_stream (app_session_t * s, u8 * buf, u32 len) { - return app_recv_stream_raw (s->rx_fifo, buf, len, 1); + return app_recv_stream_raw (s->rx_fifo, buf, len, 1, 0); } always_inline int diff --git a/src/vnet/session/session.h b/src/vnet/session/session.h index 99546cb1347..5e94c41f927 100644 --- a/src/vnet/session/session.h +++ b/src/vnet/session/session.h @@ -33,12 +33,12 @@ typedef enum { FIFO_EVENT_APP_RX, + SESSION_IO_EVT_CT_RX, FIFO_EVENT_APP_TX, + SESSION_IO_EVT_CT_TX, FIFO_EVENT_DISCONNECT, FIFO_EVENT_BUILTIN_RX, FIFO_EVENT_RPC, - SESSION_IO_EVT_CT_TX, - SESSION_IO_EVT_CT_RX, SESSION_CTRL_EVT_ACCEPTED, SESSION_CTRL_EVT_ACCEPTED_REPLY, SESSION_CTRL_EVT_CONNECTED, -- 2.16.6