diff options
Diffstat (limited to 'src/vcl')
| -rw-r--r-- | src/vcl/CMakeLists.txt | 2 | ||||
| -rw-r--r-- | src/vcl/vcl_test_client.c | 14 | ||||
| -rw-r--r-- | src/vcl/vcl_test_server.c | 652 |
3 files changed, 344 insertions, 324 deletions
diff --git a/src/vcl/CMakeLists.txt b/src/vcl/CMakeLists.txt index b0a283ce1fe..e5642b53d5e 100644 --- a/src/vcl/CMakeLists.txt +++ b/src/vcl/CMakeLists.txt @@ -59,7 +59,7 @@ if(VPP_BUILD_VCL_TESTS) test_vcl_listener_server test_vcl_listener_client ) - add_vpp_executable(${test} SOURCES ${test}.c LINK_LIBRARIES vppcom + add_vpp_executable(${test} SOURCES ${test}.c LINK_LIBRARIES vppcom pthread NO_INSTALL) endforeach() endif(VPP_BUILD_VCL_TESTS) diff --git a/src/vcl/vcl_test_client.c b/src/vcl/vcl_test_client.c index af51b66610a..4e61045b055 100644 --- a/src/vcl/vcl_test_client.c +++ b/src/vcl/vcl_test_client.c @@ -137,7 +137,6 @@ echo_test_client () FD_SET (tsock->fd, &rd_fdset); sidx = vppcom_session_index (tsock->fd); nfds = sidx > nfds ? sidx : nfds; -// nfds = ((tsock->fd + 1) > nfds) ? (tsock->fd + 1) : nfds; } nfds++; @@ -256,10 +255,8 @@ stream_test_client (sock_test_t test) sock_client_main_t *scm = &sock_client_main; sock_test_socket_t *ctrl = &scm->ctrl_socket; sock_test_socket_t *tsock; - int tx_bytes; - uint32_t i, n; - int rv; - int nfds = 0; + int tx_bytes, rv, nfds = 0; + uint32_t i, n, sidx; fd_set wr_fdset, rd_fdset; fd_set _wfdset, *wfdset = &_wfdset; fd_set _rfdset, *rfdset = (test == SOCK_TEST_TYPE_BI) ? &_rfdset : 0; @@ -297,7 +294,8 @@ stream_test_client (sock_test_t test) memset (&tsock->stats, 0, sizeof (tsock->stats)); FD_SET (tsock->fd, &wr_fdset); FD_SET (tsock->fd, &rd_fdset); - nfds = ((tsock->fd + 1) > nfds) ? (tsock->fd + 1) : nfds; + sidx = vppcom_session_index (tsock->fd); + nfds = sidx > nfds ? sidx : nfds; } nfds++; @@ -327,7 +325,7 @@ stream_test_client (sock_test_t test) continue; if ((test == SOCK_TEST_TYPE_BI) && - FD_ISSET (tsock->fd, rfdset) && + FD_ISSET (vppcom_session_index (tsock->fd), rfdset) && (tsock->stats.rx_bytes < ctrl->cfg.total_bytes)) { (void) vcl_test_read (tsock->fd, @@ -335,7 +333,7 @@ stream_test_client (sock_test_t test) tsock->rxbuf_size, &tsock->stats); } - if (FD_ISSET (tsock->fd, wfdset) + if (FD_ISSET (vppcom_session_index (tsock->fd), wfdset) && (tsock->stats.tx_bytes < ctrl->cfg.total_bytes)) { tx_bytes = vcl_test_write (tsock->fd, (uint8_t *) tsock->txbuf, diff --git a/src/vcl/vcl_test_server.c b/src/vcl/vcl_test_server.c index 7b56d497153..8b8b77fab2c 100644 --- a/src/vcl/vcl_test_server.c +++ b/src/vcl/vcl_test_server.c @@ -26,6 +26,7 @@ #include <vcl/vcl_test.h> #include <sys/epoll.h> #include <vppinfra/mem.h> +#include <pthread.h> typedef struct { @@ -37,49 +38,79 @@ typedef struct sock_test_stats_t stats; vppcom_endpt_t endpt; uint8_t ip[16]; -} sock_server_conn_t; +} vcl_test_server_conn_t; typedef struct { - uint32_t port; + uint16_t port; uint32_t address_ip6; - uint32_t transport_udp; -} sock_server_cfg_t; + u8 proto; + u8 workers; + vppcom_endpt_t endpt; +} vcl_test_server_cfg_t; + +#define SOCK_SERVER_MAX_TEST_CONN 16 +#define SOCK_SERVER_MAX_EPOLL_EVENTS 16 -#define SOCK_SERVER_MAX_TEST_CONN 10 -#define SOCK_SERVER_MAX_EPOLL_EVENTS 10 typedef struct { + uint32_t wrk_index; int listen_fd; - sock_server_cfg_t cfg; int epfd; - struct epoll_event listen_ev; struct epoll_event wait_events[SOCK_SERVER_MAX_EPOLL_EVENTS]; - size_t num_conn; size_t conn_pool_size; - sock_server_conn_t *conn_pool; + vcl_test_server_conn_t *conn_pool; int nfds; - fd_set rd_fdset; - fd_set wr_fdset; - struct timeval timeout; -} sock_server_main_t; + pthread_t thread_handle; +} vcl_test_server_worker_t; + +typedef struct +{ + vcl_test_server_cfg_t cfg; + vcl_test_server_worker_t *workers; + + struct sockaddr_storage servaddr; + volatile int worker_fails; + volatile int active_workers; +} vcl_test_server_main_t; -sock_server_main_t sock_server_main; +static __thread int __wrk_index = 0; + +static vcl_test_server_main_t sock_server_main; + +#define vtfail(_fn, _rv) \ +{ \ + errno = -_rv; \ + perror ("ERROR when calling " _fn); \ + fprintf (stderr, "\nSERVER ERROR: " _fn " failed (errno = %d)!\n", -_rv);\ + exit (1); \ +} + +#define vterr(_fn, _rv) \ +{ \ + errno = -_rv; \ + fprintf (stderr, "\nSERVER ERROR: " _fn " failed (errno = %d)!\n", -_rv);\ +} + +#define vtwrn(_fmt, _args...) \ + fprintf (stderr, "\nSERVER ERROR: " _fmt "\n", ##_args) \ + +#define vtinf(_fmt, _args...) \ + fprintf (stdout, "\nvts<w%u>: " _fmt "\n", __wrk_index, ##_args) \ static inline void -conn_pool_expand (size_t expand_size) +conn_pool_expand (vcl_test_server_worker_t * wrk, size_t expand_size) { - sock_server_main_t *ssm = &sock_server_main; - sock_server_conn_t *conn_pool; - size_t new_size = ssm->conn_pool_size + expand_size; + vcl_test_server_conn_t *conn_pool; + size_t new_size = wrk->conn_pool_size + expand_size; int i; - conn_pool = realloc (ssm->conn_pool, new_size * sizeof (*ssm->conn_pool)); + conn_pool = realloc (wrk->conn_pool, new_size * sizeof (*wrk->conn_pool)); if (conn_pool) { - for (i = ssm->conn_pool_size; i < new_size; i++) + for (i = wrk->conn_pool_size; i < new_size; i++) { - sock_server_conn_t *conn = &conn_pool[i]; + vcl_test_server_conn_t *conn = &conn_pool[i]; memset (conn, 0, sizeof (*conn)); sock_test_cfg_init (&conn->cfg); sock_test_buf_alloc (&conn->cfg, 1 /* is_rxbuf */ , @@ -87,31 +118,27 @@ conn_pool_expand (size_t expand_size) conn->cfg.txbuf_size = conn->cfg.rxbuf_size; } - ssm->conn_pool = conn_pool; - ssm->conn_pool_size = new_size; + wrk->conn_pool = conn_pool; + wrk->conn_pool_size = new_size; } else { - int errno_val = errno; - perror ("ERROR in conn_pool_expand()"); - fprintf (stderr, "SERVER: ERROR: Memory allocation " - "failed (errno = %d)!\n", errno_val); + vterr ("conn_pool_expand()", -errno); } } -static inline sock_server_conn_t * -conn_pool_alloc (void) +static inline vcl_test_server_conn_t * +conn_pool_alloc (vcl_test_server_worker_t * wrk) { - sock_server_main_t *ssm = &sock_server_main; int i; - for (i = 0; i < ssm->conn_pool_size; i++) + for (i = 0; i < wrk->conn_pool_size; i++) { - if (!ssm->conn_pool[i].is_alloc) + if (!wrk->conn_pool[i].is_alloc) { - ssm->conn_pool[i].endpt.ip = ssm->conn_pool[i].ip; - ssm->conn_pool[i].is_alloc = 1; - return (&ssm->conn_pool[i]); + wrk->conn_pool[i].endpt.ip = wrk->conn_pool[i].ip; + wrk->conn_pool[i].is_alloc = 1; + return (&wrk->conn_pool[i]); } } @@ -119,14 +146,15 @@ conn_pool_alloc (void) } static inline void -conn_pool_free (sock_server_conn_t * conn) +conn_pool_free (vcl_test_server_conn_t * conn) { conn->fd = 0; conn->is_alloc = 0; } static inline void -sync_config_and_reply (sock_server_conn_t * conn, sock_test_cfg_t * rx_cfg) +sync_config_and_reply (vcl_test_server_conn_t * conn, + sock_test_cfg_t * rx_cfg) { conn->cfg = *rx_cfg; sock_test_buf_alloc (&conn->cfg, 1 /* is_rxbuf */ , @@ -135,7 +163,7 @@ sync_config_and_reply (sock_server_conn_t * conn, sock_test_cfg_t * rx_cfg) if (conn->cfg.verbose) { - printf ("\nSERVER (fd %d): Replying to cfg message!\n", conn->fd); + vtinf ("(fd %d): Replying to cfg message!\n", conn->fd); sock_test_cfg_dump (&conn->cfg, 0 /* is_client */ ); } (void) vcl_test_write (conn->fd, (uint8_t *) & conn->cfg, @@ -143,10 +171,10 @@ sync_config_and_reply (sock_server_conn_t * conn, sock_test_cfg_t * rx_cfg) } static void -stream_test_server_start_stop (sock_server_conn_t * conn, +stream_test_server_start_stop (vcl_test_server_worker_t * wrk, + vcl_test_server_conn_t * conn, sock_test_cfg_t * rx_cfg) { - sock_server_main_t *ssm = &sock_server_main; int client_fd = conn->fd; sock_test_t test = rx_cfg->test; @@ -155,9 +183,9 @@ stream_test_server_start_stop (sock_server_conn_t * conn, int i; clock_gettime (CLOCK_REALTIME, &conn->stats.stop); - for (i = 0; i < ssm->conn_pool_size; i++) + for (i = 0; i < wrk->conn_pool_size; i++) { - sock_server_conn_t *tc = &ssm->conn_pool[i]; + vcl_test_server_conn_t *tc = &wrk->conn_pool[i]; if (tc->cfg.ctrl_handle == conn->fd) { @@ -215,7 +243,7 @@ stream_test_server_start_stop (sock_server_conn_t * conn, static inline void -stream_test_server (sock_server_conn_t * conn, int rx_bytes) +stream_test_server (vcl_test_server_conn_t * conn, int rx_bytes) { int client_fd = conn->fd; sock_test_t test = conn->cfg.test; @@ -230,36 +258,49 @@ stream_test_server (sock_server_conn_t * conn, int rx_bytes) } } +static void +vcl_test_server_echo (vcl_test_server_conn_t * conn, int rx_bytes) +{ + int tx_bytes, nbytes, pos; + + /* If it looks vaguely like a string, + * make sure it's terminated + */ + pos = rx_bytes < conn->buf_size ? rx_bytes : conn->buf_size - 1; + ((char *) conn->buf)[pos] = 0; + vtinf ("(fd %d): RX (%d bytes) - '%s'", conn->fd, rx_bytes, conn->buf); + + if (conn->cfg.verbose) + vtinf ("(fd %d): Echoing back", conn->fd); + + nbytes = strlen ((const char *) conn->buf) + 1; + tx_bytes = vcl_test_write (conn->fd, conn->buf, nbytes, &conn->stats, + conn->cfg.verbose); + if (tx_bytes >= 0) + vtinf ("(fd %d): TX (%d bytes) - '%s'", conn->fd, tx_bytes, conn->buf); +} + static inline void -new_client (void) +vcl_test_server_new_client (vcl_test_server_worker_t * wrk) { - sock_server_main_t *ssm = &sock_server_main; int client_fd; - sock_server_conn_t *conn; - - if (ssm->conn_pool_size < (ssm->num_conn + SOCK_SERVER_MAX_TEST_CONN + 1)) - conn_pool_expand (SOCK_SERVER_MAX_TEST_CONN + 1); + vcl_test_server_conn_t *conn; - conn = conn_pool_alloc (); + conn = conn_pool_alloc (wrk); if (!conn) { - fprintf (stderr, "\nSERVER: ERROR: No free connections!\n"); + vtwrn ("No free connections!"); return; } - client_fd = vppcom_session_accept (ssm->listen_fd, &conn->endpt, 0); + client_fd = vppcom_session_accept (wrk->listen_fd, &conn->endpt, 0); if (client_fd < 0) { - int errno_val; - errno_val = errno = -client_fd; - perror ("ERROR in new_client()"); - fprintf (stderr, "SERVER: ERROR: accept failed " - "(errno = %d)!\n", errno_val); + vterr ("vppcom_session_accept()", client_fd); return; } - printf ("SERVER: Got a connection -- fd = %d (0x%08x)!\n", - client_fd, client_fd); + vtinf ("Got a connection -- fd = %d (0x%08x)!", client_fd, client_fd); conn->fd = client_fd; @@ -268,18 +309,14 @@ new_client (void) int rv; ev.events = EPOLLIN; - ev.data.u64 = conn - ssm->conn_pool; - rv = vppcom_epoll_ctl (ssm->epfd, EPOLL_CTL_ADD, client_fd, &ev); + ev.data.u64 = conn - wrk->conn_pool; + rv = vppcom_epoll_ctl (wrk->epfd, EPOLL_CTL_ADD, client_fd, &ev); if (rv < 0) { - int errno_val; - errno_val = errno = -rv; - perror ("ERROR in new_client()"); - fprintf (stderr, "SERVER: ERROR: epoll_ctl failed (errno = %d)!\n", - errno_val); + vterr ("vppcom_epoll_ctl()", rv); + return; } - else - ssm->nfds++; + wrk->nfds++; } } @@ -291,30 +328,58 @@ print_usage_and_exit (void) " OPTIONS\n" " -h Print this message and exit.\n" " -6 Use IPv6\n" + " -w <num> Number of workers\n" " -u Use UDP transport layer\n"); exit (1); } -int -main (int argc, char **argv) +static void +vcl_test_init_endpoint_addr (vcl_test_server_main_t * ssm) { - sock_server_main_t *ssm = &sock_server_main; - int client_fd, rv, main_rv = 0; - int tx_bytes, rx_bytes, nbytes; - sock_server_conn_t *conn; - sock_test_cfg_t *rx_cfg; - uint32_t xtra = 0; - uint64_t xtra_bytes = 0; - struct sockaddr_storage servaddr; - int errno_val; - int c, v, i; - uint16_t port = SOCK_TEST_SERVER_PORT; - vppcom_endpt_t endpt; + struct sockaddr_storage *servaddr = &ssm->servaddr; + memset (servaddr, 0, sizeof (*servaddr)); - clib_mem_init_thread_safe (0, 64 << 20); + if (ssm->cfg.address_ip6) + { + struct sockaddr_in6 *server_addr = (struct sockaddr_in6 *) servaddr; + server_addr->sin6_family = AF_INET6; + server_addr->sin6_addr = in6addr_any; + server_addr->sin6_port = htons (ssm->cfg.port); + } + else + { + struct sockaddr_in *server_addr = (struct sockaddr_in *) servaddr; + server_addr->sin_family = AF_INET; + server_addr->sin_addr.s_addr = htonl (INADDR_ANY); + server_addr->sin_port = htons (ssm->cfg.port); + } + + if (ssm->cfg.address_ip6) + { + struct sockaddr_in6 *server_addr = (struct sockaddr_in6 *) servaddr; + ssm->cfg.endpt.is_ip4 = 0; + ssm->cfg.endpt.ip = (uint8_t *) & server_addr->sin6_addr; + ssm->cfg.endpt.port = (uint16_t) server_addr->sin6_port; + } + else + { + struct sockaddr_in *server_addr = (struct sockaddr_in *) servaddr; + ssm->cfg.endpt.is_ip4 = 1; + ssm->cfg.endpt.ip = (uint8_t *) & server_addr->sin_addr; + ssm->cfg.endpt.port = (uint16_t) server_addr->sin_port; + } +} + +void +vcl_test_server_process_opts (vcl_test_server_main_t * ssm, int argc, + char **argv) +{ + int v, c; + + ssm->cfg.proto = VPPCOM_PROTO_TCP; opterr = 0; - while ((c = getopt (argc, argv, "6D")) != -1) + while ((c = getopt (argc, argv, "6Dw:")) != -1) switch (c) { case '6': @@ -322,7 +387,15 @@ main (int argc, char **argv) break; case 'D': - ssm->cfg.transport_udp = 1; + ssm->cfg.proto = VPPCOM_PROTO_UDP; + break; + + case 'w': + v = atoi (optarg); + if (v > 1) + ssm->cfg.workers = v; + else + vtwrn ("Invalid number of workers %d", v); break; case '?': @@ -349,303 +422,252 @@ main (int argc, char **argv) } if (sscanf (argv[optind], "%d", &v) == 1) - port = (uint16_t) v; + ssm->cfg.port = (uint16_t) v; else { fprintf (stderr, "SERVER: ERROR: Invalid port (%s)!\n", argv[optind]); print_usage_and_exit (); } - conn_pool_expand (SOCK_SERVER_MAX_TEST_CONN + 1); + vcl_test_init_endpoint_addr (ssm); +} - rv = vppcom_app_create ("vcl_test_server"); - if (rv) - { - errno_val = errno = -rv; - perror ("ERROR in main()"); - fprintf (stderr, "SERVER: ERROR: vppcom_app_create() failed " - "(errno = %d)!\n", errno_val); - return -1; - } - else +int +vcl_test_server_handle_cfg (vcl_test_server_worker_t * wrk, + sock_test_cfg_t * rx_cfg, + vcl_test_server_conn_t * conn, int rx_bytes) +{ + if (rx_cfg->verbose) { - ssm->listen_fd = vppcom_session_create (ssm->cfg.transport_udp ? - VPPCOM_PROTO_UDP : - VPPCOM_PROTO_TCP, - 0 /* is_nonblocking */ ); + vtinf ("(fd %d): Received a cfg msg!", conn->fd); + sock_test_cfg_dump (rx_cfg, 0 /* is_client */ ); } - if (ssm->listen_fd < 0) + + if (rx_bytes != sizeof (*rx_cfg)) { - errno_val = errno = -ssm->listen_fd; - perror ("ERROR in main()"); - fprintf (stderr, "SERVER: ERROR: vppcom_session_create() failed " - "(errno = %d)!\n", errno_val); + vtinf ("(fd %d): Invalid cfg msg size %d expected %lu!", conn->fd, + rx_bytes, sizeof (*rx_cfg)); + conn->cfg.rxbuf_size = 0; + conn->cfg.num_writes = 0; + if (conn->cfg.verbose) + { + vtinf ("(fd %d): Replying to cfg msg", conn->fd); + sock_test_cfg_dump (rx_cfg, 0 /* is_client */ ); + } + vcl_test_write (conn->fd, (uint8_t *) & conn->cfg, + sizeof (conn->cfg), NULL, conn->cfg.verbose); return -1; } - memset (&servaddr, 0, sizeof (servaddr)); - - if (ssm->cfg.address_ip6) - { - struct sockaddr_in6 *server_addr = (struct sockaddr_in6 *) &servaddr; - server_addr->sin6_family = AF_INET6; - server_addr->sin6_addr = in6addr_any; - server_addr->sin6_port = htons (port); - } - else + switch (rx_cfg->test) { - struct sockaddr_in *server_addr = (struct sockaddr_in *) &servaddr; - server_addr->sin_family = AF_INET; - server_addr->sin_addr.s_addr = htonl (INADDR_ANY); - server_addr->sin_port = htons (port); + case SOCK_TEST_TYPE_NONE: + case SOCK_TEST_TYPE_ECHO: + sync_config_and_reply (conn, rx_cfg); + break; + + case SOCK_TEST_TYPE_BI: + case SOCK_TEST_TYPE_UNI: + stream_test_server_start_stop (wrk, conn, rx_cfg); + break; + + case SOCK_TEST_TYPE_EXIT: + vtinf ("Have a great day conn %d!", conn->fd); + vppcom_session_close (conn->fd); + conn_pool_free (conn); + vtinf ("Closed client fd %d", conn->fd); + wrk->nfds--; + break; + + default: + vtwrn ("Unknown test type %d", rx_cfg->test); + sock_test_cfg_dump (rx_cfg, 0 /* is_client */ ); + break; } - if (ssm->cfg.address_ip6) - { - struct sockaddr_in6 *server_addr = (struct sockaddr_in6 *) &servaddr; - endpt.is_ip4 = 0; - endpt.ip = (uint8_t *) & server_addr->sin6_addr; - endpt.port = (uint16_t) server_addr->sin6_port; - } - else - { - struct sockaddr_in *server_addr = (struct sockaddr_in *) &servaddr; - endpt.is_ip4 = 1; - endpt.ip = (uint8_t *) & server_addr->sin_addr; - endpt.port = (uint16_t) server_addr->sin_port; - } + return 0; +} + +static void +vcl_test_server_worker_init (vcl_test_server_worker_t * wrk) +{ + vcl_test_server_main_t *ssm = &sock_server_main; + struct epoll_event listen_ev; + int rv; + + __wrk_index = wrk->wrk_index; + + vtinf ("Initializing worker ..."); - rv = vppcom_session_bind (ssm->listen_fd, &endpt); + conn_pool_expand (wrk, SOCK_SERVER_MAX_TEST_CONN + 1); + if (wrk->wrk_index) + vppcom_worker_register (); + + wrk->listen_fd = vppcom_session_create (ssm->cfg.proto, + 0 /* is_nonblocking */ ); + if (wrk->listen_fd < 0) + vtfail ("vppcom_session_create()", wrk->listen_fd); + + rv = vppcom_session_bind (wrk->listen_fd, &ssm->cfg.endpt); if (rv < 0) - { - errno_val = errno = -rv; - perror ("ERROR in main()"); - fprintf (stderr, "SERVER: ERROR: bind failed (errno = %d)!\n", - errno_val); - return -1; - } + vtfail ("vppcom_session_bind()", rv); - if (!ssm->cfg.transport_udp) + if (!(ssm->cfg.proto == VPPCOM_PROTO_UDP)) { - rv = vppcom_session_listen (ssm->listen_fd, 10); + rv = vppcom_session_listen (wrk->listen_fd, 10); if (rv < 0) - { - errno_val = errno = -rv; - perror ("ERROR in main()"); - fprintf (stderr, "SERVER: ERROR: listen failed " - "(errno = %d)!\n", errno_val); - return -1; - } + vtfail ("vppcom_session_listen()", rv); } - ssm->epfd = vppcom_epoll_create (); - if (ssm->epfd < 0) - { - errno_val = errno = -ssm->epfd; - perror ("ERROR in main()"); - fprintf (stderr, "SERVER: ERROR: epoll_create failed (errno = %d)!\n", - errno_val); - return -1; - } + wrk->epfd = vppcom_epoll_create (); + if (wrk->epfd < 0) + vtfail ("vppcom_epoll_create()", wrk->epfd); - ssm->listen_ev.events = EPOLLIN; - ssm->listen_ev.data.u32 = ~0; - rv = vppcom_epoll_ctl (ssm->epfd, EPOLL_CTL_ADD, ssm->listen_fd, - &ssm->listen_ev); + listen_ev.events = EPOLLIN; + listen_ev.data.u32 = ~0; + rv = vppcom_epoll_ctl (wrk->epfd, EPOLL_CTL_ADD, wrk->listen_fd, + &listen_ev); if (rv < 0) - { - errno_val = errno = -rv; - perror ("ERROR in main()"); - fprintf (stderr, "SERVER: ERROR: epoll_ctl failed " - "(errno = %d)!\n", errno_val); - return -1; - } - printf ("\nSERVER: Waiting for a client to connect on port %d...\n", port); + vtfail ("vppcom_epoll_ctl", rv); + + vtinf ("Waiting for a client to connect on port %d ...", ssm->cfg.port); +} + +static void * +vcl_test_server_worker_loop (void *arg) +{ + vcl_test_server_main_t *ssm = &sock_server_main; + vcl_test_server_worker_t *wrk = arg; + vcl_test_server_conn_t *conn; + int i, rx_bytes, num_ev; + sock_test_cfg_t *rx_cfg; + + if (wrk->wrk_index) + vcl_test_server_worker_init (wrk); while (1) { - int num_ev; - num_ev = vppcom_epoll_wait (ssm->epfd, ssm->wait_events, + num_ev = vppcom_epoll_wait (wrk->epfd, wrk->wait_events, SOCK_SERVER_MAX_EPOLL_EVENTS, 60000.0); if (num_ev < 0) { - errno = -num_ev; - perror ("epoll_wait()"); - fprintf (stderr, "\nSERVER: ERROR: epoll_wait() " - "failed -- aborting!\n"); - main_rv = -1; - goto done; + vterr ("vppcom_epoll_wait()", num_ev); + goto fail; } else if (num_ev == 0) { - fprintf (stderr, "\nSERVER: epoll_wait() timeout!\n"); + vtinf ("vppcom_epoll_wait() timeout!"); continue; } for (i = 0; i < num_ev; i++) { - conn = &ssm->conn_pool[ssm->wait_events[i].data.u32]; - if (ssm->wait_events[i].events & (EPOLLHUP | EPOLLRDHUP)) + conn = &wrk->conn_pool[wrk->wait_events[i].data.u32]; + if (wrk->wait_events[i].events & (EPOLLHUP | EPOLLRDHUP)) { vppcom_session_close (conn->fd); continue; } - if (ssm->wait_events[i].data.u32 == ~0) + if (wrk->wait_events[i].data.u32 == ~0) { - new_client (); + vcl_test_server_new_client (wrk); continue; } - client_fd = conn->fd; - if (EPOLLIN & ssm->wait_events[i].events) + if (EPOLLIN & wrk->wait_events[i].events) { - rx_bytes = vcl_test_read (client_fd, conn->buf, + rx_bytes = vcl_test_read (conn->fd, conn->buf, conn->buf_size, &conn->stats); - if (rx_bytes > 0) - { - rx_cfg = (sock_test_cfg_t *) conn->buf; - if (rx_cfg->magic == SOCK_TEST_CFG_CTRL_MAGIC) - { - if (rx_cfg->verbose) - { - printf ("SERVER (fd %d): Received a cfg message!\n", - client_fd); - sock_test_cfg_dump (rx_cfg, 0 /* is_client */ ); - } - - if (rx_bytes != sizeof (*rx_cfg)) - { - printf ("SERVER (fd %d): Invalid cfg message " - "size (%d)!\n Should be %lu bytes.\n", - client_fd, rx_bytes, sizeof (*rx_cfg)); - conn->cfg.rxbuf_size = 0; - conn->cfg.num_writes = 0; - if (conn->cfg.verbose) - { - printf ("SERVER (fd %d): Replying to " - "cfg message!\n", client_fd); - sock_test_cfg_dump (rx_cfg, 0 /* is_client */ ); - } - vcl_test_write (client_fd, (uint8_t *) & conn->cfg, - sizeof (conn->cfg), NULL, - conn->cfg.verbose); - continue; - } - - switch (rx_cfg->test) - { - case SOCK_TEST_TYPE_NONE: - case SOCK_TEST_TYPE_ECHO: - sync_config_and_reply (conn, rx_cfg); - break; - - case SOCK_TEST_TYPE_BI: - case SOCK_TEST_TYPE_UNI: - stream_test_server_start_stop (conn, rx_cfg); - break; - - case SOCK_TEST_TYPE_EXIT: - printf ("SERVER: Have a great day, " - "connection %d!\n", client_fd); - vppcom_session_close (client_fd); - conn_pool_free (conn); - printf ("SERVER: Closed client fd %d\n", client_fd); - ssm->nfds--; - if (!ssm->nfds) - { - printf ("SERVER: All client connections " - "closed.\n\nSERVER: " - "May the force be with you!\n\n"); - goto done; - } - break; - - default: - fprintf (stderr, - "SERVER: ERROR: Unknown test type!\n"); - sock_test_cfg_dump (rx_cfg, 0 /* is_client */ ); - break; - } - continue; - } - else if ((conn->cfg.test == SOCK_TEST_TYPE_UNI) || - (conn->cfg.test == SOCK_TEST_TYPE_BI)) - { - stream_test_server (conn, rx_bytes); - continue; - } - - else if (isascii (conn->buf[0])) - { - /* If it looks vaguely like a string, - * make sure it's terminated. - */ - ((char *) conn->buf)[rx_bytes < - conn->buf_size ? rx_bytes : - conn->buf_size - 1] = 0; - printf ("SERVER (fd %d): RX (%d bytes) - '%s'\n", - conn->fd, rx_bytes, conn->buf); - } - } - else // rx_bytes < 0 + if (rx_bytes <= 0) { if (errno == ECONNRESET) { - printf ("\nSERVER: Connection reset by remote peer.\n" - " Y'all have a great day now!\n\n"); - break; + vtinf ("Connection reset by remote peer.\n"); + goto fail; } else continue; } - if (isascii (conn->buf[0])) + rx_cfg = (sock_test_cfg_t *) conn->buf; + if (rx_cfg->magic == SOCK_TEST_CFG_CTRL_MAGIC) + { + vcl_test_server_handle_cfg (wrk, rx_cfg, conn, rx_bytes); + if (!wrk->nfds) + { + vtinf ("All client connections closed\n"); + vtinf ("May the force be with you!\n"); + goto done; + } + continue; + } + else if ((conn->cfg.test == SOCK_TEST_TYPE_UNI) + || (conn->cfg.test == SOCK_TEST_TYPE_BI)) { - /* If it looks vaguely like a string, - * make sure it's terminated - */ - ((char *) conn->buf)[rx_bytes < - conn->buf_size ? rx_bytes : - conn->buf_size - 1] = 0; - if (xtra) - fprintf (stderr, "SERVER: ERROR: " - "FIFO not drained in previous test!\n" - " extra chunks %u (0x%x)\n" - " extra bytes %lu (0x%lx)\n", - xtra, xtra, xtra_bytes, xtra_bytes); - - xtra = 0; - xtra_bytes = 0; - - if (conn->cfg.verbose) - printf ("SERVER (fd %d): Echoing back\n", client_fd); - - nbytes = strlen ((const char *) conn->buf) + 1; - - tx_bytes = vcl_test_write (client_fd, conn->buf, - nbytes, &conn->stats, - conn->cfg.verbose); - if (tx_bytes >= 0) - printf ("SERVER (fd %d): TX (%d bytes) - '%s'\n", - conn->fd, tx_bytes, conn->buf); + stream_test_server (conn, rx_bytes); + continue; } - else // Extraneous read data from non-echo tests??? + else if (isascii (conn->buf[0])) { - xtra++; - xtra_bytes += rx_bytes; + vcl_test_server_echo (conn, rx_bytes); } + else + { + vtwrn ("FIFO not drained! extra bytes %d", rx_bytes); + } + } + else + { + vtwrn ("Unhandled event"); + goto fail; } } } +fail: + ssm->worker_fails -= 1; + done: - vppcom_session_close (ssm->listen_fd); - vppcom_app_destroy (); + vppcom_session_close (wrk->listen_fd); + if (wrk->conn_pool) + free (wrk->conn_pool); + ssm->active_workers -= 1; + return 0; +} + +int +main (int argc, char **argv) +{ + vcl_test_server_main_t *ssm = &sock_server_main; + int rv, i; + + clib_mem_init_thread_safe (0, 64 << 20); + ssm->cfg.port = SOCK_TEST_SERVER_PORT; + ssm->cfg.workers = 1; + vcl_test_server_process_opts (ssm, argc, argv); - if (ssm->conn_pool) - free (ssm->conn_pool); + rv = vppcom_app_create ("vcl_test_server"); + if (rv) + vtfail ("vppcom_app_create()", rv); + + ssm->workers = calloc (ssm->cfg.workers, sizeof (*ssm->workers)); + vcl_test_server_worker_init (&ssm->workers[0]); + for (i = 1; i < ssm->cfg.workers; i++) + { + ssm->workers[i].wrk_index = i; + rv = pthread_create (&ssm->workers[i].thread_handle, NULL, + vcl_test_server_worker_loop, + (void *) &ssm->workers[i]); + } + vcl_test_server_worker_loop (&ssm->workers[0]); + + while (ssm->active_workers) + ; + + vppcom_app_destroy (); + free (ssm->workers); - return main_rv; + return ssm->worker_fails; } /* |