summaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
-rw-r--r--src/plugins/hs_apps/proxy.c276
-rw-r--r--src/plugins/hs_apps/proxy.h7
2 files changed, 188 insertions, 95 deletions
diff --git a/src/plugins/hs_apps/proxy.c b/src/plugins/hs_apps/proxy.c
index 83b26222adb..6dd3f56ec57 100644
--- a/src/plugins/hs_apps/proxy.c
+++ b/src/plugins/hs_apps/proxy.c
@@ -61,82 +61,131 @@ 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);
+}
- if (ps)
- {
- if (CLIB_DEBUG > 0)
- clib_memset (ps, 0xFE, sizeof (*ps));
- pool_put (pm->sessions, ps);
- }
+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);
+}
- if (active_open_session)
- {
- 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);
- }
+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);
+
+ clib_spinlock_lock_if_init (&pm->sessions_lock);
- if (server_session)
+ if (is_active_open)
{
- 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);
+ 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 (ps->vpp_server_handle == SESSION_INVALID_HANDLE)
+ proxy_session_free (ps);
}
+ else
+ {
+ 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);
}
@@ -188,25 +237,34 @@ 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
@@ -227,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;
/*
@@ -266,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;
@@ -285,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;
@@ -307,16 +361,23 @@ 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 (svm_fifo_size (proxy_s->tx_fifo) >> 3, 128 << 10);
if (svm_fifo_max_enqueue (proxy_s->tx_fifo) < min_free)
@@ -327,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,
@@ -358,6 +422,7 @@ static session_cb_vft_t proxy_session_cb_vft = {
.builtin_app_rx_callback = proxy_rx_callback,
.builtin_app_tx_callback = proxy_tx_callback,
.session_reset_callback = proxy_reset_callback,
+ .session_cleanup_callback = proxy_cleanup_callback,
.fifo_tuning_callback = common_fifo_tuning_callback
};
@@ -372,6 +437,7 @@ active_open_connected_callback (u32 app_index, u32 opaque,
if (err)
{
clib_warning ("connection %d failed!", opaque);
+ ASSERT (0);
return 0;
}
@@ -382,6 +448,18 @@ active_open_connected_callback (u32 app_index, u32 opaque,
ps = pool_elt_at_index (pm->sessions, opaque);
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;
@@ -422,7 +500,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
@@ -434,7 +512,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
@@ -505,12 +583,22 @@ 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
@@ -672,6 +760,8 @@ proxy_server_create_command_fn (vlib_main_t * vm, unformat_input_t * input,
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)
{
diff --git a/src/plugins/hs_apps/proxy.h b/src/plugins/hs_apps/proxy.h
index 87a96186b22..4f74ea025fb 100644
--- a/src/plugins/hs_apps/proxy.h
+++ b/src/plugins/hs_apps/proxy.h
@@ -31,8 +31,11 @@ typedef struct
svm_fifo_t *server_rx_fifo;
svm_fifo_t *server_tx_fifo;
- u64 vpp_server_handle;
- u64 vpp_active_open_handle;
+ session_handle_t vpp_server_handle;
+ session_handle_t vpp_active_open_handle;
+ volatile int active_open_establishing;
+ volatile int po_disconnected;
+ volatile int ao_disconnected;
} proxy_session_t;
typedef struct