X-Git-Url: https://gerrit.fd.io/r/gitweb?a=blobdiff_plain;f=src%2Fplugins%2Fhs_apps%2Fproxy.c;h=9513fab20d9fe57d9160a26b4157c1e97890e706;hb=4c5343600d4fa4351816dd44a26c6a9b81e6b47e;hp=a520110dce81a4eb1850eb7bbf0c9237d83373a6;hpb=b71ef86c7429e81921997e54a526bf666a2e4089;p=vpp.git diff --git a/src/plugins/hs_apps/proxy.c b/src/plugins/hs_apps/proxy.c index a520110dce8..9513fab20d9 100644 --- a/src/plugins/hs_apps/proxy.c +++ b/src/plugins/hs_apps/proxy.c @@ -61,113 +61,215 @@ proxy_call_main_thread (vnet_connect_args_t * a) } } +static proxy_session_t * +proxy_get_active_open (proxy_main_t * pm, session_handle_t handle) +{ + proxy_session_t *ps = 0; + uword *p; + + p = hash_get (pm->proxy_session_by_active_open_handle, handle); + if (p) + ps = pool_elt_at_index (pm->sessions, p[0]); + return ps; +} + +static proxy_session_t * +proxy_get_passive_open (proxy_main_t * pm, session_handle_t handle) +{ + proxy_session_t *ps = 0; + uword *p; + + p = hash_get (pm->proxy_session_by_server_handle, handle); + if (p) + ps = pool_elt_at_index (pm->sessions, p[0]); + return ps; +} + static void -delete_proxy_session (session_t * s, int is_active_open) +proxy_try_close_session (session_t * s, int is_active_open) { proxy_main_t *pm = &proxy_main; proxy_session_t *ps = 0; vnet_disconnect_args_t _a, *a = &_a; - session_t *active_open_session = 0; - session_t *server_session = 0; - uword *p; - u64 handle; - - clib_spinlock_lock_if_init (&pm->sessions_lock); + session_handle_t handle; handle = session_handle (s); + clib_spinlock_lock_if_init (&pm->sessions_lock); + if (is_active_open) { - p = hash_get (pm->proxy_session_by_active_open_handle, handle); - if (p == 0) - { - clib_warning ("proxy session for %s handle %lld (%llx) AWOL", - is_active_open ? "active open" : "server", - handle, handle); - } - else if (!pool_is_free_index (pm->sessions, p[0])) + ps = proxy_get_active_open (pm, handle); + ASSERT (ps != 0); + + a->handle = ps->vpp_active_open_handle; + a->app_index = pm->active_open_app_index; + vnet_disconnect_session (a); + ps->ao_disconnected = 1; + + if (!ps->po_disconnected) { - active_open_session = s; - ps = pool_elt_at_index (pm->sessions, p[0]); - if (ps->vpp_server_handle != ~0) - server_session = session_get_from_handle (ps->vpp_server_handle); + ASSERT (ps->vpp_server_handle != SESSION_INVALID_HANDLE); + a->handle = ps->vpp_server_handle; + a->app_index = pm->server_app_index; + vnet_disconnect_session (a); + ps->po_disconnected = 1; } } else { - p = hash_get (pm->proxy_session_by_server_handle, handle); - if (p == 0) - { - clib_warning ("proxy session for %s handle %lld (%llx) AWOL", - is_active_open ? "active open" : "server", - handle, handle); - } - else if (!pool_is_free_index (pm->sessions, p[0])) + ps = proxy_get_passive_open (pm, handle); + ASSERT (ps != 0); + + a->handle = ps->vpp_server_handle; + a->app_index = pm->server_app_index; + vnet_disconnect_session (a); + ps->po_disconnected = 1; + + if (!ps->ao_disconnected && !ps->active_open_establishing) { - server_session = s; - ps = pool_elt_at_index (pm->sessions, p[0]); - if (ps->vpp_active_open_handle != ~0) - active_open_session = session_get_from_handle - (ps->vpp_active_open_handle); + /* Proxy session closed before active open */ + if (ps->vpp_active_open_handle != SESSION_INVALID_HANDLE) + { + a->handle = ps->vpp_active_open_handle; + a->app_index = pm->active_open_app_index; + vnet_disconnect_session (a); + } + ps->ao_disconnected = 1; } } + clib_spinlock_unlock_if_init (&pm->sessions_lock); +} + +static void +proxy_session_free (proxy_session_t * ps) +{ + proxy_main_t *pm = &proxy_main; + if (CLIB_DEBUG > 0) + clib_memset (ps, 0xFE, sizeof (*ps)); + pool_put (pm->sessions, ps); +} + +static void +proxy_try_delete_session (session_t * s, u8 is_active_open) +{ + proxy_main_t *pm = &proxy_main; + proxy_session_t *ps = 0; + session_handle_t handle; + + handle = session_handle (s); - if (ps) + clib_spinlock_lock_if_init (&pm->sessions_lock); + + if (is_active_open) { - if (CLIB_DEBUG > 0) - clib_memset (ps, 0xFE, sizeof (*ps)); - pool_put (pm->sessions, ps); - } + ps = proxy_get_active_open (pm, handle); + ASSERT (ps != 0); + + ps->vpp_active_open_handle = SESSION_INVALID_HANDLE; + hash_unset (pm->proxy_session_by_active_open_handle, handle); - if (active_open_session) + if (ps->vpp_server_handle == SESSION_INVALID_HANDLE) + proxy_session_free (ps); + } + else { - a->handle = session_handle (active_open_session); - a->app_index = pm->active_open_app_index; - hash_unset (pm->proxy_session_by_active_open_handle, - session_handle (active_open_session)); - vnet_disconnect_session (a); + ps = proxy_get_passive_open (pm, handle); + ASSERT (ps != 0); + + ps->vpp_server_handle = SESSION_INVALID_HANDLE; + hash_unset (pm->proxy_session_by_server_handle, handle); + + if (ps->vpp_active_open_handle == SESSION_INVALID_HANDLE) + { + if (!ps->active_open_establishing) + proxy_session_free (ps); + } } + clib_spinlock_unlock_if_init (&pm->sessions_lock); +} + +static int +common_fifo_tuning_callback (session_t * s, svm_fifo_t * f, + session_ft_action_t act, u32 bytes) +{ + proxy_main_t *pm = &proxy_main; + + segment_manager_t *sm = segment_manager_get (f->segment_manager); + fifo_segment_t *fs = segment_manager_get_segment (sm, f->segment_index); + + u8 seg_usage = fifo_segment_get_mem_usage (fs); + u32 fifo_in_use = svm_fifo_max_dequeue_prod (f); + u32 fifo_size = svm_fifo_size (f); + u8 fifo_usage = fifo_in_use * 100 / fifo_size; + u8 update_size = 0; - if (server_session) + ASSERT (act < SESSION_FT_ACTION_N_ACTIONS); + + if (act == SESSION_FT_ACTION_ENQUEUED) { - a->handle = session_handle (server_session); - a->app_index = pm->server_app_index; - hash_unset (pm->proxy_session_by_server_handle, - session_handle (server_session)); - vnet_disconnect_session (a); + if (seg_usage < pm->low_watermark && fifo_usage > 50) + update_size = fifo_in_use; + else if (seg_usage < pm->high_watermark && fifo_usage > 80) + update_size = fifo_in_use; + + update_size = clib_min (update_size, sm->max_fifo_size - fifo_size); + if (update_size) + svm_fifo_set_size (f, fifo_size + update_size); + } + else /* dequeued */ + { + if (seg_usage > pm->high_watermark || fifo_usage < 20) + update_size = bytes; + else if (seg_usage > pm->low_watermark && fifo_usage < 50) + update_size = (bytes / 2); + + ASSERT (fifo_size >= 4096); + update_size = clib_min (update_size, fifo_size - 4096); + if (update_size) + svm_fifo_set_size (f, fifo_size - update_size); } - clib_spinlock_unlock_if_init (&pm->sessions_lock); + return 0; } static int proxy_accept_callback (session_t * s) { proxy_main_t *pm = &proxy_main; - - s->session_state = SESSION_STATE_READY; + proxy_session_t *ps; clib_spinlock_lock_if_init (&pm->sessions_lock); + pool_get_zero (pm->sessions, ps); + ps->vpp_server_handle = session_handle (s); + ps->vpp_active_open_handle = SESSION_INVALID_HANDLE; + + hash_set (pm->proxy_session_by_server_handle, ps->vpp_server_handle, + ps - pm->sessions); + + clib_spinlock_unlock_if_init (&pm->sessions_lock); + + s->session_state = SESSION_STATE_READY; + return 0; } static void proxy_disconnect_callback (session_t * s) { - delete_proxy_session (s, 0 /* is_active_open */ ); + proxy_try_close_session (s, 0 /* is_active_open */ ); } static void proxy_reset_callback (session_t * s) { - clib_warning ("Reset session %U", format_session, s, 2); - delete_proxy_session (s, 0 /* is_active_open */ ); + proxy_try_close_session (s, 0 /* is_active_open */ ); } static int proxy_connected_callback (u32 app_index, u32 api_context, - session_t * s, u8 is_fail) + session_t * s, session_error_t err) { clib_warning ("called..."); return -1; @@ -183,25 +285,22 @@ proxy_add_segment_callback (u32 client_index, u64 segment_handle) static int proxy_rx_callback (session_t * s) { - u32 max_dequeue; - int actual_transfer __attribute__ ((unused)); - svm_fifo_t *tx_fifo, *rx_fifo; proxy_main_t *pm = &proxy_main; u32 thread_index = vlib_get_thread_index (); - vnet_connect_args_t _a, *a = &_a; - proxy_session_t *ps; - int proxy_index; - uword *p; svm_fifo_t *ao_tx_fifo; + proxy_session_t *ps; ASSERT (s->thread_index == thread_index); clib_spinlock_lock_if_init (&pm->sessions_lock); - p = hash_get (pm->proxy_session_by_server_handle, session_handle (s)); - if (PREDICT_TRUE (p != 0)) + ps = proxy_get_passive_open (pm, session_handle (s)); + ASSERT (ps != 0); + + if (PREDICT_TRUE (ps->vpp_active_open_handle != SESSION_INVALID_HANDLE)) { clib_spinlock_unlock_if_init (&pm->sessions_lock); + ao_tx_fifo = s->rx_fifo; /* @@ -222,6 +321,11 @@ proxy_rx_callback (session_t * s) } else { + vnet_connect_args_t _a, *a = &_a; + svm_fifo_t *tx_fifo, *rx_fifo; + u32 max_dequeue, proxy_index; + int actual_transfer __attribute__ ((unused)); + rx_fifo = s->rx_fifo; tx_fifo = s->tx_fifo; @@ -241,17 +345,11 @@ proxy_rx_callback (session_t * s) clib_memset (a, 0, sizeof (*a)); - pool_get (pm->sessions, ps); - clib_memset (ps, 0, sizeof (*ps)); ps->server_rx_fifo = rx_fifo; ps->server_tx_fifo = tx_fifo; - ps->vpp_server_handle = session_handle (s); - + ps->active_open_establishing = 1; proxy_index = ps - pm->sessions; - hash_set (pm->proxy_session_by_server_handle, ps->vpp_server_handle, - proxy_index); - clib_spinlock_unlock_if_init (&pm->sessions_lock); a->uri = (char *) pm->client_uri; @@ -263,18 +361,25 @@ proxy_rx_callback (session_t * s) return 0; } +static void +proxy_force_ack (void *handlep) +{ + transport_connection_t *tc; + session_t *ao_s; + + ao_s = session_get_from_handle (pointer_to_uword (handlep)); + tc = session_get_transport (ao_s); + tcp_send_ack ((tcp_connection_t *) tc); +} + static int proxy_tx_callback (session_t * proxy_s) { proxy_main_t *pm = &proxy_main; - transport_connection_t *tc; - session_handle_t handle; proxy_session_t *ps; - session_t *ao_s; u32 min_free; - uword *p; - min_free = clib_min (proxy_s->tx_fifo->nitems >> 3, 128 << 10); + min_free = clib_min (svm_fifo_size (proxy_s->tx_fifo) >> 3, 128 << 10); if (svm_fifo_max_enqueue (proxy_s->tx_fifo) < min_free) { svm_fifo_add_want_deq_ntf (proxy_s->tx_fifo, SVM_FIFO_WANT_DEQ_NOTIF); @@ -283,29 +388,32 @@ proxy_tx_callback (session_t * proxy_s) clib_spinlock_lock_if_init (&pm->sessions_lock); - handle = session_handle (proxy_s); - p = hash_get (pm->proxy_session_by_server_handle, handle); - if (!p) - return 0; + ps = proxy_get_passive_open (pm, session_handle (proxy_s)); + ASSERT (ps != 0); - if (pool_is_free_index (pm->sessions, p[0])) + if (ps->vpp_active_open_handle == SESSION_INVALID_HANDLE) return 0; - ps = pool_elt_at_index (pm->sessions, p[0]); - if (ps->vpp_active_open_handle == ~0) - return 0; - - ao_s = session_get_from_handle (ps->vpp_active_open_handle); - - /* Force ack on active open side to update rcv wnd */ - tc = session_get_transport (ao_s); - tcp_send_ack ((tcp_connection_t *) tc); + /* Force ack on active open side to update rcv wnd. Make sure it's done on + * the right thread */ + void *arg = uword_to_pointer (ps->vpp_active_open_handle, void *); + session_send_rpc_evt_to_thread (ps->server_rx_fifo->master_thread_index, + proxy_force_ack, arg); clib_spinlock_unlock_if_init (&pm->sessions_lock); return 0; } +static void +proxy_cleanup_callback (session_t * s, session_cleanup_ntf_t ntf) +{ + if (ntf == SESSION_CLEANUP_TRANSPORT) + return; + + proxy_try_delete_session (s, 0 /* is_active_open */ ); +} + static session_cb_vft_t proxy_session_cb_vft = { .session_accept_callback = proxy_accept_callback, .session_disconnect_callback = proxy_disconnect_callback, @@ -313,30 +421,52 @@ static session_cb_vft_t proxy_session_cb_vft = { .add_segment_callback = proxy_add_segment_callback, .builtin_app_rx_callback = proxy_rx_callback, .builtin_app_tx_callback = proxy_tx_callback, - .session_reset_callback = proxy_reset_callback + .session_reset_callback = proxy_reset_callback, + .session_cleanup_callback = proxy_cleanup_callback, + .fifo_tuning_callback = common_fifo_tuning_callback }; static int active_open_connected_callback (u32 app_index, u32 opaque, - session_t * s, u8 is_fail) + session_t * s, session_error_t err) { proxy_main_t *pm = &proxy_main; proxy_session_t *ps; u8 thread_index = vlib_get_thread_index (); - if (is_fail) - { - clib_warning ("connection %d failed!", opaque); - return 0; - } - /* * Setup proxy session handle. */ clib_spinlock_lock_if_init (&pm->sessions_lock); ps = pool_elt_at_index (pm->sessions, opaque); - ps->vpp_active_open_handle = session_handle (s); + + /* Connection failed */ + if (err) + { + vnet_disconnect_args_t _a, *a = &_a; + + a->handle = ps->vpp_server_handle; + a->app_index = pm->server_app_index; + vnet_disconnect_session (a); + ps->po_disconnected = 1; + } + else + { + ps->vpp_active_open_handle = session_handle (s); + ps->active_open_establishing = 0; + } + + /* Passive open session was already closed! */ + if (ps->po_disconnected) + { + /* Setup everything for the cleanup notification */ + hash_set (pm->proxy_session_by_active_open_handle, + ps->vpp_active_open_handle, opaque); + ps->ao_disconnected = 1; + clib_spinlock_unlock_if_init (&pm->sessions_lock); + return -1; + } s->tx_fifo = ps->server_rx_fifo; s->rx_fifo = ps->server_tx_fifo; @@ -356,6 +486,9 @@ active_open_connected_callback (u32 app_index, u32 opaque, s->tx_fifo->refcnt++; s->rx_fifo->refcnt++; + svm_fifo_init_ooo_lookup (s->tx_fifo, 1 /* deq ooo */ ); + svm_fifo_init_ooo_lookup (s->rx_fifo, 0 /* enq ooo */ ); + hash_set (pm->proxy_session_by_active_open_handle, ps->vpp_active_open_handle, opaque); @@ -374,7 +507,7 @@ active_open_connected_callback (u32 app_index, u32 opaque, static void active_open_reset_callback (session_t * s) { - delete_proxy_session (s, 1 /* is_active_open */ ); + proxy_try_close_session (s, 1 /* is_active_open */ ); } static int @@ -386,7 +519,7 @@ active_open_create_callback (session_t * s) static void active_open_disconnect_callback (session_t * s) { - delete_proxy_session (s, 1 /* is_active_open */ ); + proxy_try_close_session (s, 1 /* is_active_open */ ); } static int @@ -425,7 +558,7 @@ active_open_tx_callback (session_t * ao_s) u32 min_free; uword *p; - min_free = clib_min (ao_s->tx_fifo->nitems >> 3, 128 << 10); + min_free = clib_min (svm_fifo_size (ao_s->tx_fifo) >> 3, 128 << 10); if (svm_fifo_max_enqueue (ao_s->tx_fifo) < min_free) { svm_fifo_add_want_deq_ntf (ao_s->tx_fifo, SVM_FIFO_WANT_DEQ_NOTIF); @@ -457,14 +590,25 @@ active_open_tx_callback (session_t * ao_s) return 0; } +static void +active_open_cleanup_callback (session_t * s, session_cleanup_ntf_t ntf) +{ + if (ntf == SESSION_CLEANUP_TRANSPORT) + return; + + proxy_try_delete_session (s, 1 /* is_active_open */ ); +} + /* *INDENT-OFF* */ static session_cb_vft_t active_open_clients = { .session_reset_callback = active_open_reset_callback, .session_connected_callback = active_open_connected_callback, .session_accept_callback = active_open_create_callback, .session_disconnect_callback = active_open_disconnect_callback, + .session_cleanup_callback = active_open_cleanup_callback, .builtin_app_rx_callback = active_open_rx_callback, .builtin_app_tx_callback = active_open_tx_callback, + .fifo_tuning_callback = common_fifo_tuning_callback }; /* *INDENT-ON* */ @@ -488,6 +632,9 @@ proxy_server_attach () a->options[APP_OPTIONS_SEGMENT_SIZE] = segment_size; a->options[APP_OPTIONS_RX_FIFO_SIZE] = pm->fifo_size; a->options[APP_OPTIONS_TX_FIFO_SIZE] = pm->fifo_size; + a->options[APP_OPTIONS_MAX_FIFO_SIZE] = pm->max_fifo_size; + a->options[APP_OPTIONS_HIGH_WATERMARK] = (u64) pm->high_watermark; + a->options[APP_OPTIONS_LOW_WATERMARK] = (u64) pm->low_watermark; a->options[APP_OPTIONS_PRIVATE_SEGMENT_COUNT] = pm->private_segment_count; a->options[APP_OPTIONS_PREALLOC_FIFO_PAIRS] = pm->prealloc_fifos ? pm->prealloc_fifos : 0; @@ -510,7 +657,7 @@ active_open_attach (void) { proxy_main_t *pm = &proxy_main; vnet_app_attach_args_t _a, *a = &_a; - u64 options[16]; + u64 options[APP_OPTIONS_N_OPTIONS]; clib_memset (a, 0, sizeof (*a)); clib_memset (options, 0, sizeof (options)); @@ -523,6 +670,9 @@ active_open_attach (void) options[APP_OPTIONS_SEGMENT_SIZE] = 512 << 20; options[APP_OPTIONS_RX_FIFO_SIZE] = pm->fifo_size; options[APP_OPTIONS_TX_FIFO_SIZE] = pm->fifo_size; + options[APP_OPTIONS_MAX_FIFO_SIZE] = pm->max_fifo_size; + options[APP_OPTIONS_HIGH_WATERMARK] = (u64) pm->high_watermark; + options[APP_OPTIONS_LOW_WATERMARK] = (u64) pm->low_watermark; options[APP_OPTIONS_PRIVATE_SEGMENT_COUNT] = pm->private_segment_count; options[APP_OPTIONS_PREALLOC_FIFO_PAIRS] = pm->prealloc_fifos ? pm->prealloc_fifos : 0; @@ -604,21 +754,34 @@ proxy_server_create_command_fn (vlib_main_t * vm, unformat_input_t * input, proxy_main_t *pm = &proxy_main; char *default_server_uri = "tcp://0.0.0.0/23"; char *default_client_uri = "tcp://6.0.2.2/23"; - int rv; - u64 tmp; + int rv, tmp32; + u64 tmp64; pm->fifo_size = 64 << 10; + pm->max_fifo_size = 128 << 20; + pm->high_watermark = 80; + pm->low_watermark = 50; pm->rcv_buffer_size = 1024; pm->prealloc_fifos = 0; pm->private_segment_count = 0; pm->private_segment_size = 0; pm->server_uri = 0; pm->client_uri = 0; + if (vlib_num_workers ()) + clib_spinlock_init (&pm->sessions_lock); while (unformat_check_input (input) != UNFORMAT_END_OF_INPUT) { - if (unformat (input, "fifo-size %d", &pm->fifo_size)) - pm->fifo_size <<= 10; + if (unformat (input, "fifo-size %U", + unformat_memory_size, &pm->fifo_size)) + ; + else if (unformat (input, "max-fifo-size %U", + unformat_memory_size, &pm->max_fifo_size)) + ; + else if (unformat (input, "high-watermark %d", &tmp32)) + pm->high_watermark = (u8) tmp32; + else if (unformat (input, "low-watermark %d", &tmp32)) + pm->low_watermark = (u8) tmp32; else if (unformat (input, "rcv-buf-size %d", &pm->rcv_buffer_size)) ; else if (unformat (input, "prealloc-fifos %d", &pm->prealloc_fifos)) @@ -627,12 +790,12 @@ proxy_server_create_command_fn (vlib_main_t * vm, unformat_input_t * input, &pm->private_segment_count)) ; else if (unformat (input, "private-segment-size %U", - unformat_memory_size, &tmp)) + unformat_memory_size, &tmp64)) { - if (tmp >= 0x100000000ULL) + if (tmp64 >= 0x100000000ULL) return clib_error_return - (0, "private segment size %lld (%llu) too large", tmp, tmp); - pm->private_segment_size = tmp; + (0, "private segment size %lld (%llu) too large", tmp64, tmp64); + pm->private_segment_size = tmp64; } else if (unformat (input, "server-uri %s", &pm->server_uri)) vec_add1 (pm->server_uri, 0); @@ -675,9 +838,10 @@ VLIB_CLI_COMMAND (proxy_create_command, static) = { .path = "test proxy server", .short_help = "test proxy server [server-uri ]" - "[client-uri ][fifo-size ][rcv-buf-size ]" - "[prealloc-fifos ][private-segment-size ]" - "[private-segment-count ]", + "[client-uri ][fifo-size [k|m]]" + "[max-fifo-size [k|m]][high-watermark ]" + "[low-watermark ][rcv-buf-size ][prealloc-fifos ]" + "[private-segment-size ][private-segment-count ]", .function = proxy_server_create_command_fn, }; /* *INDENT-ON* */