X-Git-Url: https://gerrit.fd.io/r/gitweb?a=blobdiff_plain;f=src%2Fvnet%2Fsession%2Fsession.c;h=e9cda361f37b37299e926710375d19f7719ae3d4;hb=d8f48e2166747b529aa23762ac314dd686590c89;hp=318e01dbf54340acf3af21e3ccd10c518794c3a2;hpb=b0ffbee67bd17f6880a02184d18cd94156f2389e;p=vpp.git diff --git a/src/vnet/session/session.c b/src/vnet/session/session.c index 318e01dbf54..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) { @@ -235,7 +263,7 @@ session_delete (session_t * s) /* Delete from the main lookup table. */ if ((rv = session_lookup_del_session (s))) - clib_warning ("hash delete error, rv %d", rv); + clib_warning ("session %u hash delete rv %d", s->session_index, rv); session_free_w_fifos (s); } @@ -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); @@ -565,11 +616,45 @@ session_enqueue_notify (session_t * s) return session_enqueue_notify_inline (s); } +static void +session_enqueue_notify_rpc (void *arg) +{ + u32 session_index = pointer_to_uword (arg); + session_t *s; + + s = session_get_if_valid (session_index, vlib_get_thread_index ()); + if (!s) + return; + + session_enqueue_notify (s); +} + +/** + * Like session_enqueue_notify, but can be called from a thread that does not + * own the session. + */ +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, + uword_to_pointer (session_index, void *)); +} + int 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; @@ -582,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; } @@ -614,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++; } @@ -715,17 +801,35 @@ typedef struct _session_switch_pool_args u32 new_session_index; } session_switch_pool_args_t; +/** + * Notify old thread of the session pool switch + */ static void session_switch_pool (void *cb_args) { session_switch_pool_args_t *args = (session_switch_pool_args_t *) cb_args; + app_worker_t *app_wrk; session_t *s; + ASSERT (args->thread_index == vlib_get_thread_index ()); s = session_get (args->session_index, args->thread_index); s->tx_fifo->master_session_index = args->new_session_index; s->tx_fifo->master_thread_index = args->new_thread_index; transport_cleanup (session_get_transport_proto (s), s->connection_index, s->thread_index); + + app_wrk = app_worker_get_if_valid (s->app_wrk_index); + if (app_wrk) + { + session_handle_t new_sh; + new_sh = session_make_handle (args->new_session_index, + args->new_thread_index); + app_worker_migrate_notify (app_wrk, s, new_sh); + + /* Trigger app read on the new thread */ + session_enqueue_notify_thread (new_sh); + } + session_free (s); clib_mem_free (cb_args); } @@ -748,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)); /* @@ -812,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: @@ -835,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; @@ -923,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; } /** @@ -1144,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); } /** @@ -1184,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. * @@ -1194,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); } /** @@ -1219,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; @@ -1249,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++) { @@ -1274,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 * @@ -1408,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); @@ -1451,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; } @@ -1523,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) @@ -1607,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);