diff options
| -rw-r--r-- | src/svm/svm_fifo.c | 3 | ||||
| -rw-r--r-- | src/vnet/session/session.c | 32 | ||||
| -rw-r--r-- | src/vnet/session/session_node.c | 15 | ||||
| -rw-r--r-- | src/vnet/session/stream_session.h | 1 |
4 files changed, 39 insertions, 12 deletions
diff --git a/src/svm/svm_fifo.c b/src/svm/svm_fifo.c index dbdb813a7df..10c319236ee 100644 --- a/src/svm/svm_fifo.c +++ b/src/svm/svm_fifo.c @@ -174,7 +174,8 @@ format_svm_fifo (u8 * s, va_list * args) s = format (s, "cursize %u nitems %u has_event %d\n", f->cursize, f->nitems, f->has_event); - s = format (s, " head %d tail %d\n", f->head, f->tail); + s = format (s, " head %d tail %d segment manager %u\n", f->head, f->tail, + f->segment_manager); if (verbose > 1) s = format diff --git a/src/vnet/session/session.c b/src/vnet/session/session.c index c0163255fb6..ee02526c190 100644 --- a/src/vnet/session/session.c +++ b/src/vnet/session/session.c @@ -31,9 +31,9 @@ static void session_send_evt_to_thread (u64 session_handle, fifo_event_type_t evt_type, u32 thread_index, void *fp, void *rpc_args) { - u32 tries = 0; session_fifo_event_t evt = { {0}, }; svm_queue_t *q; + u32 tries = 0, max_tries; evt.event_type = evt_type; if (evt_type == FIFO_EVENT_RPC) @@ -47,7 +47,8 @@ session_send_evt_to_thread (u64 session_handle, fifo_event_type_t evt_type, q = session_manager_get_vpp_event_queue (thread_index); while (svm_queue_add (q, (u8 *) & evt, 1)) { - if (tries++ == 3) + max_tries = vlib_get_current_process (vlib_get_main ())? 1e6 : 3; + if (tries++ == max_tries) { SESSION_DBG ("failed to enqueue evt"); break; @@ -450,7 +451,7 @@ session_enqueue_notify (stream_session_t * s, u8 block) session_fifo_event_t evt; svm_queue_t *q; - if (PREDICT_FALSE (s->session_state == SESSION_STATE_CLOSED)) + if (PREDICT_FALSE (s->session_state >= SESSION_STATE_CLOSING)) { /* Session is closed so app will never clean up. Flush rx fifo */ u32 to_dequeue = svm_fifo_max_dequeue (s->server_rx_fifo); @@ -1059,11 +1060,28 @@ stream_session_stop_listen (stream_session_t * s) void stream_session_disconnect (stream_session_t * s) { - if (!s || s->session_state == SESSION_STATE_CLOSED) + u32 thread_index = vlib_get_thread_index (); + session_manager_main_t *smm = &session_manager_main; + session_fifo_event_t *evt; + + if (!s || s->session_state >= SESSION_STATE_CLOSING) return; - s->session_state = SESSION_STATE_CLOSED; - session_send_session_evt_to_thread (session_handle (s), - FIFO_EVENT_DISCONNECT, s->thread_index); + s->session_state = SESSION_STATE_CLOSING; + + /* If we are in the handler thread, or being called with the worker barrier + * held (api/cli), just append a new event pending disconnects vector. */ + if (thread_index > 0 || !vlib_get_current_process (vlib_get_main ())) + { + ASSERT (s->thread_index == thread_index || thread_index == 0); + vec_add2 (smm->pending_disconnects[s->thread_index], evt, 1); + memset (evt, 0, sizeof (*evt)); + evt->session_handle = session_handle (s); + evt->event_type = FIFO_EVENT_DISCONNECT; + } + else + session_send_session_evt_to_thread (session_handle (s), + FIFO_EVENT_DISCONNECT, + s->thread_index); } /** diff --git a/src/vnet/session/session_node.c b/src/vnet/session/session_node.c index e046efba81f..b6c1b2f04f3 100644 --- a/src/vnet/session/session_node.c +++ b/src/vnet/session/session_node.c @@ -692,7 +692,7 @@ session_queue_node_fn (vlib_main_t * vm, vlib_node_runtime_t * node, vlib_frame_t * frame) { session_manager_main_t *smm = vnet_get_session_manager_main (); - session_fifo_event_t *my_pending_event_vector, *pending_disconnects, *e; + session_fifo_event_t *my_pending_event_vector, *e; session_fifo_event_t *my_fifo_events; u32 n_to_dequeue, n_events; svm_queue_t *q; @@ -722,10 +722,9 @@ session_queue_node_fn (vlib_main_t * vm, vlib_node_runtime_t * node, /* min number of events we can dequeue without blocking */ n_to_dequeue = q->cursize; my_pending_event_vector = smm->pending_event_vector[thread_index]; - pending_disconnects = smm->pending_disconnects[thread_index]; if (!n_to_dequeue && !vec_len (my_pending_event_vector) - && !vec_len (pending_disconnects)) + && !vec_len (smm->pending_disconnects[thread_index])) return 0; SESSION_EVT_DBG (SESSION_EVT_DEQ_NODE, 0); @@ -802,12 +801,20 @@ skip_dequeue: case FIFO_EVENT_DISCONNECT: /* Make sure stream disconnects run after the pending list is drained */ s0 = session_get_from_handle (e0->session_handle); - if (!e0->postponed || svm_fifo_max_dequeue (s0->server_tx_fifo)) + if (!e0->postponed) { e0->postponed = 1; vec_add1 (smm->pending_disconnects[thread_index], *e0); continue; } + /* If tx queue is still not empty, wait a bit */ + if (svm_fifo_max_dequeue (s0->server_tx_fifo) + && e0->postponed < 200) + { + e0->postponed += 1; + vec_add1 (smm->pending_disconnects[thread_index], *e0); + continue; + } stream_session_disconnect_transport (s0); break; diff --git a/src/vnet/session/stream_session.h b/src/vnet/session/stream_session.h index 9e0e4d98990..1673356769c 100644 --- a/src/vnet/session/stream_session.h +++ b/src/vnet/session/stream_session.h @@ -32,6 +32,7 @@ typedef enum SESSION_STATE_ACCEPTING, SESSION_STATE_READY, SESSION_STATE_OPENED, + SESSION_STATE_CLOSING, SESSION_STATE_CLOSED, SESSION_STATE_N_STATES, } stream_session_state_t; |