TCP proxy prototype
[vpp.git] / src / vnet / tcp / builtin_server.c
1 /*
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:
6 *
7 *     http://www.apache.org/licenses/LICENSE-2.0
8 *
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.
14 */
15
16 #include <vnet/vnet.h>
17 #include <vlibmemory/api.h>
18 #include <vnet/session/application.h>
19 #include <vnet/session/application_interface.h>
20
21 typedef struct
22 {
23   /*
24    * Server app parameters
25    */
26   unix_shared_memory_queue_t **vpp_queue;
27   unix_shared_memory_queue_t *vl_input_queue;   /**< Sever's event queue */
28
29   u32 app_index;                /**< Server app index */
30   u32 my_client_index;          /**< API client handle */
31   u32 node_index;               /**< process node index for evnt scheduling */
32
33   /*
34    * Config params
35    */
36   u8 no_echo;                   /**< Don't echo traffic */
37   u32 fifo_size;                /**< Fifo size */
38   u32 rcv_buffer_size;          /**< Rcv buffer size */
39   u32 prealloc_fifos;           /**< Preallocate fifos */
40   u32 private_segment_count;    /**< Number of private segments  */
41   u32 private_segment_size;     /**< Size of private segments  */
42
43   /*
44    * Test state
45    */
46   u8 **rx_buf;                  /**< Per-thread RX buffer */
47   u64 byte_index;
48   u32 **rx_retries;
49
50   vlib_main_t *vlib_main;
51 } builtin_server_main_t;
52
53 builtin_server_main_t builtin_server_main;
54
55 int
56 builtin_session_accept_callback (stream_session_t * s)
57 {
58   builtin_server_main_t *bsm = &builtin_server_main;
59
60   bsm->vpp_queue[s->thread_index] =
61     session_manager_get_vpp_event_queue (s->thread_index);
62   s->session_state = SESSION_STATE_READY;
63   bsm->byte_index = 0;
64   vec_validate (bsm->rx_retries[s->thread_index], s->session_index);
65   bsm->rx_retries[s->thread_index][s->session_index] = 0;
66   return 0;
67 }
68
69 void
70 builtin_session_disconnect_callback (stream_session_t * s)
71 {
72   builtin_server_main_t *bsm = &builtin_server_main;
73   vnet_disconnect_args_t _a, *a = &_a;
74
75   a->handle = stream_session_handle (s);
76   a->app_index = bsm->app_index;
77   vnet_disconnect_session (a);
78 }
79
80 void
81 builtin_session_reset_callback (stream_session_t * s)
82 {
83   clib_warning ("Reset session %U", format_stream_session, s, 2);
84   stream_session_cleanup (s);
85 }
86
87
88 int
89 builtin_session_connected_callback (u32 app_index, u32 api_context,
90                                     stream_session_t * s, u8 is_fail)
91 {
92   clib_warning ("called...");
93   return -1;
94 }
95
96 int
97 builtin_add_segment_callback (u32 client_index,
98                               const u8 * seg_name, u32 seg_size)
99 {
100   clib_warning ("called...");
101   return -1;
102 }
103
104 int
105 builtin_redirect_connect_callback (u32 client_index, void *mp)
106 {
107   clib_warning ("called...");
108   return -1;
109 }
110
111 void
112 test_bytes (builtin_server_main_t * bsm, int actual_transfer)
113 {
114   int i;
115   u32 my_thread_id = vlib_get_thread_index ();
116
117   for (i = 0; i < actual_transfer; i++)
118     {
119       if (bsm->rx_buf[my_thread_id][i] != ((bsm->byte_index + i) & 0xff))
120         {
121           clib_warning ("at %lld expected %d got %d", bsm->byte_index + i,
122                         (bsm->byte_index + i) & 0xff,
123                         bsm->rx_buf[my_thread_id][i]);
124         }
125     }
126   bsm->byte_index += actual_transfer;
127 }
128
129 /*
130  * If no-echo, just read the data and be done with it
131  */
132 int
133 builtin_server_rx_callback_no_echo (stream_session_t * s)
134 {
135   builtin_server_main_t *bsm = &builtin_server_main;
136   u32 my_thread_id = vlib_get_thread_index ();
137   int actual_transfer;
138   svm_fifo_t *rx_fifo;
139
140   rx_fifo = s->server_rx_fifo;
141
142   do
143     {
144       actual_transfer =
145         svm_fifo_dequeue_nowait (rx_fifo, bsm->rcv_buffer_size,
146                                  bsm->rx_buf[my_thread_id]);
147     }
148   while (actual_transfer > 0);
149   return 0;
150 }
151
152 int
153 builtin_server_rx_callback (stream_session_t * s)
154 {
155   u32 n_written, max_dequeue, max_enqueue, max_transfer;
156   int actual_transfer;
157   svm_fifo_t *tx_fifo, *rx_fifo;
158   builtin_server_main_t *bsm = &builtin_server_main;
159   session_fifo_event_t evt;
160   static int serial_number = 0;
161   u32 thread_index = vlib_get_thread_index ();
162
163   ASSERT (s->thread_index == thread_index);
164
165   rx_fifo = s->server_rx_fifo;
166   tx_fifo = s->server_tx_fifo;
167
168   ASSERT (rx_fifo->master_thread_index == thread_index);
169   ASSERT (tx_fifo->master_thread_index == thread_index);
170
171   max_dequeue = svm_fifo_max_dequeue (s->server_rx_fifo);
172   max_enqueue = svm_fifo_max_enqueue (s->server_tx_fifo);
173
174   if (PREDICT_FALSE (max_dequeue == 0))
175     return 0;
176
177   /* Number of bytes we're going to copy */
178   max_transfer = (max_dequeue < max_enqueue) ? max_dequeue : max_enqueue;
179
180   /* No space in tx fifo */
181   if (PREDICT_FALSE (max_transfer == 0))
182     {
183       /* XXX timeout for session that are stuck */
184
185     rx_event:
186       /* Program self-tap to retry */
187       if (svm_fifo_set_event (rx_fifo))
188         {
189           unix_shared_memory_queue_t *q;
190           evt.fifo = rx_fifo;
191           evt.event_type = FIFO_EVENT_BUILTIN_RX;
192           evt.event_id = 0;
193
194           q = bsm->vpp_queue[thread_index];
195           if (PREDICT_FALSE (q->cursize == q->maxsize))
196             clib_warning ("out of event queue space");
197           else if (unix_shared_memory_queue_add (q, (u8 *) & evt, 0))
198             clib_warning ("failed to enqueue self-tap");
199
200           if (bsm->rx_retries[thread_index][s->session_index] == 500000)
201             {
202               clib_warning ("session stuck: %U", format_stream_session, s, 2);
203             }
204           if (bsm->rx_retries[thread_index][s->session_index] < 500001)
205             bsm->rx_retries[thread_index][s->session_index]++;
206         }
207
208       return 0;
209     }
210
211   _vec_len (bsm->rx_buf[thread_index]) = max_transfer;
212
213   actual_transfer = svm_fifo_dequeue_nowait (rx_fifo, max_transfer,
214                                              bsm->rx_buf[thread_index]);
215   ASSERT (actual_transfer == max_transfer);
216
217 //  test_bytes (bsm, actual_transfer);
218
219   /*
220    * Echo back
221    */
222
223   n_written = svm_fifo_enqueue_nowait (tx_fifo, actual_transfer,
224                                        bsm->rx_buf[thread_index]);
225
226   if (n_written != max_transfer)
227     clib_warning ("short trout!");
228
229   if (svm_fifo_set_event (tx_fifo))
230     {
231       /* Fabricate TX event, send to vpp */
232       evt.fifo = tx_fifo;
233       evt.event_type = FIFO_EVENT_APP_TX;
234       evt.event_id = serial_number++;
235
236       if (unix_shared_memory_queue_add (bsm->vpp_queue[s->thread_index],
237                                         (u8 *) & evt,
238                                         0 /* do wait for mutex */ ))
239         clib_warning ("failed to enqueue tx evt");
240     }
241
242   if (PREDICT_FALSE (n_written < max_dequeue))
243     goto rx_event;
244
245   return 0;
246 }
247
248 static session_cb_vft_t builtin_session_cb_vft = {
249   .session_accept_callback = builtin_session_accept_callback,
250   .session_disconnect_callback = builtin_session_disconnect_callback,
251   .session_connected_callback = builtin_session_connected_callback,
252   .add_segment_callback = builtin_add_segment_callback,
253   .redirect_connect_callback = builtin_redirect_connect_callback,
254   .builtin_server_rx_callback = builtin_server_rx_callback,
255   .session_reset_callback = builtin_session_reset_callback
256 };
257
258 /* Abuse VPP's input queue */
259 static int
260 create_api_loopback (vlib_main_t * vm)
261 {
262   builtin_server_main_t *bsm = &builtin_server_main;
263   api_main_t *am = &api_main;
264   vl_shmem_hdr_t *shmem_hdr;
265
266   shmem_hdr = am->shmem_hdr;
267   bsm->vl_input_queue = shmem_hdr->vl_input_queue;
268   bsm->my_client_index =
269     vl_api_memclnt_create_internal ("tcp_test_server", bsm->vl_input_queue);
270   return 0;
271 }
272
273 static int
274 server_attach ()
275 {
276   builtin_server_main_t *bsm = &builtin_server_main;
277   u8 segment_name[128];
278   u64 options[SESSION_OPTIONS_N_OPTIONS];
279   vnet_app_attach_args_t _a, *a = &_a;
280
281   memset (a, 0, sizeof (*a));
282   memset (options, 0, sizeof (options));
283
284   if (bsm->no_echo)
285     builtin_session_cb_vft.builtin_server_rx_callback =
286       builtin_server_rx_callback_no_echo;
287   else
288     builtin_session_cb_vft.builtin_server_rx_callback =
289       builtin_server_rx_callback;
290   a->api_client_index = bsm->my_client_index;
291   a->session_cb_vft = &builtin_session_cb_vft;
292   a->options = options;
293   a->options[SESSION_OPTIONS_SEGMENT_SIZE] = 512 << 20;
294   a->options[SESSION_OPTIONS_RX_FIFO_SIZE] = bsm->fifo_size;
295   a->options[SESSION_OPTIONS_TX_FIFO_SIZE] = bsm->fifo_size;
296   a->options[APP_OPTIONS_PRIVATE_SEGMENT_COUNT] = bsm->private_segment_count;
297   a->options[APP_OPTIONS_PRIVATE_SEGMENT_SIZE] = bsm->private_segment_size;
298   a->options[APP_OPTIONS_PREALLOC_FIFO_PAIRS] =
299     bsm->prealloc_fifos ? bsm->prealloc_fifos : 1;
300
301   a->options[APP_OPTIONS_FLAGS] = APP_OPTIONS_FLAGS_BUILTIN_APP;
302
303   a->segment_name = segment_name;
304   a->segment_name_length = ARRAY_LEN (segment_name);
305
306   if (vnet_application_attach (a))
307     {
308       clib_warning ("failed to attach server");
309       return -1;
310     }
311   bsm->app_index = a->app_index;
312   return 0;
313 }
314
315 static int
316 server_listen ()
317 {
318   builtin_server_main_t *bsm = &builtin_server_main;
319   vnet_bind_args_t _a, *a = &_a;
320   memset (a, 0, sizeof (*a));
321   a->app_index = bsm->app_index;
322   a->uri = "tcp://0.0.0.0/1234";
323   return vnet_bind_uri (a);
324 }
325
326 static int
327 server_create (vlib_main_t * vm)
328 {
329   builtin_server_main_t *bsm = &builtin_server_main;
330   vlib_thread_main_t *vtm = vlib_get_thread_main ();
331   u32 num_threads;
332   int i;
333
334   if (bsm->my_client_index == (u32) ~ 0)
335     {
336       if (create_api_loopback (vm))
337         {
338           clib_warning ("failed to create api loopback");
339           return -1;
340         }
341     }
342
343   num_threads = 1 /* main thread */  + vtm->n_threads;
344   vec_validate (builtin_server_main.vpp_queue, num_threads - 1);
345   vec_validate (bsm->rx_buf, num_threads - 1);
346   vec_validate (bsm->rx_retries, num_threads - 1);
347
348   for (i = 0; i < num_threads; i++)
349     vec_validate (bsm->rx_buf[i], bsm->rcv_buffer_size);
350
351   if (server_attach ())
352     {
353       clib_warning ("failed to attach server");
354       return -1;
355     }
356   if (server_listen ())
357     {
358       clib_warning ("failed to start listening");
359       return -1;
360     }
361   return 0;
362 }
363
364 static clib_error_t *
365 server_create_command_fn (vlib_main_t * vm, unformat_input_t * input,
366                           vlib_cli_command_t * cmd)
367 {
368   builtin_server_main_t *bsm = &builtin_server_main;
369   int rv;
370   u32 tmp;
371
372   bsm->no_echo = 0;
373   bsm->fifo_size = 64 << 10;
374   bsm->rcv_buffer_size = 128 << 10;
375   bsm->prealloc_fifos = 0;
376   bsm->private_segment_count = 0;
377   bsm->private_segment_size = 0;
378
379   while (unformat_check_input (input) != UNFORMAT_END_OF_INPUT)
380     {
381       if (unformat (input, "no-echo"))
382         bsm->no_echo = 1;
383       else if (unformat (input, "fifo-size %d", &bsm->fifo_size))
384         bsm->fifo_size <<= 10;
385       else if (unformat (input, "rcv-buf-size %d", &bsm->rcv_buffer_size))
386         ;
387       else if (unformat (input, "prealloc-fifos %d", &bsm->prealloc_fifos))
388         ;
389       else if (unformat (input, "private-segment-count %d",
390                          &bsm->private_segment_count))
391         ;
392       else if (unformat (input, "private-segment-size %dm", &tmp))
393         bsm->private_segment_size = tmp << 20;
394       else if (unformat (input, "private-segment-size %dg", &tmp))
395         bsm->private_segment_size = tmp << 30;
396       else if (unformat (input, "private-segment-size %d", &tmp))
397         bsm->private_segment_size = tmp;
398       else
399         return clib_error_return (0, "unknown input `%U'",
400                                   format_unformat_error, input);
401     }
402
403   vnet_session_enable_disable (vm, 1 /* turn on TCP, etc. */ );
404
405   rv = server_create (vm);
406   switch (rv)
407     {
408     case 0:
409       break;
410     default:
411       return clib_error_return (0, "server_create returned %d", rv);
412     }
413
414   return 0;
415 }
416
417 /* *INDENT-OFF* */
418 VLIB_CLI_COMMAND (server_create_command, static) =
419 {
420   .path = "test tcp server",
421   .short_help = "test tcp server",
422   .function = server_create_command_fn,
423 };
424 /* *INDENT-ON* */
425
426 clib_error_t *
427 builtin_tcp_server_main_init (vlib_main_t * vm)
428 {
429   builtin_server_main_t *bsm = &builtin_server_main;
430   bsm->my_client_index = ~0;
431   return 0;
432 }
433
434 VLIB_INIT_FUNCTION (builtin_tcp_server_main_init);
435
436 /*
437 * fd.io coding-style-patch-verification: ON
438 *
439 * Local Variables:
440 * eval: (c-set-style "gnu")
441 * End:
442 */