Session layer refactoring
[vpp.git] / src / uri / uri_tcp_test.c
1 /*
2  * Copyright (c) 2016 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 <stdio.h>
17 #include <signal.h>
18 #include <svm/svm_fifo_segment.h>
19 #include <vlibmemory/api.h>
20 #include <vpp/api/vpe_msg_enum.h>
21 #include <vnet/session/application_interface.h>
22
23 #define vl_typedefs             /* define message structures */
24 #include <vpp/api/vpe_all_api_h.h>
25 #undef vl_typedefs
26
27 /* declare message handlers for each api */
28
29 #define vl_endianfun            /* define message structures */
30 #include <vpp/api/vpe_all_api_h.h>
31 #undef vl_endianfun
32
33 /* instantiate all the print functions we know about */
34 #define vl_print(handle, ...)
35 #define vl_printfun
36 #include <vpp/api/vpe_all_api_h.h>
37 #undef vl_printfun
38
39 /* Satisfy external references when not linking with -lvlib */
40 vlib_main_t vlib_global_main;
41 vlib_main_t **vlib_mains;
42
43 typedef struct
44 {
45   svm_fifo_t *server_rx_fifo;
46   svm_fifo_t *server_tx_fifo;
47
48   u32 vpp_session_handle;
49 } session_t;
50
51 typedef enum
52 {
53   STATE_START,
54   STATE_READY,
55   STATE_DISCONNECTING,
56   STATE_FAILED
57 } connection_state_t;
58
59 typedef struct
60 {
61   /* vpe input queue */
62   unix_shared_memory_queue_t *vl_input_queue;
63
64   /* API client handle */
65   u32 my_client_index;
66
67   /* The URI we're playing with */
68   u8 *uri;
69
70   /* Session pool */
71   session_t *sessions;
72
73   /* Hash table for disconnect processing */
74   uword *session_index_by_vpp_handles;
75
76   /* intermediate rx buffer */
77   u8 *rx_buf;
78
79   /* URI for slave's connect */
80   u8 *connect_uri;
81
82   u32 connected_session_index;
83
84   int i_am_master;
85
86   /* drop all packets */
87   int drop_packets;
88
89   /* Our event queue */
90   unix_shared_memory_queue_t *our_event_queue;
91
92   /* $$$ single thread only for the moment */
93   unix_shared_memory_queue_t *vpp_event_queue;
94
95   pid_t my_pid;
96
97   /* For deadman timers */
98   clib_time_t clib_time;
99
100   /* State of the connection, shared between msg RX thread and main thread */
101   volatile connection_state_t state;
102
103   /* Signal variables */
104   volatile int time_to_stop;
105   volatile int time_to_print_stats;
106
107   u32 configured_segment_size;
108
109   /* VNET_API_ERROR_FOO -> "Foo" hash table */
110   uword *error_string_by_error_number;
111
112   u8 *connect_test_data;
113   pthread_t client_rx_thread_handle;
114   u32 client_bytes_received;
115   u8 test_return_packets;
116   u64 bytes_to_send;
117
118   /* convenience */
119   svm_fifo_segment_main_t *segment_main;
120 } uri_tcp_test_main_t;
121
122 uri_tcp_test_main_t uri_tcp_test_main;
123
124 #if CLIB_DEBUG > 0
125 #define NITER 10000
126 #else
127 #define NITER 4000000
128 #endif
129
130 int
131 wait_for_state_change (uri_tcp_test_main_t * utm, connection_state_t state)
132 {
133 #if CLIB_DEBUG > 0
134 #define TIMEOUT 600.0
135 #else
136 #define TIMEOUT 600.0
137 #endif
138
139   f64 timeout = clib_time_now (&utm->clib_time) + TIMEOUT;
140
141   while (clib_time_now (&utm->clib_time) < timeout)
142     {
143       if (utm->state == state)
144         return 0;
145       if (utm->state == STATE_FAILED)
146         return -1;
147     }
148   clib_warning ("timeout waiting for STATE_READY");
149   return -1;
150 }
151
152 void
153 application_attach (uri_tcp_test_main_t * utm)
154 {
155   vl_api_application_attach_t *bmp;
156   u32 fifo_size = 3 << 20;
157   bmp = vl_msg_api_alloc (sizeof (*bmp));
158   memset (bmp, 0, sizeof (*bmp));
159
160   bmp->_vl_msg_id = ntohs (VL_API_APPLICATION_ATTACH);
161   bmp->client_index = utm->my_client_index;
162   bmp->context = ntohl (0xfeedface);
163   bmp->options[SESSION_OPTIONS_FLAGS] =
164     SESSION_OPTIONS_FLAGS_USE_FIFO | SESSION_OPTIONS_FLAGS_ADD_SEGMENT;
165   bmp->options[SESSION_OPTIONS_RX_FIFO_SIZE] = fifo_size;
166   bmp->options[SESSION_OPTIONS_TX_FIFO_SIZE] = fifo_size;
167   bmp->options[SESSION_OPTIONS_ADD_SEGMENT_SIZE] = 128 << 20;
168   bmp->options[SESSION_OPTIONS_SEGMENT_SIZE] = 256 << 20;
169   vl_msg_api_send_shmem (utm->vl_input_queue, (u8 *) & bmp);
170 }
171
172 void
173 application_detach (uri_tcp_test_main_t * utm)
174 {
175   vl_api_application_detach_t *bmp;
176   bmp = vl_msg_api_alloc (sizeof (*bmp));
177   memset (bmp, 0, sizeof (*bmp));
178
179   bmp->_vl_msg_id = ntohs (VL_API_APPLICATION_DETACH);
180   bmp->client_index = utm->my_client_index;
181   bmp->context = ntohl (0xfeedface);
182   vl_msg_api_send_shmem (utm->vl_input_queue, (u8 *) & bmp);
183 }
184
185 static void
186 vl_api_application_attach_reply_t_handler (vl_api_application_attach_reply_t *
187                                            mp)
188 {
189   uri_tcp_test_main_t *utm = &uri_tcp_test_main;
190   svm_fifo_segment_create_args_t _a, *a = &_a;
191   int rv;
192
193   if (mp->retval)
194     {
195       uword *errp = hash_get (utm->error_string_by_error_number, mp->retval);
196       clib_warning ("attach failed: %s", *errp);
197       utm->state = STATE_FAILED;
198       return;
199     }
200
201   if (mp->segment_name_length == 0)
202     {
203       clib_warning ("segment_name_length zero");
204       return;
205     }
206
207   a->segment_name = (char *) mp->segment_name;
208   a->segment_size = mp->segment_size;
209
210   ASSERT (mp->app_event_queue_address);
211
212   /* Attach to the segment vpp created */
213   rv = svm_fifo_segment_attach (a);
214   if (rv)
215     {
216       clib_warning ("svm_fifo_segment_attach ('%s') failed",
217                     mp->segment_name);
218       return;
219     }
220
221   utm->our_event_queue =
222     (unix_shared_memory_queue_t *) mp->app_event_queue_address;
223
224 }
225
226 static void
227 vl_api_application_detach_reply_t_handler (vl_api_application_detach_reply_t *
228                                            mp)
229 {
230   if (mp->retval)
231     clib_warning ("detach returned with err: %d", mp->retval);
232 }
233
234 static void
235 init_error_string_table (uri_tcp_test_main_t * utm)
236 {
237   utm->error_string_by_error_number = hash_create (0, sizeof (uword));
238
239 #define _(n,v,s) hash_set (utm->error_string_by_error_number, -v, s);
240   foreach_vnet_api_error;
241 #undef _
242
243   hash_set (utm->error_string_by_error_number, 99, "Misc");
244 }
245
246 static void
247 stop_signal (int signum)
248 {
249   uri_tcp_test_main_t *um = &uri_tcp_test_main;
250
251   um->time_to_stop = 1;
252 }
253
254 static void
255 stats_signal (int signum)
256 {
257   uri_tcp_test_main_t *um = &uri_tcp_test_main;
258
259   um->time_to_print_stats = 1;
260 }
261
262 static clib_error_t *
263 setup_signal_handlers (void)
264 {
265   signal (SIGINT, stats_signal);
266   signal (SIGQUIT, stop_signal);
267   signal (SIGTERM, stop_signal);
268
269   return 0;
270 }
271
272 void
273 vlib_cli_output (struct vlib_main_t *vm, char *fmt, ...)
274 {
275   clib_warning ("BUG");
276 }
277
278 int
279 connect_to_vpp (char *name)
280 {
281   uri_tcp_test_main_t *utm = &uri_tcp_test_main;
282   api_main_t *am = &api_main;
283
284   if (vl_client_connect_to_vlib ("/vpe-api", name, 32) < 0)
285     return -1;
286
287   utm->vl_input_queue = am->shmem_hdr->vl_input_queue;
288   utm->my_client_index = am->my_client_index;
289
290   return 0;
291 }
292
293 static void
294 vl_api_map_another_segment_t_handler (vl_api_map_another_segment_t * mp)
295 {
296   svm_fifo_segment_create_args_t _a, *a = &_a;
297   int rv;
298
299   a->segment_name = (char *) mp->segment_name;
300   a->segment_size = mp->segment_size;
301   /* Attach to the segment vpp created */
302   rv = svm_fifo_segment_attach (a);
303   if (rv)
304     {
305       clib_warning ("svm_fifo_segment_attach ('%s') failed",
306                     mp->segment_name);
307       return;
308     }
309   clib_warning ("Mapped new segment '%s' size %d", mp->segment_name,
310                 mp->segment_size);
311 }
312
313 static void
314 vl_api_disconnect_session_t_handler (vl_api_disconnect_session_t * mp)
315 {
316   uri_tcp_test_main_t *utm = &uri_tcp_test_main;
317   session_t *session;
318   vl_api_disconnect_session_reply_t *rmp;
319   uword *p;
320   int rv = 0;
321
322   p = hash_get (utm->session_index_by_vpp_handles, mp->handle);
323
324   if (p)
325     {
326       session = pool_elt_at_index (utm->sessions, p[0]);
327       hash_unset (utm->session_index_by_vpp_handles, mp->handle);
328       pool_put (utm->sessions, session);
329     }
330   else
331     {
332       clib_warning ("couldn't find session key %llx", mp->handle);
333       rv = -11;
334     }
335
336   utm->time_to_stop = 1;
337
338   rmp = vl_msg_api_alloc (sizeof (*rmp));
339   memset (rmp, 0, sizeof (*rmp));
340
341   rmp->_vl_msg_id = ntohs (VL_API_DISCONNECT_SESSION_REPLY);
342   rmp->retval = rv;
343   rmp->handle = mp->handle;
344   vl_msg_api_send_shmem (utm->vl_input_queue, (u8 *) & rmp);
345 }
346
347 static void
348 vl_api_reset_session_t_handler (vl_api_reset_session_t * mp)
349 {
350   uri_tcp_test_main_t *utm = &uri_tcp_test_main;
351   session_t *session;
352   vl_api_reset_session_reply_t *rmp;
353   uword *p;
354   int rv = 0;
355
356   p = hash_get (utm->session_index_by_vpp_handles, mp->handle);
357
358   if (p)
359     {
360       session = pool_elt_at_index (utm->sessions, p[0]);
361       hash_unset (utm->session_index_by_vpp_handles, mp->handle);
362       pool_put (utm->sessions, session);
363       utm->time_to_stop = 1;
364     }
365   else
366     {
367       clib_warning ("couldn't find session key %llx", mp->handle);
368       rv = -11;
369     }
370
371   rmp = vl_msg_api_alloc (sizeof (*rmp));
372   memset (rmp, 0, sizeof (*rmp));
373   rmp->_vl_msg_id = ntohs (VL_API_RESET_SESSION_REPLY);
374   rmp->retval = rv;
375   rmp->handle = mp->handle;
376   vl_msg_api_send_shmem (utm->vl_input_queue, (u8 *) & rmp);
377 }
378
379 void
380 client_handle_fifo_event_rx (uri_tcp_test_main_t * utm,
381                              session_fifo_event_t * e)
382 {
383   svm_fifo_t *rx_fifo;
384   int n_read, bytes, i;
385
386   rx_fifo = e->fifo;
387
388   bytes = svm_fifo_max_dequeue (rx_fifo);
389   /* Allow enqueuing of new event */
390   svm_fifo_unset_event (rx_fifo);
391
392   /* Read the bytes */
393   do
394     {
395       n_read = svm_fifo_dequeue_nowait (rx_fifo, 0,
396                                         clib_min (vec_len (utm->rx_buf),
397                                                   bytes), utm->rx_buf);
398       if (n_read > 0)
399         {
400           bytes -= n_read;
401           for (i = 0; i < n_read; i++)
402             {
403               if (utm->rx_buf[i] != ((utm->client_bytes_received + i) & 0xff))
404                 {
405                   clib_warning ("error at byte %lld, 0x%x not 0x%x",
406                                 utm->client_bytes_received + i,
407                                 utm->rx_buf[i],
408                                 ((utm->client_bytes_received + i) & 0xff));
409                 }
410             }
411           utm->client_bytes_received += n_read;
412         }
413       else
414         {
415           if (n_read == -2)
416             {
417 //            clib_warning ("weird!");
418               break;
419             }
420         }
421
422     }
423   while (bytes > 0);
424 }
425
426 void
427 client_handle_event_queue (uri_tcp_test_main_t * utm)
428 {
429   session_fifo_event_t _e, *e = &_e;;
430
431   unix_shared_memory_queue_sub (utm->our_event_queue, (u8 *) e,
432                                 0 /* nowait */ );
433   switch (e->event_type)
434     {
435     case FIFO_EVENT_SERVER_RX:
436       client_handle_fifo_event_rx (utm, e);
437       break;
438
439     case FIFO_EVENT_SERVER_EXIT:
440       return;
441
442     default:
443       clib_warning ("unknown event type %d", e->event_type);
444       break;
445     }
446 }
447
448 static void *
449 client_rx_thread_fn (void *arg)
450 {
451   session_fifo_event_t _e, *e = &_e;
452   uri_tcp_test_main_t *utm = &uri_tcp_test_main;
453
454   utm->client_bytes_received = 0;
455   while (1)
456     {
457       unix_shared_memory_queue_sub (utm->our_event_queue, (u8 *) e,
458                                     0 /* nowait */ );
459       switch (e->event_type)
460         {
461         case FIFO_EVENT_SERVER_RX:
462           client_handle_fifo_event_rx (utm, e);
463           break;
464
465         case FIFO_EVENT_SERVER_EXIT:
466           return 0;
467         default:
468           clib_warning ("unknown event type %d", e->event_type);
469           break;
470         }
471
472       if (PREDICT_FALSE (utm->time_to_stop == 1))
473         break;
474     }
475   pthread_exit (0);
476 }
477
478
479 static void
480 vl_api_connect_uri_reply_t_handler (vl_api_connect_uri_reply_t * mp)
481 {
482   uri_tcp_test_main_t *utm = &uri_tcp_test_main;
483   session_t *session;
484   u32 session_index;
485   svm_fifo_t *rx_fifo, *tx_fifo;
486   int rv;
487
488   if (mp->retval)
489     {
490       uword *errp = hash_get (utm->error_string_by_error_number, -mp->retval);
491       clib_warning ("connection failed with code: %s", *errp);
492       utm->state = STATE_FAILED;
493       return;
494     }
495
496   utm->vpp_event_queue = (unix_shared_memory_queue_t *)
497     mp->vpp_event_queue_address;
498
499   /*
500    * Setup session
501    */
502
503   pool_get (utm->sessions, session);
504   session_index = session - utm->sessions;
505
506   rx_fifo = (svm_fifo_t *) mp->server_rx_fifo;
507   rx_fifo->client_session_index = session_index;
508   tx_fifo = (svm_fifo_t *) mp->server_tx_fifo;
509   tx_fifo->client_session_index = session_index;
510
511   session->server_rx_fifo = rx_fifo;
512   session->server_tx_fifo = tx_fifo;
513   session->vpp_session_handle = mp->handle;
514
515   /* Save handle */
516   utm->connected_session_index = session_index;
517   utm->state = STATE_READY;
518
519   /* Add it to lookup table */
520   hash_set (utm->session_index_by_vpp_handles, mp->handle, session_index);
521
522   /* Start RX thread */
523   rv = pthread_create (&utm->client_rx_thread_handle,
524                        NULL /*attr */ , client_rx_thread_fn, 0);
525   if (rv)
526     {
527       clib_warning ("pthread_create returned %d", rv);
528       rv = VNET_API_ERROR_SYSCALL_ERROR_1;
529     }
530 }
531
532 static void
533 send_test_chunk (uri_tcp_test_main_t * utm, svm_fifo_t * tx_fifo, int mypid,
534                  u32 bytes)
535 {
536   u8 *test_data = utm->connect_test_data;
537   u64 bytes_sent = 0;
538   int test_buf_offset = 0;
539   u32 bytes_to_snd;
540   u32 queue_max_chunk = 64 << 10, actual_write;
541   session_fifo_event_t evt;
542   static int serial_number = 0;
543   int rv;
544
545   bytes_to_snd = (bytes == 0) ? vec_len (test_data) : bytes;
546   if (bytes_to_snd > vec_len (test_data))
547     bytes_to_snd = vec_len (test_data);
548
549   while (bytes_to_snd > 0)
550     {
551       actual_write =
552         bytes_to_snd > queue_max_chunk ? queue_max_chunk : bytes_to_snd;
553       rv = svm_fifo_enqueue_nowait (tx_fifo, mypid, actual_write,
554                                     test_data + test_buf_offset);
555
556       if (rv > 0)
557         {
558           bytes_to_snd -= rv;
559           test_buf_offset += rv;
560           bytes_sent += rv;
561
562           if (svm_fifo_set_event (tx_fifo))
563             {
564               /* Fabricate TX event, send to vpp */
565               evt.fifo = tx_fifo;
566               evt.event_type = FIFO_EVENT_SERVER_TX;
567               evt.event_id = serial_number++;
568
569               unix_shared_memory_queue_add (utm->vpp_event_queue,
570                                             (u8 *) & evt,
571                                             0 /* do wait for mutex */ );
572             }
573         }
574     }
575 }
576
577 void
578 client_send_data (uri_tcp_test_main_t * utm)
579 {
580   u8 *test_data = utm->connect_test_data;
581   int mypid = getpid ();
582   session_t *session;
583   svm_fifo_t *tx_fifo;
584   u32 n_iterations, leftover;
585   int i;
586
587   session = pool_elt_at_index (utm->sessions, utm->connected_session_index);
588   tx_fifo = session->server_tx_fifo;
589
590   vec_validate (utm->rx_buf, vec_len (test_data) - 1);
591   n_iterations = utm->bytes_to_send / vec_len (test_data);
592
593   for (i = 0; i < n_iterations; i++)
594     {
595       send_test_chunk (utm, tx_fifo, mypid, 0);
596     }
597
598   leftover = utm->bytes_to_send % vec_len (test_data);
599   if (leftover)
600     send_test_chunk (utm, tx_fifo, mypid, leftover);
601
602   if (utm->test_return_packets)
603     {
604       f64 timeout = clib_time_now (&utm->clib_time) + 2;
605
606       /* Wait for the outstanding packets */
607       while (utm->client_bytes_received <
608              vec_len (test_data) * n_iterations + leftover)
609         {
610           if (clib_time_now (&utm->clib_time) > timeout)
611             {
612               clib_warning ("timed out waiting for the missing packets");
613               break;
614             }
615         }
616     }
617   utm->time_to_stop = 1;
618 }
619
620 void
621 client_connect (uri_tcp_test_main_t * utm)
622 {
623   vl_api_connect_uri_t *cmp;
624   cmp = vl_msg_api_alloc (sizeof (*cmp));
625   memset (cmp, 0, sizeof (*cmp));
626
627   cmp->_vl_msg_id = ntohs (VL_API_CONNECT_URI);
628   cmp->client_index = utm->my_client_index;
629   cmp->context = ntohl (0xfeedface);
630   memcpy (cmp->uri, utm->connect_uri, vec_len (utm->connect_uri));
631   vl_msg_api_send_shmem (utm->vl_input_queue, (u8 *) & cmp);
632 }
633
634 void
635 client_disconnect (uri_tcp_test_main_t * utm)
636 {
637   session_t *connected_session;
638   vl_api_disconnect_session_t *dmp;
639   connected_session = pool_elt_at_index (utm->sessions,
640                                          utm->connected_session_index);
641   dmp = vl_msg_api_alloc (sizeof (*dmp));
642   memset (dmp, 0, sizeof (*dmp));
643   dmp->_vl_msg_id = ntohs (VL_API_DISCONNECT_SESSION);
644   dmp->client_index = utm->my_client_index;
645   dmp->handle = connected_session->vpp_session_handle;
646   vl_msg_api_send_shmem (utm->vl_input_queue, (u8 *) & dmp);
647 }
648
649 static void
650 client_test (uri_tcp_test_main_t * utm)
651 {
652   int i;
653
654   application_attach (utm);
655   client_connect (utm);
656
657   if (wait_for_state_change (utm, STATE_READY))
658     {
659       return;
660     }
661
662   /* Init test data */
663   vec_validate (utm->connect_test_data, 64 * 1024 - 1);
664   for (i = 0; i < vec_len (utm->connect_test_data); i++)
665     utm->connect_test_data[i] = i & 0xff;
666
667   /* Start send */
668   client_send_data (utm);
669
670   /* Disconnect */
671   client_disconnect (utm);
672
673   if (wait_for_state_change (utm, STATE_START))
674     {
675       clib_warning ("Disconnect failed");
676       return;
677     }
678   application_detach (utm);
679 }
680
681 static void
682 vl_api_bind_uri_reply_t_handler (vl_api_bind_uri_reply_t * mp)
683 {
684   uri_tcp_test_main_t *utm = &uri_tcp_test_main;
685
686   if (mp->retval)
687     {
688       uword *errp = hash_get (utm->error_string_by_error_number,
689                               -clib_net_to_host_u32 (mp->retval));
690       clib_warning ("bind failed: %s", (char *) *errp);
691       utm->state = STATE_FAILED;
692       return;
693     }
694
695   utm->state = STATE_READY;
696 }
697
698 static void
699 vl_api_unbind_uri_reply_t_handler (vl_api_unbind_uri_reply_t * mp)
700 {
701   uri_tcp_test_main_t *utm = &uri_tcp_test_main;
702
703   if (mp->retval != 0)
704     clib_warning ("returned %d", ntohl (mp->retval));
705
706   utm->state = STATE_START;
707 }
708
709 u8 *
710 format_ip4_address (u8 * s, va_list * args)
711 {
712   u8 *a = va_arg (*args, u8 *);
713   return format (s, "%d.%d.%d.%d", a[0], a[1], a[2], a[3]);
714 }
715
716 u8 *
717 format_ip6_address (u8 * s, va_list * args)
718 {
719   ip6_address_t *a = va_arg (*args, ip6_address_t *);
720   u32 i, i_max_n_zero, max_n_zeros, i_first_zero, n_zeros, last_double_colon;
721
722   i_max_n_zero = ARRAY_LEN (a->as_u16);
723   max_n_zeros = 0;
724   i_first_zero = i_max_n_zero;
725   n_zeros = 0;
726   for (i = 0; i < ARRAY_LEN (a->as_u16); i++)
727     {
728       u32 is_zero = a->as_u16[i] == 0;
729       if (is_zero && i_first_zero >= ARRAY_LEN (a->as_u16))
730         {
731           i_first_zero = i;
732           n_zeros = 0;
733         }
734       n_zeros += is_zero;
735       if ((!is_zero && n_zeros > max_n_zeros)
736           || (i + 1 >= ARRAY_LEN (a->as_u16) && n_zeros > max_n_zeros))
737         {
738           i_max_n_zero = i_first_zero;
739           max_n_zeros = n_zeros;
740           i_first_zero = ARRAY_LEN (a->as_u16);
741           n_zeros = 0;
742         }
743     }
744
745   last_double_colon = 0;
746   for (i = 0; i < ARRAY_LEN (a->as_u16); i++)
747     {
748       if (i == i_max_n_zero && max_n_zeros > 1)
749         {
750           s = format (s, "::");
751           i += max_n_zeros - 1;
752           last_double_colon = 1;
753         }
754       else
755         {
756           s = format (s, "%s%x",
757                       (last_double_colon || i == 0) ? "" : ":",
758                       clib_net_to_host_u16 (a->as_u16[i]));
759           last_double_colon = 0;
760         }
761     }
762
763   return s;
764 }
765
766 /* Format an IP46 address. */
767 u8 *
768 format_ip46_address (u8 * s, va_list * args)
769 {
770   ip46_address_t *ip46 = va_arg (*args, ip46_address_t *);
771   ip46_type_t type = va_arg (*args, ip46_type_t);
772   int is_ip4 = 1;
773
774   switch (type)
775     {
776     case IP46_TYPE_ANY:
777       is_ip4 = ip46_address_is_ip4 (ip46);
778       break;
779     case IP46_TYPE_IP4:
780       is_ip4 = 1;
781       break;
782     case IP46_TYPE_IP6:
783       is_ip4 = 0;
784       break;
785     }
786
787   return is_ip4 ?
788     format (s, "%U", format_ip4_address, &ip46->ip4) :
789     format (s, "%U", format_ip6_address, &ip46->ip6);
790 }
791
792 static void
793 vl_api_accept_session_t_handler (vl_api_accept_session_t * mp)
794 {
795   uri_tcp_test_main_t *utm = &uri_tcp_test_main;
796   vl_api_accept_session_reply_t *rmp;
797   svm_fifo_t *rx_fifo, *tx_fifo;
798   session_t *session;
799   static f64 start_time;
800   u32 session_index;
801   u8 *ip_str;
802
803   if (start_time == 0.0)
804     start_time = clib_time_now (&utm->clib_time);
805
806   ip_str = format (0, "%U", format_ip46_address, &mp->ip, mp->is_ip4);
807   clib_warning ("Accepted session from: %s:%d", ip_str,
808                 clib_net_to_host_u16 (mp->port));
809   utm->vpp_event_queue = (unix_shared_memory_queue_t *)
810     mp->vpp_event_queue_address;
811
812   /* Allocate local session and set it up */
813   pool_get (utm->sessions, session);
814   session_index = session - utm->sessions;
815
816   rx_fifo = (svm_fifo_t *) mp->server_rx_fifo;
817   rx_fifo->client_session_index = session_index;
818   tx_fifo = (svm_fifo_t *) mp->server_tx_fifo;
819   tx_fifo->client_session_index = session_index;
820
821   session->server_rx_fifo = rx_fifo;
822   session->server_tx_fifo = tx_fifo;
823
824   /* Add it to lookup table */
825   hash_set (utm->session_index_by_vpp_handles, mp->handle, session_index);
826
827   utm->state = STATE_READY;
828
829   /* Stats printing */
830   if (pool_elts (utm->sessions) && (pool_elts (utm->sessions) % 20000) == 0)
831     {
832       f64 now = clib_time_now (&utm->clib_time);
833       fformat (stdout, "%d active sessions in %.2f seconds, %.2f/sec...\n",
834                pool_elts (utm->sessions), now - start_time,
835                (f64) pool_elts (utm->sessions) / (now - start_time));
836     }
837
838   /*
839    * Send accept reply to vpp
840    */
841   rmp = vl_msg_api_alloc (sizeof (*rmp));
842   memset (rmp, 0, sizeof (*rmp));
843   rmp->_vl_msg_id = ntohs (VL_API_ACCEPT_SESSION_REPLY);
844   rmp->handle = mp->handle;
845   vl_msg_api_send_shmem (utm->vl_input_queue, (u8 *) & rmp);
846 }
847
848 void
849 server_handle_fifo_event_rx (uri_tcp_test_main_t * utm,
850                              session_fifo_event_t * e)
851 {
852   svm_fifo_t *rx_fifo, *tx_fifo;
853   int n_read;
854   session_fifo_event_t evt;
855   unix_shared_memory_queue_t *q;
856   int rv, bytes;
857
858   rx_fifo = e->fifo;
859   tx_fifo = utm->sessions[rx_fifo->client_session_index].server_tx_fifo;
860
861   bytes = svm_fifo_max_dequeue (rx_fifo);
862   /* Allow enqueuing of a new event */
863   svm_fifo_unset_event (rx_fifo);
864
865   if (bytes == 0)
866     return;
867
868   /* Read the bytes */
869   do
870     {
871       n_read = svm_fifo_dequeue_nowait (rx_fifo, 0, vec_len (utm->rx_buf),
872                                         utm->rx_buf);
873       if (n_read > 0)
874         bytes -= n_read;
875
876       if (utm->drop_packets)
877         continue;
878
879       /* Reflect if a non-drop session */
880       if (n_read > 0)
881         {
882           do
883             {
884               rv = svm_fifo_enqueue_nowait (tx_fifo, 0, n_read, utm->rx_buf);
885             }
886           while (rv <= 0 && !utm->time_to_stop);
887
888           /* If event wasn't set, add one */
889           if (svm_fifo_set_event (tx_fifo))
890             {
891               /* Fabricate TX event, send to vpp */
892               evt.fifo = tx_fifo;
893               evt.event_type = FIFO_EVENT_SERVER_TX;
894               evt.event_id = e->event_id;
895
896               q = utm->vpp_event_queue;
897               unix_shared_memory_queue_add (q, (u8 *) & evt,
898                                             0 /* do wait for mutex */ );
899             }
900         }
901     }
902   while ((n_read < 0 || bytes > 0) && !utm->time_to_stop);
903 }
904
905 void
906 server_handle_event_queue (uri_tcp_test_main_t * utm)
907 {
908   session_fifo_event_t _e, *e = &_e;;
909
910   while (1)
911     {
912       unix_shared_memory_queue_sub (utm->our_event_queue, (u8 *) e,
913                                     0 /* nowait */ );
914       switch (e->event_type)
915         {
916         case FIFO_EVENT_SERVER_RX:
917           server_handle_fifo_event_rx (utm, e);
918           break;
919
920         case FIFO_EVENT_SERVER_EXIT:
921           return;
922
923         default:
924           clib_warning ("unknown event type %d", e->event_type);
925           break;
926         }
927       if (PREDICT_FALSE (utm->time_to_stop == 1))
928         break;
929       if (PREDICT_FALSE (utm->time_to_print_stats == 1))
930         {
931           utm->time_to_print_stats = 0;
932           fformat (stdout, "%d connections\n", pool_elts (utm->sessions));
933         }
934     }
935 }
936
937 void
938 server_listen (uri_tcp_test_main_t * utm)
939 {
940   vl_api_bind_uri_t *bmp;
941   bmp = vl_msg_api_alloc (sizeof (*bmp));
942   memset (bmp, 0, sizeof (*bmp));
943
944   bmp->_vl_msg_id = ntohs (VL_API_BIND_URI);
945   bmp->client_index = utm->my_client_index;
946   bmp->context = ntohl (0xfeedface);
947   memcpy (bmp->uri, utm->uri, vec_len (utm->uri));
948   vl_msg_api_send_shmem (utm->vl_input_queue, (u8 *) & bmp);
949 }
950
951 void
952 server_unbind (uri_tcp_test_main_t * utm)
953 {
954   vl_api_unbind_uri_t *ump;
955
956   ump = vl_msg_api_alloc (sizeof (*ump));
957   memset (ump, 0, sizeof (*ump));
958
959   ump->_vl_msg_id = ntohs (VL_API_UNBIND_URI);
960   ump->client_index = utm->my_client_index;
961   memcpy (ump->uri, utm->uri, vec_len (utm->uri));
962   vl_msg_api_send_shmem (utm->vl_input_queue, (u8 *) & ump);
963 }
964
965 void
966 server_test (uri_tcp_test_main_t * utm)
967 {
968   application_attach (utm);
969
970   /* Bind to uri */
971   server_listen (utm);
972
973   if (wait_for_state_change (utm, STATE_READY))
974     {
975       clib_warning ("timeout waiting for STATE_READY");
976       return;
977     }
978
979   /* Enter handle event loop */
980   server_handle_event_queue (utm);
981
982   /* Cleanup */
983   server_unbind (utm);
984
985   if (wait_for_state_change (utm, STATE_START))
986     {
987       clib_warning ("timeout waiting for STATE_START");
988       return;
989     }
990
991   application_detach (utm);
992
993   fformat (stdout, "Test complete...\n");
994 }
995
996 static void
997 vl_api_disconnect_session_reply_t_handler (vl_api_disconnect_session_reply_t *
998                                            mp)
999 {
1000   uri_tcp_test_main_t *utm = &uri_tcp_test_main;
1001
1002   clib_warning ("retval %d", ntohl (mp->retval));
1003   utm->state = STATE_START;
1004 }
1005
1006 #define foreach_uri_msg                                 \
1007 _(BIND_URI_REPLY, bind_uri_reply)                       \
1008 _(UNBIND_URI_REPLY, unbind_uri_reply)                   \
1009 _(ACCEPT_SESSION, accept_session)                       \
1010 _(CONNECT_URI_REPLY, connect_uri_reply)                 \
1011 _(DISCONNECT_SESSION, disconnect_session)               \
1012 _(DISCONNECT_SESSION_REPLY, disconnect_session_reply)   \
1013 _(RESET_SESSION, reset_session)                         \
1014 _(APPLICATION_ATTACH_REPLY, application_attach_reply)   \
1015 _(APPLICATION_DETACH_REPLY, application_detach_reply)   \
1016 _(MAP_ANOTHER_SEGMENT, map_another_segment)             \
1017
1018 void
1019 uri_api_hookup (uri_tcp_test_main_t * utm)
1020 {
1021 #define _(N,n)                                                  \
1022     vl_msg_api_set_handlers(VL_API_##N, #n,                     \
1023                            vl_api_##n##_t_handler,              \
1024                            vl_noop_handler,                     \
1025                            vl_api_##n##_t_endian,               \
1026                            vl_api_##n##_t_print,                \
1027                            sizeof(vl_api_##n##_t), 1);
1028   foreach_uri_msg;
1029 #undef _
1030 }
1031
1032 int
1033 main (int argc, char **argv)
1034 {
1035   uri_tcp_test_main_t *utm = &uri_tcp_test_main;
1036   unformat_input_t _argv, *a = &_argv;
1037   u8 *chroot_prefix;
1038   u8 *heap, *uri = 0;
1039   u8 *bind_uri = (u8 *) "tcp://0.0.0.0/1234";
1040   u8 *connect_uri = (u8 *) "tcp://6.0.1.2/1234";
1041   u64 bytes_to_send = 64 << 10, mbytes;
1042   u32 tmp;
1043   mheap_t *h;
1044   session_t *session;
1045   int i;
1046   int i_am_master = 1, drop_packets = 0, test_return_packets = 0;
1047
1048   clib_mem_init (0, 256 << 20);
1049
1050   heap = clib_mem_get_per_cpu_heap ();
1051   h = mheap_header (heap);
1052
1053   /* make the main heap thread-safe */
1054   h->flags |= MHEAP_FLAG_THREAD_SAFE;
1055
1056   vec_validate (utm->rx_buf, 65536);
1057
1058   utm->session_index_by_vpp_handles = hash_create (0, sizeof (uword));
1059
1060   utm->my_pid = getpid ();
1061   utm->configured_segment_size = 1 << 20;
1062
1063   clib_time_init (&utm->clib_time);
1064   init_error_string_table (utm);
1065   svm_fifo_segment_init (0x200000000ULL, 20);
1066   unformat_init_command_line (a, argv);
1067
1068   while (unformat_check_input (a) != UNFORMAT_END_OF_INPUT)
1069     {
1070       if (unformat (a, "chroot prefix %s", &chroot_prefix))
1071         {
1072           vl_set_memory_root_path ((char *) chroot_prefix);
1073         }
1074       else if (unformat (a, "uri %s", &uri))
1075         ;
1076       else if (unformat (a, "segment-size %dM", &tmp))
1077         utm->configured_segment_size = tmp << 20;
1078       else if (unformat (a, "segment-size %dG", &tmp))
1079         utm->configured_segment_size = tmp << 30;
1080       else if (unformat (a, "master"))
1081         i_am_master = 1;
1082       else if (unformat (a, "slave"))
1083         i_am_master = 0;
1084       else if (unformat (a, "drop"))
1085         drop_packets = 1;
1086       else if (unformat (a, "test"))
1087         test_return_packets = 1;
1088       else if (unformat (a, "mbytes %lld", &mbytes))
1089         {
1090           bytes_to_send = mbytes << 20;
1091         }
1092       else if (unformat (a, "gbytes %lld", &mbytes))
1093         {
1094           bytes_to_send = mbytes << 30;
1095         }
1096       else
1097         {
1098           fformat (stderr, "%s: usage [master|slave]\n");
1099           exit (1);
1100         }
1101     }
1102
1103   if (uri)
1104     {
1105       utm->uri = format (0, "%s%c", uri, 0);
1106       utm->connect_uri = format (0, "%s%c", uri, 0);
1107     }
1108   else
1109     {
1110       utm->uri = format (0, "%s%c", bind_uri, 0);
1111       utm->connect_uri = format (0, "%s%c", connect_uri, 0);
1112     }
1113
1114   utm->i_am_master = i_am_master;
1115   utm->segment_main = &svm_fifo_segment_main;
1116   utm->drop_packets = drop_packets;
1117   utm->test_return_packets = test_return_packets;
1118   utm->bytes_to_send = bytes_to_send;
1119
1120   setup_signal_handlers ();
1121   uri_api_hookup (utm);
1122
1123   if (connect_to_vpp (i_am_master ? "uri_tcp_server" : "uri_tcp_client") < 0)
1124     {
1125       svm_region_exit ();
1126       fformat (stderr, "Couldn't connect to vpe, exiting...\n");
1127       exit (1);
1128     }
1129
1130   if (i_am_master == 0)
1131     {
1132       client_test (utm);
1133       vl_client_disconnect_from_vlib ();
1134       exit (0);
1135     }
1136
1137   /* $$$$ hack preallocation */
1138   for (i = 0; i < 200000; i++)
1139     {
1140       pool_get (utm->sessions, session);
1141       memset (session, 0, sizeof (*session));
1142     }
1143   for (i = 0; i < 200000; i++)
1144     pool_put_index (utm->sessions, i);
1145
1146   server_test (utm);
1147
1148   vl_client_disconnect_from_vlib ();
1149   exit (0);
1150 }
1151
1152 /*
1153  * fd.io coding-style-patch-verification: ON
1154  *
1155  * Local Variables:
1156  * eval: (c-set-style "gnu")
1157  * End:
1158  */