X-Git-Url: https://gerrit.fd.io/r/gitweb?a=blobdiff_plain;f=src%2Fvnet%2Fsession%2Fsession.c;h=e9cda361f37b37299e926710375d19f7719ae3d4;hb=d8f48e2166747b529aa23762ac314dd686590c89;hp=7b53a47023a2a9f167dd56cbe1db0e76a773425e;hpb=d09236d17d86a5d50166b2017f8f30a560c6e1b8;p=vpp.git diff --git a/src/vnet/session/session.c b/src/vnet/session/session.c index 7b53a47023a..e9cda361f37 100644 --- a/src/vnet/session/session.c +++ b/src/vnet/session/session.c @@ -36,22 +36,17 @@ session_send_evt_to_thread (void *data, void *args, u32 thread_index, mq = session_main_get_vpp_event_queue (thread_index); if (PREDICT_FALSE (svm_msg_q_lock (mq))) return -1; - if (PREDICT_FALSE (svm_msg_q_ring_is_full (mq, SESSION_MQ_IO_EVT_RING))) + if (PREDICT_FALSE (svm_msg_q_is_full (mq) + || svm_msg_q_ring_is_full (mq, SESSION_MQ_IO_EVT_RING))) { svm_msg_q_unlock (mq); return -2; } - msg = svm_msg_q_alloc_msg_w_ring (mq, SESSION_MQ_IO_EVT_RING); - if (PREDICT_FALSE (svm_msg_q_msg_is_invalid (&msg))) - { - svm_msg_q_unlock (mq); - return -2; - } - evt = (session_event_t *) svm_msg_q_msg_data (mq, &msg); - evt->event_type = evt_type; switch (evt_type) { case SESSION_CTRL_EVT_RPC: + msg = svm_msg_q_alloc_msg_w_ring (mq, SESSION_MQ_IO_EVT_RING); + evt = (session_event_t *) svm_msg_q_msg_data (mq, &msg); evt->rpc_args.fp = data; evt->rpc_args.arg = args; break; @@ -59,10 +54,15 @@ session_send_evt_to_thread (void *data, void *args, u32 thread_index, case SESSION_IO_EVT_TX: case SESSION_IO_EVT_TX_FLUSH: case SESSION_IO_EVT_BUILTIN_RX: + msg = svm_msg_q_alloc_msg_w_ring (mq, SESSION_MQ_IO_EVT_RING); + evt = (session_event_t *) svm_msg_q_msg_data (mq, &msg); evt->session_index = *(u32 *) data; break; case SESSION_IO_EVT_BUILTIN_TX: case SESSION_CTRL_EVT_CLOSE: + case SESSION_CTRL_EVT_RESET: + msg = svm_msg_q_alloc_msg_w_ring (mq, SESSION_MQ_IO_EVT_RING); + evt = (session_event_t *) svm_msg_q_msg_data (mq, &msg); evt->session_handle = session_handle ((session_t *) data); break; default: @@ -70,6 +70,7 @@ session_send_evt_to_thread (void *data, void *args, u32 thread_index, svm_msg_q_unlock (mq); return -1; } + evt->event_type = evt_type; svm_msg_q_add_and_unlock (mq, &msg); return 0; @@ -92,10 +93,10 @@ session_send_io_evt_to_thread_custom (void *data, u32 thread_index, int session_send_ctrl_evt_to_thread (session_t * s, session_evt_type_t evt_type) { - /* only event supported for now is disconnect */ - ASSERT (evt_type == SESSION_CTRL_EVT_CLOSE); - return session_send_evt_to_thread (s, 0, s->thread_index, - SESSION_CTRL_EVT_CLOSE); + /* only events supported are disconnect and reset */ + ASSERT (evt_type == SESSION_CTRL_EVT_CLOSE + || evt_type == SESSION_CTRL_EVT_RESET); + return session_send_evt_to_thread (s, 0, s->thread_index, evt_type); } void @@ -125,6 +126,7 @@ session_add_self_custom_tx_evt (transport_connection_t * tc, u8 has_prio) s = session_get (tc->s_index, tc->thread_index); ASSERT (s->thread_index == vlib_get_thread_index ()); + ASSERT (s->session_state != SESSION_STATE_TRANSPORT_DELETED); if (!(s->flags & SESSION_F_CUSTOM_TX)) { s->flags |= SESSION_F_CUSTOM_TX; @@ -144,7 +146,7 @@ session_add_self_custom_tx_evt (transport_connection_t * tc, u8 has_prio) } static void -session_program_transport_close (session_t * s) +session_program_transport_ctrl_evt (session_t * s, session_evt_type_t evt) { u32 thread_index = vlib_get_thread_index (); session_evt_elt_t *elt; @@ -158,10 +160,10 @@ session_program_transport_close (session_t * s) elt = session_evt_alloc_ctrl (wrk); clib_memset (&elt->evt, 0, sizeof (session_event_t)); elt->evt.session_handle = session_handle (s); - elt->evt.event_type = SESSION_CTRL_EVT_CLOSE; + elt->evt.event_type = evt; } else - session_send_ctrl_evt_to_thread (s, SESSION_CTRL_EVT_CLOSE); + session_send_ctrl_evt_to_thread (s, evt); } session_t * @@ -204,6 +206,32 @@ session_free (session_t * s) pool_put (session_main.wrk[s->thread_index].sessions, s); } +u8 +session_is_valid (u32 si, u8 thread_index) +{ + session_t *s; + transport_connection_t *tc; + + s = pool_elt_at_index (session_main.wrk[thread_index].sessions, si); + + if (!s) + return 1; + + if (s->thread_index != thread_index || s->session_index != si) + return 0; + + if (s->session_state == SESSION_STATE_TRANSPORT_DELETED + || s->session_state <= SESSION_STATE_LISTENING) + return 1; + + tc = session_get_transport (s); + if (s->connection_index != tc->c_index + || s->thread_index != tc->thread_index || tc->s_index != si) + return 0; + + return 1; +} + static void session_cleanup_notify (session_t * s, session_cleanup_ntf_t ntf) { @@ -368,6 +396,24 @@ session_enqueue_chain_tail (session_t * s, vlib_buffer_t * b, return 0; } +void +session_fifo_tuning (session_t * s, svm_fifo_t * f, + session_ft_action_t act, u32 len) +{ + if (s->flags & SESSION_F_CUSTOM_FIFO_TUNING) + { + app_worker_t *app_wrk = app_worker_get (s->app_wrk_index); + app_worker_session_fifo_tuning (app_wrk, s, f, act, len); + if (CLIB_ASSERT_ENABLE) + { + segment_manager_t *sm; + sm = segment_manager_get (f->segment_manager); + ASSERT (f->size >= 4096); + ASSERT (f->size <= sm->max_fifo_size); + } + } +} + /* * Enqueue data for delivery to session peer. Does not notify peer of enqueue * event but on request can queue notification events for later delivery by @@ -430,6 +476,8 @@ session_enqueue_stream_connection (transport_connection_t * tc, s->flags |= SESSION_F_RX_EVT; vec_add1 (wrk->session_to_enqueue[tc->proto], s->session_index); } + + session_fifo_tuning (s, s->rx_fifo, SESSION_FT_ACTION_ENQUEUED, 0); } return enqueued; @@ -467,6 +515,8 @@ session_enqueue_dgram_connection (session_t * s, s->flags |= SESSION_F_RX_EVT; vec_add1 (wrk->session_to_enqueue[proto], s->session_index); } + + session_fifo_tuning (s, s->rx_fifo, SESSION_FT_ACTION_ENQUEUED, 0); } return enqueued; } @@ -486,6 +536,7 @@ session_tx_fifo_dequeue_drop (transport_connection_t * tc, u32 max_bytes) u32 rv; rv = svm_fifo_dequeue_drop (s->tx_fifo, max_bytes); + session_fifo_tuning (s, s->tx_fifo, SESSION_FT_ACTION_DEQUEUED, rv); if (svm_fifo_needs_deq_ntf (s->tx_fifo, max_bytes)) session_dequeue_notify (s); @@ -568,10 +619,10 @@ session_enqueue_notify (session_t * s) static void session_enqueue_notify_rpc (void *arg) { - session_handle_t sh = (session_handle_t) arg; + u32 session_index = pointer_to_uword (arg); session_t *s; - s = session_get_from_handle_if_valid (sh); + s = session_get_if_valid (session_index, vlib_get_thread_index ()); if (!s) return; @@ -586,8 +637,15 @@ void session_enqueue_notify_thread (session_handle_t sh) { u32 thread_index = session_thread_from_handle (sh); + u32 session_index = session_index_from_handle (sh); + + /* + * Pass session index (u32) as opposed to handle (u64) in case pointers + * are not 64-bit. + */ session_send_rpc_evt_to_thread (thread_index, - session_enqueue_notify_rpc, (void *) sh); + session_enqueue_notify_rpc, + uword_to_pointer (session_index, void *)); } int @@ -595,6 +653,8 @@ session_dequeue_notify (session_t * s) { app_worker_t *app_wrk; + svm_fifo_clear_deq_ntf (s->tx_fifo); + app_wrk = app_worker_get_if_valid (s->app_wrk_index); if (PREDICT_FALSE (!app_wrk)) return -1; @@ -607,8 +667,6 @@ session_dequeue_notify (session_t * s) return session_notify_subscribers (app_wrk->app_index, s, s->tx_fifo, SESSION_IO_EVT_TX); - svm_fifo_clear_deq_ntf (s->tx_fifo); - return 0; } @@ -639,6 +697,9 @@ session_main_flush_enqueue_events (u8 transport_proto, u32 thread_index) continue; } + session_fifo_tuning (s, s->rx_fifo, SESSION_FT_ACTION_ENQUEUED, + 0 /* TODO/not needed */ ); + if (PREDICT_FALSE (session_enqueue_notify_inline (s))) errors++; } @@ -791,6 +852,7 @@ session_dgram_connect_notify (transport_connection_t * tc, new_s->rx_fifo->master_session_index = new_s->session_index; new_s->rx_fifo->master_thread_index = new_s->thread_index; new_s->session_state = SESSION_STATE_READY; + new_s->flags |= SESSION_F_IS_MIGRATING; session_lookup_add_connection (tc, session_handle (new_s)); /* @@ -855,7 +917,8 @@ session_transport_delete_notify (transport_connection_t * tc) /* Session was created but accept notification was not yet sent to the * app. Cleanup everything. */ session_lookup_del_session (s); - session_free_w_fifos (s); + segment_manager_dealloc_fifos (s->rx_fifo, s->tx_fifo); + session_free (s); break; case SESSION_STATE_ACCEPTING: case SESSION_STATE_TRANSPORT_CLOSING: @@ -878,10 +941,10 @@ session_transport_delete_notify (transport_connection_t * tc) * session is just removed because both transport and app have * confirmed the close*/ session_lookup_del_session (s); - s->session_state = SESSION_STATE_CLOSED; + s->session_state = SESSION_STATE_TRANSPORT_DELETED; session_cleanup_notify (s, SESSION_CLEANUP_TRANSPORT); svm_fifo_dequeue_drop_all (s->tx_fifo); - session_program_transport_close (s); + session_program_transport_ctrl_evt (s, SESSION_CTRL_EVT_CLOSE); break; case SESSION_STATE_TRANSPORT_DELETED: break; @@ -966,7 +1029,14 @@ session_stream_accept_notify (transport_connection_t * tc) if (!app_wrk) return -1; s->session_state = SESSION_STATE_ACCEPTING; - return app_worker_accept_notify (app_wrk, s); + if (app_worker_accept_notify (app_wrk, s)) + { + /* On transport delete, no notifications should be sent. Unless, the + * accept is retried and successful. */ + s->session_state = SESSION_STATE_CREATED; + return -1; + } + return 0; } /** @@ -1187,12 +1257,26 @@ session_close (session_t * s) * acknowledge the close */ if (s->session_state == SESSION_STATE_TRANSPORT_CLOSED || s->session_state == SESSION_STATE_TRANSPORT_DELETED) - session_program_transport_close (s); + session_program_transport_ctrl_evt (s, SESSION_CTRL_EVT_CLOSE); return; } s->session_state = SESSION_STATE_CLOSING; - session_program_transport_close (s); + session_program_transport_ctrl_evt (s, SESSION_CTRL_EVT_CLOSE); +} + +/** + * Force a close without waiting for data to be flushed + */ +void +session_reset (session_t * s) +{ + if (s->session_state >= SESSION_STATE_CLOSING) + return; + /* Drop all outstanding tx data */ + svm_fifo_dequeue_drop_all (s->tx_fifo); + s->session_state = SESSION_STATE_CLOSING; + session_program_transport_ctrl_evt (s, SESSION_CTRL_EVT_RESET); } /** @@ -1227,6 +1311,26 @@ session_transport_close (session_t * s) s->thread_index); } +/** + * Force transport close + */ +void +session_transport_reset (session_t * s) +{ + if (s->session_state >= SESSION_STATE_APP_CLOSED) + { + if (s->session_state == SESSION_STATE_TRANSPORT_CLOSED) + s->session_state = SESSION_STATE_CLOSED; + else if (s->session_state >= SESSION_STATE_TRANSPORT_DELETED) + session_free_w_fifos (s); + return; + } + + s->session_state = SESSION_STATE_APP_CLOSED; + transport_reset (session_get_transport_proto (s), s->connection_index, + s->thread_index); +} + /** * Cleanup transport and session state. * @@ -1237,15 +1341,15 @@ session_transport_close (session_t * s) void session_transport_cleanup (session_t * s) { - s->session_state = SESSION_STATE_CLOSED; - /* Delete from main lookup table before we axe the the transport */ session_lookup_del_session (s); - transport_cleanup (session_get_transport_proto (s), s->connection_index, - s->thread_index); + if (s->session_state != SESSION_STATE_TRANSPORT_DELETED) + transport_cleanup (session_get_transport_proto (s), s->connection_index, + s->thread_index); /* Since we called cleanup, no delete notification will come. So, make * sure the session is properly freed. */ - session_free_w_fifos (s); + segment_manager_dealloc_fifos (s->rx_fifo, s->tx_fifo); + session_free (s); } /** @@ -1262,7 +1366,6 @@ session_vpp_event_queues_allocate (session_main_t * smm) { u32 evt_q_length = 2048, evt_size = sizeof (session_event_t); ssvm_private_t *eqs = &smm->evt_qs_segment; - api_main_t *am = &api_main; uword eqs_size = 64 << 20; pid_t vpp_pid = getpid (); void *oldheap; @@ -1292,7 +1395,7 @@ session_vpp_event_queues_allocate (session_main_t * smm) if (smm->evt_qs_use_memfd_seg) oldheap = ssvm_push_heap (eqs->sh); else - oldheap = svm_push_data_heap (am->vlib_rp); + oldheap = vl_msg_push_heap (); for (i = 0; i < vec_len (smm->wrk); i++) { @@ -1317,7 +1420,7 @@ session_vpp_event_queues_allocate (session_main_t * smm) if (smm->evt_qs_use_memfd_seg) ssvm_pop_heap (oldheap); else - svm_pop_heap (oldheap); + vl_msg_pop_heap (oldheap); } ssvm_private_t * @@ -1451,6 +1554,7 @@ session_manager_main_enable (vlib_main_t * vm) wrk->old_head = clib_llist_make_head (wrk->event_elts, evt_list); wrk->vm = vlib_mains[i]; wrk->last_vlib_time = vlib_time_now (vlib_mains[i]); + wrk->last_vlib_us_time = wrk->last_vlib_time * CLIB_US_TIME_FREQ; if (num_threads > 1) clib_rwlock_init (&smm->wrk[i].peekers_rw_locks); @@ -1494,7 +1598,6 @@ session_manager_main_enable (vlib_main_t * vm) /* Enable transports */ transport_enable_disable (vm, 1); - transport_init_tx_pacers_period (); return 0; } @@ -1566,10 +1669,25 @@ session_manager_main_init (vlib_main_t * vm) smm->evt_qs_segment_size = 1 << 20; #endif smm->is_enabled = 0; + smm->session_enable_asap = 0; + return 0; +} + +static clib_error_t * +session_main_init (vlib_main_t * vm) +{ + session_main_t *smm = &session_main; + if (smm->session_enable_asap) + { + vlib_worker_thread_barrier_sync (vm); + vnet_session_enable_disable (vm, 1 /* is_en */ ); + vlib_worker_thread_barrier_release (vm); + } return 0; } VLIB_INIT_FUNCTION (session_manager_main_init); +VLIB_MAIN_LOOP_ENTER_FUNCTION (session_main_init); static clib_error_t * session_config_fn (vlib_main_t * vm, unformat_input_t * input) @@ -1650,6 +1768,8 @@ session_config_fn (vlib_main_t * vm, unformat_input_t * input) else if (unformat (input, "evt_qs_seg_size %U", unformat_memory_size, &smm->evt_qs_segment_size)) ; + else if (unformat (input, "enable")) + smm->session_enable_asap = 1; else return clib_error_return (0, "unknown input `%U'", format_unformat_error, input);