X-Git-Url: https://gerrit.fd.io/r/gitweb?a=blobdiff_plain;f=src%2Fvnet%2Fsession%2Fsession_node.c;h=605fd584e2dcdabcfd06024ab9a20c279eb4cea5;hb=317b8e08367c769b90463613231b9fcfad486098;hp=d0936c7e13dfad5dae13918a432fd3f430466b79;hpb=31c9955a4e83578422b3328ef0d7b2b248536644;p=vpp.git diff --git a/src/vnet/session/session_node.c b/src/vnet/session/session_node.c index d0936c7e13d..605fd584e2d 100644 --- a/src/vnet/session/session_node.c +++ b/src/vnet/session/session_node.c @@ -74,22 +74,22 @@ session_mq_accepted_reply_handler (void *data) if (!session_has_transport (s)) { + s->session_state = SESSION_STATE_READY; if (ct_session_connect_notify (s)) return; - s->session_state = SESSION_STATE_READY; } else { old_state = s->session_state; s->session_state = SESSION_STATE_READY; - if (!svm_fifo_is_empty (s->rx_fifo)) + + if (!svm_fifo_is_empty_prod (s->rx_fifo)) app_worker_lock_and_send_event (app_wrk, s, SESSION_IO_EVT_RX); /* Closed while waiting for app to reply. Resend disconnect */ if (old_state >= SESSION_STATE_TRANSPORT_CLOSING) { - application_t *app = application_get (app_wrk->app_index); - app->cb_fns.session_disconnect_callback (s); + app_worker_close_notify (app_wrk, s); s->session_state = old_state; return; } @@ -289,7 +289,7 @@ session_mq_worker_update_handler (void *data) app_worker_lock_and_send_event (app_wrk, s, SESSION_IO_EVT_RX); if (s->session_state >= SESSION_STATE_TRANSPORT_CLOSING) - app->cb_fns.session_disconnect_callback (s); + app_worker_close_notify (app_wrk, s); } vlib_node_registration_t session_queue_node; @@ -558,7 +558,7 @@ session_tx_set_dequeue_params (vlib_main_t * vm, session_tx_context_t * ctx, u32 max_segs, u8 peek_data) { u32 n_bytes_per_buf, n_bytes_per_seg; - ctx->max_dequeue = svm_fifo_max_dequeue (ctx->s->tx_fifo); + ctx->max_dequeue = svm_fifo_max_dequeue_cons (ctx->s->tx_fifo); if (peek_data) { /* Offset in rx fifo from where to peek data */ @@ -789,7 +789,7 @@ session_tx_fifo_read_and_snd_i (vlib_main_t * vm, vlib_node_runtime_t * node, svm_fifo_overwrite_head (ctx->s->tx_fifo, (u8 *) & ctx->hdr, sizeof (session_dgram_pre_hdr_t)); /* More data needs to be read */ - else if (svm_fifo_max_dequeue (ctx->s->tx_fifo) > 0) + else if (svm_fifo_max_dequeue_cons (ctx->s->tx_fifo) > 0) if (svm_fifo_set_event (ctx->s->tx_fifo)) vec_add1 (wrk->pending_event_vector, *e); } @@ -819,19 +819,17 @@ session_tx_fifo_dequeue_internal (vlib_main_t * vm, session_event_t * e, int *n_tx_pkts) { session_t *s = wrk->ctx.s; - application_t *app; - if (PREDICT_FALSE (s->session_state == SESSION_STATE_CLOSED)) + if (PREDICT_FALSE (s->session_state >= SESSION_STATE_TRANSPORT_CLOSED)) return 0; - app = application_get (s->t_app_index); svm_fifo_unset_event (s->tx_fifo); - return app->cb_fns.builtin_app_tx_callback (s); + return transport_custom_tx (session_get_transport_proto (s), s); } always_inline session_t * session_event_get_session (session_event_t * e, u8 thread_index) { - return session_get_if_valid (e->fifo->master_session_index, thread_index); + return session_get_if_valid (e->session_index, thread_index); } static void @@ -858,7 +856,6 @@ session_queue_node_fn (vlib_main_t * vm, vlib_node_runtime_t * node, f64 now = vlib_time_now (vm); int n_tx_packets = 0, i, rv; app_worker_t *app_wrk; - application_t *app; svm_msg_q_t *mq; void (*fp) (void *); @@ -930,6 +927,7 @@ session_queue_node_fn (vlib_main_t * vm, vlib_node_runtime_t * node, clib_warning ("session was freed!"); continue; } + CLIB_PREFETCH (s->tx_fifo, 2 * CLIB_CACHE_LINE_BYTES, LOAD); wrk->ctx.s = s; /* Spray packets in per session type frames, since they go to * different nodes */ @@ -949,6 +947,13 @@ session_queue_node_fn (vlib_main_t * vm, vlib_node_runtime_t * node, continue; } break; + case SESSION_IO_EVT_RX: + s = session_event_get_session (e, thread_index); + if (!s) + break; + transport_app_rx_evt (session_get_transport_proto (s), + s->connection_index, s->thread_index); + break; case SESSION_CTRL_EVT_CLOSE: s = session_get_from_handle_if_valid (e->session_handle); if (PREDICT_FALSE (!s)) @@ -959,7 +964,8 @@ session_queue_node_fn (vlib_main_t * vm, vlib_node_runtime_t * node, * and the tx queue is still not empty, try to wait for some * dispatch cycles */ if (!e->postponed - || (e->postponed < 200 && svm_fifo_max_dequeue (s->tx_fifo))) + || (e->postponed < 200 + && svm_fifo_max_dequeue_cons (s->tx_fifo))) { e->postponed += 1; vec_add1 (wrk->pending_disconnects, *e); @@ -974,8 +980,7 @@ session_queue_node_fn (vlib_main_t * vm, vlib_node_runtime_t * node, continue; svm_fifo_unset_event (s->rx_fifo); app_wrk = app_worker_get (s->app_wrk_index); - app = application_get (app_wrk->app_index); - app->cb_fns.builtin_app_rx_callback (s); + app_worker_builtin_rx (app_wrk, s); break; case SESSION_IO_EVT_BUILTIN_TX: s = session_get_from_handle_if_valid (e->session_handle); @@ -1103,7 +1108,7 @@ session_node_cmp_event (session_event_t * e, svm_fifo_t * f) case SESSION_IO_EVT_RX: case SESSION_IO_EVT_TX: case SESSION_IO_EVT_BUILTIN_RX: - if (e->fifo == f) + if (e->session_index == f->master_session_index) return 1; break; case SESSION_CTRL_EVT_CLOSE: @@ -1234,6 +1239,26 @@ VLIB_REGISTER_NODE (session_queue_process_node) = }; /* *INDENT-ON* */ +static_always_inline uword +session_queue_pre_input_inline (vlib_main_t * vm, vlib_node_runtime_t * node, + vlib_frame_t * frame) +{ + session_main_t *sm = &session_main; + if (!sm->wrk[0].vpp_event_queue) + return 0; + return session_queue_node_fn (vm, node, frame); +} + +/* *INDENT-OFF* */ +VLIB_REGISTER_NODE (session_queue_pre_input_node) = +{ + .function = session_queue_pre_input_inline, + .type = VLIB_NODE_TYPE_PRE_INPUT, + .name = "session-queue-main", + .state = VLIB_NODE_STATE_DISABLED, +}; +/* *INDENT-ON* */ + /* * fd.io coding-style-patch-verification: ON *