2 * Copyright (c) 2015-2017 Cisco and/or its affiliates.
3 * Licensed under the Apache License, Version 2.0 (the "License");
4 * you may not use this file except in compliance with the License.
5 * You may obtain a copy of the License at:
7 * http://www.apache.org/licenses/LICENSE-2.0
9 * Unless required by applicable law or agreed to in writing, software
10 * distributed under the License is distributed on an "AS IS" BASIS,
11 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12 * See the License for the specific language governing permissions and
13 * limitations under the License.
16 #include <vnet/vnet.h>
17 #include <vlibmemory/api.h>
18 #include <vnet/session/application.h>
19 #include <vnet/session/application_interface.h>
20 #include <vnet/session-apps/proxy.h>
22 proxy_main_t proxy_main;
25 delete_proxy_session (stream_session_t * s, int is_active_open)
27 proxy_main_t *pm = &proxy_main;
28 proxy_session_t *ps = 0;
29 vnet_disconnect_args_t _a, *a = &_a;
30 stream_session_t *active_open_session = 0;
31 stream_session_t *server_session = 0;
35 handle = session_handle (s);
37 clib_spinlock_lock_if_init (&pm->sessions_lock);
40 active_open_session = s;
42 p = hash_get (pm->proxy_session_by_active_open_handle, handle);
45 clib_warning ("proxy session for %s handle %lld (%llx) AWOL",
46 is_active_open ? "active open" : "server",
51 ps = pool_elt_at_index (pm->sessions, p[0]);
52 if (ps->vpp_server_handle != ~0)
53 server_session = session_get_from_handle (ps->vpp_server_handle);
62 p = hash_get (pm->proxy_session_by_server_handle, handle);
65 clib_warning ("proxy session for %s handle %lld (%llx) AWOL",
66 is_active_open ? "active open" : "server",
71 ps = pool_elt_at_index (pm->sessions, p[0]);
72 if (ps->vpp_server_handle != ~0)
73 active_open_session = session_get_from_handle
74 (ps->vpp_server_handle);
76 active_open_session = 0;
83 memset (ps, 0xFE, sizeof (*ps));
84 pool_put (pm->sessions, ps);
87 clib_spinlock_unlock_if_init (&pm->sessions_lock);
89 if (active_open_session)
91 a->handle = session_handle (active_open_session);
92 a->app_index = pm->active_open_app_index;
93 hash_unset (pm->proxy_session_by_active_open_handle,
94 session_handle (active_open_session));
95 vnet_disconnect_session (a);
100 a->handle = session_handle (server_session);
101 a->app_index = pm->server_app_index;
102 hash_unset (pm->proxy_session_by_server_handle,
103 session_handle (server_session));
104 vnet_disconnect_session (a);
109 proxy_accept_callback (stream_session_t * s)
111 proxy_main_t *pm = &proxy_main;
113 s->session_state = SESSION_STATE_READY;
115 clib_spinlock_lock_if_init (&pm->sessions_lock);
121 proxy_disconnect_callback (stream_session_t * s)
123 delete_proxy_session (s, 0 /* is_active_open */ );
127 proxy_reset_callback (stream_session_t * s)
129 clib_warning ("Reset session %U", format_stream_session, s, 2);
130 delete_proxy_session (s, 0 /* is_active_open */ );
134 proxy_connected_callback (u32 app_index, u32 api_context,
135 stream_session_t * s, u8 is_fail)
137 clib_warning ("called...");
142 proxy_add_segment_callback (u32 client_index, const ssvm_private_t * sp)
144 clib_warning ("called...");
149 proxy_rx_callback (stream_session_t * s)
152 int actual_transfer __attribute__ ((unused));
153 svm_fifo_t *tx_fifo, *rx_fifo;
154 proxy_main_t *pm = &proxy_main;
155 u32 thread_index = vlib_get_thread_index ();
156 vnet_connect_args_t _a, *a = &_a;
160 svm_fifo_t *active_open_tx_fifo;
161 session_fifo_event_t evt;
163 ASSERT (s->thread_index == thread_index);
165 clib_spinlock_lock_if_init (&pm->sessions_lock);
166 p = hash_get (pm->proxy_session_by_server_handle, session_handle (s));
168 if (PREDICT_TRUE (p != 0))
170 clib_spinlock_unlock_if_init (&pm->sessions_lock);
171 active_open_tx_fifo = s->server_rx_fifo;
174 * Send event for active open tx fifo
176 if (svm_fifo_set_event (active_open_tx_fifo))
178 evt.fifo = active_open_tx_fifo;
179 evt.event_type = FIFO_EVENT_APP_TX;
181 (pm->active_open_event_queue[thread_index], (u8 *) & evt,
182 0 /* do wait for mutex */ ))
183 clib_warning ("failed to enqueue tx evt");
188 rx_fifo = s->server_rx_fifo;
189 tx_fifo = s->server_tx_fifo;
191 ASSERT (rx_fifo->master_thread_index == thread_index);
192 ASSERT (tx_fifo->master_thread_index == thread_index);
194 max_dequeue = svm_fifo_max_dequeue (s->server_rx_fifo);
196 if (PREDICT_FALSE (max_dequeue == 0))
199 actual_transfer = svm_fifo_peek (rx_fifo, 0 /* relative_offset */ ,
200 max_dequeue, pm->rx_buf[thread_index]);
202 /* $$$ your message in this space: parse url, etc. */
204 memset (a, 0, sizeof (*a));
206 clib_spinlock_lock_if_init (&pm->sessions_lock);
207 pool_get (pm->sessions, ps);
208 memset (ps, 0, sizeof (*ps));
209 ps->server_rx_fifo = rx_fifo;
210 ps->server_tx_fifo = tx_fifo;
211 ps->vpp_server_handle = session_handle (s);
213 proxy_index = ps - pm->sessions;
215 hash_set (pm->proxy_session_by_server_handle, ps->vpp_server_handle,
218 clib_spinlock_unlock_if_init (&pm->sessions_lock);
220 a->uri = (char *) pm->client_uri;
221 a->api_context = proxy_index;
222 a->app_index = pm->active_open_app_index;
223 vnet_connect_uri (a);
229 static session_cb_vft_t proxy_session_cb_vft = {
230 .session_accept_callback = proxy_accept_callback,
231 .session_disconnect_callback = proxy_disconnect_callback,
232 .session_connected_callback = proxy_connected_callback,
233 .add_segment_callback = proxy_add_segment_callback,
234 .builtin_app_rx_callback = proxy_rx_callback,
235 .session_reset_callback = proxy_reset_callback
239 active_open_connected_callback (u32 app_index, u32 opaque,
240 stream_session_t * s, u8 is_fail)
242 proxy_main_t *pm = &proxy_main;
244 u8 thread_index = vlib_get_thread_index ();
245 session_fifo_event_t evt;
249 clib_warning ("connection %d failed!", opaque);
254 * Setup proxy session handle.
256 clib_spinlock_lock_if_init (&pm->sessions_lock);
258 ps = pool_elt_at_index (pm->sessions, opaque);
259 ps->vpp_active_open_handle = session_handle (s);
261 s->server_tx_fifo = ps->server_rx_fifo;
262 s->server_rx_fifo = ps->server_tx_fifo;
265 * Reset the active-open tx-fifo master indices so the active-open session
266 * will receive data, etc.
268 s->server_tx_fifo->master_session_index = s->session_index;
269 s->server_tx_fifo->master_thread_index = s->thread_index;
272 * Account for the active-open session's use of the fifos
273 * so they won't disappear until the last session which uses
276 s->server_tx_fifo->refcnt++;
277 s->server_rx_fifo->refcnt++;
279 hash_set (pm->proxy_session_by_active_open_handle,
280 ps->vpp_active_open_handle, opaque);
282 clib_spinlock_unlock_if_init (&pm->sessions_lock);
285 * Send event for active open tx fifo
287 if (svm_fifo_set_event (s->server_tx_fifo))
289 evt.fifo = s->server_tx_fifo;
290 evt.event_type = FIFO_EVENT_APP_TX;
292 (pm->active_open_event_queue[thread_index], (u8 *) & evt,
293 0 /* do wait for mutex */ ))
294 clib_warning ("failed to enqueue tx evt");
301 active_open_reset_callback (stream_session_t * s)
303 delete_proxy_session (s, 1 /* is_active_open */ );
307 active_open_create_callback (stream_session_t * s)
313 active_open_disconnect_callback (stream_session_t * s)
315 delete_proxy_session (s, 1 /* is_active_open */ );
319 active_open_rx_callback (stream_session_t * s)
321 proxy_main_t *pm = &proxy_main;
322 session_fifo_event_t evt;
323 svm_fifo_t *server_rx_fifo;
324 u32 thread_index = vlib_get_thread_index ();
326 server_rx_fifo = s->server_rx_fifo;
329 * Send event for server tx fifo
331 if (svm_fifo_set_event (server_rx_fifo))
333 evt.fifo = server_rx_fifo;
334 evt.event_type = FIFO_EVENT_APP_TX;
336 (pm->server_event_queue[thread_index], (u8 *) & evt,
337 0 /* do wait for mutex */ ))
338 clib_warning ("failed to enqueue server rx evt");
345 static session_cb_vft_t active_open_clients = {
346 .session_reset_callback = active_open_reset_callback,
347 .session_connected_callback = active_open_connected_callback,
348 .session_accept_callback = active_open_create_callback,
349 .session_disconnect_callback = active_open_disconnect_callback,
350 .builtin_app_rx_callback = active_open_rx_callback
356 create_api_loopbacks (vlib_main_t * vm)
358 proxy_main_t *pm = &proxy_main;
359 api_main_t *am = &api_main;
360 vl_shmem_hdr_t *shmem_hdr;
362 shmem_hdr = am->shmem_hdr;
363 pm->vl_input_queue = shmem_hdr->vl_input_queue;
364 pm->server_client_index =
365 vl_api_memclnt_create_internal ("proxy_server", pm->vl_input_queue);
366 pm->active_open_client_index =
367 vl_api_memclnt_create_internal ("proxy_active_open", pm->vl_input_queue);
371 proxy_server_attach ()
373 proxy_main_t *pm = &proxy_main;
374 u64 options[APP_OPTIONS_N_OPTIONS];
375 vnet_app_attach_args_t _a, *a = &_a;
376 u32 segment_size = 512 << 20;
378 memset (a, 0, sizeof (*a));
379 memset (options, 0, sizeof (options));
381 if (pm->private_segment_size)
382 segment_size = pm->private_segment_size;
383 a->api_client_index = pm->server_client_index;
384 a->session_cb_vft = &proxy_session_cb_vft;
385 a->options = options;
386 a->options[APP_OPTIONS_SEGMENT_SIZE] = segment_size;
387 a->options[APP_OPTIONS_RX_FIFO_SIZE] = pm->fifo_size;
388 a->options[APP_OPTIONS_TX_FIFO_SIZE] = pm->fifo_size;
389 a->options[APP_OPTIONS_PRIVATE_SEGMENT_COUNT] = pm->private_segment_count;
390 a->options[APP_OPTIONS_PREALLOC_FIFO_PAIRS] =
391 pm->prealloc_fifos ? pm->prealloc_fifos : 1;
393 a->options[APP_OPTIONS_FLAGS] = APP_OPTIONS_FLAGS_IS_BUILTIN;
395 if (vnet_application_attach (a))
397 clib_warning ("failed to attach server");
400 pm->server_app_index = a->app_index;
406 active_open_attach (void)
408 proxy_main_t *pm = &proxy_main;
409 vnet_app_attach_args_t _a, *a = &_a;
412 memset (a, 0, sizeof (*a));
413 memset (options, 0, sizeof (options));
415 a->api_client_index = pm->active_open_client_index;
416 a->session_cb_vft = &active_open_clients;
418 options[APP_OPTIONS_ACCEPT_COOKIE] = 0x12345678;
419 options[APP_OPTIONS_SEGMENT_SIZE] = 512 << 20;
420 options[APP_OPTIONS_RX_FIFO_SIZE] = pm->fifo_size;
421 options[APP_OPTIONS_TX_FIFO_SIZE] = pm->fifo_size;
422 options[APP_OPTIONS_PRIVATE_SEGMENT_COUNT] = pm->private_segment_count;
423 options[APP_OPTIONS_PREALLOC_FIFO_PAIRS] =
424 pm->prealloc_fifos ? pm->prealloc_fifos : 1;
426 options[APP_OPTIONS_FLAGS] = APP_OPTIONS_FLAGS_IS_BUILTIN
427 | APP_OPTIONS_FLAGS_IS_PROXY;
429 a->options = options;
431 if (vnet_application_attach (a))
434 pm->active_open_app_index = a->app_index;
440 proxy_server_listen ()
442 proxy_main_t *pm = &proxy_main;
443 vnet_bind_args_t _a, *a = &_a;
444 memset (a, 0, sizeof (*a));
445 a->app_index = pm->server_app_index;
446 a->uri = (char *) pm->server_uri;
447 return vnet_bind_uri (a);
451 proxy_server_create (vlib_main_t * vm)
453 proxy_main_t *pm = &proxy_main;
454 vlib_thread_main_t *vtm = vlib_get_thread_main ();
458 if (pm->server_client_index == (u32) ~ 0)
459 create_api_loopbacks (vm);
461 num_threads = 1 /* main thread */ + vtm->n_threads;
462 vec_validate (proxy_main.server_event_queue, num_threads - 1);
463 vec_validate (proxy_main.active_open_event_queue, num_threads - 1);
464 vec_validate (pm->rx_buf, num_threads - 1);
466 for (i = 0; i < num_threads; i++)
467 vec_validate (pm->rx_buf[i], pm->rcv_buffer_size);
469 if (proxy_server_attach ())
471 clib_warning ("failed to attach server app");
474 if (proxy_server_listen ())
476 clib_warning ("failed to start listening");
479 if (active_open_attach ())
481 clib_warning ("failed to attach active open app");
485 for (i = 0; i < num_threads; i++)
487 pm->active_open_event_queue[i] =
488 session_manager_get_vpp_event_queue (i);
490 ASSERT (pm->active_open_event_queue[i]);
492 pm->server_event_queue[i] = session_manager_get_vpp_event_queue (i);
498 static clib_error_t *
499 proxy_server_create_command_fn (vlib_main_t * vm, unformat_input_t * input,
500 vlib_cli_command_t * cmd)
502 proxy_main_t *pm = &proxy_main;
503 char *default_server_uri = "tcp://0.0.0.0/23";
504 char *default_client_uri = "tcp://6.0.2.2/23";
508 pm->fifo_size = 64 << 10;
509 pm->rcv_buffer_size = 1024;
510 pm->prealloc_fifos = 0;
511 pm->private_segment_count = 0;
512 pm->private_segment_size = 0;
515 while (unformat_check_input (input) != UNFORMAT_END_OF_INPUT)
517 if (unformat (input, "fifo-size %d", &pm->fifo_size))
518 pm->fifo_size <<= 10;
519 else if (unformat (input, "rcv-buf-size %d", &pm->rcv_buffer_size))
521 else if (unformat (input, "prealloc-fifos %d", &pm->prealloc_fifos))
523 else if (unformat (input, "private-segment-count %d",
524 &pm->private_segment_count))
526 else if (unformat (input, "private-segment-size %U",
527 unformat_memory_size, &tmp))
529 if (tmp >= 0x100000000ULL)
530 return clib_error_return
531 (0, "private segment size %lld (%llu) too large", tmp, tmp);
532 pm->private_segment_size = tmp;
534 else if (unformat (input, "server-uri %s", &pm->server_uri))
536 else if (unformat (input, "client-uri %s", &pm->client_uri))
539 return clib_error_return (0, "unknown input `%U'",
540 format_unformat_error, input);
545 clib_warning ("No server-uri provided, Using default: %s",
547 pm->server_uri = format (0, "%s%c", default_server_uri, 0);
551 clib_warning ("No client-uri provided, Using default: %s",
553 pm->client_uri = format (0, "%s%c", default_client_uri, 0);
556 vnet_session_enable_disable (vm, 1 /* turn on session and transport */ );
558 rv = proxy_server_create (vm);
564 return clib_error_return (0, "server_create returned %d", rv);
571 VLIB_CLI_COMMAND (proxy_create_command, static) =
573 .path = "test proxy server",
574 .short_help = "test proxy server [server-uri <tcp://ip/port>]"
575 "[client-uri <tcp://ip/port>][fifo-size <nn>][rcv-buf-size <nn>]"
576 "[prealloc-fifos <nn>][private-segment-size <mem>]"
577 "[private-segment-count <nn>]",
578 .function = proxy_server_create_command_fn,
583 proxy_main_init (vlib_main_t * vm)
585 proxy_main_t *pm = &proxy_main;
586 pm->server_client_index = ~0;
587 pm->active_open_client_index = ~0;
588 pm->proxy_session_by_active_open_handle = hash_create (0, sizeof (uword));
589 pm->proxy_session_by_server_handle = hash_create (0, sizeof (uword));
594 VLIB_INIT_FUNCTION (proxy_main_init);
597 * fd.io coding-style-patch-verification: ON
600 * eval: (c-set-style "gnu")