X-Git-Url: https://gerrit.fd.io/r/gitweb?a=blobdiff_plain;f=src%2Fvnet%2Fsession%2Fsession.c;h=c6f9d0a75ae1f9e0718db5e930503287eb913924;hb=458089bba;hp=5b4e840f053f40a2d10000ad67e491a88d7adadc;hpb=70f26d5425f72ebf207ad4ca6e83c1b7ff959c18;p=vpp.git diff --git a/src/vnet/session/session.c b/src/vnet/session/session.c index 5b4e840f053..c6f9d0a75ae 100644 --- a/src/vnet/session/session.c +++ b/src/vnet/session/session.c @@ -92,10 +92,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 @@ -118,25 +118,50 @@ session_send_rpc_evt_to_thread (u32 thread_index, void *fp, void *rpc_args) } } +void +session_add_self_custom_tx_evt (transport_connection_t * tc, u8 has_prio) +{ + session_t *s; + + s = session_get (tc->s_index, tc->thread_index); + ASSERT (s->thread_index == vlib_get_thread_index ()); + if (!(s->flags & SESSION_F_CUSTOM_TX)) + { + s->flags |= SESSION_F_CUSTOM_TX; + if (svm_fifo_set_event (s->tx_fifo)) + { + session_worker_t *wrk; + session_evt_elt_t *elt; + wrk = session_main_get_worker (tc->thread_index); + if (has_prio) + elt = session_evt_alloc_new (wrk); + else + elt = session_evt_alloc_old (wrk); + elt->evt.session_index = tc->s_index; + elt->evt.event_type = SESSION_IO_EVT_TX; + } + } +} + 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; session_worker_t *wrk; - session_event_t *evt; /* If we are in the handler thread, or being called with the worker barrier * held, just append a new event to pending disconnects vector. */ if (vlib_thread_is_main_w_barrier () || thread_index == s->thread_index) { wrk = session_main_get_worker (s->thread_index); - vec_add2 (wrk->pending_disconnects, evt, 1); - clib_memset (evt, 0, sizeof (*evt)); - evt->session_handle = session_handle (s); - evt->event_type = SESSION_CTRL_EVT_CLOSE; + 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 = evt; } else - session_send_ctrl_evt_to_thread (s, SESSION_CTRL_EVT_CLOSE); + session_send_ctrl_evt_to_thread (s, evt); } session_t * @@ -175,7 +200,7 @@ session_free (session_t * s) pool_put (session_main.wrk[thread_index].sessions, s); return; } - SESSION_EVT_DBG (SESSION_EVT_FREE, s); + SESSION_EVT (SESSION_EVT_FREE, s); pool_put (session_main.wrk[s->thread_index].sessions, s); } @@ -210,7 +235,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); } @@ -517,12 +542,7 @@ session_enqueue_notify_inline (session_t * s) return 0; } - /* *INDENT-OFF* */ - SESSION_EVT_DBG(SESSION_EVT_ENQ, s, ({ - ed->data[0] = SESSION_IO_EVT_RX; - ed->data[1] = svm_fifo_max_dequeue_prod (s->rx_fifo); - })); - /* *INDENT-ON* */ + SESSION_EVT (SESSION_EVT_ENQ, s, svm_fifo_max_dequeue_prod (s->rx_fifo)); s->flags &= ~SESSION_F_RX_EVT; if (PREDICT_FALSE (app_worker_lock_and_send_event (app_wrk, s, @@ -545,6 +565,38 @@ 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) { @@ -695,17 +747,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); } @@ -796,30 +866,31 @@ session_transport_delete_notify (transport_connection_t * tc) break; case SESSION_STATE_ACCEPTING: case SESSION_STATE_TRANSPORT_CLOSING: + case SESSION_STATE_CLOSING: + case SESSION_STATE_TRANSPORT_CLOSED: /* If transport finishes or times out before we get a reply * from the app, mark transport as closed and wait for reply * before removing the session. Cleanup session table in advance * because transport will soon be closed and closed sessions * are assumed to have been removed from the lookup table */ session_lookup_del_session (s); - s->session_state = SESSION_STATE_TRANSPORT_CLOSED; + s->session_state = SESSION_STATE_TRANSPORT_DELETED; session_cleanup_notify (s, SESSION_CLEANUP_TRANSPORT); svm_fifo_dequeue_drop_all (s->tx_fifo); break; - case SESSION_STATE_CLOSING: - case SESSION_STATE_CLOSED_WAITING: + case SESSION_STATE_APP_CLOSED: /* Cleanup lookup table as transport needs to still be valid. * Program transport close to ensure that all session events * have been cleaned up. Once transport close is called, the * session is just removed because both transport and app have * confirmed the close*/ session_lookup_del_session (s); - s->session_state = SESSION_STATE_TRANSPORT_CLOSED; - session_program_transport_close (s); + s->session_state = SESSION_STATE_CLOSED; session_cleanup_notify (s, SESSION_CLEANUP_TRANSPORT); svm_fifo_dequeue_drop_all (s->tx_fifo); + session_program_transport_ctrl_evt (s, SESSION_CTRL_EVT_CLOSE); break; - case SESSION_STATE_TRANSPORT_CLOSED: + case SESSION_STATE_TRANSPORT_DELETED: break; case SESSION_STATE_CLOSED: session_cleanup_notify (s, SESSION_CLEANUP_TRANSPORT); @@ -834,16 +905,17 @@ session_transport_delete_notify (transport_connection_t * tc) } /** - * Notification from transport that session can be closed + * Notification from transport that it is closed * - * Should be called by transport only if it was closed with non-empty - * tx fifo and once it decides to begin the closing procedure prior to - * issuing a delete notify. This gives the chance to the session layer - * to cleanup any outstanding events. + * Should be called by transport, prior to calling delete notify, once it + * knows that no more data will be exchanged. This could serve as an + * early acknowledgment of an active close especially if transport delete + * can be delayed a long time, e.g., tcp time-wait. */ void session_transport_closed_notify (transport_connection_t * tc) { + app_worker_t *app_wrk; session_t *s; if (!(s = session_get_if_valid (tc->s_index, tc->thread_index))) @@ -855,16 +927,21 @@ session_transport_closed_notify (transport_connection_t * tc) { session_transport_closing_notify (tc); svm_fifo_dequeue_drop_all (s->tx_fifo); + s->session_state = SESSION_STATE_TRANSPORT_CLOSED; } /* If app close has not been received or has not yet resulted in * a transport close, only mark the session transport as closed */ else if (s->session_state <= SESSION_STATE_CLOSING) { - session_lookup_del_session (s); s->session_state = SESSION_STATE_TRANSPORT_CLOSED; } - else + /* If app also closed, switch to closed */ + else if (s->session_state == SESSION_STATE_APP_CLOSED) s->session_state = SESSION_STATE_CLOSED; + + app_wrk = app_worker_get_if_valid (s->app_wrk_index); + if (app_wrk) + app_worker_transport_closed_notify (app_wrk, s); } /** @@ -1115,17 +1192,28 @@ session_close (session_t * s) { /* Session will only be removed once both app and transport * acknowledge the close */ - if (s->session_state == SESSION_STATE_TRANSPORT_CLOSED) - session_program_transport_close (s); - - /* Session already closed. Clear the tx fifo */ - if (s->session_state == SESSION_STATE_CLOSED) - svm_fifo_dequeue_drop_all (s->tx_fifo); + if (s->session_state == SESSION_STATE_TRANSPORT_CLOSED + || s->session_state == SESSION_STATE_TRANSPORT_DELETED) + 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); } /** @@ -1138,28 +1226,48 @@ session_close (session_t * s) void session_transport_close (session_t * s) { - /* If transport is already closed, just free the session */ - if (s->session_state >= SESSION_STATE_TRANSPORT_CLOSED) + if (s->session_state >= SESSION_STATE_APP_CLOSED) { - session_free_w_fifos (s); + if (s->session_state == SESSION_STATE_TRANSPORT_CLOSED) + s->session_state = SESSION_STATE_CLOSED; + /* If transport is already deleted, just free the session */ + else if (s->session_state >= SESSION_STATE_TRANSPORT_DELETED) + session_free_w_fifos (s); return; } - /* If tx queue wasn't drained, change state to closed waiting for transport. - * This way, the transport, if it so wishes, can continue to try sending the - * outstanding data (in closed state it cannot). It MUST however at one - * point, either after sending everything or after a timeout, call delete - * notify. This will finally lead to the complete cleanup of the session. + /* If the tx queue wasn't drained, the transport can continue to try + * sending the outstanding data (in closed state it cannot). It MUST however + * at one point, either after sending everything or after a timeout, call + * delete notify. This will finally lead to the complete cleanup of the + * session. */ - if (svm_fifo_max_dequeue_cons (s->tx_fifo)) - s->session_state = SESSION_STATE_CLOSED_WAITING; - else - s->session_state = SESSION_STATE_CLOSED; + s->session_state = SESSION_STATE_APP_CLOSED; transport_close (session_get_transport_proto (s), s->connection_index, 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. * @@ -1379,26 +1487,16 @@ session_manager_main_enable (vlib_main_t * vm) for (i = 0; i < num_threads; i++) { wrk = &smm->wrk[i]; - vec_validate (wrk->free_event_vector, 128); - _vec_len (wrk->free_event_vector) = 0; - vec_validate (wrk->pending_event_vector, 128); - _vec_len (wrk->pending_event_vector) = 0; - vec_validate (wrk->pending_disconnects, 128); - _vec_len (wrk->pending_disconnects) = 0; - vec_validate (wrk->postponed_event_vector, 128); - _vec_len (wrk->postponed_event_vector) = 0; - + wrk->ctrl_head = clib_llist_make_head (wrk->event_elts, evt_list); + wrk->new_head = clib_llist_make_head (wrk->event_elts, evt_list); + 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->dispatch_period = 500e-6; if (num_threads > 1) clib_rwlock_init (&smm->wrk[i].peekers_rw_locks); } -#if SESSION_DEBUG - vec_validate (smm->last_event_poll_by_thread, num_threads - 1); -#endif - /* Allocate vpp event queues segment and queue */ session_vpp_event_queues_allocate (smm); @@ -1484,8 +1582,8 @@ vnet_session_enable_disable (vlib_main_t * vm, u8 is_en) if (session_main.is_enabled) return 0; - session_node_enable_disable (is_en); error = session_manager_main_enable (vm); + session_node_enable_disable (is_en); } else {