diff options
Diffstat (limited to 'src/uri')
| -rw-r--r-- | src/uri/sock_test.h | 415 | ||||
| -rw-r--r-- | src/uri/sock_test_client.c | 1076 | ||||
| -rw-r--r-- | src/uri/sock_test_server.c | 596 | ||||
| -rw-r--r-- | src/uri/uri_socket_server.c | 231 | ||||
| -rw-r--r-- | src/uri/uri_socket_test.c | 181 | ||||
| -rwxr-xr-x | src/uri/uri_tcp_test.c | 1274 | ||||
| -rw-r--r-- | src/uri/uri_udp_test.c | 1040 | ||||
| -rw-r--r-- | src/uri/vcl_test_client.c | 27 | ||||
| -rw-r--r-- | src/uri/vcl_test_server.c | 27 | ||||
| -rw-r--r-- | src/uri/vppcom.c | 2440 | ||||
| -rw-r--r-- | src/uri/vppcom.h | 152 | ||||
| -rw-r--r-- | src/uri/vppcom_test.conf | 25 |
12 files changed, 7484 insertions, 0 deletions
diff --git a/src/uri/sock_test.h b/src/uri/sock_test.h new file mode 100644 index 00000000..281ba6fd --- /dev/null +++ b/src/uri/sock_test.h @@ -0,0 +1,415 @@ +/* + * Copyright (c) 2017 Cisco and/or its affiliates. + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at: + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +#ifndef __sock_test_h__ +#define __sock_test_h__ + +#include <netdb.h> +#include <errno.h> +#include <stdlib.h> +#include <string.h> + +#define SOCK_TEST_TOKEN_HELP "#H" +#define SOCK_TEST_TOKEN_EXIT "#X" +#define SOCK_TEST_TOKEN_VERBOSE "#V" +#define SOCK_TEST_TOKEN_TXBUF_SIZE "#T:" +#define SOCK_TEST_TOKEN_NUM_TEST_SCKTS "#I:" +#define SOCK_TEST_TOKEN_NUM_WRITES "#N:" +#define SOCK_TEST_TOKEN_RXBUF_SIZE "#R:" +#define SOCK_TEST_TOKEN_SHOW_CFG "#C" +#define SOCK_TEST_TOKEN_RUN_UNI "#U" +#define SOCK_TEST_TOKEN_RUN_BI "#B" + +#define SOCK_TEST_BANNER_STRING \ + "============================================\n" +#define SOCK_TEST_SEPARATOR_STRING \ + " -----------------------------\n" + +#define ONE_GIG (1024*1024*1024) +#define SOCK_TEST_SERVER_PORT 22000 +#define SOCK_TEST_LOCALHOST_IPADDR "127.0.0.1" + +#define SOCK_TEST_CFG_CTRL_MAGIC 0xfeedface +#define SOCK_TEST_CFG_NUM_WRITES_DEF 1000000 +#define SOCK_TEST_CFG_TXBUF_SIZE_DEF 8192 +#define SOCK_TEST_CFG_RXBUF_SIZE_DEF (64*SOCK_TEST_CFG_TXBUF_SIZE_DEF) +#define SOCK_TEST_CFG_BUF_SIZE_MIN 128 +#define SOCK_TEST_CFG_MAX_TEST_SCKTS 5 + +typedef enum +{ + SOCK_TEST_TYPE_NONE, + SOCK_TEST_TYPE_ECHO, + SOCK_TEST_TYPE_UNI, + SOCK_TEST_TYPE_BI, + SOCK_TEST_TYPE_EXIT, +} sock_test_t; + +typedef struct __attribute__ ((packed)) +{ + uint32_t magic; + uint32_t test; + uint32_t ctrl_handle; + uint32_t num_test_sockets; + uint32_t verbose; + uint64_t rxbuf_size; + uint64_t txbuf_size; + uint64_t num_writes; + uint64_t total_bytes; +} sock_test_cfg_t; + +typedef struct +{ + uint64_t rx_xacts; + uint64_t rx_bytes; + uint32_t rx_eagain; + uint32_t rx_incomp; + uint64_t tx_xacts; + uint64_t tx_bytes; + uint32_t tx_eagain; + uint32_t tx_incomp; + struct timespec start; + struct timespec stop; +} sock_test_stats_t; + +typedef struct +{ + int fd; + uint32_t txbuf_size; + char *txbuf; + uint32_t rxbuf_size; + char *rxbuf; + sock_test_cfg_t cfg; + sock_test_stats_t stats; +} sock_test_socket_t; + +static inline void +sock_test_stats_accumulate (sock_test_stats_t * accum, + sock_test_stats_t * incr) +{ + accum->rx_xacts += incr->rx_xacts; + accum->rx_bytes += incr->rx_bytes; + accum->rx_eagain += incr->rx_eagain; + accum->rx_incomp += incr->rx_incomp; + accum->tx_xacts += incr->tx_xacts; + accum->tx_bytes += incr->tx_bytes; + accum->tx_eagain += incr->tx_eagain; + accum->tx_incomp += incr->tx_incomp; +} + +static inline void +sock_test_cfg_init (sock_test_cfg_t *cfg) +{ + cfg->magic = SOCK_TEST_CFG_CTRL_MAGIC; + cfg->test = SOCK_TEST_TYPE_NONE; + cfg->ctrl_handle = ~0; + cfg->num_test_sockets = 1; + cfg->verbose = 0; + cfg->rxbuf_size = SOCK_TEST_CFG_RXBUF_SIZE_DEF; + cfg->num_writes = SOCK_TEST_CFG_NUM_WRITES_DEF; + cfg->txbuf_size = SOCK_TEST_CFG_TXBUF_SIZE_DEF; + cfg->total_bytes = cfg->num_writes * cfg->txbuf_size; +} + +static inline int +sock_test_cfg_verify (sock_test_cfg_t *cfg, sock_test_cfg_t *valid_cfg) +{ + /* Note: txbuf & rxbuf on server are the same buffer, + * so txbuf_size is not included in this check. + */ + return ((cfg->magic == valid_cfg->magic) + && (cfg->test == valid_cfg->test) + && (cfg->verbose == valid_cfg->verbose) + && (cfg->rxbuf_size == valid_cfg->rxbuf_size) + && (cfg->num_writes == valid_cfg->num_writes) + && (cfg->total_bytes == valid_cfg->total_bytes)); +} + +static inline void +sock_test_buf_alloc (sock_test_cfg_t *cfg, uint8_t is_rxbuf, uint8_t **buf, + uint32_t *bufsize) +{ + uint32_t alloc_size = is_rxbuf ? cfg->rxbuf_size : cfg->txbuf_size; + uint8_t *lb = realloc (*buf, (size_t) alloc_size); + + if (lb) + { + if (is_rxbuf) + cfg->rxbuf_size = *bufsize = alloc_size; + else + cfg->txbuf_size = *bufsize = alloc_size; + + *buf = lb; + } + else + { + int errno_val = errno; + perror ("ERROR in sock_test_buf_alloc()"); + fprintf (stderr, "ERROR: Buffer allocation failed (errno = %d)!\n" + " Using buffer size %d instead of desired" + " size (%d)\n", errno_val, *bufsize, alloc_size); + } +} + +static inline void +sock_test_socket_buf_alloc (sock_test_socket_t *socket) +{ + socket->rxbuf_size = socket->cfg.rxbuf_size; + socket->txbuf_size = socket->cfg.txbuf_size; + sock_test_buf_alloc (&socket->cfg, 0 /* is_rxbuf */ , + (uint8_t **) &socket->txbuf, &socket->txbuf_size); + sock_test_buf_alloc (&socket->cfg, 1 /* is_rxbuf */ , + (uint8_t **) &socket->rxbuf, &socket->rxbuf_size); +} + +static inline char * +sock_test_type_str (sock_test_t t) +{ + switch (t) + { + case SOCK_TEST_TYPE_NONE: + return "NONE"; + + case SOCK_TEST_TYPE_ECHO: + return "ECHO"; + + case SOCK_TEST_TYPE_UNI: + return "UNI"; + + case SOCK_TEST_TYPE_BI: + return "BI"; + + case SOCK_TEST_TYPE_EXIT: + return "EXIT"; + + default: + return "Unknown"; + } +} + +static inline void +sock_test_cfg_dump (sock_test_cfg_t * cfg, uint8_t is_client) +{ + char *spc = " "; + + printf (" test config (%p):\n" + SOCK_TEST_SEPARATOR_STRING + " magic: 0x%08x\n" + "%-5s test: %s (%d)\n" + " ctrl handle: %d (0x%x)\n" + "%-5s num test sockets: %u (0x%08x)\n" + "%-5s verbose: %s (%d)\n" + "%-5s rxbuf size: %lu (0x%08lx)\n" + "%-5s txbuf size: %lu (0x%08lx)\n" + "%-5s num writes: %lu (0x%08lx)\n" + " client tx bytes: %lu (0x%08lx)\n" + SOCK_TEST_SEPARATOR_STRING, + (void *) cfg, cfg->magic, + is_client && (cfg->test == SOCK_TEST_TYPE_UNI) ? + "'"SOCK_TEST_TOKEN_RUN_UNI"'" : + is_client && (cfg->test == SOCK_TEST_TYPE_BI) ? + "'"SOCK_TEST_TOKEN_RUN_BI"'" : spc, + sock_test_type_str (cfg->test), cfg->test, + cfg->ctrl_handle, cfg->ctrl_handle, + is_client ? "'"SOCK_TEST_TOKEN_NUM_TEST_SCKTS"'" : spc, + cfg->num_test_sockets, cfg->num_test_sockets, + is_client ? "'"SOCK_TEST_TOKEN_VERBOSE"'" : spc, + cfg->verbose ? "on" : "off", cfg->verbose, + is_client ? "'"SOCK_TEST_TOKEN_RXBUF_SIZE"'" : spc, + cfg->rxbuf_size, cfg->rxbuf_size, + is_client ? "'"SOCK_TEST_TOKEN_TXBUF_SIZE"'" : spc, + cfg->txbuf_size, cfg->txbuf_size, + is_client ? "'"SOCK_TEST_TOKEN_NUM_WRITES"'" : spc, + cfg->num_writes, cfg->num_writes, + cfg->total_bytes, cfg->total_bytes); +} + +static inline void +sock_test_stats_dump (char * header, sock_test_stats_t * stats, + uint8_t show_rx, uint8_t show_tx, + uint8_t verbose) +{ + struct timespec diff; + double duration, rate; + uint64_t total_bytes; + + if ((stats->stop.tv_nsec - stats->start.tv_nsec) < 0) + { + diff.tv_sec = stats->stop.tv_sec - stats->start.tv_sec - 1; + diff.tv_nsec = stats->stop.tv_nsec - stats->start.tv_nsec + 1000000000; + } + else + { + diff.tv_sec = stats->stop.tv_sec - stats->start.tv_sec; + diff.tv_nsec = stats->stop.tv_nsec - stats->start.tv_nsec; + } + duration = (double) diff.tv_sec + (1e-9 * diff.tv_nsec); + + total_bytes = stats->tx_bytes + stats->rx_bytes; + rate = (double) total_bytes * 8 / duration / ONE_GIG; + printf ("\n%s: Streamed %lu bytes\n" + " in %lf seconds (%lf Gbps %s-duplex)!\n", + header, total_bytes, duration, rate, + (show_rx && show_tx) ? "full" : "half"); + + if (show_tx) + { + printf (SOCK_TEST_SEPARATOR_STRING + " tx stats (0x%p):\n" + SOCK_TEST_SEPARATOR_STRING + " writes: %lu (0x%08lx)\n" + " tx bytes: %lu (0x%08lx)\n" + " tx eagain: %u (0x%08x)\n" + " tx incomplete: %u (0x%08x)\n", + (void *)stats, stats->tx_xacts, stats->tx_xacts, + stats->tx_bytes, stats->tx_bytes, + stats->tx_eagain, stats->tx_eagain, + stats->tx_incomp, stats->tx_incomp); + } + if (show_rx) + { + printf (SOCK_TEST_SEPARATOR_STRING + " rx stats (0x%p):\n" + SOCK_TEST_SEPARATOR_STRING + " reads: %lu (0x%08lx)\n" + " rx bytes: %lu (0x%08lx)\n" + " rx eagain: %u (0x%08x)\n" + " rx incomplete: %u (0x%08x)\n", + (void *)stats, stats->rx_xacts, stats->rx_xacts, + stats->rx_bytes, stats->rx_bytes, + stats->rx_eagain, stats->rx_eagain, + stats->rx_incomp, stats->rx_incomp); + } + if (verbose) + printf (" start.tv_sec: %ld\n" + " start.tv_nsec: %ld\n" + " stop.tv_sec: %ld\n" + " stop.tv_nsec: %ld\n", + stats->start.tv_sec, stats->start.tv_nsec, + stats->stop.tv_sec, stats->stop.tv_nsec); + + printf (SOCK_TEST_SEPARATOR_STRING); +} + +static inline int +sock_test_read (int fd, uint8_t *buf, uint32_t nbytes, + sock_test_stats_t *stats) +{ + int rx_bytes, errno_val; + + do + { + if (stats) + stats->rx_xacts++; +#ifdef VCL_TEST + rx_bytes = vppcom_session_read (fd, buf, nbytes); + + if (rx_bytes < 0) + { + errno = -rx_bytes; + rx_bytes = -1; + } +#else + rx_bytes = read (fd, buf, nbytes); +#endif + if (stats) + { + if ((rx_bytes == 0) || + ((rx_bytes < 0) && ((errno == EAGAIN) || (errno == EWOULDBLOCK)))) + stats->rx_eagain++; + else if (rx_bytes < nbytes) + stats->rx_incomp++; + } + } + while ((rx_bytes == 0) || + ((rx_bytes < 0) && ((errno == EAGAIN) || (errno == EWOULDBLOCK)))); + + if (rx_bytes < 0) + { + errno_val = errno; + perror ("ERROR in sock_test_read()"); + fprintf (stderr, "ERROR: socket read failed (errno = %d)!\n", + errno_val); + errno = errno_val; + } + else if (stats) + stats->rx_bytes += rx_bytes; + + return (rx_bytes); +} + +static inline int +sock_test_write (int fd, uint8_t *buf, uint32_t nbytes, + sock_test_stats_t *stats, uint32_t verbose) +{ + int tx_bytes = 0; + int nbytes_left = nbytes; + int rv, errno_val; + + do + { + if (stats) + stats->tx_xacts++; +#ifdef VCL_TEST + rv = vppcom_session_write (fd, buf, nbytes_left); + if (rv < 0) + { + errno = -rv; + rv = -1; + } +#else + rv = write (fd, buf, nbytes_left); +#endif + if (rv < 0) + { + if ((errno == EAGAIN) || (errno == EWOULDBLOCK)) + { + if (stats) + stats->tx_eagain++; + continue; + } + else + break; + } + tx_bytes += rv; + + if (tx_bytes != nbytes) + { + nbytes_left = nbytes_left - rv; + if (stats) + stats->tx_incomp++; + if (verbose) + { + printf ("WARNING: bytes written (%d) != bytes to write (%d)!\n", + tx_bytes, nbytes); + } + } + + } while (tx_bytes != nbytes); + + if (tx_bytes < 0) + { + errno_val = errno; + perror ("ERROR in sock_test_write()"); + fprintf (stderr, "ERROR: socket write failed (errno = %d)!\n", + errno_val); + } + else if (stats) + stats->tx_bytes += tx_bytes; + + return (tx_bytes); +} + +#endif /* __sock_test_h__ */ diff --git a/src/uri/sock_test_client.c b/src/uri/sock_test_client.c new file mode 100644 index 00000000..151c90b2 --- /dev/null +++ b/src/uri/sock_test_client.c @@ -0,0 +1,1076 @@ +/* + * Copyright (c) 2017 Cisco and/or its affiliates. + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at: + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +#include <unistd.h> +#include <errno.h> +#include <stdlib.h> +#include <ctype.h> +#include <sys/types.h> +#include <sys/socket.h> +#include <stdio.h> +#include <time.h> +#include <arpa/inet.h> +#include <uri/sock_test.h> + +typedef struct +{ +#ifdef VCL_TEST + vppcom_endpt_t server_endpt; +#endif + struct sockaddr_in server_addr; + sock_test_socket_t ctrl_socket; + sock_test_socket_t *test_socket; + uint32_t num_test_sockets; + uint8_t dump_cfg; +} sock_client_main_t; + +sock_client_main_t sock_client_main; + + +static int +sock_test_cfg_sync (sock_test_socket_t * socket) +{ + sock_client_main_t *scm = &sock_client_main; + sock_test_socket_t *ctrl = &scm->ctrl_socket; + sock_test_cfg_t *rl_cfg = (sock_test_cfg_t *) socket->rxbuf; + int rx_bytes, tx_bytes; + + if (socket->cfg.verbose) + sock_test_cfg_dump (&socket->cfg, 1 /* is_client */ ); + + tx_bytes = sock_test_write (socket->fd, (uint8_t *) & ctrl->cfg, + sizeof (ctrl->cfg), NULL, ctrl->cfg.verbose); + if (tx_bytes < 0) + { + fprintf (stderr, "ERROR: write test cfg failed (%d)!\n", tx_bytes); + return tx_bytes; + } + + rx_bytes = sock_test_read (socket->fd, (uint8_t *) socket->rxbuf, + sizeof (sock_test_cfg_t), NULL); + if (rx_bytes < 0) + return rx_bytes; + + if (rl_cfg->magic != SOCK_TEST_CFG_CTRL_MAGIC) + { + fprintf (stderr, "ERROR: Bad server reply cfg -- aborting!\n"); + return -1; + } + if (socket->cfg.verbose) + { + printf ("CLIENT (fd %d): Got config back from server.\n", socket->fd); + sock_test_cfg_dump (rl_cfg, 1 /* is_client */ ); + } + if ((rx_bytes != sizeof (sock_test_cfg_t)) + || !sock_test_cfg_verify (rl_cfg, &ctrl->cfg)) + { + fprintf (stderr, + "ERROR: Invalid config received from server -- aborting!\n"); + sock_test_cfg_dump (rl_cfg, 1 /* is_client */ ); + return -1; + } + ctrl->cfg.ctrl_handle = ((ctrl->cfg.ctrl_handle == ~0) ? + rl_cfg->ctrl_handle : ctrl->cfg.ctrl_handle); + + return 0; +} + +static void +echo_test_client () +{ + sock_client_main_t *scm = &sock_client_main; + sock_test_socket_t *ctrl = &scm->ctrl_socket; + sock_test_socket_t *tsock; + int rx_bytes, tx_bytes, nbytes; + uint32_t i, n; + int rv; + int nfds = 0; + fd_set wr_fdset, rd_fdset; + fd_set _wfdset, *wfdset = &_wfdset; + fd_set _rfdset, *rfdset = &_rfdset; + + FD_ZERO (&wr_fdset); + FD_ZERO (&rd_fdset); + memset (&ctrl->stats, 0, sizeof (ctrl->stats)); + ctrl->cfg.total_bytes = nbytes = strlen (ctrl->txbuf) + 1; + for (n = 0; n != ctrl->cfg.num_test_sockets; n++) + { + tsock = &scm->test_socket[n]; + tsock->cfg = ctrl->cfg; + sock_test_socket_buf_alloc (tsock); + sock_test_cfg_sync (tsock); + + memcpy (tsock->txbuf, ctrl->txbuf, nbytes); + 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; + } + + nfds++; + clock_gettime (CLOCK_REALTIME, &ctrl->stats.start); + while (n) + { + _wfdset = wr_fdset; + _rfdset = rd_fdset; + +#ifdef VCL_TEST + rv = vppcom_select (nfds, (uint64_t *) rfdset, (uint64_t *) wfdset, + NULL, 0); +#else + { + struct timeval timeout; + timeout.tv_sec = 0; + timeout.tv_usec = 0; + rv = select (nfds, rfdset, wfdset, NULL, &timeout); + } +#endif + if (rv < 0) + { + perror ("select()"); + fprintf (stderr, "\nERROR: select() failed -- aborting test!\n"); + return; + } + else if (rv == 0) + continue; + + for (i = 0; i < ctrl->cfg.num_test_sockets; i++) + { + tsock = &scm->test_socket[i]; + if (!((tsock->stats.stop.tv_sec == 0) && + (tsock->stats.stop.tv_nsec == 0))) + continue; + + if (FD_ISSET (tsock->fd, wfdset) && + (tsock->stats.tx_bytes < ctrl->cfg.total_bytes)) + + { + tx_bytes = + sock_test_write (tsock->fd, (uint8_t *) tsock->txbuf, nbytes, + &tsock->stats, ctrl->cfg.verbose); + if (tx_bytes < 0) + { + fprintf (stderr, "\nERROR: sock_test_write(%d) failed " + "-- aborting test!\n", tsock->fd); + return; + } + + printf ("CLIENT (fd %d): TX (%d bytes) - '%s'\n", + tsock->fd, tx_bytes, tsock->txbuf); + } + + if ((FD_ISSET (tsock->fd, rfdset)) && + (tsock->stats.rx_bytes < ctrl->cfg.total_bytes)) + { + rx_bytes = + sock_test_read (tsock->fd, (uint8_t *) tsock->rxbuf, + nbytes, &tsock->stats); + if (rx_bytes > 0) + { + printf ("CLIENT (fd %d): RX (%d bytes) - '%s'\n", + tsock->fd, rx_bytes, tsock->rxbuf); + + if (tsock->stats.rx_bytes != tsock->stats.tx_bytes) + printf + ("WARNING: bytes read (%lu) != bytes written (%lu)!\n", + tsock->stats.rx_bytes, tsock->stats.tx_bytes); + } + } + + if (tsock->stats.rx_bytes >= ctrl->cfg.total_bytes) + { + clock_gettime (CLOCK_REALTIME, &tsock->stats.stop); + n--; + } + } + } + clock_gettime (CLOCK_REALTIME, &ctrl->stats.stop); + + for (i = 0; i < ctrl->cfg.num_test_sockets; i++) + { + tsock = &scm->test_socket[i]; + tsock->stats.start = ctrl->stats.start; + + if (ctrl->cfg.verbose) + { + static char buf[64]; + + sprintf (buf, "CLIENT (fd %d) RESULTS", tsock->fd); + sock_test_stats_dump (buf, &tsock->stats, + 1 /* show_rx */ , 1 /* show tx */ , + ctrl->cfg.verbose); + } + + sock_test_stats_accumulate (&ctrl->stats, &tsock->stats); + } + + if (ctrl->cfg.verbose) + { + sock_test_stats_dump ("CLIENT RESULTS", &ctrl->stats, + 1 /* show_rx */ , 1 /* show tx */ , + ctrl->cfg.verbose); + sock_test_cfg_dump (&ctrl->cfg, 1 /* is_client */ ); + + if (ctrl->cfg.verbose > 1) + { + printf (" ctrl socket info\n" + SOCK_TEST_SEPARATOR_STRING + " fd: %d (0x%08x)\n" + " rxbuf: %p\n" + " rxbuf size: %u (0x%08x)\n" + " txbuf: %p\n" + " txbuf size: %u (0x%08x)\n" + SOCK_TEST_SEPARATOR_STRING, + ctrl->fd, (uint32_t) ctrl->fd, + ctrl->rxbuf, ctrl->rxbuf_size, ctrl->rxbuf_size, + ctrl->txbuf, ctrl->txbuf_size, ctrl->txbuf_size); + } + } +} + +static void +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; + fd_set wr_fdset, rd_fdset; + fd_set _wfdset, *wfdset = &_wfdset; + fd_set _rfdset, *rfdset = (test == SOCK_TEST_TYPE_BI) ? &_rfdset : 0; + + ctrl->cfg.total_bytes = ctrl->cfg.num_writes * ctrl->cfg.txbuf_size; + ctrl->cfg.ctrl_handle = ~0; + + printf ("\n" SOCK_TEST_BANNER_STRING + "CLIENT (fd %d): %s-directional Stream Test!\n\n" + "CLIENT (fd %d): Sending config to server on ctrl socket...\n", + ctrl->fd, test == SOCK_TEST_TYPE_BI ? "Bi" : "Uni", ctrl->fd); + + if (sock_test_cfg_sync (ctrl)) + { + fprintf (stderr, "ERROR: test cfg sync failed -- aborting!"); + return; + } + + FD_ZERO (&wr_fdset); + FD_ZERO (&rd_fdset); + memset (&ctrl->stats, 0, sizeof (ctrl->stats)); + for (n = 0; n != ctrl->cfg.num_test_sockets; n++) + { + tsock = &scm->test_socket[n]; + tsock->cfg = ctrl->cfg; + sock_test_socket_buf_alloc (tsock); + printf ("CLIENT (fd %d): Sending config to server on " + "test socket %d...\n", tsock->fd, n); + sock_test_cfg_sync (tsock); + + /* Fill payload with incrementing uint32's */ + for (i = 0; i < tsock->txbuf_size; i++) + tsock->txbuf[i] = i & 0xff; + + 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; + } + + nfds++; + clock_gettime (CLOCK_REALTIME, &ctrl->stats.start); + while (n) + { + _wfdset = wr_fdset; + _rfdset = rd_fdset; + +#ifdef VCL_TEST + rv = vppcom_select (nfds, (uint64_t *) rfdset, (uint64_t *) wfdset, + NULL, 0); +#else + { + struct timeval timeout; + timeout.tv_sec = 0; + timeout.tv_usec = 0; + rv = select (nfds, rfdset, wfdset, NULL, &timeout); + } +#endif + if (rv < 0) + { + perror ("select()"); + fprintf (stderr, "\nERROR: select() failed -- aborting test!\n"); + return; + } + else if (rv == 0) + continue; + + for (i = 0; i < ctrl->cfg.num_test_sockets; i++) + { + tsock = &scm->test_socket[i]; + if (!((tsock->stats.stop.tv_sec == 0) && + (tsock->stats.stop.tv_nsec == 0))) + continue; + + if (FD_ISSET (tsock->fd, wfdset) && + (tsock->stats.tx_bytes < ctrl->cfg.total_bytes)) + { + tx_bytes = + sock_test_write (tsock->fd, (uint8_t *) tsock->txbuf, + ctrl->cfg.txbuf_size, &tsock->stats, + ctrl->cfg.verbose); + if (tx_bytes < 0) + { + fprintf (stderr, "\nERROR: sock_test_write(%d) failed " + "-- aborting test!\n", tsock->fd); + return; + } + } + + if ((test == SOCK_TEST_TYPE_BI) && + FD_ISSET (tsock->fd, rfdset) && + (tsock->stats.rx_bytes < ctrl->cfg.total_bytes)) + { + (void) sock_test_read (tsock->fd, + (uint8_t *) tsock->rxbuf, + tsock->rxbuf_size, &tsock->stats); + } + + if (((test == SOCK_TEST_TYPE_UNI) && + (tsock->stats.tx_bytes >= ctrl->cfg.total_bytes)) || + ((test == SOCK_TEST_TYPE_BI) && + (tsock->stats.rx_bytes >= ctrl->cfg.total_bytes))) + { + clock_gettime (CLOCK_REALTIME, &tsock->stats.stop); + n--; + } + } + } + clock_gettime (CLOCK_REALTIME, &ctrl->stats.stop); + + printf ("CLIENT (fd %d): Sending config to server on ctrl socket...\n", + ctrl->fd); + + if (sock_test_cfg_sync (ctrl)) + { + fprintf (stderr, "ERROR: test cfg sync failed -- aborting!"); + return; + } + + for (i = 0; i < ctrl->cfg.num_test_sockets; i++) + { + tsock = &scm->test_socket[i]; + + if (ctrl->cfg.verbose) + { + static char buf[64]; + + sprintf (buf, "CLIENT (fd %d) RESULTS", tsock->fd); + sock_test_stats_dump (buf, &tsock->stats, + test == SOCK_TEST_TYPE_BI /* show_rx */ , + 1 /* show tx */ , ctrl->cfg.verbose); + } + + sock_test_stats_accumulate (&ctrl->stats, &tsock->stats); + } + + sock_test_stats_dump ("CLIENT RESULTS", &ctrl->stats, + test == SOCK_TEST_TYPE_BI /* show_rx */ , + 1 /* show tx */ , ctrl->cfg.verbose); + sock_test_cfg_dump (&ctrl->cfg, 1 /* is_client */ ); + + if (ctrl->cfg.verbose) + { + printf (" ctrl socket info\n" + SOCK_TEST_SEPARATOR_STRING + " fd: %d (0x%08x)\n" + " rxbuf: %p\n" + " rxbuf size: %u (0x%08x)\n" + " txbuf: %p\n" + " txbuf size: %u (0x%08x)\n" + SOCK_TEST_SEPARATOR_STRING, + ctrl->fd, (uint32_t) ctrl->fd, + ctrl->rxbuf, ctrl->rxbuf_size, ctrl->rxbuf_size, + ctrl->txbuf, ctrl->txbuf_size, ctrl->txbuf_size); + } + + ctrl->cfg.test = SOCK_TEST_TYPE_ECHO; + if (sock_test_cfg_sync (ctrl)) + fprintf (stderr, "ERROR: post-test cfg sync failed!"); + + printf ("CLIENT (fd %d): %s-directional Stream Test Complete!\n" + SOCK_TEST_BANNER_STRING "\n", ctrl->fd, + test == SOCK_TEST_TYPE_BI ? "Bi" : "Uni"); +} + +static void +exit_client (void) +{ + sock_client_main_t *scm = &sock_client_main; + sock_test_socket_t *ctrl = &scm->ctrl_socket; + sock_test_socket_t *tsock; + int i; + + for (i = 0; i < ctrl->cfg.num_test_sockets; i++) + { + tsock = &scm->test_socket[i]; + tsock->cfg.test = SOCK_TEST_TYPE_EXIT; + + /* coverity[COPY_PASTE_ERROR] */ + if (ctrl->cfg.verbose) + { + printf ("\nCLIENT (fd %d): Sending exit cfg to server...\n", + tsock->fd); + sock_test_cfg_dump (&tsock->cfg, 1 /* is_client */ ); + } + (void) sock_test_write (tsock->fd, (uint8_t *) & tsock->cfg, + sizeof (tsock->cfg), &tsock->stats, + ctrl->cfg.verbose); + } + + ctrl->cfg.test = SOCK_TEST_TYPE_EXIT; + if (ctrl->cfg.verbose) + { + printf ("\nCLIENT (fd %d): Sending exit cfg to server...\n", ctrl->fd); + sock_test_cfg_dump (&ctrl->cfg, 1 /* is_client */ ); + } + (void) sock_test_write (ctrl->fd, (uint8_t *) & ctrl->cfg, + sizeof (ctrl->cfg), &ctrl->stats, + ctrl->cfg.verbose); + printf ("\nCLIENT: So long and thanks for all the fish!\n\n"); + sleep (1); +} + +static int +sock_test_connect_test_sockets (uint32_t num_test_sockets) +{ + sock_client_main_t *scm = &sock_client_main; + sock_test_socket_t *ctrl = &scm->ctrl_socket; + sock_test_socket_t *tsock; + int i, rv, errno_val; + + if (num_test_sockets < 1) + { + errno = EINVAL; + return -1; + } + + if (num_test_sockets < scm->num_test_sockets) + { + for (i = scm->num_test_sockets - 1; i >= num_test_sockets; i--) + { + tsock = &scm->test_socket[i]; +#ifdef VCL_TEST + vppcom_session_close (tsock->fd); +#else + close (tsock->fd); +#endif + free (tsock->txbuf); + free (tsock->rxbuf); + } + } + + else if (num_test_sockets > scm->num_test_sockets) + { + tsock = realloc (scm->test_socket, + sizeof (sock_test_socket_t) * num_test_sockets); + if (!tsock) + { + errno_val = errno; + perror ("ERROR in sock_test_connect_test_sockets()"); + fprintf (stderr, "ERROR: socket failed (errno = %d)!\n", errno_val); + return -1; + } + + if (!scm->test_socket) + memset (tsock, 0, sizeof (*tsock)); + + scm->test_socket = tsock; + for (i = scm->num_test_sockets; i < num_test_sockets; i++) + { + tsock = &scm->test_socket[i]; +#ifdef VCL_TEST + tsock->fd = + vppcom_session_create (VPPCOM_VRF_DEFAULT, VPPCOM_PROTO_TCP, + 0 /* is_nonblocking */ ); + if (tsock->fd < 0) + { + errno = -tsock->fd; + tsock->fd = -1; + } +#else + tsock->fd = socket (AF_INET, SOCK_STREAM, 0); +#endif + if (tsock->fd < 0) + { + errno_val = errno; + perror ("ERROR in sock_test_connect_test_sockets()"); + fprintf (stderr, "ERROR: socket failed (errno = %d)!\n", + errno_val); + return tsock->fd; + } + +#ifdef VCL_TEST + rv = vppcom_session_connect (tsock->fd, &scm->server_endpt); +#else + rv = + connect (tsock->fd, (struct sockaddr *) &scm->server_addr, + sizeof (scm->server_addr)); +#endif + if (rv < 0) + { + errno_val = errno; + perror ("ERROR in main()"); + fprintf (stderr, "ERROR: connect failed (errno = %d)!\n", + errno_val); + } + tsock->cfg = ctrl->cfg; + sock_test_socket_buf_alloc (tsock); + sock_test_cfg_sync (tsock); + + printf ("CLIENT (fd %d): Test socket %d connected.\n", + tsock->fd, i); + } + } + + scm->num_test_sockets = num_test_sockets; + printf ("CLIENT: All sockets (%d) connected!\n", scm->num_test_sockets + 1); + return 0; +} + +static void +dump_help (void) +{ +#define INDENT "\n " + + printf ("Test configuration commands:" + INDENT SOCK_TEST_TOKEN_HELP + "\t\t\tDisplay help." + INDENT SOCK_TEST_TOKEN_EXIT + "\t\t\tExit test client & server." + INDENT SOCK_TEST_TOKEN_SHOW_CFG + "\t\t\tShow the current test cfg." + INDENT SOCK_TEST_TOKEN_RUN_UNI + "\t\t\tRun the Uni-directional test." + INDENT SOCK_TEST_TOKEN_RUN_BI + "\t\t\tRun the Bi-directional test." + INDENT SOCK_TEST_TOKEN_VERBOSE + "\t\t\tToggle verbose setting." + INDENT SOCK_TEST_TOKEN_RXBUF_SIZE + "<rxbuf size>\tRx buffer size (bytes)." + INDENT SOCK_TEST_TOKEN_TXBUF_SIZE + "<txbuf size>\tTx buffer size (bytes)." + INDENT SOCK_TEST_TOKEN_NUM_WRITES + "<# of writes>\tNumber of txbuf writes to server." "\n"); +} + +static void +cfg_txbuf_size_set (void) +{ + sock_client_main_t *scm = &sock_client_main; + sock_test_socket_t *ctrl = &scm->ctrl_socket; + char *p = ctrl->txbuf + strlen (SOCK_TEST_TOKEN_TXBUF_SIZE); + uint64_t txbuf_size = strtoull ((const char *) p, NULL, 10); + + if (txbuf_size >= SOCK_TEST_CFG_BUF_SIZE_MIN) + { + ctrl->cfg.txbuf_size = txbuf_size; + ctrl->cfg.total_bytes = ctrl->cfg.num_writes * ctrl->cfg.txbuf_size; + sock_test_buf_alloc (&ctrl->cfg, 0 /* is_rxbuf */ , + (uint8_t **) & ctrl->txbuf, &ctrl->txbuf_size); + sock_test_cfg_dump (&ctrl->cfg, 1 /* is_client */ ); + } + else + fprintf (stderr, + "ERROR: Invalid txbuf size (%lu) < minimum buf size (%u)!\n", + txbuf_size, SOCK_TEST_CFG_BUF_SIZE_MIN); +} + +static void +cfg_num_writes_set (void) +{ + sock_client_main_t *scm = &sock_client_main; + sock_test_socket_t *ctrl = &scm->ctrl_socket; + char *p = ctrl->txbuf + strlen (SOCK_TEST_TOKEN_NUM_WRITES); + uint32_t num_writes = strtoul ((const char *) p, NULL, 10); + + if (num_writes > 0) + { + ctrl->cfg.num_writes = num_writes; + ctrl->cfg.total_bytes = ctrl->cfg.num_writes * ctrl->cfg.txbuf_size; + sock_test_cfg_dump (&ctrl->cfg, 1 /* is_client */ ); + } + else + { + fprintf (stderr, "ERROR: invalid num writes: %u\n", num_writes); + } +} + +static void +cfg_num_test_sockets_set (void) +{ + sock_client_main_t *scm = &sock_client_main; + sock_test_socket_t *ctrl = &scm->ctrl_socket; + char *p = ctrl->txbuf + strlen (SOCK_TEST_TOKEN_NUM_TEST_SCKTS); + uint32_t num_test_sockets = strtoul ((const char *) p, NULL, 10); + + if ((num_test_sockets > 0) && + (num_test_sockets <= SOCK_TEST_CFG_MAX_TEST_SCKTS)) + { + ctrl->cfg.num_test_sockets = num_test_sockets; + sock_test_connect_test_sockets (num_test_sockets); + sock_test_cfg_dump (&ctrl->cfg, 1 /* is_client */ ); + } + else + { + fprintf (stderr, "ERROR: invalid num test sockets: %u, (%d max)\n", + num_test_sockets, SOCK_TEST_CFG_MAX_TEST_SCKTS); + } +} + +static void +cfg_rxbuf_size_set (void) +{ + sock_client_main_t *scm = &sock_client_main; + sock_test_socket_t *ctrl = &scm->ctrl_socket; + char *p = ctrl->txbuf + strlen (SOCK_TEST_TOKEN_RXBUF_SIZE); + uint64_t rxbuf_size = strtoull ((const char *) p, NULL, 10); + + if (rxbuf_size >= SOCK_TEST_CFG_BUF_SIZE_MIN) + { + ctrl->cfg.rxbuf_size = rxbuf_size; + sock_test_buf_alloc (&ctrl->cfg, 1 /* is_rxbuf */ , + (uint8_t **) & ctrl->rxbuf, &ctrl->rxbuf_size); + sock_test_cfg_dump (&ctrl->cfg, 1 /* is_client */ ); + } + else + fprintf (stderr, + "ERROR: Invalid rxbuf size (%lu) < minimum buf size (%u)!\n", + rxbuf_size, SOCK_TEST_CFG_BUF_SIZE_MIN); +} + +static void +cfg_verbose_toggle (void) +{ + sock_client_main_t *scm = &sock_client_main; + sock_test_socket_t *ctrl = &scm->ctrl_socket; + + ctrl->cfg.verbose = ctrl->cfg.verbose ? 0 : 1; + sock_test_cfg_dump (&ctrl->cfg, 1 /* is_client */ ); + +} + +static sock_test_t +parse_input () +{ + sock_client_main_t *scm = &sock_client_main; + sock_test_socket_t *ctrl = &scm->ctrl_socket; + sock_test_t rv = SOCK_TEST_TYPE_NONE; + + if (!strcmp (SOCK_TEST_TOKEN_EXIT, ctrl->txbuf)) + rv = SOCK_TEST_TYPE_EXIT; + + else if (!strcmp (SOCK_TEST_TOKEN_HELP, ctrl->txbuf)) + dump_help (); + + else if (!strcmp (SOCK_TEST_TOKEN_SHOW_CFG, ctrl->txbuf)) + scm->dump_cfg = 1; + + else if (!strcmp (SOCK_TEST_TOKEN_VERBOSE, ctrl->txbuf)) + cfg_verbose_toggle (); + + else if (!strncmp (SOCK_TEST_TOKEN_TXBUF_SIZE, ctrl->txbuf, + strlen (SOCK_TEST_TOKEN_TXBUF_SIZE))) + cfg_txbuf_size_set (); + + else if (!strncmp (SOCK_TEST_TOKEN_NUM_TEST_SCKTS, ctrl->txbuf, + strlen (SOCK_TEST_TOKEN_NUM_TEST_SCKTS))) + cfg_num_test_sockets_set (); + + else if (!strncmp (SOCK_TEST_TOKEN_NUM_WRITES, ctrl->txbuf, + strlen (SOCK_TEST_TOKEN_NUM_WRITES))) + cfg_num_writes_set (); + + else if (!strncmp (SOCK_TEST_TOKEN_RXBUF_SIZE, ctrl->txbuf, + strlen (SOCK_TEST_TOKEN_RXBUF_SIZE))) + cfg_rxbuf_size_set (); + + else if (!strncmp (SOCK_TEST_TOKEN_RUN_UNI, ctrl->txbuf, + strlen (SOCK_TEST_TOKEN_RUN_UNI))) + rv = ctrl->cfg.test = SOCK_TEST_TYPE_UNI; + + else if (!strncmp (SOCK_TEST_TOKEN_RUN_BI, ctrl->txbuf, + strlen (SOCK_TEST_TOKEN_RUN_BI))) + rv = ctrl->cfg.test = SOCK_TEST_TYPE_BI; + + else + rv = SOCK_TEST_TYPE_ECHO; + + return rv; +} + +void +print_usage_and_exit (void) +{ + fprintf (stderr, + "sock_test_client [OPTIONS] <ipaddr> <port>\n" + " OPTIONS\n" + " -h Print this message and exit.\n" + " -c Print test config before test.\n" + " -w <dir> Write test results to <dir>.\n" + " -X Exit after running test.\n" + " -E Run Echo test.\n" + " -N <num-writes> Test Cfg: number of writes.\n" + " -R <rxbuf-size> Test Cfg: rx buffer size.\n" + " -T <txbuf-size> Test Cfg: tx buffer size.\n" + " -U Run Uni-directional test.\n" + " -B Run Bi-directional test.\n" + " -V Verbose mode.\n"); + exit (1); +} + +int +main (int argc, char **argv) +{ + sock_client_main_t *scm = &sock_client_main; + sock_test_socket_t *ctrl = &scm->ctrl_socket; + int c, rv, errno_val; + sock_test_t post_test = SOCK_TEST_TYPE_NONE; + + sock_test_cfg_init (&ctrl->cfg); + sock_test_socket_buf_alloc (ctrl); + + opterr = 0; + while ((c = getopt (argc, argv, "chn:w:XE:I:N:R:T:UBV")) != -1) + switch (c) + { + case 'c': + scm->dump_cfg = 1; + break; + + case 's': + if (sscanf (optarg, "0x%x", &ctrl->cfg.num_test_sockets) != 1) + if (sscanf (optarg, "%u", &ctrl->cfg.num_test_sockets) != 1) + { + fprintf (stderr, "ERROR: Invalid value for option -%c!\n", c); + print_usage_and_exit (); + } + if (!ctrl->cfg.num_test_sockets || + (ctrl->cfg.num_test_sockets > FD_SETSIZE)) + { + fprintf (stderr, "ERROR: Invalid number of sockets (%d)" + "specified for option -%c!\n" + " Valid range is 1 - %d\n", + ctrl->cfg.num_test_sockets, c, FD_SETSIZE); + print_usage_and_exit (); + } + break; + + case 'w': + fprintf (stderr, "Writing test results to files is TBD.\n"); + break; + + case 'X': + post_test = SOCK_TEST_TYPE_EXIT; + break; + + case 'E': + if (strlen (optarg) > ctrl->txbuf_size) + { + fprintf (stderr, + "ERROR: Option -%c value larger than txbuf size (%d)!\n", + optopt, ctrl->txbuf_size); + print_usage_and_exit (); + } + strcpy (ctrl->txbuf, optarg); + ctrl->cfg.test = SOCK_TEST_TYPE_ECHO; + break; + + case 'I': + if (sscanf (optarg, "0x%x", &ctrl->cfg.num_test_sockets) != 1) + if (sscanf (optarg, "%d", &ctrl->cfg.num_test_sockets) != 1) + { + fprintf (stderr, "ERROR: Invalid value for option -%c!\n", c); + print_usage_and_exit (); + } + if (ctrl->cfg.num_test_sockets > SOCK_TEST_CFG_MAX_TEST_SCKTS) + { + fprintf (stderr, "ERROR: value greater than max number test" + " sockets (%d)!", SOCK_TEST_CFG_MAX_TEST_SCKTS); + print_usage_and_exit (); + } + break; + + case 'N': + if (sscanf (optarg, "0x%lx", &ctrl->cfg.num_writes) != 1) + if (sscanf (optarg, "%ld", &ctrl->cfg.num_writes) != 1) + { + fprintf (stderr, "ERROR: Invalid value for option -%c!\n", c); + print_usage_and_exit (); + } + ctrl->cfg.total_bytes = ctrl->cfg.num_writes * ctrl->cfg.txbuf_size; + break; + + case 'R': + if (sscanf (optarg, "0x%lx", &ctrl->cfg.rxbuf_size) != 1) + if (sscanf (optarg, "%ld", &ctrl->cfg.rxbuf_size) != 1) + { + fprintf (stderr, "ERROR: Invalid value for option -%c!\n", c); + print_usage_and_exit (); + } + if (ctrl->cfg.rxbuf_size >= SOCK_TEST_CFG_BUF_SIZE_MIN) + { + ctrl->rxbuf_size = ctrl->cfg.rxbuf_size; + sock_test_buf_alloc (&ctrl->cfg, 1 /* is_rxbuf */ , + (uint8_t **) & ctrl->rxbuf, + &ctrl->rxbuf_size); + } + else + { + fprintf (stderr, + "ERROR: rxbuf size (%lu) less than minumum (%u)\n", + ctrl->cfg.rxbuf_size, SOCK_TEST_CFG_BUF_SIZE_MIN); + print_usage_and_exit (); + } + + break; + + case 'T': + if (sscanf (optarg, "0x%lx", &ctrl->cfg.txbuf_size) != 1) + if (sscanf (optarg, "%ld", &ctrl->cfg.txbuf_size) != 1) + { + fprintf (stderr, "ERROR: Invalid value for option -%c!\n", c); + print_usage_and_exit (); + } + if (ctrl->cfg.txbuf_size >= SOCK_TEST_CFG_BUF_SIZE_MIN) + { + ctrl->txbuf_size = ctrl->cfg.txbuf_size; + sock_test_buf_alloc (&ctrl->cfg, 0 /* is_rxbuf */ , + (uint8_t **) & ctrl->txbuf, + &ctrl->txbuf_size); + ctrl->cfg.total_bytes = + ctrl->cfg.num_writes * ctrl->cfg.txbuf_size; + } + else + { + fprintf (stderr, + "ERROR: txbuf size (%lu) less than minumum (%u)!\n", + ctrl->cfg.txbuf_size, SOCK_TEST_CFG_BUF_SIZE_MIN); + print_usage_and_exit (); + } + break; + + case 'U': + ctrl->cfg.test = SOCK_TEST_TYPE_UNI; + break; + + case 'B': + ctrl->cfg.test = SOCK_TEST_TYPE_BI; + break; + + case 'V': + ctrl->cfg.verbose = 1; + break; + + case '?': + switch (optopt) + { + case 'E': + case 'I': + case 'N': + case 'R': + case 'T': + case 'w': + fprintf (stderr, "ERROR: Option -%c requires an argument.\n", + optopt); + break; + + default: + if (isprint (optopt)) + fprintf (stderr, "ERROR: Unknown option `-%c'.\n", optopt); + else + fprintf (stderr, "ERROR: Unknown option character `\\x%x'.\n", + optopt); + } + /* fall thru */ + case 'h': + default: + print_usage_and_exit (); + } + + if (argc < (optind + 2)) + { + fprintf (stderr, "ERROR: Insufficient number of arguments!\n"); + print_usage_and_exit (); + } + +#ifdef VCL_TEST + ctrl->fd = vppcom_app_create ("vcl_test_client"); + if (ctrl->fd < 0) + { + errno = -ctrl->fd; + ctrl->fd = -1; + } + else + { + ctrl->fd = vppcom_session_create (VPPCOM_VRF_DEFAULT, VPPCOM_PROTO_TCP, + 0 /* is_nonblocking */ ); + if (ctrl->fd < 0) + { + errno = -ctrl->fd; + ctrl->fd = -1; + } + } +#else + ctrl->fd = socket (AF_INET, SOCK_STREAM, 0); +#endif + + if (ctrl->fd < 0) + { + errno_val = errno; + perror ("ERROR in main()"); + fprintf (stderr, "ERROR: socket failed (errno = %d)!\n", errno_val); + return ctrl->fd; + } + + memset (&scm->server_addr, 0, sizeof (scm->server_addr)); + + scm->server_addr.sin_family = AF_INET; + inet_pton (AF_INET, argv[optind++], &(scm->server_addr.sin_addr)); + scm->server_addr.sin_port = htons (atoi (argv[optind])); + +#ifdef VCL_TEST + scm->server_endpt.vrf = VPPCOM_VRF_DEFAULT; + scm->server_endpt.is_ip4 = (scm->server_addr.sin_family == AF_INET); + scm->server_endpt.ip = (uint8_t *) & scm->server_addr.sin_addr; + scm->server_endpt.port = (uint16_t) scm->server_addr.sin_port; +#endif + + do + { + printf ("\nCLIENT: Connecting to server...\n"); + +#ifdef VCL_TEST + rv = vppcom_session_connect (ctrl->fd, &scm->server_endpt); +#else + rv = + connect (ctrl->fd, (struct sockaddr *) &scm->server_addr, + sizeof (scm->server_addr)); +#endif + if (rv < 0) + { + errno_val = errno; + perror ("ERROR in main()"); + fprintf (stderr, "ERROR: connect failed (errno = %d)!\n", + errno_val); + } + + sock_test_cfg_sync (ctrl); + printf ("CLIENT (fd %d): Control socket connected.\n", ctrl->fd); + } + while (rv < 0); + + sock_test_connect_test_sockets (ctrl->cfg.num_test_sockets); + + while (ctrl->cfg.test != SOCK_TEST_TYPE_EXIT) + { + if (scm->dump_cfg) + { + sock_test_cfg_dump (&ctrl->cfg, 1 /* is_client */ ); + scm->dump_cfg = 0; + } + + switch (ctrl->cfg.test) + { + case SOCK_TEST_TYPE_ECHO: + echo_test_client (); + break; + + case SOCK_TEST_TYPE_UNI: + case SOCK_TEST_TYPE_BI: + stream_test_client (ctrl->cfg.test); + break; + + case SOCK_TEST_TYPE_EXIT: + continue; + + case SOCK_TEST_TYPE_NONE: + default: + break; + } + switch (post_test) + { + case SOCK_TEST_TYPE_EXIT: + switch (ctrl->cfg.test) + { + case SOCK_TEST_TYPE_EXIT: + case SOCK_TEST_TYPE_UNI: + case SOCK_TEST_TYPE_BI: + case SOCK_TEST_TYPE_ECHO: + ctrl->cfg.test = SOCK_TEST_TYPE_EXIT; + continue; + + case SOCK_TEST_TYPE_NONE: + default: + break; + } + break; + + case SOCK_TEST_TYPE_NONE: + case SOCK_TEST_TYPE_ECHO: + case SOCK_TEST_TYPE_UNI: + case SOCK_TEST_TYPE_BI: + default: + break; + } + + memset (ctrl->txbuf, 0, ctrl->txbuf_size); + memset (ctrl->rxbuf, 0, ctrl->rxbuf_size); + + printf ("\nType some characters and hit <return>\n" + "('" SOCK_TEST_TOKEN_HELP "' for help): "); + + if (fgets (ctrl->txbuf, ctrl->txbuf_size, stdin) != NULL) + { + if (strlen (ctrl->txbuf) == 1) + { + printf ("\nCLIENT: Nothing to send! Please try again...\n"); + continue; + } + ctrl->txbuf[strlen (ctrl->txbuf) - 1] = 0; // chomp the newline. + + /* Parse input for keywords */ + ctrl->cfg.test = parse_input (); + } + } + + exit_client (); +#ifdef VCL_TEST + vppcom_session_close (ctrl->fd); + vppcom_app_destroy (); +#else + close (ctrl->fd); +#endif + return 0; +} + +/* + * fd.io coding-style-patch-verification: ON + * + * Local Variables: + * eval: (c-set-style "gnu") + * End: + */ diff --git a/src/uri/sock_test_server.c b/src/uri/sock_test_server.c new file mode 100644 index 00000000..35046aa0 --- /dev/null +++ b/src/uri/sock_test_server.c @@ -0,0 +1,596 @@ +/* + * Copyright (c) 2017 Cisco and/or its affiliates. + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at: + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +#include <unistd.h> +#include <errno.h> +#include <sys/types.h> +#include <sys/socket.h> +#include <stdio.h> +#include <string.h> +#include <time.h> +#include <ctype.h> +#include <uri/sock_test.h> + +typedef struct +{ + uint8_t is_alloc; + int fd; + uint8_t *buf; + uint32_t buf_size; + sock_test_cfg_t cfg; + sock_test_stats_t stats; +#ifdef VCL_TEST + vppcom_endpt_t endpt; + uint8_t ip[16]; +#endif +} sock_server_conn_t; + +#define SOCK_SERVER_MAX_TEST_CONN 10 +typedef struct +{ + int listen_fd; + size_t num_conn; + size_t conn_pool_size; + sock_server_conn_t *conn_pool; + int nfds; + fd_set rd_fdset; + fd_set wr_fdset; + struct timeval timeout; +} sock_server_main_t; + +sock_server_main_t sock_server_main; + +static inline int +get_nfds (void) +{ + sock_server_main_t *ssm = &sock_server_main; + int i, nfds; + + for (nfds = i = 0; i < FD_SETSIZE; i++) + { + if (FD_ISSET (i, &ssm->rd_fdset) || FD_ISSET (i, &ssm->wr_fdset)) + nfds = i + 1; + } + return nfds; +} + +static inline void +conn_fdset_set (sock_server_conn_t * conn, fd_set * fdset) +{ + sock_server_main_t *ssm = &sock_server_main; + + FD_SET (conn->fd, fdset); + ssm->nfds = get_nfds (); +} + +static inline void +conn_fdset_clr (sock_server_conn_t * conn, fd_set * fdset) +{ + sock_server_main_t *ssm = &sock_server_main; + + FD_CLR (conn->fd, fdset); + ssm->nfds = get_nfds (); +} + +static inline void +conn_pool_expand (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; + int i; + + conn_pool = realloc (ssm->conn_pool, new_size * sizeof (*ssm->conn_pool)); + if (conn_pool) + { + for (i = ssm->conn_pool_size; i < new_size; i++) + { + sock_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 */ , + &conn->buf, &conn->buf_size); + conn->cfg.txbuf_size = conn->cfg.rxbuf_size; + } + + ssm->conn_pool = conn_pool; + ssm->conn_pool_size = new_size; + } + else + { + int errno_val = errno; + perror ("ERROR in conn_pool_expand()"); + fprintf (stderr, "ERROR: Memory allocation failed (errno = %d)!\n", + errno_val); + } +} + +static inline sock_server_conn_t * +conn_pool_alloc (void) +{ + sock_server_main_t *ssm = &sock_server_main; + int i; + + for (i = 0; i < ssm->conn_pool_size; i++) + { + if (!ssm->conn_pool[i].is_alloc) + { +#ifdef VCL_TEST + ssm->conn_pool[i].endpt.ip = ssm->conn_pool[i].ip; +#endif + ssm->conn_pool[i].is_alloc = 1; + return (&ssm->conn_pool[i]); + } + } + + return 0; +} + +static inline void +conn_pool_free (sock_server_conn_t * conn) +{ + sock_server_main_t *ssm = &sock_server_main; + + conn_fdset_clr (conn, &ssm->rd_fdset); + conn_fdset_clr (conn, &ssm->wr_fdset); + 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) +{ + conn->cfg = *rx_cfg; + sock_test_buf_alloc (&conn->cfg, 1 /* is_rxbuf */ , + &conn->buf, &conn->buf_size); + conn->cfg.txbuf_size = conn->cfg.rxbuf_size; + + if (conn->cfg.verbose) + { + printf ("\nSERVER (fd %d): Replying to cfg message!\n", conn->fd); + sock_test_cfg_dump (&conn->cfg, 0 /* is_client */ ); + } + (void) sock_test_write (conn->fd, (uint8_t *) & conn->cfg, + sizeof (conn->cfg), NULL, conn->cfg.verbose); +} + +static void +stream_test_server_start_stop (sock_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; + + if (rx_cfg->ctrl_handle == conn->fd) + { + int i; + clock_gettime (CLOCK_REALTIME, &conn->stats.stop); + + for (i = 0; i < ssm->conn_pool_size; i++) + { + sock_server_conn_t *tc = &ssm->conn_pool[i]; + + if (tc->cfg.ctrl_handle == conn->fd) + { + sock_test_stats_accumulate (&conn->stats, &tc->stats); + + if (conn->cfg.verbose) + { + static char buf[64]; + + sprintf (buf, "SERVER (fd %d) RESULTS", tc->fd); + sock_test_stats_dump (buf, &tc->stats, 1 /* show_rx */ , + test == SOCK_TEST_TYPE_BI + /* show tx */ , + conn->cfg.verbose); + } + } + } + + sock_test_stats_dump ("SERVER RESULTS", &conn->stats, 1 /* show_rx */ , + (test == SOCK_TEST_TYPE_BI) /* show_tx */ , + conn->cfg.verbose); + sock_test_cfg_dump (&conn->cfg, 0 /* is_client */ ); + if (conn->cfg.verbose) + { + printf (" sock server main\n" + SOCK_TEST_SEPARATOR_STRING + " buf: %p\n" + " buf size: %u (0x%08x)\n" + SOCK_TEST_SEPARATOR_STRING, + conn->buf, conn->buf_size, conn->buf_size); + } + + sync_config_and_reply (conn, rx_cfg); + printf ("\nSERVER (fd %d): %s-directional Stream Test Complete!\n" + SOCK_TEST_BANNER_STRING "\n", conn->fd, + test == SOCK_TEST_TYPE_BI ? "Bi" : "Uni"); + } + else + { + printf ("\n" SOCK_TEST_BANNER_STRING + "SERVER (fd %d): %s-directional Stream Test!\n" + " Sending client the test cfg to start streaming data...\n", + client_fd, test == SOCK_TEST_TYPE_BI ? "Bi" : "Uni"); + + rx_cfg->ctrl_handle = (rx_cfg->ctrl_handle == ~0) ? conn->fd : + rx_cfg->ctrl_handle; + + sync_config_and_reply (conn, rx_cfg); + + /* read the 1st chunk, record start time */ + memset (&conn->stats, 0, sizeof (conn->stats)); + clock_gettime (CLOCK_REALTIME, &conn->stats.start); + } +} + + +static inline void +stream_test_server (sock_server_conn_t * conn, int rx_bytes) +{ + int client_fd = conn->fd; + sock_test_t test = conn->cfg.test; + + if (test == SOCK_TEST_TYPE_BI) + (void) sock_test_write (client_fd, conn->buf, rx_bytes, &conn->stats, + conn->cfg.verbose); + + if (conn->stats.rx_bytes >= conn->cfg.total_bytes) + { + clock_gettime (CLOCK_REALTIME, &conn->stats.stop); + } +} + +static inline void +new_client (void) +{ + 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); + + conn = conn_pool_alloc (); + if (!conn) + { + fprintf (stderr, "\nERROR: No free connections!\n"); + return; + } + +#ifdef VCL_TEST + client_fd = vppcom_session_accept (ssm->listen_fd, &conn->endpt, + -1.0 /* wait forever */ ); +#else + client_fd = accept (ssm->listen_fd, (struct sockaddr *) NULL, NULL); +#endif + if (client_fd < 0) + { + int errno_val; + errno_val = errno; + perror ("ERROR in main()"); + fprintf (stderr, "ERROR: accept failed (errno = %d)!\n", errno_val); + } + + printf ("SERVER: Got a connection -- fd = %d (0x%08x)!\n", + client_fd, client_fd); + + conn->fd = client_fd; + conn_fdset_set (conn, &ssm->rd_fdset); +} + +int +main (int argc, char **argv) +{ + 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_in servaddr; + int errno_val; + int v, i; + uint16_t port = SOCK_TEST_SERVER_PORT; + fd_set _rfdset, *rfdset = &_rfdset; +#ifdef VCL_TEST + vppcom_endpt_t endpt; +#else + fd_set _wfdset, *wfdset = &_wfdset; +#endif + + if ((argc == 2) && (sscanf (argv[1], "%d", &v) == 1)) + port = (uint16_t) v; + + conn_pool_expand (SOCK_SERVER_MAX_TEST_CONN + 1); + +#ifdef VCL_TEST + rv = vppcom_app_create ("vcl_test_server"); + if (rv) + { + errno = -rv; + ssm->listen_fd = -1; + } + else + { + ssm->listen_fd = + vppcom_session_create (VPPCOM_VRF_DEFAULT, VPPCOM_PROTO_TCP, + 0 /* is_nonblocking */ ); + } +#else + ssm->listen_fd = socket (AF_INET, SOCK_STREAM, 0); +#endif + if (ssm->listen_fd < 0) + { + errno_val = errno; + perror ("ERROR in main()"); + fprintf (stderr, "ERROR: socket() failed (errno = %d)!\n", errno_val); + return ssm->listen_fd; + } + + memset (&servaddr, 0, sizeof (servaddr)); + + servaddr.sin_family = AF_INET; + servaddr.sin_addr.s_addr = htonl (INADDR_ANY); + servaddr.sin_port = htons (port); + +#ifdef VCL_TEST + endpt.vrf = VPPCOM_VRF_DEFAULT; + endpt.is_ip4 = (servaddr.sin_family == AF_INET); + endpt.ip = (uint8_t *) & servaddr.sin_addr; + endpt.port = (uint16_t) servaddr.sin_port; + + rv = vppcom_session_bind (ssm->listen_fd, &endpt); + if (rv) + { + errno = -rv; + rv = -1; + } +#else + rv = + bind (ssm->listen_fd, (struct sockaddr *) &servaddr, sizeof (servaddr)); +#endif + if (rv < 0) + { + errno_val = errno; + perror ("ERROR in main()"); + fprintf (stderr, "ERROR: bind failed (errno = %d)!\n", errno_val); + return rv; + } + +#ifdef VCL_TEST + rv = vppcom_session_listen (ssm->listen_fd, 10); + if (rv) + { + errno = -rv; + rv = -1; + } +#else + rv = listen (ssm->listen_fd, 10); +#endif + if (rv < 0) + { + errno_val = errno; + perror ("ERROR in main()"); + fprintf (stderr, "ERROR: listen failed (errno = %d)!\n", errno_val); + return rv; + } + + FD_ZERO (&ssm->wr_fdset); + FD_ZERO (&ssm->rd_fdset); + + FD_SET (ssm->listen_fd, &ssm->rd_fdset); + ssm->nfds = ssm->listen_fd + 1; + + printf ("\nSERVER: Waiting for a client to connect on port %d...\n", port); + + while (1) + { + _rfdset = ssm->rd_fdset; + +#ifdef VCL_TEST + rv = vppcom_select (ssm->nfds, (uint64_t *) rfdset, NULL, NULL, 0); +#else + { + struct timeval timeout; + timeout = ssm->timeout; + _wfdset = ssm->wr_fdset; + rv = select (ssm->nfds, rfdset, wfdset, NULL, &timeout); + } +#endif + if (rv < 0) + { + perror ("select()"); + fprintf (stderr, "\nERROR: select() failed -- aborting!\n"); + main_rv = -1; + goto done; + } + else if (rv == 0) + continue; + + if (FD_ISSET (ssm->listen_fd, rfdset)) + new_client (); + + for (i = 0; i < ssm->conn_pool_size; i++) + { + if (!ssm->conn_pool[i].is_alloc) + continue; + + conn = &ssm->conn_pool[i]; + client_fd = conn->fd; + + if (FD_ISSET (client_fd, rfdset)) + { + rx_bytes = sock_test_read (client_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 */ ); + } + sock_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); +#ifdef VCL_TEST + vppcom_session_close (client_fd); +#else + close (client_fd); +#endif + conn_pool_free (conn); + + if (ssm->nfds == (ssm->listen_fd + 1)) + { + printf ("SERVER: All client connections " + "closed.\n\nSERVER: " + "May the force be with you!\n\n"); + goto done; + } + break; + + default: + fprintf (stderr, "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 ("\nSERVER (fd %d): RX (%d bytes) - '%s'\n", + conn->fd, rx_bytes, conn->buf); + } + } + else // rx_bytes < 0 + { + if (errno == ECONNRESET) + { + printf ("\nSERVER: Connection reset by remote peer.\n" + " Y'all have a great day now!\n\n"); + break; + } + else + continue; + } + + 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; + if (xtra) + fprintf (stderr, + "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 = sock_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); + } + + else // Extraneous read data from non-echo tests??? + { + xtra++; + xtra_bytes += rx_bytes; + } + } + } + } + +done: +#ifdef VCL_TEST + vppcom_session_close (ssm->listen_fd); + vppcom_app_destroy (); +#else + close (ssm->listen_fd); +#endif + if (ssm->conn_pool) + free (ssm->conn_pool); + + return main_rv; +} + +/* + * fd.io coding-style-patch-verification: ON + * + * Local Variables: + * eval: (c-set-style "gnu") + * End: + */ diff --git a/src/uri/uri_socket_server.c b/src/uri/uri_socket_server.c new file mode 100644 index 00000000..4f4c5f30 --- /dev/null +++ b/src/uri/uri_socket_server.c @@ -0,0 +1,231 @@ +/* + * Copyright (c) 2017 Cisco and/or its affiliates. + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at: + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +#include <stdio.h> +#include <sys/types.h> +#include <sys/socket.h> +#include <netinet/in.h> +#include <arpa/inet.h> +#include <netdb.h> +#include <vppinfra/format.h> +#include <signal.h> +#include <sys/ucontext.h> +#include <sys/time.h> + +volatile int signal_received; + +static void +unix_signal_handler (int signum, siginfo_t * si, ucontext_t * uc) +{ + signal_received = 1; +} + +static void +setup_signal_handler (void) +{ + uword i; + struct sigaction sa; + + for (i = 1; i < 32; i++) + { + memset (&sa, 0, sizeof (sa)); + sa.sa_sigaction = (void *) unix_signal_handler; + sa.sa_flags = SA_SIGINFO; + + switch (i) + { + /* these signals take the default action */ + case SIGABRT: + case SIGKILL: + case SIGSTOP: + case SIGUSR1: + case SIGUSR2: + continue; + + /* ignore SIGPIPE, SIGCHLD */ + case SIGPIPE: + case SIGCHLD: + sa.sa_sigaction = (void *) SIG_IGN; + break; + + /* catch and handle all other signals */ + default: + break; + } + + if (sigaction (i, &sa, 0) < 0) + clib_unix_warning ("sigaction %U", format_signal, i); + } +} + + +int +main (int argc, char *argv[]) +{ + int sockfd, portno, n, sent, accfd, reuse; + socklen_t client_addr_len; + struct sockaddr_in serv_addr; + struct sockaddr_in client; + struct hostent *server; + u8 *rx_buffer = 0, no_echo = 0; + struct timeval start, end; + long rcvd = 0; + double deltat; + + if (argc > 1 && argc < 3) + { + fformat (stderr, "usage %s host port\n", argv[0]); + exit (0); + } + + if (argc >= 4) + { + no_echo = atoi (argv[3]); + portno = atoi (argv[2]); + server = gethostbyname (argv[1]); + if (server == NULL) + { + clib_unix_warning ("gethostbyname"); + exit (1); + } + } + else + { + /* Defaults */ + portno = 1234; + server = gethostbyname ("6.0.1.1"); + if (server == NULL) + { + clib_unix_warning ("gethostbyname"); + exit (1); + } + } + + + setup_signal_handler (); + + sockfd = socket (AF_INET, SOCK_STREAM, 0); + if (sockfd < 0) + { + clib_unix_error ("socket"); + exit (1); + } + + reuse = 1; + if (setsockopt (sockfd, SOL_SOCKET, SO_REUSEADDR, (const char *) &reuse, + sizeof (reuse)) < 0) + { + clib_unix_error ("setsockopt(SO_REUSEADDR) failed"); + exit (1); + } + + bzero ((char *) &serv_addr, sizeof (serv_addr)); + serv_addr.sin_family = AF_INET; + bcopy ((char *) server->h_addr, + (char *) &serv_addr.sin_addr.s_addr, server->h_length); + serv_addr.sin_port = htons (portno); + if (bind (sockfd, (const void *) &serv_addr, sizeof (serv_addr)) < 0) + { + clib_unix_warning ("bind"); + exit (1); + } + + vec_validate (rx_buffer, 128 << 10); + + if (listen (sockfd, 5 /* backlog */ ) < 0) + { + clib_unix_warning ("listen"); + close (sockfd); + return 1; + } + + while (1) + { + if (signal_received) + break; + + client_addr_len = sizeof (struct sockaddr); + accfd = accept (sockfd, (struct sockaddr *) &client, &client_addr_len); + if (accfd < 0) + { + clib_unix_warning ("accept"); + continue; + } + fformat (stderr, "Accepted connection from: %s : %d\n", + inet_ntoa (client.sin_addr), client.sin_port); + gettimeofday (&start, NULL); + + while (1) + { + n = recv (accfd, rx_buffer, vec_len (rx_buffer), 0 /* flags */ ); + if (n == 0) + { + /* Graceful exit */ + close (accfd); + gettimeofday (&end, NULL); + deltat = (end.tv_sec - start.tv_sec); + deltat += (end.tv_usec - start.tv_usec) / 1000000.0; + clib_warning ("Finished in %.6f", deltat); + clib_warning ("%.4f Gbit/second %s", + (((f64) rcvd * 8.0) / deltat / 1e9), + no_echo ? "half" : "full"); + rcvd = 0; + break; + } + if (n < 0) + { + clib_unix_warning ("recv"); + close (accfd); + break; + } + + if (signal_received) + break; + + rcvd += n; + if (no_echo) + continue; + + sent = send (accfd, rx_buffer, n, 0 /* flags */ ); + if (n < 0) + { + clib_unix_warning ("send"); + close (accfd); + break; + } + + if (sent != n) + { + clib_warning ("sent %d not %d", sent, n); + } + + if (signal_received) + break; + } + } + + close (sockfd); + + return 0; +} + + +/* + * fd.io coding-style-patch-verification: ON + * + * Local Variables: + * eval: (c-set-style "gnu") + * End: + */ diff --git a/src/uri/uri_socket_test.c b/src/uri/uri_socket_test.c new file mode 100644 index 00000000..4469b03d --- /dev/null +++ b/src/uri/uri_socket_test.c @@ -0,0 +1,181 @@ +/* + * Copyright (c) 2017 Cisco and/or its affiliates. + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at: + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +#include <stdio.h> +#include <sys/types.h> +#include <sys/socket.h> +#include <netinet/in.h> +#include <netdb.h> +#include <vppinfra/format.h> +#include <sys/time.h> + +int +main (int argc, char *argv[]) +{ + int sockfd, portno, n; + struct sockaddr_in serv_addr; + struct hostent *server; + u8 *rx_buffer = 0, *tx_buffer = 0, no_echo = 0, test_bytes = 0; + u32 offset; + long bytes = 1 << 20, to_send; + int i; + struct timeval start, end; + double deltat; + + if (argc >= 3) + { + portno = atoi (argv[2]); + server = gethostbyname (argv[1]); + if (server == NULL) + { + clib_unix_warning ("gethostbyname"); + exit (1); + } + + argc -= 3; + argv += 3; + + if (argc) + { + bytes = ((long) atoi (argv[0])) << 20; + argc--; + argv++; + } + if (argc) + { + no_echo = atoi (argv[0]); + argc--; + argv++; + } + if (argc) + { + test_bytes = atoi (argv[0]); + argc--; + argv++; + } + } + else + { + portno = 1234; // atoi(argv[2]); + server = gethostbyname ("6.0.1.1" /* argv[1] */ ); + if (server == NULL) + { + clib_unix_warning ("gethostbyname"); + exit (1); + } + } + + to_send = bytes; + sockfd = socket (AF_INET, SOCK_STREAM, 0); + if (sockfd < 0) + { + clib_unix_error ("socket"); + exit (1); + } + + bzero ((char *) &serv_addr, sizeof (serv_addr)); + serv_addr.sin_family = AF_INET; + bcopy ((char *) server->h_addr, + (char *) &serv_addr.sin_addr.s_addr, server->h_length); + serv_addr.sin_port = htons (portno); + if (connect (sockfd, (const void *) &serv_addr, sizeof (serv_addr)) < 0) + { + clib_unix_warning ("connect"); + exit (1); + } + + vec_validate (rx_buffer, 128 << 10); + vec_validate (tx_buffer, 128 << 10); + + for (i = 0; i < vec_len (tx_buffer); i++) + tx_buffer[i] = (i + 1) % 0xff; + + /* + * Send one packet to warm up the RX pipeline + */ + n = send (sockfd, tx_buffer, vec_len (tx_buffer), 0 /* flags */ ); + if (n != vec_len (tx_buffer)) + { + clib_unix_warning ("write"); + exit (0); + } + + gettimeofday (&start, NULL); + while (bytes > 0) + { + /* + * TX + */ + n = send (sockfd, tx_buffer, vec_len (tx_buffer), 0 /* flags */ ); + if (n != vec_len (tx_buffer)) + { + clib_unix_warning ("write"); + exit (0); + } + bytes -= n; + + if (no_echo) + continue; + + /* + * RX + */ + + offset = 0; + do + { + n = recv (sockfd, rx_buffer + offset, + vec_len (rx_buffer) - offset, 0 /* flags */ ); + if (n < 0) + { + clib_unix_warning ("read"); + exit (0); + } + offset += n; + } + while (offset < vec_len (rx_buffer)); + + if (test_bytes) + { + for (i = 0; i < vec_len (rx_buffer); i++) + { + if (rx_buffer[i] != tx_buffer[i]) + { + clib_warning ("[%d] read 0x%x not 0x%x", rx_buffer[i], + tx_buffer[i]); + exit (1); + } + } + } + } + close (sockfd); + gettimeofday (&end, NULL); + + deltat = (end.tv_sec - start.tv_sec); + deltat += (end.tv_usec - start.tv_usec) / 1000000.0; // us to ms + clib_warning ("Finished in %.6f", deltat); + clib_warning ("%.4f Gbit/second %s", (((f64) to_send * 8.0) / deltat / 1e9), + no_echo ? "half" : "full"); + return 0; +} + + +/* + * fd.io coding-style-patch-verification: ON + * + * Local Variables: + * eval: (c-set-style "gnu") + * End: + */ diff --git a/src/uri/uri_tcp_test.c b/src/uri/uri_tcp_test.c new file mode 100755 index 00000000..cb297b55 --- /dev/null +++ b/src/uri/uri_tcp_test.c @@ -0,0 +1,1274 @@ +/* + * Copyright (c) 2016 Cisco and/or its affiliates. + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at: + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +#include <stdio.h> +#include <signal.h> +#include <svm/svm_fifo_segment.h> +#include <vlibmemory/api.h> +#include <vpp/api/vpe_msg_enum.h> +#include <vnet/session/application_interface.h> + +#define vl_typedefs /* define message structures */ +#include <vpp/api/vpe_all_api_h.h> +#undef vl_typedefs + +/* declare message handlers for each api */ + +#define vl_endianfun /* define message structures */ +#include <vpp/api/vpe_all_api_h.h> +#undef vl_endianfun + +/* instantiate all the print functions we know about */ +#define vl_print(handle, ...) +#define vl_printfun +#include <vpp/api/vpe_all_api_h.h> +#undef vl_printfun + +typedef struct +{ + svm_fifo_t *server_rx_fifo; + svm_fifo_t *server_tx_fifo; + + u64 vpp_session_handle; + u64 bytes_received; + f64 start; +} session_t; + +typedef enum +{ + STATE_START, + STATE_ATTACHED, + STATE_READY, + STATE_DISCONNECTING, + STATE_FAILED +} connection_state_t; + +typedef struct +{ + /* vpe input queue */ + unix_shared_memory_queue_t *vl_input_queue; + + /* API client handle */ + u32 my_client_index; + + /* The URI we're playing with */ + u8 *uri; + + /* Session pool */ + session_t *sessions; + + /* Hash table for disconnect processing */ + uword *session_index_by_vpp_handles; + + /* intermediate rx buffer */ + u8 *rx_buf; + + /* URI for slave's connect */ + u8 *connect_uri; + + u32 connected_session_index; + + int i_am_master; + + /* drop all packets */ + int drop_packets; + + /* Our event queue */ + unix_shared_memory_queue_t *our_event_queue; + + /* $$$ single thread only for the moment */ + unix_shared_memory_queue_t *vpp_event_queue; + + pid_t my_pid; + + /* For deadman timers */ + clib_time_t clib_time; + + /* State of the connection, shared between msg RX thread and main thread */ + volatile connection_state_t state; + + /* Signal variables */ + volatile int time_to_stop; + volatile int time_to_print_stats; + + u32 configured_segment_size; + + /* VNET_API_ERROR_FOO -> "Foo" hash table */ + uword *error_string_by_error_number; + + u8 *connect_test_data; + pthread_t client_rx_thread_handle; + u32 client_bytes_received; + u8 test_return_packets; + u64 bytes_to_send; + + /* convenience */ + svm_fifo_segment_main_t *segment_main; +} uri_tcp_test_main_t; + +uri_tcp_test_main_t uri_tcp_test_main; + +#if CLIB_DEBUG > 0 +#define NITER 10000 +#else +#define NITER 4000000 +#endif + +static u8 * +format_api_error (u8 * s, va_list * args) +{ + uri_tcp_test_main_t *utm = &uri_tcp_test_main; + i32 error = va_arg (*args, u32); + uword *p; + + p = hash_get (utm->error_string_by_error_number, -error); + + if (p) + s = format (s, "%s", p[0]); + else + s = format (s, "%d", error); + return s; +} + +static void +init_error_string_table (uri_tcp_test_main_t * utm) +{ + utm->error_string_by_error_number = hash_create (0, sizeof (uword)); + +#define _(n,v,s) hash_set (utm->error_string_by_error_number, -v, s); + foreach_vnet_api_error; +#undef _ + + hash_set (utm->error_string_by_error_number, 99, "Misc"); +} + +int +wait_for_state_change (uri_tcp_test_main_t * utm, connection_state_t state) +{ +#if CLIB_DEBUG > 0 +#define TIMEOUT 600.0 +#else +#define TIMEOUT 600.0 +#endif + + f64 timeout = clib_time_now (&utm->clib_time) + TIMEOUT; + + while (clib_time_now (&utm->clib_time) < timeout) + { + if (utm->state == state) + return 0; + if (utm->state == STATE_FAILED) + return -1; + if (utm->time_to_stop == 1) + return 0; + } + clib_warning ("timeout waiting for STATE_READY"); + return -1; +} + +void +application_send_attach (uri_tcp_test_main_t * utm) +{ + vl_api_application_attach_t *bmp; + u32 fifo_size = 4 << 20; + bmp = vl_msg_api_alloc (sizeof (*bmp)); + memset (bmp, 0, sizeof (*bmp)); + + bmp->_vl_msg_id = ntohs (VL_API_APPLICATION_ATTACH); + bmp->client_index = utm->my_client_index; + bmp->context = ntohl (0xfeedface); + bmp->options[APP_OPTIONS_FLAGS] = + APP_OPTIONS_FLAGS_USE_FIFO | APP_OPTIONS_FLAGS_ADD_SEGMENT; + bmp->options[APP_OPTIONS_PREALLOC_FIFO_PAIRS] = 16; + bmp->options[SESSION_OPTIONS_RX_FIFO_SIZE] = fifo_size; + bmp->options[SESSION_OPTIONS_TX_FIFO_SIZE] = fifo_size; + bmp->options[SESSION_OPTIONS_ADD_SEGMENT_SIZE] = 128 << 20; + bmp->options[SESSION_OPTIONS_SEGMENT_SIZE] = 256 << 20; + vl_msg_api_send_shmem (utm->vl_input_queue, (u8 *) & bmp); +} + +int +application_attach (uri_tcp_test_main_t * utm) +{ + application_send_attach (utm); + if (wait_for_state_change (utm, STATE_ATTACHED)) + { + clib_warning ("timeout waiting for STATE_ATTACHED"); + return -1; + } + return 0; +} + +void +application_detach (uri_tcp_test_main_t * utm) +{ + vl_api_application_detach_t *bmp; + bmp = vl_msg_api_alloc (sizeof (*bmp)); + memset (bmp, 0, sizeof (*bmp)); + + bmp->_vl_msg_id = ntohs (VL_API_APPLICATION_DETACH); + bmp->client_index = utm->my_client_index; + bmp->context = ntohl (0xfeedface); + vl_msg_api_send_shmem (utm->vl_input_queue, (u8 *) & bmp); +} + +static void +vl_api_application_attach_reply_t_handler (vl_api_application_attach_reply_t * + mp) +{ + uri_tcp_test_main_t *utm = &uri_tcp_test_main; + svm_fifo_segment_create_args_t _a, *a = &_a; + int rv; + + if (mp->retval) + { + clib_warning ("attach failed: %U", format_api_error, + clib_net_to_host_u32 (mp->retval)); + utm->state = STATE_FAILED; + return; + } + + if (mp->segment_name_length == 0) + { + clib_warning ("segment_name_length zero"); + return; + } + + a->segment_name = (char *) mp->segment_name; + a->segment_size = mp->segment_size; + + ASSERT (mp->app_event_queue_address); + + /* Attach to the segment vpp created */ + rv = svm_fifo_segment_attach (a); + if (rv) + { + clib_warning ("svm_fifo_segment_attach ('%s') failed", + mp->segment_name); + return; + } + + utm->our_event_queue = + uword_to_pointer (mp->app_event_queue_address, + unix_shared_memory_queue_t *); + utm->state = STATE_ATTACHED; +} + +static void +vl_api_application_detach_reply_t_handler (vl_api_application_detach_reply_t * + mp) +{ + if (mp->retval) + clib_warning ("detach returned with err: %d", mp->retval); +} + +static void +stop_signal (int signum) +{ + uri_tcp_test_main_t *um = &uri_tcp_test_main; + + um->time_to_stop = 1; +} + +static void +stats_signal (int signum) +{ + uri_tcp_test_main_t *um = &uri_tcp_test_main; + + um->time_to_print_stats = 1; +} + +static clib_error_t * +setup_signal_handlers (void) +{ + signal (SIGINT, stats_signal); + signal (SIGQUIT, stop_signal); + signal (SIGTERM, stop_signal); + + return 0; +} + +void +vlib_cli_output (struct vlib_main_t *vm, char *fmt, ...) +{ + clib_warning ("BUG"); +} + +int +connect_to_vpp (char *name) +{ + uri_tcp_test_main_t *utm = &uri_tcp_test_main; + api_main_t *am = &api_main; + + if (vl_client_connect_to_vlib ("/vpe-api", name, 32) < 0) + return -1; + + utm->vl_input_queue = am->shmem_hdr->vl_input_queue; + utm->my_client_index = am->my_client_index; + + return 0; +} + +static void +vl_api_map_another_segment_t_handler (vl_api_map_another_segment_t * mp) +{ + svm_fifo_segment_create_args_t _a, *a = &_a; + int rv; + + a->segment_name = (char *) mp->segment_name; + a->segment_size = mp->segment_size; + /* Attach to the segment vpp created */ + rv = svm_fifo_segment_attach (a); + if (rv) + { + clib_warning ("svm_fifo_segment_attach ('%s') failed", + mp->segment_name); + return; + } + clib_warning ("Mapped new segment '%s' size %d", mp->segment_name, + mp->segment_size); +} + +static void +session_print_stats (uri_tcp_test_main_t * utm, session_t * session) +{ + f64 deltat; + u64 bytes; + + deltat = clib_time_now (&utm->clib_time) - session->start; + bytes = utm->i_am_master ? session->bytes_received : utm->bytes_to_send; + fformat (stdout, "Finished in %.6f\n", deltat); + fformat (stdout, "%.4f Gbit/second\n", (bytes * 8.0) / deltat / 1e9); +} + +static void +vl_api_disconnect_session_t_handler (vl_api_disconnect_session_t * mp) +{ + uri_tcp_test_main_t *utm = &uri_tcp_test_main; + session_t *session = 0; + vl_api_disconnect_session_reply_t *rmp; + uword *p; + int rv = 0; + + p = hash_get (utm->session_index_by_vpp_handles, mp->handle); + + if (p) + { + session = pool_elt_at_index (utm->sessions, p[0]); + hash_unset (utm->session_index_by_vpp_handles, mp->handle); + pool_put (utm->sessions, session); + } + else + { + clib_warning ("couldn't find session key %llx", mp->handle); + rv = -11; + } + +// utm->time_to_stop = 1; + + rmp = vl_msg_api_alloc (sizeof (*rmp)); + memset (rmp, 0, sizeof (*rmp)); + + rmp->_vl_msg_id = ntohs (VL_API_DISCONNECT_SESSION_REPLY); + rmp->retval = rv; + rmp->handle = mp->handle; + vl_msg_api_send_shmem (utm->vl_input_queue, (u8 *) & rmp); + + if (session) + session_print_stats (utm, session); +} + +static void +vl_api_reset_session_t_handler (vl_api_reset_session_t * mp) +{ + uri_tcp_test_main_t *utm = &uri_tcp_test_main; + vl_api_reset_session_reply_t *rmp; + uword *p; + int rv = 0; + + p = hash_get (utm->session_index_by_vpp_handles, mp->handle); + + if (p) + { + clib_warning ("got reset"); + /* Cleanup later */ + utm->time_to_stop = 1; + } + else + { + clib_warning ("couldn't find session key %llx", mp->handle); + rv = -11; + } + + rmp = vl_msg_api_alloc (sizeof (*rmp)); + memset (rmp, 0, sizeof (*rmp)); + rmp->_vl_msg_id = ntohs (VL_API_RESET_SESSION_REPLY); + rmp->retval = rv; + rmp->handle = mp->handle; + vl_msg_api_send_shmem (utm->vl_input_queue, (u8 *) & rmp); +} + +void +client_handle_fifo_event_rx (uri_tcp_test_main_t * utm, + session_fifo_event_t * e) +{ + svm_fifo_t *rx_fifo; + int n_read, bytes, i; + + rx_fifo = e->fifo; + + bytes = svm_fifo_max_dequeue (rx_fifo); + /* Allow enqueuing of new event */ + svm_fifo_unset_event (rx_fifo); + + /* Read the bytes */ + do + { + n_read = svm_fifo_dequeue_nowait (rx_fifo, + clib_min (vec_len (utm->rx_buf), + bytes), utm->rx_buf); + if (n_read > 0) + { + bytes -= n_read; + if (utm->test_return_packets) + { + for (i = 0; i < n_read; i++) + { + if (utm->rx_buf[i] + != ((utm->client_bytes_received + i) & 0xff)) + { + clib_warning ("error at byte %lld, 0x%x not 0x%x", + utm->client_bytes_received + i, + utm->rx_buf[i], + ((utm->client_bytes_received + + i) & 0xff)); + } + } + } + utm->client_bytes_received += n_read; + } + else + { + if (n_read == -2) + { +// clib_warning ("weird!"); + break; + } + } + + } + while (bytes > 0); +} + +void +client_handle_event_queue (uri_tcp_test_main_t * utm) +{ + session_fifo_event_t _e, *e = &_e;; + + unix_shared_memory_queue_sub (utm->our_event_queue, (u8 *) e, + 0 /* nowait */ ); + switch (e->event_type) + { + case FIFO_EVENT_APP_RX: + client_handle_fifo_event_rx (utm, e); + break; + + case FIFO_EVENT_DISCONNECT: + return; + + default: + clib_warning ("unknown event type %d", e->event_type); + break; + } +} + +static void * +client_rx_thread_fn (void *arg) +{ + session_fifo_event_t _e, *e = &_e; + uri_tcp_test_main_t *utm = &uri_tcp_test_main; + + utm->client_bytes_received = 0; + while (1) + { + unix_shared_memory_queue_sub (utm->our_event_queue, (u8 *) e, + 0 /* nowait */ ); + switch (e->event_type) + { + case FIFO_EVENT_APP_RX: + client_handle_fifo_event_rx (utm, e); + break; + + case FIFO_EVENT_DISCONNECT: + return 0; + default: + clib_warning ("unknown event type %d", e->event_type); + break; + } + + if (PREDICT_FALSE (utm->time_to_stop == 1)) + break; + } + pthread_exit (0); +} + + +static void +vl_api_connect_session_reply_t_handler (vl_api_connect_session_reply_t * mp) +{ + uri_tcp_test_main_t *utm = &uri_tcp_test_main; + session_t *session; + u32 session_index; + svm_fifo_t *rx_fifo, *tx_fifo; + int rv; + + if (mp->retval) + { + clib_warning ("connection failed with code: %U", format_api_error, + clib_net_to_host_u32 (mp->retval)); + utm->state = STATE_FAILED; + return; + } + + utm->vpp_event_queue = + uword_to_pointer (mp->vpp_event_queue_address, + unix_shared_memory_queue_t *); + + /* + * Setup session + */ + + pool_get (utm->sessions, session); + session_index = session - utm->sessions; + + rx_fifo = uword_to_pointer (mp->server_rx_fifo, svm_fifo_t *); + rx_fifo->client_session_index = session_index; + tx_fifo = uword_to_pointer (mp->server_tx_fifo, svm_fifo_t *); + tx_fifo->client_session_index = session_index; + + session->server_rx_fifo = rx_fifo; + session->server_tx_fifo = tx_fifo; + session->vpp_session_handle = mp->handle; + session->start = clib_time_now (&utm->clib_time); + + /* Save handle */ + utm->connected_session_index = session_index; + utm->state = STATE_READY; + + /* Add it to lookup table */ + hash_set (utm->session_index_by_vpp_handles, mp->handle, session_index); + + /* Start RX thread */ + rv = pthread_create (&utm->client_rx_thread_handle, + NULL /*attr */ , client_rx_thread_fn, 0); + if (rv) + { + clib_warning ("pthread_create returned %d", rv); + rv = VNET_API_ERROR_SYSCALL_ERROR_1; + } +} + +static void +send_test_chunk (uri_tcp_test_main_t * utm, svm_fifo_t * tx_fifo, int mypid, + u32 bytes) +{ + u8 *test_data = utm->connect_test_data; + u64 bytes_sent = 0; + int test_buf_offset = 0; + u32 bytes_to_snd; + u32 queue_max_chunk = 128 << 10, actual_write; + session_fifo_event_t evt; + static int serial_number = 0; + int rv; + + bytes_to_snd = (bytes == 0) ? vec_len (test_data) : bytes; + if (bytes_to_snd > vec_len (test_data)) + bytes_to_snd = vec_len (test_data); + + while (bytes_to_snd > 0 && !utm->time_to_stop) + { + actual_write = (bytes_to_snd > queue_max_chunk) ? + queue_max_chunk : bytes_to_snd; + rv = svm_fifo_enqueue_nowait (tx_fifo, actual_write, + test_data + test_buf_offset); + + if (rv > 0) + { + bytes_to_snd -= rv; + test_buf_offset += rv; + bytes_sent += rv; + + if (svm_fifo_set_event (tx_fifo)) + { + /* Fabricate TX event, send to vpp */ + evt.fifo = tx_fifo; + evt.event_type = FIFO_EVENT_APP_TX; + evt.event_id = serial_number++; + + unix_shared_memory_queue_add (utm->vpp_event_queue, + (u8 *) & evt, + 0 /* do wait for mutex */ ); + } + } + } +} + +void +client_send_data (uri_tcp_test_main_t * utm) +{ + u8 *test_data = utm->connect_test_data; + int mypid = getpid (); + session_t *session; + svm_fifo_t *tx_fifo; + u32 n_iterations, leftover; + int i; + + session = pool_elt_at_index (utm->sessions, utm->connected_session_index); + tx_fifo = session->server_tx_fifo; + + ASSERT (vec_len (test_data) > 0); + + vec_validate (utm->rx_buf, vec_len (test_data) - 1); + n_iterations = utm->bytes_to_send / vec_len (test_data); + + for (i = 0; i < n_iterations; i++) + { + send_test_chunk (utm, tx_fifo, mypid, 0); + if (utm->time_to_stop) + break; + } + + leftover = utm->bytes_to_send % vec_len (test_data); + if (leftover) + send_test_chunk (utm, tx_fifo, mypid, leftover); + + if (!utm->drop_packets) + { + f64 timeout = clib_time_now (&utm->clib_time) + 10; + + /* Wait for the outstanding packets */ + while (utm->client_bytes_received < + vec_len (test_data) * n_iterations + leftover) + { + if (clib_time_now (&utm->clib_time) > timeout) + { + clib_warning ("timed out waiting for the missing packets"); + break; + } + } + } + utm->time_to_stop = 1; +} + +void +client_send_connect (uri_tcp_test_main_t * utm) +{ + vl_api_connect_uri_t *cmp; + cmp = vl_msg_api_alloc (sizeof (*cmp)); + memset (cmp, 0, sizeof (*cmp)); + + cmp->_vl_msg_id = ntohs (VL_API_CONNECT_URI); + cmp->client_index = utm->my_client_index; + cmp->context = ntohl (0xfeedface); + memcpy (cmp->uri, utm->connect_uri, vec_len (utm->connect_uri)); + vl_msg_api_send_shmem (utm->vl_input_queue, (u8 *) & cmp); +} + +int +client_connect (uri_tcp_test_main_t * utm) +{ + client_send_connect (utm); + if (wait_for_state_change (utm, STATE_READY)) + { + clib_warning ("Connect failed"); + return -1; + } + return 0; +} + +void +client_send_disconnect (uri_tcp_test_main_t * utm) +{ + session_t *connected_session; + vl_api_disconnect_session_t *dmp; + connected_session = pool_elt_at_index (utm->sessions, + utm->connected_session_index); + dmp = vl_msg_api_alloc (sizeof (*dmp)); + memset (dmp, 0, sizeof (*dmp)); + dmp->_vl_msg_id = ntohs (VL_API_DISCONNECT_SESSION); + dmp->client_index = utm->my_client_index; + dmp->handle = connected_session->vpp_session_handle; + vl_msg_api_send_shmem (utm->vl_input_queue, (u8 *) & dmp); +} + +int +client_disconnect (uri_tcp_test_main_t * utm) +{ + client_send_disconnect (utm); + clib_warning ("Sent disconnect"); + if (wait_for_state_change (utm, STATE_START)) + { + clib_warning ("Disconnect failed"); + return -1; + } + return 0; +} + +static void +client_test (uri_tcp_test_main_t * utm) +{ + int i; + + if (application_attach (utm)) + return; + + if (client_connect (utm)) + { + application_detach (utm); + return; + } + + /* Init test data */ + vec_validate (utm->connect_test_data, 128 * 1024 - 1); + for (i = 0; i < vec_len (utm->connect_test_data); i++) + utm->connect_test_data[i] = i & 0xff; + + /* Start send */ + client_send_data (utm); + + /* Disconnect */ + client_disconnect (utm); + + application_detach (utm); +} + +static void +vl_api_bind_uri_reply_t_handler (vl_api_bind_uri_reply_t * mp) +{ + uri_tcp_test_main_t *utm = &uri_tcp_test_main; + + if (mp->retval) + { + clib_warning ("bind failed: %U", format_api_error, + clib_net_to_host_u32 (mp->retval)); + utm->state = STATE_FAILED; + return; + } + + utm->state = STATE_READY; +} + +static void +vl_api_unbind_uri_reply_t_handler (vl_api_unbind_uri_reply_t * mp) +{ + uri_tcp_test_main_t *utm = &uri_tcp_test_main; + + if (mp->retval != 0) + clib_warning ("returned %d", ntohl (mp->retval)); + + utm->state = STATE_START; +} + +u8 * +format_ip4_address (u8 * s, va_list * args) +{ + u8 *a = va_arg (*args, u8 *); + return format (s, "%d.%d.%d.%d", a[0], a[1], a[2], a[3]); +} + +u8 * +format_ip6_address (u8 * s, va_list * args) +{ + ip6_address_t *a = va_arg (*args, ip6_address_t *); + u32 i, i_max_n_zero, max_n_zeros, i_first_zero, n_zeros, last_double_colon; + + i_max_n_zero = ARRAY_LEN (a->as_u16); + max_n_zeros = 0; + i_first_zero = i_max_n_zero; + n_zeros = 0; + for (i = 0; i < ARRAY_LEN (a->as_u16); i++) + { + u32 is_zero = a->as_u16[i] == 0; + if (is_zero && i_first_zero >= ARRAY_LEN (a->as_u16)) + { + i_first_zero = i; + n_zeros = 0; + } + n_zeros += is_zero; + if ((!is_zero && n_zeros > max_n_zeros) + || (i + 1 >= ARRAY_LEN (a->as_u16) && n_zeros > max_n_zeros)) + { + i_max_n_zero = i_first_zero; + max_n_zeros = n_zeros; + i_first_zero = ARRAY_LEN (a->as_u16); + n_zeros = 0; + } + } + + last_double_colon = 0; + for (i = 0; i < ARRAY_LEN (a->as_u16); i++) + { + if (i == i_max_n_zero && max_n_zeros > 1) + { + s = format (s, "::"); + i += max_n_zeros - 1; + last_double_colon = 1; + } + else + { + s = format (s, "%s%x", + (last_double_colon || i == 0) ? "" : ":", + clib_net_to_host_u16 (a->as_u16[i])); + last_double_colon = 0; + } + } + + return s; +} + +/* Format an IP46 address. */ +u8 * +format_ip46_address (u8 * s, va_list * args) +{ + ip46_address_t *ip46 = va_arg (*args, ip46_address_t *); + ip46_type_t type = va_arg (*args, ip46_type_t); + int is_ip4 = 1; + + switch (type) + { + case IP46_TYPE_ANY: + is_ip4 = ip46_address_is_ip4 (ip46); + break; + case IP46_TYPE_IP4: + is_ip4 = 1; + break; + case IP46_TYPE_IP6: + is_ip4 = 0; + break; + } + + return is_ip4 ? + format (s, "%U", format_ip4_address, &ip46->ip4) : + format (s, "%U", format_ip6_address, &ip46->ip6); +} + +static void +vl_api_accept_session_t_handler (vl_api_accept_session_t * mp) +{ + uri_tcp_test_main_t *utm = &uri_tcp_test_main; + vl_api_accept_session_reply_t *rmp; + svm_fifo_t *rx_fifo, *tx_fifo; + session_t *session; + static f64 start_time; + u32 session_index; + u8 *ip_str; + + if (start_time == 0.0) + start_time = clib_time_now (&utm->clib_time); + + ip_str = format (0, "%U", format_ip46_address, &mp->ip, mp->is_ip4); + clib_warning ("Accepted session from: %s:%d", ip_str, + clib_net_to_host_u16 (mp->port)); + utm->vpp_event_queue = + uword_to_pointer (mp->vpp_event_queue_address, + unix_shared_memory_queue_t *); + + /* Allocate local session and set it up */ + pool_get (utm->sessions, session); + session_index = session - utm->sessions; + + rx_fifo = uword_to_pointer (mp->server_rx_fifo, svm_fifo_t *); + rx_fifo->client_session_index = session_index; + tx_fifo = uword_to_pointer (mp->server_tx_fifo, svm_fifo_t *); + tx_fifo->client_session_index = session_index; + + session->server_rx_fifo = rx_fifo; + session->server_tx_fifo = tx_fifo; + + /* Add it to lookup table */ + hash_set (utm->session_index_by_vpp_handles, mp->handle, session_index); + + utm->state = STATE_READY; + + /* Stats printing */ + if (pool_elts (utm->sessions) && (pool_elts (utm->sessions) % 20000) == 0) + { + f64 now = clib_time_now (&utm->clib_time); + fformat (stdout, "%d active sessions in %.2f seconds, %.2f/sec...\n", + pool_elts (utm->sessions), now - start_time, + (f64) pool_elts (utm->sessions) / (now - start_time)); + } + + /* + * Send accept reply to vpp + */ + rmp = vl_msg_api_alloc (sizeof (*rmp)); + memset (rmp, 0, sizeof (*rmp)); + rmp->_vl_msg_id = ntohs (VL_API_ACCEPT_SESSION_REPLY); + rmp->handle = mp->handle; + vl_msg_api_send_shmem (utm->vl_input_queue, (u8 *) & rmp); + + session->bytes_received = 0; + session->start = clib_time_now (&utm->clib_time); +} + +void +server_handle_fifo_event_rx (uri_tcp_test_main_t * utm, + session_fifo_event_t * e) +{ + svm_fifo_t *rx_fifo, *tx_fifo; + int n_read; + session_fifo_event_t evt; + unix_shared_memory_queue_t *q; + session_t *session; + int rv; + u32 max_dequeue, offset, max_transfer, rx_buf_len; + + rx_buf_len = vec_len (utm->rx_buf); + rx_fifo = e->fifo; + session = &utm->sessions[rx_fifo->client_session_index]; + tx_fifo = session->server_tx_fifo; + + max_dequeue = svm_fifo_max_dequeue (rx_fifo); + /* Allow enqueuing of a new event */ + svm_fifo_unset_event (rx_fifo); + + if (PREDICT_FALSE (max_dequeue == 0)) + { + return; + } + + /* Read the max_dequeue */ + do + { + max_transfer = clib_min (rx_buf_len, max_dequeue); + n_read = svm_fifo_dequeue_nowait (rx_fifo, max_transfer, utm->rx_buf); + if (n_read > 0) + { + max_dequeue -= n_read; + session->bytes_received += n_read; + } + + /* Reflect if a non-drop session */ + if (!utm->drop_packets && n_read > 0) + { + offset = 0; + do + { + rv = svm_fifo_enqueue_nowait (tx_fifo, n_read, + &utm->rx_buf[offset]); + if (rv > 0) + { + n_read -= rv; + offset += rv; + } + } + while ((rv <= 0 || n_read > 0) && !utm->time_to_stop); + + /* If event wasn't set, add one */ + if (svm_fifo_set_event (tx_fifo)) + { + /* Fabricate TX event, send to vpp */ + evt.fifo = tx_fifo; + evt.event_type = FIFO_EVENT_APP_TX; + evt.event_id = e->event_id; + + q = utm->vpp_event_queue; + unix_shared_memory_queue_add (q, (u8 *) & evt, + 1 /* do wait for mutex */ ); + } + } + } + while ((n_read < 0 || max_dequeue > 0) && !utm->time_to_stop); +} + +void +server_handle_event_queue (uri_tcp_test_main_t * utm) +{ + session_fifo_event_t _e, *e = &_e;; + + while (1) + { + unix_shared_memory_queue_sub (utm->our_event_queue, (u8 *) e, + 0 /* nowait */ ); + switch (e->event_type) + { + case FIFO_EVENT_APP_RX: + server_handle_fifo_event_rx (utm, e); + break; + + case FIFO_EVENT_DISCONNECT: + return; + + default: + clib_warning ("unknown event type %d", e->event_type); + break; + } + if (PREDICT_FALSE (utm->time_to_stop == 1)) + break; + if (PREDICT_FALSE (utm->time_to_print_stats == 1)) + { + utm->time_to_print_stats = 0; + fformat (stdout, "%d connections\n", pool_elts (utm->sessions)); + } + } +} + +void +server_send_listen (uri_tcp_test_main_t * utm) +{ + vl_api_bind_uri_t *bmp; + bmp = vl_msg_api_alloc (sizeof (*bmp)); + memset (bmp, 0, sizeof (*bmp)); + + bmp->_vl_msg_id = ntohs (VL_API_BIND_URI); + bmp->client_index = utm->my_client_index; + bmp->context = ntohl (0xfeedface); + memcpy (bmp->uri, utm->uri, vec_len (utm->uri)); + vl_msg_api_send_shmem (utm->vl_input_queue, (u8 *) & bmp); +} + +int +server_listen (uri_tcp_test_main_t * utm) +{ + server_send_listen (utm); + if (wait_for_state_change (utm, STATE_READY)) + { + clib_warning ("timeout waiting for STATE_READY"); + return -1; + } + return 0; +} + +void +server_send_unbind (uri_tcp_test_main_t * utm) +{ + vl_api_unbind_uri_t *ump; + + ump = vl_msg_api_alloc (sizeof (*ump)); + memset (ump, 0, sizeof (*ump)); + + ump->_vl_msg_id = ntohs (VL_API_UNBIND_URI); + ump->client_index = utm->my_client_index; + memcpy (ump->uri, utm->uri, vec_len (utm->uri)); + vl_msg_api_send_shmem (utm->vl_input_queue, (u8 *) & ump); +} + +int +server_unbind (uri_tcp_test_main_t * utm) +{ + server_send_unbind (utm); + if (wait_for_state_change (utm, STATE_START)) + { + clib_warning ("timeout waiting for STATE_START"); + return -1; + } + return 0; +} + +void +server_test (uri_tcp_test_main_t * utm) +{ + if (application_attach (utm)) + return; + + /* Bind to uri */ + if (server_listen (utm)) + return; + + /* Enter handle event loop */ + server_handle_event_queue (utm); + + /* Cleanup */ + server_send_unbind (utm); + + application_detach (utm); + + fformat (stdout, "Test complete...\n"); +} + +static void +vl_api_disconnect_session_reply_t_handler (vl_api_disconnect_session_reply_t * + mp) +{ + uri_tcp_test_main_t *utm = &uri_tcp_test_main; + session_t *session; + + if (mp->retval) + { + clib_warning ("vpp complained about disconnect: %d", + ntohl (mp->retval)); + } + + utm->state = STATE_START; + session = pool_elt_at_index (utm->sessions, utm->connected_session_index); + if (session) + session_print_stats (utm, session); +} + +#define foreach_uri_msg \ +_(BIND_URI_REPLY, bind_uri_reply) \ +_(UNBIND_URI_REPLY, unbind_uri_reply) \ +_(ACCEPT_SESSION, accept_session) \ +_(CONNECT_SESSION_REPLY, connect_session_reply) \ +_(DISCONNECT_SESSION, disconnect_session) \ +_(DISCONNECT_SESSION_REPLY, disconnect_session_reply) \ +_(RESET_SESSION, reset_session) \ +_(APPLICATION_ATTACH_REPLY, application_attach_reply) \ +_(APPLICATION_DETACH_REPLY, application_detach_reply) \ +_(MAP_ANOTHER_SEGMENT, map_another_segment) \ + +void +uri_api_hookup (uri_tcp_test_main_t * utm) +{ +#define _(N,n) \ + vl_msg_api_set_handlers(VL_API_##N, #n, \ + vl_api_##n##_t_handler, \ + vl_noop_handler, \ + vl_api_##n##_t_endian, \ + vl_api_##n##_t_print, \ + sizeof(vl_api_##n##_t), 1); + foreach_uri_msg; +#undef _ +} + +int +main (int argc, char **argv) +{ + uri_tcp_test_main_t *utm = &uri_tcp_test_main; + unformat_input_t _argv, *a = &_argv; + u8 *chroot_prefix; + u8 *heap, *uri = 0; + u8 *bind_uri = (u8 *) "tcp://0.0.0.0/1234"; + u8 *connect_uri = (u8 *) "tcp://6.0.1.2/1234"; + u64 bytes_to_send = 64 << 10, mbytes; + u32 tmp; + mheap_t *h; + session_t *session; + int i; + int i_am_master = 1, drop_packets = 0, test_return_packets = 0; + + clib_mem_init (0, 256 << 20); + + heap = clib_mem_get_per_cpu_heap (); + h = mheap_header (heap); + + /* make the main heap thread-safe */ + h->flags |= MHEAP_FLAG_THREAD_SAFE; + + vec_validate (utm->rx_buf, 128 << 10); + + utm->session_index_by_vpp_handles = hash_create (0, sizeof (uword)); + + utm->my_pid = getpid (); + utm->configured_segment_size = 1 << 20; + + clib_time_init (&utm->clib_time); + init_error_string_table (utm); + svm_fifo_segment_init (0x200000000ULL, 20); + unformat_init_command_line (a, argv); + + while (unformat_check_input (a) != UNFORMAT_END_OF_INPUT) + { + if (unformat (a, "chroot prefix %s", &chroot_prefix)) + { + vl_set_memory_root_path ((char *) chroot_prefix); + } + else if (unformat (a, "uri %s", &uri)) + ; + else if (unformat (a, "segment-size %dM", &tmp)) + utm->configured_segment_size = tmp << 20; + else if (unformat (a, "segment-size %dG", &tmp)) + utm->configured_segment_size = tmp << 30; + else if (unformat (a, "master")) + i_am_master = 1; + else if (unformat (a, "slave")) + i_am_master = 0; + else if (unformat (a, "drop")) + drop_packets = 1; + else if (unformat (a, "test")) + test_return_packets = 1; + else if (unformat (a, "mbytes %lld", &mbytes)) + { + bytes_to_send = mbytes << 20; + } + else if (unformat (a, "gbytes %lld", &mbytes)) + { + bytes_to_send = mbytes << 30; + } + else + { + fformat (stderr, "%s: usage [master|slave]\n"); + exit (1); + } + } + + if (uri) + { + utm->uri = format (0, "%s%c", uri, 0); + utm->connect_uri = format (0, "%s%c", uri, 0); + } + else + { + utm->uri = format (0, "%s%c", bind_uri, 0); + utm->connect_uri = format (0, "%s%c", connect_uri, 0); + } + + utm->i_am_master = i_am_master; + utm->segment_main = &svm_fifo_segment_main; + utm->drop_packets = drop_packets; + utm->test_return_packets = test_return_packets; + utm->bytes_to_send = bytes_to_send; + utm->time_to_stop = 0; + + setup_signal_handlers (); + uri_api_hookup (utm); + + if (connect_to_vpp (i_am_master ? "uri_tcp_server" : "uri_tcp_client") < 0) + { + svm_region_exit (); + fformat (stderr, "Couldn't connect to vpe, exiting...\n"); + exit (1); + } + + if (i_am_master == 0) + { + client_test (utm); + vl_client_disconnect_from_vlib (); + exit (0); + } + + /* $$$$ hack preallocation */ + for (i = 0; i < 200000; i++) + { + pool_get (utm->sessions, session); + memset (session, 0, sizeof (*session)); + } + for (i = 0; i < 200000; i++) + pool_put_index (utm->sessions, i); + + server_test (utm); + + vl_client_disconnect_from_vlib (); + exit (0); +} + +/* + * fd.io coding-style-patch-verification: ON + * + * Local Variables: + * eval: (c-set-style "gnu") + * End: + */ diff --git a/src/uri/uri_udp_test.c b/src/uri/uri_udp_test.c new file mode 100644 index 00000000..f50ee688 --- /dev/null +++ b/src/uri/uri_udp_test.c @@ -0,0 +1,1040 @@ +/* + * Copyright (c) 2016 Cisco and/or its affiliates. + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at: + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +#include <stdio.h> +#include <setjmp.h> +#include <signal.h> +#include <vppinfra/clib.h> +#include <vppinfra/format.h> +#include <vppinfra/error.h> +#include <vppinfra/time.h> +#include <vppinfra/macros.h> +#include <vnet/vnet.h> +#include <vlib/vlib.h> +#include <vlib/unix/unix.h> +#include <vlibapi/api.h> +#include <vlibmemory/api.h> +#include <vpp/api/vpe_msg_enum.h> +#include <svm/svm_fifo_segment.h> +#include <pthread.h> +#include <vnet/session/application_interface.h> + +#define vl_typedefs /* define message structures */ +#include <vpp/api/vpe_all_api_h.h> +#undef vl_typedefs + +/* declare message handlers for each api */ + +#define vl_endianfun /* define message structures */ +#include <vpp/api/vpe_all_api_h.h> +#undef vl_endianfun + +/* instantiate all the print functions we know about */ +#define vl_print(handle, ...) +#define vl_printfun +#include <vpp/api/vpe_all_api_h.h> +#undef vl_printfun + +typedef enum +{ + STATE_START, + STATE_READY, + STATE_FAILED, + STATE_DISCONNECTING, +} connection_state_t; + +typedef struct +{ + svm_fifo_t *server_rx_fifo; + svm_fifo_t *server_tx_fifo; +} session_t; + +typedef struct +{ + /* vpe input queue */ + unix_shared_memory_queue_t *vl_input_queue; + + /* API client handle */ + u32 my_client_index; + + /* The URI we're playing with */ + u8 *uri; + + /* Session pool */ + session_t *sessions; + + /* Hash table for disconnect processing */ + uword *session_index_by_vpp_handles; + + /* fifo segment */ + svm_fifo_segment_private_t *seg; + + /* intermediate rx buffer */ + u8 *rx_buf; + + /* URI for connect */ + u8 *connect_uri; + + int i_am_master; + + /* Our event queue */ + unix_shared_memory_queue_t *our_event_queue; + + /* $$$ single thread only for the moment */ + unix_shared_memory_queue_t *vpp_event_queue; + + /* $$$$ hack: cut-through session index */ + volatile u32 cut_through_session_index; + + /* unique segment name counter */ + u32 unique_segment_index; + + pid_t my_pid; + + /* pthread handle */ + pthread_t cut_through_thread_handle; + + /* For deadman timers */ + clib_time_t clib_time; + + /* State of the connection, shared between msg RX thread and main thread */ + volatile connection_state_t state; + + volatile int time_to_stop; + volatile int time_to_print_stats; + + u32 configured_segment_size; + + /* VNET_API_ERROR_FOO -> "Foo" hash table */ + uword *error_string_by_error_number; + + /* convenience */ + svm_fifo_segment_main_t *segment_main; + +} uri_udp_test_main_t; + +#if CLIB_DEBUG > 0 +#define NITER 10000 +#else +#define NITER 4000000 +#endif + +uri_udp_test_main_t uri_udp_test_main; + +static void +stop_signal (int signum) +{ + uri_udp_test_main_t *um = &uri_udp_test_main; + + um->time_to_stop = 1; +} + +static void +stats_signal (int signum) +{ + uri_udp_test_main_t *um = &uri_udp_test_main; + + um->time_to_print_stats = 1; +} + +static clib_error_t * +setup_signal_handlers (void) +{ + signal (SIGINT, stats_signal); + signal (SIGQUIT, stop_signal); + signal (SIGTERM, stop_signal); + + return 0; +} + +void +application_send_attach (uri_udp_test_main_t * utm) +{ + vl_api_application_attach_t *bmp; + u32 fifo_size = 3 << 20; + bmp = vl_msg_api_alloc (sizeof (*bmp)); + memset (bmp, 0, sizeof (*bmp)); + + bmp->_vl_msg_id = ntohs (VL_API_APPLICATION_ATTACH); + bmp->client_index = utm->my_client_index; + bmp->context = ntohl (0xfeedface); + bmp->options[APP_OPTIONS_FLAGS] = + APP_OPTIONS_FLAGS_USE_FIFO | APP_OPTIONS_FLAGS_ADD_SEGMENT; + bmp->options[APP_OPTIONS_PREALLOC_FIFO_PAIRS] = 16; + bmp->options[SESSION_OPTIONS_RX_FIFO_SIZE] = fifo_size; + bmp->options[SESSION_OPTIONS_TX_FIFO_SIZE] = fifo_size; + bmp->options[SESSION_OPTIONS_ADD_SEGMENT_SIZE] = 128 << 20; + bmp->options[SESSION_OPTIONS_SEGMENT_SIZE] = 256 << 20; + vl_msg_api_send_shmem (utm->vl_input_queue, (u8 *) & bmp); +} + +void +application_detach (uri_udp_test_main_t * utm) +{ + vl_api_application_detach_t *bmp; + bmp = vl_msg_api_alloc (sizeof (*bmp)); + memset (bmp, 0, sizeof (*bmp)); + + bmp->_vl_msg_id = ntohs (VL_API_APPLICATION_DETACH); + bmp->client_index = utm->my_client_index; + bmp->context = ntohl (0xfeedface); + vl_msg_api_send_shmem (utm->vl_input_queue, (u8 *) & bmp); +} + +static void +vl_api_application_attach_reply_t_handler (vl_api_application_attach_reply_t * + mp) +{ + uri_udp_test_main_t *utm = &uri_udp_test_main; + svm_fifo_segment_create_args_t _a, *a = &_a; + int rv; + + if (mp->retval) + { + clib_warning ("attach failed: %d", mp->retval); + utm->state = STATE_FAILED; + return; + } + + if (mp->segment_name_length == 0) + { + clib_warning ("segment_name_length zero"); + return; + } + + a->segment_name = (char *) mp->segment_name; + a->segment_size = mp->segment_size; + + ASSERT (mp->app_event_queue_address); + + /* Attach to the segment vpp created */ + rv = svm_fifo_segment_attach (a); + if (rv) + { + clib_warning ("svm_fifo_segment_attach ('%s') failed", + mp->segment_name); + return; + } + + utm->our_event_queue = + uword_to_pointer (mp->app_event_queue_address, + unix_shared_memory_queue_t *); +} + +static void +vl_api_application_detach_reply_t_handler (vl_api_application_detach_reply_t * + mp) +{ + if (mp->retval) + clib_warning ("detach returned with err: %d", mp->retval); +} + +u8 * +format_api_error (u8 * s, va_list * args) +{ + uri_udp_test_main_t *utm = va_arg (*args, uri_udp_test_main_t *); + i32 error = va_arg (*args, u32); + uword *p; + + p = hash_get (utm->error_string_by_error_number, -error); + + if (p) + s = format (s, "%s", p[0]); + else + s = format (s, "%d", error); + return s; +} + +int +wait_for_state_change (uri_udp_test_main_t * utm, connection_state_t state) +{ +#if CLIB_DEBUG > 0 +#define TIMEOUT 600.0 +#else +#define TIMEOUT 600.0 +#endif + + f64 timeout = clib_time_now (&utm->clib_time) + TIMEOUT; + + while (clib_time_now (&utm->clib_time) < timeout) + { + if (utm->state == state) + return 0; + } + return -1; +} + +u64 server_bytes_received, server_bytes_sent; + +static void * +cut_through_thread_fn (void *arg) +{ + session_t *s; + svm_fifo_t *rx_fifo; + svm_fifo_t *tx_fifo; + u8 *my_copy_buffer = 0; + uri_udp_test_main_t *utm = &uri_udp_test_main; + i32 actual_transfer; + int rv; + u32 buffer_offset; + + while (utm->cut_through_session_index == ~0) + ; + + s = pool_elt_at_index (utm->sessions, utm->cut_through_session_index); + + rx_fifo = s->server_rx_fifo; + tx_fifo = s->server_tx_fifo; + + vec_validate (my_copy_buffer, 64 * 1024 - 1); + + while (true) + { + /* We read from the tx fifo and write to the rx fifo */ + do + { + actual_transfer = svm_fifo_dequeue_nowait (tx_fifo, + vec_len (my_copy_buffer), + my_copy_buffer); + } + while (actual_transfer <= 0); + + server_bytes_received += actual_transfer; + + buffer_offset = 0; + while (actual_transfer > 0) + { + rv = svm_fifo_enqueue_nowait (rx_fifo, actual_transfer, + my_copy_buffer + buffer_offset); + if (rv > 0) + { + actual_transfer -= rv; + buffer_offset += rv; + server_bytes_sent += rv; + } + + } + if (PREDICT_FALSE (utm->time_to_stop)) + break; + } + + pthread_exit (0); +} + +static void +udp_client_connect (uri_udp_test_main_t * utm) +{ + vl_api_connect_uri_t *cmp; + cmp = vl_msg_api_alloc (sizeof (*cmp)); + memset (cmp, 0, sizeof (*cmp)); + + cmp->_vl_msg_id = ntohs (VL_API_CONNECT_URI); + cmp->client_index = utm->my_client_index; + cmp->context = ntohl (0xfeedface); + memcpy (cmp->uri, utm->connect_uri, vec_len (utm->connect_uri)); + vl_msg_api_send_shmem (utm->vl_input_queue, (u8 *) & cmp); +} + +static void +client_send (uri_udp_test_main_t * utm, session_t * session) +{ + int i; + u8 *test_data = 0; + u64 bytes_received = 0, bytes_sent = 0; + i32 bytes_to_read; + int rv; + f64 before, after, delta, bytes_per_second; + svm_fifo_t *rx_fifo, *tx_fifo; + int buffer_offset, bytes_to_send = 0; + + /* + * Prepare test data + */ + vec_validate (test_data, 64 * 1024 - 1); + for (i = 0; i < vec_len (test_data); i++) + test_data[i] = i & 0xff; + + rx_fifo = session->server_rx_fifo; + tx_fifo = session->server_tx_fifo; + + before = clib_time_now (&utm->clib_time); + + vec_validate (utm->rx_buf, vec_len (test_data) - 1); + + for (i = 0; i < NITER; i++) + { + bytes_to_send = vec_len (test_data); + buffer_offset = 0; + while (bytes_to_send > 0) + { + rv = svm_fifo_enqueue_nowait (tx_fifo, bytes_to_send, + test_data + buffer_offset); + + if (rv > 0) + { + bytes_to_send -= rv; + buffer_offset += rv; + bytes_sent += rv; + } + } + + bytes_to_read = svm_fifo_max_dequeue (rx_fifo); + + bytes_to_read = vec_len (utm->rx_buf) > bytes_to_read ? + bytes_to_read : vec_len (utm->rx_buf); + + buffer_offset = 0; + while (bytes_to_read > 0) + { + rv = svm_fifo_dequeue_nowait (rx_fifo, + bytes_to_read, + utm->rx_buf + buffer_offset); + if (rv > 0) + { + bytes_to_read -= rv; + buffer_offset += rv; + bytes_received += rv; + } + } + } + while (bytes_received < bytes_sent) + { + rv = + svm_fifo_dequeue_nowait (rx_fifo, vec_len (utm->rx_buf), utm->rx_buf); + if (rv > 0) + { +#if CLIB_DEBUG > 0 + int j; + for (j = 0; j < rv; j++) + { + if (utm->rx_buf[j] != ((bytes_received + j) & 0xff)) + { + clib_warning ("error at byte %lld, 0x%x not 0x%x", + bytes_received + j, + utm->rx_buf[j], + ((bytes_received + j) & 0xff)); + } + } +#endif + bytes_received += (u64) rv; + } + } + + after = clib_time_now (&utm->clib_time); + delta = after - before; + bytes_per_second = 0.0; + + if (delta > 0.0) + bytes_per_second = (f64) bytes_received / delta; + + fformat (stdout, + "Done: %lld recv bytes in %.2f seconds, %.2f bytes/sec...\n\n", + bytes_received, delta, bytes_per_second); + fformat (stdout, + "Done: %lld sent bytes in %.2f seconds, %.2f bytes/sec...\n\n", + bytes_sent, delta, bytes_per_second); + fformat (stdout, + "client -> server -> client round trip: %.2f Gbit/sec \n\n", + (bytes_per_second * 8.0) / 1e9); +} + +static void +uri_udp_client_test (uri_udp_test_main_t * utm) +{ + session_t *session; + + application_send_attach (utm); + udp_client_connect (utm); + + if (wait_for_state_change (utm, STATE_READY)) + { + clib_warning ("timeout waiting for STATE_READY"); + return; + } + + /* Only works with cut through sessions */ + session = pool_elt_at_index (utm->sessions, utm->cut_through_session_index); + + client_send (utm, session); + application_detach (utm); +} + +static void +vl_api_bind_uri_reply_t_handler (vl_api_bind_uri_reply_t * mp) +{ + uri_udp_test_main_t *utm = &uri_udp_test_main; + + if (mp->retval) + { + clib_warning ("bind failed: %d", mp->retval); + utm->state = STATE_FAILED; + return; + } + + utm->state = STATE_READY; +} + +static void +vl_api_map_another_segment_t_handler (vl_api_map_another_segment_t * mp) +{ + svm_fifo_segment_create_args_t _a, *a = &_a; + int rv; + + a->segment_name = (char *) mp->segment_name; + a->segment_size = mp->segment_size; + /* Attach to the segment vpp created */ + rv = svm_fifo_segment_attach (a); + if (rv) + { + clib_warning ("svm_fifo_segment_attach ('%s') failed", + mp->segment_name); + return; + } + clib_warning ("Mapped new segment '%s' size %d", mp->segment_name, + mp->segment_size); +} + +/** + * Acting as server for redirected connect requests + */ +static void +vl_api_connect_uri_t_handler (vl_api_connect_uri_t * mp) +{ + u32 segment_index; + uri_udp_test_main_t *utm = &uri_udp_test_main; + svm_fifo_segment_main_t *sm = &svm_fifo_segment_main; + svm_fifo_segment_create_args_t _a, *a = &_a; + svm_fifo_segment_private_t *seg; + unix_shared_memory_queue_t *client_q; + vl_api_connect_session_reply_t *rmp; + session_t *session = 0; + int rv = 0; + + /* Create the segment */ + a->segment_name = (char *) format (0, "%d:segment%d%c", utm->my_pid, + utm->unique_segment_index++, 0); + a->segment_size = utm->configured_segment_size; + + rv = svm_fifo_segment_create (a); + if (rv) + { + clib_warning ("sm_fifo_segment_create ('%s') failed", a->segment_name); + rv = VNET_API_ERROR_URI_FIFO_CREATE_FAILED; + goto send_reply; + } + + vec_add2 (utm->seg, seg, 1); + + segment_index = vec_len (sm->segments) - 1; + memcpy (seg, sm->segments + segment_index, sizeof (utm->seg[0])); + + pool_get (utm->sessions, session); + + session->server_rx_fifo = svm_fifo_segment_alloc_fifo + (utm->seg, 128 * 1024, FIFO_SEGMENT_RX_FREELIST); + ASSERT (session->server_rx_fifo); + + session->server_tx_fifo = svm_fifo_segment_alloc_fifo + (utm->seg, 128 * 1024, FIFO_SEGMENT_TX_FREELIST); + ASSERT (session->server_tx_fifo); + + session->server_rx_fifo->master_session_index = session - utm->sessions; + session->server_tx_fifo->master_session_index = session - utm->sessions; + utm->cut_through_session_index = session - utm->sessions; + + rv = pthread_create (&utm->cut_through_thread_handle, + NULL /*attr */ , cut_through_thread_fn, 0); + if (rv) + { + clib_warning ("pthread_create returned %d", rv); + rv = VNET_API_ERROR_SYSCALL_ERROR_1; + } + +send_reply: + rmp = vl_msg_api_alloc (sizeof (*rmp)); + memset (rmp, 0, sizeof (*rmp)); + + rmp->_vl_msg_id = ntohs (VL_API_CONNECT_SESSION_REPLY); + rmp->context = mp->context; + rmp->retval = ntohl (rv); + rmp->segment_name_length = vec_len (a->segment_name); + if (session) + { + rmp->server_rx_fifo = pointer_to_uword (session->server_rx_fifo); + rmp->server_tx_fifo = pointer_to_uword (session->server_tx_fifo); + } + + memcpy (rmp->segment_name, a->segment_name, vec_len (a->segment_name)); + + vec_free (a->segment_name); + + client_q = + uword_to_pointer (mp->client_queue_address, unix_shared_memory_queue_t *); + vl_msg_api_send_shmem (client_q, (u8 *) & rmp); +} + +static void +vl_api_unbind_uri_reply_t_handler (vl_api_unbind_uri_reply_t * mp) +{ + uri_udp_test_main_t *utm = &uri_udp_test_main; + + if (mp->retval != 0) + clib_warning ("returned %d", ntohl (mp->retval)); + + utm->state = STATE_START; +} + +static void +vl_api_accept_session_t_handler (vl_api_accept_session_t * mp) +{ + uri_udp_test_main_t *utm = &uri_udp_test_main; + vl_api_accept_session_reply_t *rmp; + svm_fifo_t *rx_fifo, *tx_fifo; + session_t *session; + static f64 start_time; + + if (start_time == 0.0) + start_time = clib_time_now (&utm->clib_time); + + utm->vpp_event_queue = + uword_to_pointer (mp->vpp_event_queue_address, + unix_shared_memory_queue_t *); + + pool_get (utm->sessions, session); + + rx_fifo = uword_to_pointer (mp->server_rx_fifo, svm_fifo_t *); + rx_fifo->client_session_index = session - utm->sessions; + tx_fifo = uword_to_pointer (mp->server_tx_fifo, svm_fifo_t *); + tx_fifo->client_session_index = session - utm->sessions; + + session->server_rx_fifo = rx_fifo; + session->server_tx_fifo = tx_fifo; + + hash_set (utm->session_index_by_vpp_handles, mp->handle, + session - utm->sessions); + + utm->state = STATE_READY; + + if (pool_elts (utm->sessions) && (pool_elts (utm->sessions) % 20000) == 0) + { + f64 now = clib_time_now (&utm->clib_time); + fformat (stdout, "%d active sessions in %.2f seconds, %.2f/sec...\n", + pool_elts (utm->sessions), now - start_time, + (f64) pool_elts (utm->sessions) / (now - start_time)); + } + + rmp = vl_msg_api_alloc (sizeof (*rmp)); + memset (rmp, 0, sizeof (*rmp)); + rmp->_vl_msg_id = ntohs (VL_API_ACCEPT_SESSION_REPLY); + rmp->handle = mp->handle; + vl_msg_api_send_shmem (utm->vl_input_queue, (u8 *) & rmp); +} + +static void +vl_api_disconnect_session_t_handler (vl_api_disconnect_session_t * mp) +{ + uri_udp_test_main_t *utm = &uri_udp_test_main; + session_t *session; + vl_api_disconnect_session_reply_t *rmp; + uword *p; + int rv = 0; + + p = hash_get (utm->session_index_by_vpp_handles, mp->handle); + + if (p) + { + session = pool_elt_at_index (utm->sessions, p[0]); + hash_unset (utm->session_index_by_vpp_handles, mp->handle); + pool_put (utm->sessions, session); + } + else + { + clib_warning ("couldn't find session key %llx", mp->handle); + rv = -11; + } + + rmp = vl_msg_api_alloc (sizeof (*rmp)); + memset (rmp, 0, sizeof (*rmp)); + rmp->_vl_msg_id = ntohs (VL_API_DISCONNECT_SESSION_REPLY); + rmp->retval = rv; + rmp->handle = mp->handle; + vl_msg_api_send_shmem (utm->vl_input_queue, (u8 *) & rmp); +} + +static void +vl_api_connect_session_reply_t_handler (vl_api_connect_session_reply_t * mp) +{ + uri_udp_test_main_t *utm = &uri_udp_test_main; + + ASSERT (utm->i_am_master == 0); + + /* We've been redirected */ + if (mp->segment_name_length > 0) + { + svm_fifo_segment_main_t *sm = &svm_fifo_segment_main; + svm_fifo_segment_create_args_t _a, *a = &_a; + u32 segment_index; + session_t *session; + svm_fifo_segment_private_t *seg; + int rv; + + memset (a, 0, sizeof (*a)); + a->segment_name = (char *) mp->segment_name; + + sleep (1); + + rv = svm_fifo_segment_attach (a); + if (rv) + { + clib_warning ("sm_fifo_segment_create ('%v') failed", + mp->segment_name); + return; + } + + segment_index = a->new_segment_indices[0]; + vec_add2 (utm->seg, seg, 1); + memcpy (seg, sm->segments + segment_index, sizeof (*seg)); + sleep (1); + + pool_get (utm->sessions, session); + utm->cut_through_session_index = session - utm->sessions; + + session->server_rx_fifo = uword_to_pointer (mp->server_rx_fifo, + svm_fifo_t *); + ASSERT (session->server_rx_fifo); + session->server_tx_fifo = uword_to_pointer (mp->server_tx_fifo, + svm_fifo_t *); + ASSERT (session->server_tx_fifo); + } + + /* security: could unlink /dev/shm/<mp->segment_name> here, maybe */ + + utm->state = STATE_READY; +} + +#define foreach_uri_msg \ +_(BIND_URI_REPLY, bind_uri_reply) \ +_(CONNECT_URI, connect_uri) \ +_(CONNECT_SESSION_REPLY, connect_session_reply) \ +_(UNBIND_URI_REPLY, unbind_uri_reply) \ +_(ACCEPT_SESSION, accept_session) \ +_(DISCONNECT_SESSION, disconnect_session) \ +_(MAP_ANOTHER_SEGMENT, map_another_segment) \ +_(APPLICATION_ATTACH_REPLY, application_attach_reply) \ +_(APPLICATION_DETACH_REPLY, application_detach_reply) \ + +void +uri_api_hookup (uri_udp_test_main_t * utm) +{ +#define _(N,n) \ + vl_msg_api_set_handlers(VL_API_##N, #n, \ + vl_api_##n##_t_handler, \ + vl_noop_handler, \ + vl_api_##n##_t_endian, \ + vl_api_##n##_t_print, \ + sizeof(vl_api_##n##_t), 1); + foreach_uri_msg; +#undef _ + +} + +int +connect_to_vpp (char *name) +{ + uri_udp_test_main_t *utm = &uri_udp_test_main; + api_main_t *am = &api_main; + + if (vl_client_connect_to_vlib ("/vpe-api", name, 32) < 0) + return -1; + + utm->vl_input_queue = am->shmem_hdr->vl_input_queue; + utm->my_client_index = am->my_client_index; + + return 0; +} + +void +vlib_cli_output (struct vlib_main_t *vm, char *fmt, ...) +{ + clib_warning ("BUG"); +} + +static void +init_error_string_table (uri_udp_test_main_t * utm) +{ + utm->error_string_by_error_number = hash_create (0, sizeof (uword)); + +#define _(n,v,s) hash_set (utm->error_string_by_error_number, -v, s); + foreach_vnet_api_error; +#undef _ + + hash_set (utm->error_string_by_error_number, 99, "Misc"); +} + +void +server_handle_fifo_event_rx (uri_udp_test_main_t * utm, + session_fifo_event_t * e) +{ + svm_fifo_t *rx_fifo, *tx_fifo; + int nbytes; + + session_fifo_event_t evt; + unix_shared_memory_queue_t *q; + int rv; + + rx_fifo = e->fifo; + tx_fifo = utm->sessions[rx_fifo->client_session_index].server_tx_fifo; + + do + { + nbytes = svm_fifo_dequeue_nowait (rx_fifo, vec_len (utm->rx_buf), + utm->rx_buf); + } + while (nbytes <= 0); + do + { + rv = svm_fifo_enqueue_nowait (tx_fifo, nbytes, utm->rx_buf); + } + while (rv == -2); + + /* Fabricate TX event, send to vpp */ + evt.fifo = tx_fifo; + evt.event_type = FIFO_EVENT_APP_TX; + evt.event_id = e->event_id; + + if (svm_fifo_set_event (tx_fifo)) + { + q = utm->vpp_event_queue; + unix_shared_memory_queue_add (q, (u8 *) & evt, + 0 /* do wait for mutex */ ); + } +} + +void +server_handle_event_queue (uri_udp_test_main_t * utm) +{ + session_fifo_event_t _e, *e = &_e; + + while (1) + { + unix_shared_memory_queue_sub (utm->our_event_queue, (u8 *) e, + 0 /* nowait */ ); + switch (e->event_type) + { + case FIFO_EVENT_APP_RX: + server_handle_fifo_event_rx (utm, e); + break; + + case FIFO_EVENT_DISCONNECT: + return; + + default: + clib_warning ("unknown event type %d", e->event_type); + break; + } + if (PREDICT_FALSE (utm->time_to_stop == 1)) + break; + if (PREDICT_FALSE (utm->time_to_print_stats == 1)) + { + utm->time_to_print_stats = 0; + fformat (stdout, "%d connections\n", pool_elts (utm->sessions)); + } + } +} + +static void +server_unbind (uri_udp_test_main_t * utm) +{ + vl_api_unbind_uri_t *ump; + + ump = vl_msg_api_alloc (sizeof (*ump)); + memset (ump, 0, sizeof (*ump)); + + ump->_vl_msg_id = ntohs (VL_API_UNBIND_URI); + ump->client_index = utm->my_client_index; + memcpy (ump->uri, utm->uri, vec_len (utm->uri)); + vl_msg_api_send_shmem (utm->vl_input_queue, (u8 *) & ump); +} + +static void +server_listen (uri_udp_test_main_t * utm) +{ + vl_api_bind_uri_t *bmp; + + bmp = vl_msg_api_alloc (sizeof (*bmp)); + memset (bmp, 0, sizeof (*bmp)); + + bmp->_vl_msg_id = ntohs (VL_API_BIND_URI); + bmp->client_index = utm->my_client_index; + bmp->context = ntohl (0xfeedface); + memcpy (bmp->uri, utm->uri, vec_len (utm->uri)); + vl_msg_api_send_shmem (utm->vl_input_queue, (u8 *) & bmp); +} + +void +udp_server_test (uri_udp_test_main_t * utm) +{ + + application_send_attach (utm); + + /* Bind to uri */ + server_listen (utm); + + if (wait_for_state_change (utm, STATE_READY)) + { + clib_warning ("timeout waiting for STATE_READY"); + return; + } + + server_handle_event_queue (utm); + + /* Cleanup */ + server_unbind (utm); + + if (wait_for_state_change (utm, STATE_START)) + { + clib_warning ("timeout waiting for STATE_START"); + return; + } + + application_detach (utm); + + fformat (stdout, "Test complete...\n"); +} + +int +main (int argc, char **argv) +{ + uri_udp_test_main_t *utm = &uri_udp_test_main; + unformat_input_t _argv, *a = &_argv; + u8 *chroot_prefix; + u8 *heap; + u8 *bind_name = (u8 *) "udp://0.0.0.0/1234"; + u32 tmp; + mheap_t *h; + session_t *session; + int i; + int i_am_master = 1; + + clib_mem_init (0, 256 << 20); + + heap = clib_mem_get_per_cpu_heap (); + h = mheap_header (heap); + + /* make the main heap thread-safe */ + h->flags |= MHEAP_FLAG_THREAD_SAFE; + + vec_validate (utm->rx_buf, 8192); + + utm->session_index_by_vpp_handles = hash_create (0, sizeof (uword)); + + utm->my_pid = getpid (); + utm->configured_segment_size = 1 << 20; + + clib_time_init (&utm->clib_time); + init_error_string_table (utm); + svm_fifo_segment_init (0x200000000ULL, 20); + unformat_init_command_line (a, argv); + + while (unformat_check_input (a) != UNFORMAT_END_OF_INPUT) + { + if (unformat (a, "chroot prefix %s", &chroot_prefix)) + { + vl_set_memory_root_path ((char *) chroot_prefix); + } + else if (unformat (a, "uri %s", &bind_name)) + ; + else if (unformat (a, "segment-size %dM", &tmp)) + utm->configured_segment_size = tmp << 20; + else if (unformat (a, "segment-size %dG", &tmp)) + utm->configured_segment_size = tmp << 30; + else if (unformat (a, "master")) + i_am_master = 1; + else if (unformat (a, "slave")) + i_am_master = 0; + else + { + fformat (stderr, "%s: usage [master|slave]\n"); + exit (1); + } + } + + utm->cut_through_session_index = ~0; + utm->uri = format (0, "%s%c", bind_name, 0); + utm->i_am_master = i_am_master; + utm->segment_main = &svm_fifo_segment_main; + + utm->connect_uri = format (0, "udp://6.0.0.1/1234%c", 0); + + setup_signal_handlers (); + + uri_api_hookup (utm); + + if (connect_to_vpp (i_am_master ? "uri_udp_master" : "uri_udp_slave") < 0) + { + svm_region_exit (); + fformat (stderr, "Couldn't connect to vpe, exiting...\n"); + exit (1); + } + + if (i_am_master == 0) + { + uri_udp_client_test (utm); + exit (0); + } + + /* $$$$ hack preallocation */ + for (i = 0; i < 200000; i++) + { + pool_get (utm->sessions, session); + memset (session, 0, sizeof (*session)); + } + for (i = 0; i < 200000; i++) + pool_put_index (utm->sessions, i); + + udp_server_test (utm); + + vl_client_disconnect_from_vlib (); + exit (0); +} + +#undef vl_api_version +#define vl_api_version(n,v) static u32 vpe_api_version = v; +#include <vpp/api/vpe.api.h> +#undef vl_api_version + +void +vl_client_add_api_signatures (vl_api_memclnt_create_t * mp) +{ + /* + * Send the main API signature in slot 0. This bit of code must + * match the checks in ../vpe/api/api.c: vl_msg_api_version_check(). + */ + mp->api_versions[0] = clib_host_to_net_u32 (vpe_api_version); +} + +u32 +vl (void *p) +{ + return vec_len (p); +} + +/* + * fd.io coding-style-patch-verification: ON + * + * Local Variables: + * eval: (c-set-style "gnu") + * End: + */ diff --git a/src/uri/vcl_test_client.c b/src/uri/vcl_test_client.c new file mode 100644 index 00000000..7ab8824f --- /dev/null +++ b/src/uri/vcl_test_client.c @@ -0,0 +1,27 @@ +/* + * Copyright (c) 2017 Cisco and/or its affiliates. + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at: + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +#define VCL_TEST + +#include <uri/vppcom.h> +#include <uri/sock_test_client.c> + +/* + * fd.io coding-style-patch-verification: ON + * + * Local Variables: + * eval: (c-set-style "gnu") + * End: + */ diff --git a/src/uri/vcl_test_server.c b/src/uri/vcl_test_server.c new file mode 100644 index 00000000..660d705e --- /dev/null +++ b/src/uri/vcl_test_server.c @@ -0,0 +1,27 @@ +/* + * Copyright (c) 2017 Cisco and/or its affiliates. + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at: + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +#define VCL_TEST + +#include <uri/vppcom.h> +#include <uri/sock_test_server.c> + +/* + * fd.io coding-style-patch-verification: ON + * + * Local Variables: + * eval: (c-set-style "gnu") + * End: + */ diff --git a/src/uri/vppcom.c b/src/uri/vppcom.c new file mode 100644 index 00000000..c7ae0ea5 --- /dev/null +++ b/src/uri/vppcom.c @@ -0,0 +1,2440 @@ +/* + * Copyright (c) 2017 Cisco and/or its affiliates. + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this + * You may obtain a copy of the License at: + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +#include <stdio.h> +#include <stdlib.h> +#include <signal.h> +#include <svm/svm_fifo_segment.h> +#include <vlibmemory/api.h> +#include <vpp/api/vpe_msg_enum.h> +#include <vnet/session/application_interface.h> +#include <uri/vppcom.h> +#include <vlib/unix/unix.h> +#include <vppinfra/vec_bootstrap.h> + +#define vl_typedefs /* define message structures */ +#include <vpp/api/vpe_all_api_h.h> +#undef vl_typedefs + +/* declare message handlers for each api */ + +#define vl_endianfun /* define message structures */ +#include <vpp/api/vpe_all_api_h.h> +#undef vl_endianfun + +/* instantiate all the print functions we know about */ +#define vl_print(handle, ...) +#define vl_printfun +#include <vpp/api/vpe_all_api_h.h> +#undef vl_printfun + +#if (CLIB_DEBUG > 0) +/* Set VPPCOM_DEBUG 2 for connection debug, 3 for read/write debug output */ +#define VPPCOM_DEBUG 1 +#else +#define VPPCOM_DEBUG 0 +#endif + +/* + * VPPCOM Private definitions and functions. + */ +typedef enum +{ + STATE_APP_START, + STATE_APP_CONN_VPP, + STATE_APP_ENABLED, + STATE_APP_ATTACHED, +} app_state_t; + +typedef enum +{ + STATE_START, + STATE_CONNECT, + STATE_LISTEN, + STATE_ACCEPT, + STATE_DISCONNECT, + STATE_FAILED +} session_state_t; + +typedef struct +{ + volatile session_state_t state; + + svm_fifo_t *server_rx_fifo; + svm_fifo_t *server_tx_fifo; + u32 sm_seg_index; + u64 vpp_session_handle; + unix_shared_memory_queue_t *vpp_event_queue; + + /* Socket configuration state */ + u8 is_server; + u8 is_listen; + u8 is_cut_thru; + u8 is_nonblocking; + u32 vrf; + u8 is_ip4; + u8 ip[16]; + u16 port; + u8 proto; + u64 client_queue_address; + u64 options[16]; +} session_t; + +typedef struct vppcom_cfg_t_ +{ + u64 heapsize; + u64 segment_baseva; + u32 segment_size; + u32 add_segment_size; + u32 preallocated_fifo_pairs; + u32 rx_fifo_size; + u32 tx_fifo_size; + u32 event_queue_size; + u32 listen_queue_size; + f64 app_timeout; + f64 session_timeout; + f64 accept_timeout; +} vppcom_cfg_t; + +typedef struct vppcom_main_t_ +{ + u8 init; + u32 *client_session_index_fifo; + volatile u32 bind_session_index; + u32 tx_event_id; + int main_cpu; + + /* vpe input queue */ + unix_shared_memory_queue_t *vl_input_queue; + + /* API client handle */ + u32 my_client_index; + + /* Session pool */ + clib_spinlock_t sessions_lockp; + session_t *sessions; + + /* Hash table for disconnect processing */ + uword *session_index_by_vpp_handles; + + /* Select bitmaps */ + clib_bitmap_t *rd_bitmap; + clib_bitmap_t *wr_bitmap; + clib_bitmap_t *ex_bitmap; + + /* Our event queue */ + unix_shared_memory_queue_t *app_event_queue; + + /* unique segment name counter */ + u32 unique_segment_index; + + pid_t my_pid; + + /* For deadman timers */ + clib_time_t clib_time; + + /* State of the connection, shared between msg RX thread and main thread */ + volatile app_state_t app_state; + + vppcom_cfg_t cfg; + + /* VNET_API_ERROR_FOO -> "Foo" hash table */ + uword *error_string_by_error_number; +} vppcom_main_t; + +vppcom_main_t vppcom_main = {.my_client_index = ~0 }; + +static const char * +vppcom_app_state_str (app_state_t state) +{ + char *st; + + switch (state) + { + case STATE_APP_START: + st = "STATE_APP_START"; + break; + + case STATE_APP_CONN_VPP: + st = "STATE_APP_CONN_VPP"; + break; + + case STATE_APP_ENABLED: + st = "STATE_APP_ENABLED"; + break; + + case STATE_APP_ATTACHED: + st = "STATE_APP_ATTACHED"; + break; + + default: + st = "UNKNOWN_APP_STATE"; + break; + } + + return st; +} + +static const char * +vppcom_session_state_str (session_state_t state) +{ + char *st; + + switch (state) + { + case STATE_START: + st = "STATE_START"; + break; + + case STATE_CONNECT: + st = "STATE_CONNECT"; + break; + + case STATE_LISTEN: + st = "STATE_LISTEN"; + break; + + case STATE_ACCEPT: + st = "STATE_ACCEPT"; + break; + + case STATE_DISCONNECT: + st = "STATE_DISCONNECT"; + break; + + case STATE_FAILED: + st = "STATE_FAILED"; + break; + + default: + st = "UNKNOWN_STATE"; + break; + } + + return st; +} + +/* + * VPPCOM Utility Functions + */ +static inline int +vppcom_session_at_index (u32 session_index, session_t * volatile *sess) +{ + vppcom_main_t *vcm = &vppcom_main; + + /* Assumes that caller has acquired spinlock: vcm->sessions_lockp */ + if (PREDICT_FALSE ((session_index == ~0) || + pool_is_free_index (vcm->sessions, session_index))) + { + clib_warning ("[%d] invalid session, sid (%d) has been closed!", + vcm->my_pid, session_index); + return VPPCOM_EBADFD; + } + *sess = pool_elt_at_index (vcm->sessions, session_index); + return VPPCOM_OK; +} + +static int +vppcom_connect_to_vpp (char *app_name) +{ + api_main_t *am = &api_main; + vppcom_main_t *vcm = &vppcom_main; + + if (VPPCOM_DEBUG > 0) + printf ("\nConnecting to VPP api..."); + if (vl_client_connect_to_vlib ("/vpe-api", app_name, 32) < 0) + { + clib_warning ("[%d] connect to vpp (%s) failed!", + vcm->my_pid, app_name); + return VPPCOM_ECONNREFUSED; + } + + vcm->vl_input_queue = am->shmem_hdr->vl_input_queue; + vcm->my_client_index = am->my_client_index; + if (VPPCOM_DEBUG > 0) + printf (" connected!\n"); + + vcm->app_state = STATE_APP_CONN_VPP; + return VPPCOM_OK; +} + +static u8 * +format_api_error (u8 * s, va_list * args) +{ + vppcom_main_t *vcm = &vppcom_main; + i32 error = va_arg (*args, u32); + uword *p; + + p = hash_get (vcm->error_string_by_error_number, -error); + + if (p) + s = format (s, "%s (%d)", p[0], error); + else + s = format (s, "%d", error); + return s; +} + +static void +vppcom_init_error_string_table (void) +{ + vppcom_main_t *vcm = &vppcom_main; + + vcm->error_string_by_error_number = hash_create (0, sizeof (uword)); + +#define _(n,v,s) hash_set (vcm->error_string_by_error_number, -v, s); + foreach_vnet_api_error; +#undef _ + + hash_set (vcm->error_string_by_error_number, 99, "Misc"); +} + +static inline int +vppcom_wait_for_app_state_change (app_state_t app_state) +{ + vppcom_main_t *vcm = &vppcom_main; + f64 timeout = clib_time_now (&vcm->clib_time) + vcm->cfg.app_timeout; + + while (clib_time_now (&vcm->clib_time) < timeout) + { + if (vcm->app_state == app_state) + return VPPCOM_OK; + } + if (VPPCOM_DEBUG > 0) + clib_warning ("[%d] timeout waiting for state %s (%d)", vcm->my_pid, + vppcom_app_state_str (app_state), app_state); + return VPPCOM_ETIMEDOUT; +} + +static inline int +vppcom_wait_for_session_state_change (u32 session_index, + session_state_t state, + f64 wait_for_time) +{ + vppcom_main_t *vcm = &vppcom_main; + f64 timeout = clib_time_now (&vcm->clib_time) + wait_for_time; + session_t *volatile session; + int rv; + + do + { + clib_spinlock_lock (&vcm->sessions_lockp); + rv = vppcom_session_at_index (session_index, &session); + if (PREDICT_FALSE (rv)) + { + clib_spinlock_unlock (&vcm->sessions_lockp); + return rv; + } + if (session->state == state) + { + clib_spinlock_unlock (&vcm->sessions_lockp); + return VPPCOM_OK; + } + clib_spinlock_unlock (&vcm->sessions_lockp); + } + while (clib_time_now (&vcm->clib_time) < timeout); + + if (VPPCOM_DEBUG > 0) + clib_warning ("[%d] timeout waiting for state %s (%d)", vcm->my_pid, + vppcom_session_state_str (state), state); + return VPPCOM_ETIMEDOUT; +} + +static inline int +vppcom_wait_for_client_session_index (f64 wait_for_time) +{ + vppcom_main_t *vcm = &vppcom_main; + f64 timeout = clib_time_now (&vcm->clib_time) + wait_for_time; + + do + { + if (clib_fifo_elts (vcm->client_session_index_fifo)) + return VPPCOM_OK; + } + while (clib_time_now (&vcm->clib_time) < timeout); + + if (wait_for_time == 0) + return VPPCOM_EAGAIN; + + if (VPPCOM_DEBUG > 0) + clib_warning ("[%d] timeout waiting for client_session_index", + vcm->my_pid); + return VPPCOM_ETIMEDOUT; +} + +/* + * VPP-API message functions + */ +static void +vppcom_send_session_enable_disable (u8 is_enable) +{ + vppcom_main_t *vcm = &vppcom_main; + vl_api_session_enable_disable_t *bmp; + bmp = vl_msg_api_alloc (sizeof (*bmp)); + memset (bmp, 0, sizeof (*bmp)); + + bmp->_vl_msg_id = ntohs (VL_API_SESSION_ENABLE_DISABLE); + bmp->client_index = vcm->my_client_index; + bmp->context = htonl (0xfeedface); + bmp->is_enable = is_enable; + vl_msg_api_send_shmem (vcm->vl_input_queue, (u8 *) & bmp); +} + +static int +vppcom_app_session_enable (void) +{ + vppcom_main_t *vcm = &vppcom_main; + int rv; + + if (vcm->app_state != STATE_APP_ENABLED) + { + vppcom_send_session_enable_disable (1 /* is_enabled == TRUE */ ); + rv = vppcom_wait_for_app_state_change (STATE_APP_ENABLED); + if (PREDICT_FALSE (rv)) + { + if (VPPCOM_DEBUG > 0) + clib_warning ("[%d] Session enable timed out, rv = %s (%d)", + vcm->my_pid, vppcom_retval_str (rv), rv); + return rv; + } + } + return VPPCOM_OK; +} + +static void + vl_api_session_enable_disable_reply_t_handler + (vl_api_session_enable_disable_reply_t * mp) +{ + vppcom_main_t *vcm = &vppcom_main; + + if (mp->retval) + { + clib_warning ("[%d] session_enable_disable failed: %U", vcm->my_pid, + format_api_error, ntohl (mp->retval)); + } + else + vcm->app_state = STATE_APP_ENABLED; +} + +static void +vppcom_app_send_attach (void) +{ + vppcom_main_t *vcm = &vppcom_main; + vl_api_application_attach_t *bmp; + bmp = vl_msg_api_alloc (sizeof (*bmp)); + memset (bmp, 0, sizeof (*bmp)); + + bmp->_vl_msg_id = ntohs (VL_API_APPLICATION_ATTACH); + bmp->client_index = vcm->my_client_index; + bmp->context = htonl (0xfeedface); + bmp->options[APP_OPTIONS_FLAGS] = + APP_OPTIONS_FLAGS_USE_FIFO | APP_OPTIONS_FLAGS_ADD_SEGMENT; + bmp->options[SESSION_OPTIONS_SEGMENT_SIZE] = vcm->cfg.segment_size; + bmp->options[SESSION_OPTIONS_ADD_SEGMENT_SIZE] = vcm->cfg.add_segment_size; + bmp->options[SESSION_OPTIONS_RX_FIFO_SIZE] = vcm->cfg.rx_fifo_size; + bmp->options[SESSION_OPTIONS_TX_FIFO_SIZE] = vcm->cfg.tx_fifo_size; + vl_msg_api_send_shmem (vcm->vl_input_queue, (u8 *) & bmp); +} + +static int +vppcom_app_attach (void) +{ + vppcom_main_t *vcm = &vppcom_main; + int rv; + + vppcom_app_send_attach (); + rv = vppcom_wait_for_app_state_change (STATE_APP_ATTACHED); + if (PREDICT_FALSE (rv)) + { + if (VPPCOM_DEBUG > 0) + clib_warning ("[%d] application attach timed out, rv = %s (%d)", + vcm->my_pid, vppcom_retval_str (rv), rv); + return rv; + } + return VPPCOM_OK; +} + +static void +vppcom_app_detach (void) +{ + vppcom_main_t *vcm = &vppcom_main; + vl_api_application_detach_t *bmp; + bmp = vl_msg_api_alloc (sizeof (*bmp)); + memset (bmp, 0, sizeof (*bmp)); + + bmp->_vl_msg_id = ntohs (VL_API_APPLICATION_DETACH); + bmp->client_index = vcm->my_client_index; + bmp->context = htonl (0xfeedface); + vl_msg_api_send_shmem (vcm->vl_input_queue, (u8 *) & bmp); +} + +static void +vl_api_application_attach_reply_t_handler (vl_api_application_attach_reply_t * + mp) +{ + vppcom_main_t *vcm = &vppcom_main; + static svm_fifo_segment_create_args_t _a; + svm_fifo_segment_create_args_t *a = &_a; + int rv; + + memset (a, 0, sizeof (*a)); + if (mp->retval) + { + clib_warning ("[%d] attach failed: %U", vcm->my_pid, + format_api_error, ntohl (mp->retval)); + return; + } + + if (mp->segment_name_length == 0) + { + clib_warning ("[%d] segment_name_length zero", vcm->my_pid); + return; + } + + a->segment_name = (char *) mp->segment_name; + a->segment_size = mp->segment_size; + + ASSERT (mp->app_event_queue_address); + + /* Attach to the segment vpp created */ + rv = svm_fifo_segment_attach (a); + vec_reset_length (a->new_segment_indices); + if (PREDICT_FALSE (rv)) + { + clib_warning ("[%d] svm_fifo_segment_attach ('%s') failed", vcm->my_pid, + mp->segment_name); + return; + } + + vcm->app_event_queue = + uword_to_pointer (mp->app_event_queue_address, + unix_shared_memory_queue_t *); + + vcm->app_state = STATE_APP_ATTACHED; +} + +static void +vl_api_application_detach_reply_t_handler (vl_api_application_detach_reply_t * + mp) +{ + vppcom_main_t *vcm = &vppcom_main; + + if (mp->retval) + clib_warning ("[%d] detach failed: %U", vcm->my_pid, format_api_error, + ntohl (mp->retval)); + + vcm->app_state = STATE_APP_ENABLED; +} + +static void +vl_api_disconnect_session_reply_t_handler (vl_api_disconnect_session_reply_t * + mp) +{ + vppcom_main_t *vcm = &vppcom_main; + uword *p; + + p = hash_get (vcm->session_index_by_vpp_handles, mp->handle); + if (p) + { + session_t *session = 0; + int rv; + clib_spinlock_lock (&vcm->sessions_lockp); + rv = vppcom_session_at_index (p[0], &session); + if (PREDICT_FALSE (rv)) + { + if (VPPCOM_DEBUG > 1) + clib_warning ("[%d] invalid session, sid (%d) has been closed!", + vcm->my_pid, p[0]); + } + hash_unset (vcm->session_index_by_vpp_handles, mp->handle); + session->state = STATE_DISCONNECT; + clib_spinlock_unlock (&vcm->sessions_lockp); + } + else + { + if (VPPCOM_DEBUG > 1) + clib_warning ("[%d] couldn't find session key %llx", vcm->my_pid, + mp->handle); + } + + if (mp->retval) + clib_warning ("[%d] disconnect_session failed: %U", vcm->my_pid, + format_api_error, ntohl (mp->retval)); +} + +static void +vl_api_map_another_segment_t_handler (vl_api_map_another_segment_t * mp) +{ + vppcom_main_t *vcm = &vppcom_main; + static svm_fifo_segment_create_args_t _a; + svm_fifo_segment_create_args_t *a = &_a; + int rv; + + memset (a, 0, sizeof (*a)); + a->segment_name = (char *) mp->segment_name; + a->segment_size = mp->segment_size; + /* Attach to the segment vpp created */ + rv = svm_fifo_segment_attach (a); + vec_reset_length (a->new_segment_indices); + if (PREDICT_FALSE (rv)) + { + clib_warning ("[%d] svm_fifo_segment_attach ('%s') failed", + vcm->my_pid, mp->segment_name); + return; + } + if (VPPCOM_DEBUG > 1) + clib_warning ("[%d] mapped new segment '%s' size %d", vcm->my_pid, + mp->segment_name, mp->segment_size); +} + +static void +vl_api_disconnect_session_t_handler (vl_api_disconnect_session_t * mp) +{ + vppcom_main_t *vcm = &vppcom_main; + session_t *session = 0; + vl_api_disconnect_session_reply_t *rmp; + uword *p; + int rv = 0; + + p = hash_get (vcm->session_index_by_vpp_handles, mp->handle); + if (p) + { + int rval; + clib_spinlock_lock (&vcm->sessions_lockp); + rval = vppcom_session_at_index (p[0], &session); + if (PREDICT_FALSE (rval)) + { + if (VPPCOM_DEBUG > 1) + clib_warning ("[%d] invalid session, sid (%d) has been closed!", + vcm->my_pid, p[0]); + } + else + pool_put (vcm->sessions, session); + clib_spinlock_unlock (&vcm->sessions_lockp); + hash_unset (vcm->session_index_by_vpp_handles, mp->handle); + } + else + { + clib_warning ("[%d] couldn't find session key %llx", vcm->my_pid, + mp->handle); + rv = -11; + } + + rmp = vl_msg_api_alloc (sizeof (*rmp)); + memset (rmp, 0, sizeof (*rmp)); + + rmp->_vl_msg_id = ntohs (VL_API_DISCONNECT_SESSION_REPLY); + rmp->retval = htonl (rv); + rmp->handle = mp->handle; + vl_msg_api_send_shmem (vcm->vl_input_queue, (u8 *) & rmp); +} + +static void +vl_api_reset_session_t_handler (vl_api_reset_session_t * mp) +{ + vppcom_main_t *vcm = &vppcom_main; + session_t *session = 0; + vl_api_reset_session_reply_t *rmp; + uword *p; + int rv = 0; + + p = hash_get (vcm->session_index_by_vpp_handles, mp->handle); + if (p) + { + int rval; + clib_spinlock_lock (&vcm->sessions_lockp); + rval = vppcom_session_at_index (p[0], &session); + if (PREDICT_FALSE (rval)) + { + if (VPPCOM_DEBUG > 1) + clib_warning ("[%d] invalid session, sid (%d) has been closed!", + vcm->my_pid, p[0]); + } + else + pool_put (vcm->sessions, session); + clib_spinlock_unlock (&vcm->sessions_lockp); + hash_unset (vcm->session_index_by_vpp_handles, mp->handle); + } + else + { + clib_warning ("[%d] couldn't find session key %llx", vcm->my_pid, + mp->handle); + rv = -11; + } + + rmp = vl_msg_api_alloc (sizeof (*rmp)); + memset (rmp, 0, sizeof (*rmp)); + rmp->_vl_msg_id = ntohs (VL_API_RESET_SESSION_REPLY); + rmp->retval = htonl (rv); + rmp->handle = mp->handle; + vl_msg_api_send_shmem (vcm->vl_input_queue, (u8 *) & rmp); +} + +static void +vl_api_connect_session_reply_t_handler (vl_api_connect_session_reply_t * mp) +{ + vppcom_main_t *vcm = &vppcom_main; + session_t *session; + u32 session_index; + svm_fifo_t *rx_fifo, *tx_fifo; + u8 is_cut_thru = 0; + int rv; + + if (mp->retval) + { + clib_warning ("[%d] connect failed: %U", vcm->my_pid, format_api_error, + ntohl (mp->retval)); + return; + } + + session_index = mp->context; + if (VPPCOM_DEBUG > 1) + clib_warning ("[%d] session_index = %d 0x%08x", vcm->my_pid, + session_index, session_index); + + clib_spinlock_lock (&vcm->sessions_lockp); + if (pool_is_free_index (vcm->sessions, session_index)) + { + clib_spinlock_unlock (&vcm->sessions_lockp); + if (VPPCOM_DEBUG > 1) + clib_warning ("[%d] invalid session, sid %d is closed!", + vcm->my_pid, session_index); + return; + } + + /* We've been redirected */ + if (mp->segment_name_length > 0) + { + static svm_fifo_segment_create_args_t _a; + svm_fifo_segment_create_args_t *a = &_a; + + is_cut_thru = 1; + memset (a, 0, sizeof (*a)); + a->segment_name = (char *) mp->segment_name; + if (VPPCOM_DEBUG > 1) + clib_warning ("[%d] cut-thru segment: %s", vcm->my_pid, + a->segment_name); + rv = svm_fifo_segment_attach (a); + vec_reset_length (a->new_segment_indices); + if (PREDICT_FALSE (rv)) + { + clib_warning ("[%d] sm_fifo_segment_attach ('%s') failed", + vcm->my_pid, a->segment_name); + return; + } + } + + /* + * Setup session + */ + if (VPPCOM_DEBUG > 1) + clib_warning ("[%d] client sid %d", vcm->my_pid, session_index); + + session = pool_elt_at_index (vcm->sessions, session_index); + session->is_cut_thru = is_cut_thru; + session->vpp_event_queue = uword_to_pointer (mp->vpp_event_queue_address, + unix_shared_memory_queue_t *); + + rx_fifo = uword_to_pointer (mp->server_rx_fifo, svm_fifo_t *); + rx_fifo->client_session_index = session_index; + tx_fifo = uword_to_pointer (mp->server_tx_fifo, svm_fifo_t *); + tx_fifo->client_session_index = session_index; + + session->server_rx_fifo = rx_fifo; + session->server_tx_fifo = tx_fifo; + session->vpp_session_handle = mp->handle; + session->state = STATE_CONNECT; + + /* Add it to lookup table */ + hash_set (vcm->session_index_by_vpp_handles, mp->handle, session_index); + clib_spinlock_unlock (&vcm->sessions_lockp); +} + +static void +vppcom_send_connect_sock (session_t * session, u32 session_index) +{ + vppcom_main_t *vcm = &vppcom_main; + vl_api_connect_sock_t *cmp; + + /* Assumes caller as acquired the spinlock: vcm->sessions_lockp */ + session->is_server = 0; + cmp = vl_msg_api_alloc (sizeof (*cmp)); + memset (cmp, 0, sizeof (*cmp)); + cmp->_vl_msg_id = ntohs (VL_API_CONNECT_SOCK); + cmp->client_index = vcm->my_client_index; + cmp->context = session_index; + + if (VPPCOM_DEBUG > 1) + clib_warning ("[%d] session_index = %d 0x%08x", + vcm->my_pid, session_index, session_index); + + cmp->vrf = session->vrf; + cmp->is_ip4 = session->is_ip4; + clib_memcpy (cmp->ip, session->ip, sizeof (cmp->ip)); + cmp->port = session->port; + cmp->proto = session->proto; + clib_memcpy (cmp->options, session->options, sizeof (cmp->options)); + vl_msg_api_send_shmem (vcm->vl_input_queue, (u8 *) & cmp); +} + +static int +vppcom_send_disconnect (u32 session_index) +{ + vppcom_main_t *vcm = &vppcom_main; + vl_api_disconnect_session_t *dmp; + session_t *session = 0; + int rv; + + clib_spinlock_lock (&vcm->sessions_lockp); + rv = vppcom_session_at_index (session_index, &session); + if (PREDICT_FALSE (rv)) + { + clib_spinlock_unlock (&vcm->sessions_lockp); + if (VPPCOM_DEBUG > 1) + clib_warning ("[%d] invalid session, sid (%d) has been closed!", + vcm->my_pid, session_index); + return rv; + } + + dmp = vl_msg_api_alloc (sizeof (*dmp)); + memset (dmp, 0, sizeof (*dmp)); + dmp->_vl_msg_id = ntohs (VL_API_DISCONNECT_SESSION); + dmp->client_index = vcm->my_client_index; + dmp->handle = session->vpp_session_handle; + clib_spinlock_unlock (&vcm->sessions_lockp); + vl_msg_api_send_shmem (vcm->vl_input_queue, (u8 *) & dmp); + return VPPCOM_OK; +} + +static void +vl_api_bind_sock_reply_t_handler (vl_api_bind_sock_reply_t * mp) +{ + vppcom_main_t *vcm = &vppcom_main; + session_t *session = 0; + int rv; + + if (mp->retval) + clib_warning ("[%d] bind failed: %U", vcm->my_pid, format_api_error, + ntohl (mp->retval)); + + ASSERT (vcm->bind_session_index != ~0); + + clib_spinlock_lock (&vcm->sessions_lockp); + rv = vppcom_session_at_index (vcm->bind_session_index, &session); + if (rv == VPPCOM_OK) + { + session->vpp_session_handle = mp->handle; + hash_set (vcm->session_index_by_vpp_handles, mp->handle, + vcm->bind_session_index); + session->state = mp->retval ? STATE_FAILED : STATE_LISTEN; + vcm->bind_session_index = ~0; + } + clib_spinlock_unlock (&vcm->sessions_lockp); +} + +static void +vl_api_unbind_sock_reply_t_handler (vl_api_unbind_sock_reply_t * mp) +{ + vppcom_main_t *vcm = &vppcom_main; + session_t *session = 0; + int rv; + + clib_spinlock_lock (&vcm->sessions_lockp); + rv = vppcom_session_at_index (vcm->bind_session_index, &session); + if (rv == VPPCOM_OK) + { + if ((VPPCOM_DEBUG > 1) && (mp->retval)) + clib_warning ("[%d] unbind failed: %U", vcm->my_pid, format_api_error, + ntohl (mp->retval)); + + vcm->bind_session_index = ~0; + session->state = STATE_START; + } + clib_spinlock_unlock (&vcm->sessions_lockp); +} + +u8 * +format_ip4_address (u8 * s, va_list * args) +{ + u8 *a = va_arg (*args, u8 *); + return format (s, "%d.%d.%d.%d", a[0], a[1], a[2], a[3]); +} + +u8 * +format_ip6_address (u8 * s, va_list * args) +{ + ip6_address_t *a = va_arg (*args, ip6_address_t *); + u32 i, i_max_n_zero, max_n_zeros, i_first_zero, n_zeros, last_double_colon; + + i_max_n_zero = ARRAY_LEN (a->as_u16); + max_n_zeros = 0; + i_first_zero = i_max_n_zero; + n_zeros = 0; + for (i = 0; i < ARRAY_LEN (a->as_u16); i++) + { + u32 is_zero = a->as_u16[i] == 0; + if (is_zero && i_first_zero >= ARRAY_LEN (a->as_u16)) + { + i_first_zero = i; + n_zeros = 0; + } + n_zeros += is_zero; + if ((!is_zero && n_zeros > max_n_zeros) + || (i + 1 >= ARRAY_LEN (a->as_u16) && n_zeros > max_n_zeros)) + { + i_max_n_zero = i_first_zero; + max_n_zeros = n_zeros; + i_first_zero = ARRAY_LEN (a->as_u16); + n_zeros = 0; + } + } + + last_double_colon = 0; + for (i = 0; i < ARRAY_LEN (a->as_u16); i++) + { + if (i == i_max_n_zero && max_n_zeros > 1) + { + s = format (s, "::"); + i += max_n_zeros - 1; + last_double_colon = 1; + } + else + { + s = format (s, "%s%x", + (last_double_colon || i == 0) ? "" : ":", + clib_net_to_host_u16 (a->as_u16[i])); + last_double_colon = 0; + } + } + + return s; +} + +/* Format an IP46 address. */ +u8 * +format_ip46_address (u8 * s, va_list * args) +{ + ip46_address_t *ip46 = va_arg (*args, ip46_address_t *); + ip46_type_t type = va_arg (*args, ip46_type_t); + int is_ip4 = 1; + + switch (type) + { + case IP46_TYPE_ANY: + is_ip4 = ip46_address_is_ip4 (ip46); + break; + case IP46_TYPE_IP4: + is_ip4 = 1; + break; + case IP46_TYPE_IP6: + is_ip4 = 0; + break; + } + + return is_ip4 ? + format (s, "%U", format_ip4_address, &ip46->ip4) : + format (s, "%U", format_ip6_address, &ip46->ip6); +} + +static void +vl_api_accept_session_t_handler (vl_api_accept_session_t * mp) +{ + vppcom_main_t *vcm = &vppcom_main; + vl_api_accept_session_reply_t *rmp; + svm_fifo_t *rx_fifo, *tx_fifo; + session_t *session; + u32 session_index; + int rv = 0; + + if (!clib_fifo_free_elts (vcm->client_session_index_fifo)) + { + clib_warning ("[%d] client session queue is full!", vcm->my_pid); + rv = VNET_API_ERROR_QUEUE_FULL; + goto send_reply; + } + + if (VPPCOM_DEBUG > 1) + { + u8 *ip_str = format (0, "%U", format_ip46_address, &mp->ip, mp->is_ip4); + clib_warning ("[%d] accepted session from: %s:%d", vcm->my_pid, ip_str, + clib_net_to_host_u16 (mp->port)); + vec_free (ip_str); + } + + clib_spinlock_lock (&vcm->sessions_lockp); + /* Allocate local session and set it up */ + pool_get (vcm->sessions, session); + memset (session, 0, sizeof (*session)); + session_index = session - vcm->sessions; + + rx_fifo = uword_to_pointer (mp->server_rx_fifo, svm_fifo_t *); + rx_fifo->client_session_index = session_index; + tx_fifo = uword_to_pointer (mp->server_tx_fifo, svm_fifo_t *); + tx_fifo->client_session_index = session_index; + + session->server_rx_fifo = rx_fifo; + session->server_tx_fifo = tx_fifo; + session->vpp_event_queue = uword_to_pointer (mp->vpp_event_queue_address, + unix_shared_memory_queue_t *); + session->state = STATE_ACCEPT; + session->is_cut_thru = 0; + session->is_server = 1; + session->port = ntohs (mp->port); + session->is_ip4 = mp->is_ip4; + clib_memcpy (session->ip, mp->ip, sizeof (session->ip)); + clib_spinlock_unlock (&vcm->sessions_lockp); + + /* Add it to lookup table */ + hash_set (vcm->session_index_by_vpp_handles, mp->handle, session_index); + + clib_fifo_add1 (vcm->client_session_index_fifo, session_index); + + /* + * Send accept reply to vpp + */ +send_reply: + rmp = vl_msg_api_alloc (sizeof (*rmp)); + memset (rmp, 0, sizeof (*rmp)); + rmp->_vl_msg_id = ntohs (VL_API_ACCEPT_SESSION_REPLY); + rmp->retval = htonl (rv); + rmp->handle = mp->handle; + vl_msg_api_send_shmem (vcm->vl_input_queue, (u8 *) & rmp); +} + +/* + * Acting as server for redirected connect requests + */ +static void +vl_api_connect_sock_t_handler (vl_api_connect_sock_t * mp) +{ + static svm_fifo_segment_create_args_t _a; + svm_fifo_segment_create_args_t *a = &_a; + vppcom_main_t *vcm = &vppcom_main; + u32 session_index; + svm_fifo_segment_private_t *seg; + unix_shared_memory_queue_t *client_q; + vl_api_connect_session_reply_t *rmp; + session_t *session = 0; + int rv = 0; + svm_fifo_t *rx_fifo; + svm_fifo_t *tx_fifo; + unix_shared_memory_queue_t *event_q = 0; + + if (!clib_fifo_free_elts (vcm->client_session_index_fifo)) + { + if (VPPCOM_DEBUG > 1) + clib_warning ("[%d] client session queue is full!", vcm->my_pid); + rv = VNET_API_ERROR_QUEUE_FULL; + goto send_reply; + } + + /* Create the segment */ + memset (a, 0, sizeof (*a)); + a->segment_name = (char *) format ((u8 *) a->segment_name, "%d:segment%d%c", + vcm->my_pid, vcm->unique_segment_index++, + 0); + a->segment_size = vcm->cfg.segment_size; + a->preallocated_fifo_pairs = vcm->cfg.preallocated_fifo_pairs; + a->rx_fifo_size = vcm->cfg.rx_fifo_size; + a->tx_fifo_size = vcm->cfg.tx_fifo_size; + + rv = svm_fifo_segment_create (a); + if (PREDICT_FALSE (rv)) + { + if (VPPCOM_DEBUG > 1) + clib_warning ("[%d] svm_fifo_segment_create ('%s') failed", + vcm->my_pid, a->segment_name); + vec_reset_length (a->new_segment_indices); + rv = VNET_API_ERROR_URI_FIFO_CREATE_FAILED; + goto send_reply; + } + + if (VPPCOM_DEBUG > 1) + clib_warning ("[%d] created segment '%s'", vcm->my_pid, a->segment_name); + + clib_spinlock_lock (&vcm->sessions_lockp); + pool_get (vcm->sessions, session); + memset (session, 0, sizeof (*session)); + session_index = session - vcm->sessions; + + session->sm_seg_index = a->new_segment_indices[0]; + vec_reset_length (a->new_segment_indices); + + seg = svm_fifo_segment_get_segment (session->sm_seg_index); + rx_fifo = session->server_rx_fifo = + svm_fifo_segment_alloc_fifo (seg, vcm->cfg.rx_fifo_size, + FIFO_SEGMENT_RX_FREELIST); + if (PREDICT_FALSE (!session->server_rx_fifo)) + { + svm_fifo_segment_delete (seg); + clib_warning ("[%d] rx fifo alloc failed, size %ld (0x%lx)", + vcm->my_pid, vcm->cfg.rx_fifo_size, + vcm->cfg.rx_fifo_size); + rv = VNET_API_ERROR_URI_FIFO_CREATE_FAILED; + clib_spinlock_unlock (&vcm->sessions_lockp); + goto send_reply; + } + + tx_fifo = session->server_tx_fifo = + svm_fifo_segment_alloc_fifo (seg, vcm->cfg.tx_fifo_size, + FIFO_SEGMENT_TX_FREELIST); + if (PREDICT_FALSE (!session->server_tx_fifo)) + { + svm_fifo_segment_delete (seg); + if (VPPCOM_DEBUG > 1) + clib_warning ("[%d] tx fifo alloc failed, size %ld (0x%lx)", + vcm->my_pid, vcm->cfg.tx_fifo_size, + vcm->cfg.tx_fifo_size); + rv = VNET_API_ERROR_URI_FIFO_CREATE_FAILED; + clib_spinlock_unlock (&vcm->sessions_lockp); + goto send_reply; + } + + session->server_rx_fifo->master_session_index = session_index; + session->server_tx_fifo->master_session_index = session_index; + session->client_queue_address = mp->client_queue_address; + session->is_cut_thru = 1; + session->is_server = 1; + session->is_ip4 = mp->is_ip4; + session->port = mp->port; + { + void *oldheap; + ssvm_shared_header_t *sh = seg->ssvm.sh; + + ssvm_lock_non_recursive (sh, 1); + oldheap = ssvm_push_heap (sh); + event_q = session->vpp_event_queue = + unix_shared_memory_queue_init (vcm->cfg.event_queue_size, + sizeof (session_fifo_event_t), + vcm->my_pid, 0 /* signal not sent */ ); + ssvm_pop_heap (oldheap); + ssvm_unlock_non_recursive (sh); + } + clib_memcpy (session->ip, mp->ip, sizeof (session->ip)); + + session->state = STATE_ACCEPT; + if (VPPCOM_DEBUG > 1) + clib_warning ("[%d] Connected cut-thru to client: sid %d", + vcm->my_pid, session_index); + clib_spinlock_unlock (&vcm->sessions_lockp); + clib_fifo_add1 (vcm->client_session_index_fifo, session_index); + +send_reply: + rmp = vl_msg_api_alloc (sizeof (*rmp)); + memset (rmp, 0, sizeof (*rmp)); + + rmp->_vl_msg_id = ntohs (VL_API_CONNECT_SESSION_REPLY); + rmp->context = mp->context; + rmp->retval = htonl (rv); + rmp->segment_name_length = vec_len (a->segment_name); + clib_memcpy (rmp->segment_name, a->segment_name, vec_len (a->segment_name)); + vec_reset_length (a->segment_name); + + if (event_q) + { + rmp->vpp_event_queue_address = pointer_to_uword (event_q); + rmp->server_rx_fifo = pointer_to_uword (rx_fifo); + rmp->server_tx_fifo = pointer_to_uword (tx_fifo); + } + client_q = + uword_to_pointer (mp->client_queue_address, unix_shared_memory_queue_t *); + + ASSERT (client_q); + vl_msg_api_send_shmem (client_q, (u8 *) & rmp); +} + +static void +vppcom_send_bind_sock (session_t * session) +{ + vppcom_main_t *vcm = &vppcom_main; + vl_api_bind_sock_t *bmp; + + /* Assumes caller has acquired spinlock: vcm->sessions_lockp */ + session->is_server = 1; + bmp = vl_msg_api_alloc (sizeof (*bmp)); + memset (bmp, 0, sizeof (*bmp)); + + bmp->_vl_msg_id = ntohs (VL_API_BIND_SOCK); + bmp->client_index = vcm->my_client_index; + bmp->context = htonl (0xfeedface); + bmp->vrf = session->vrf; + bmp->is_ip4 = session->is_ip4; + clib_memcpy (bmp->ip, session->ip, sizeof (bmp->ip)); + bmp->port = session->port; + bmp->proto = session->proto; + clib_memcpy (bmp->options, session->options, sizeof (bmp->options)); + vl_msg_api_send_shmem (vcm->vl_input_queue, (u8 *) & bmp); +} + +static void +vppcom_send_unbind_sock (u32 session_index) +{ + vppcom_main_t *vcm = &vppcom_main; + vl_api_unbind_sock_t *ump; + session_t *session = 0; + int rv; + + clib_spinlock_lock (&vcm->sessions_lockp); + rv = vppcom_session_at_index (session_index, &session); + if (PREDICT_FALSE (rv)) + { + clib_spinlock_unlock (&vcm->sessions_lockp); + if (VPPCOM_DEBUG > 0) + clib_warning ("[%d] invalid session, sid (%d) has been closed!", + vcm->my_pid, session_index); + return; + } + + ump = vl_msg_api_alloc (sizeof (*ump)); + memset (ump, 0, sizeof (*ump)); + + ump->_vl_msg_id = ntohs (VL_API_UNBIND_SOCK); + ump->client_index = vcm->my_client_index; + ump->handle = session->vpp_session_handle; + clib_spinlock_unlock (&vcm->sessions_lockp); + vl_msg_api_send_shmem (vcm->vl_input_queue, (u8 *) & ump); +} + +static int +vppcom_session_unbind_cut_thru (session_t * session) +{ + svm_fifo_segment_main_t *sm = &svm_fifo_segment_main; + svm_fifo_segment_private_t *seg; + int rv = VPPCOM_OK; + + seg = vec_elt_at_index (sm->segments, session->sm_seg_index); + svm_fifo_segment_free_fifo (seg, session->server_rx_fifo, + FIFO_SEGMENT_RX_FREELIST); + svm_fifo_segment_free_fifo (seg, session->server_tx_fifo, + FIFO_SEGMENT_TX_FREELIST); + svm_fifo_segment_delete (seg); + + return rv; +} + +static int +vppcom_session_unbind (u32 session_index) +{ + vppcom_main_t *vcm = &vppcom_main; + int rv; + + clib_spinlock_lock (&vcm->sessions_lockp); + if (PREDICT_FALSE (pool_is_free_index (vcm->sessions, session_index))) + { + clib_spinlock_unlock (&vcm->sessions_lockp); + if (VPPCOM_DEBUG > 1) + clib_warning ("[%d] invalid session, sid (%d) has been closed!", + vcm->my_pid, session_index); + return VPPCOM_EBADFD; + } + clib_spinlock_unlock (&vcm->sessions_lockp); + + vcm->bind_session_index = session_index; + vppcom_send_unbind_sock (session_index); + rv = vppcom_wait_for_session_state_change (session_index, STATE_START, + vcm->cfg.session_timeout); + if (PREDICT_FALSE (rv)) + { + vcm->bind_session_index = ~0; + if (VPPCOM_DEBUG > 0) + clib_warning ("[%d] server unbind timed out, rv = %s (%d)", + vcm->my_pid, vppcom_retval_str (rv), rv); + return rv; + } + return VPPCOM_OK; +} + +static int +vppcom_session_disconnect (u32 session_index) +{ + vppcom_main_t *vcm = &vppcom_main; + int rv; + + rv = vppcom_send_disconnect (session_index); + if (PREDICT_FALSE (rv)) + return rv; + + rv = vppcom_wait_for_session_state_change (session_index, STATE_DISCONNECT, + vcm->cfg.session_timeout); + if (PREDICT_FALSE (rv)) + { + if (VPPCOM_DEBUG > 0) + clib_warning ("[%d] client disconnect timed out, rv = %s (%d)", + vcm->my_pid, vppcom_retval_str (rv), rv); + return rv; + } + return VPPCOM_OK; +} + +#define foreach_sock_msg \ +_(SESSION_ENABLE_DISABLE_REPLY, session_enable_disable_reply) \ +_(BIND_SOCK_REPLY, bind_sock_reply) \ +_(UNBIND_SOCK_REPLY, unbind_sock_reply) \ +_(ACCEPT_SESSION, accept_session) \ +_(CONNECT_SOCK, connect_sock) \ +_(CONNECT_SESSION_REPLY, connect_session_reply) \ +_(DISCONNECT_SESSION, disconnect_session) \ +_(DISCONNECT_SESSION_REPLY, disconnect_session_reply) \ +_(RESET_SESSION, reset_session) \ +_(APPLICATION_ATTACH_REPLY, application_attach_reply) \ +_(APPLICATION_DETACH_REPLY, application_detach_reply) \ +_(MAP_ANOTHER_SEGMENT, map_another_segment) + +static void +vppcom_api_hookup (void) +{ +#define _(N,n) \ + vl_msg_api_set_handlers(VL_API_##N, #n, \ + vl_api_##n##_t_handler, \ + vl_noop_handler, \ + vl_api_##n##_t_endian, \ + vl_api_##n##_t_print, \ + sizeof(vl_api_##n##_t), 1); + foreach_sock_msg; +#undef _ +} + +static void +vppcom_cfg_init (vppcom_cfg_t * vcl_cfg) +{ + ASSERT (vcl_cfg); + + vcl_cfg->heapsize = (256ULL << 20); + vcl_cfg->segment_baseva = 0x200000000ULL; + vcl_cfg->segment_size = (256 << 20); + vcl_cfg->add_segment_size = (128 << 20); + vcl_cfg->preallocated_fifo_pairs = 8; + vcl_cfg->rx_fifo_size = (1 << 20); + vcl_cfg->tx_fifo_size = (1 << 20); + vcl_cfg->event_queue_size = 2048; + vcl_cfg->listen_queue_size = CLIB_CACHE_LINE_BYTES / sizeof (u32); + vcl_cfg->app_timeout = 10 * 60.0; + vcl_cfg->session_timeout = 10 * 60.0; + vcl_cfg->accept_timeout = 60.0; +} + +static void +vppcom_cfg_heapsize (char *conf_fname) +{ + vppcom_main_t *vcm = &vppcom_main; + vppcom_cfg_t *vcl_cfg = &vcm->cfg; + FILE *fp; + char inbuf[4096]; + int argc = 1; + char **argv = NULL; + char *arg = NULL; + char *p; + int i; + u8 *sizep; + u32 size; + + fp = fopen (conf_fname, "r"); + if (fp == NULL) + { + if (VPPCOM_DEBUG > 0) + fprintf (stderr, "open configuration file '%s' failed\n", conf_fname); + goto defaulted; + } + argv = calloc (1, sizeof (char *)); + if (argv == NULL) + goto defaulted; + + while (1) + { + if (fgets (inbuf, 4096, fp) == 0) + break; + p = strtok (inbuf, " \t\n"); + while (p != NULL) + { + if (*p == '#') + break; + argc++; + char **tmp = realloc (argv, argc * sizeof (char *)); + if (tmp == NULL) + goto defaulted; + argv = tmp; + arg = strndup (p, 1024); + if (arg == NULL) + goto defaulted; + argv[argc - 1] = arg; + p = strtok (NULL, " \t\n"); + } + } + + fclose (fp); + fp = NULL; + + char **tmp = realloc (argv, (argc + 1) * sizeof (char *)); + if (tmp == NULL) + goto defaulted; + argv = tmp; + argv[argc] = NULL; + + /* + * Look for and parse the "heapsize" config parameter. + * Manual since none of the clib infra has been bootstrapped yet. + * + * Format: heapsize <nn>[mM][gG] + */ + + for (i = 1; i < (argc - 1); i++) + { + if (!strncmp (argv[i], "heapsize", 8)) + { + sizep = (u8 *) argv[i + 1]; + size = 0; + while (*sizep >= '0' && *sizep <= '9') + { + size *= 10; + size += *sizep++ - '0'; + } + if (size == 0) + { + if (VPPCOM_DEBUG > 0) + clib_warning ("[%d] parse error '%s %s', " + "using default heapsize %lld (0x%llx)", + vcm->my_pid, argv[i], argv[i + 1], + vcl_cfg->heapsize, vcl_cfg->heapsize); + goto defaulted; + } + + if (*sizep == 'g' || *sizep == 'G') + vcl_cfg->heapsize = size << 30; + else if (*sizep == 'm' || *sizep == 'M') + vcl_cfg->heapsize = size << 20; + else + { + if (VPPCOM_DEBUG > 0) + clib_warning ("[%d] parse error '%s %s', " + "using default heapsize %lld (0x%llx)", + vcm->my_pid, argv[i], argv[i + 1], + vcl_cfg->heapsize, vcl_cfg->heapsize); + goto defaulted; + } + } + } + +defaulted: + if (fp != NULL) + fclose (fp); + if (argv != NULL) + free (argv); + if (!clib_mem_init (0, vcl_cfg->heapsize)) + clib_warning ("[%d] vppcom heap allocation failure!", vcm->my_pid); + else if (VPPCOM_DEBUG > 0) + clib_warning ("[%d] allocated vppcom heapsize %lld (0x%llx)", + vcm->my_pid, vcl_cfg->heapsize, vcl_cfg->heapsize); +} + +static void +vppcom_cfg_read (char *conf_fname) +{ + vppcom_main_t *vcm = &vppcom_main; + vppcom_cfg_t *vcl_cfg = &vcm->cfg; + int fd; + unformat_input_t _input, *input = &_input; + unformat_input_t _line_input, *line_input = &_line_input; + u8 vc_cfg_input = 0; + u8 *chroot_path; + struct stat s; + u32 uid, gid; + + fd = open (conf_fname, O_RDONLY); + if (fd < 0) + { + if (VPPCOM_DEBUG > 0) + clib_warning ("[%d] open configuration file '%s' failed!", + vcm->my_pid, conf_fname); + goto file_done; + } + + if (fstat (fd, &s) < 0) + { + if (VPPCOM_DEBUG > 0) + clib_warning ("[%d] failed to stat `%s'", vcm->my_pid, conf_fname); + goto file_done; + } + + if (!(S_ISREG (s.st_mode) || S_ISLNK (s.st_mode))) + { + if (VPPCOM_DEBUG > 0) + clib_warning ("[%d] not a regular file `%s'", vcm->my_pid, + conf_fname); + goto file_done; + } + + unformat_init_unix_file (input, fd); + + while (unformat_check_input (input) != UNFORMAT_END_OF_INPUT) + { + (void) unformat_user (input, unformat_line_input, line_input); + unformat_skip_white_space (line_input); + + if (unformat (line_input, "vppcom {")) + { + vc_cfg_input = 1; + continue; + } + + if (vc_cfg_input) + { + if (unformat (line_input, "heapsize %s", &chroot_path)) + { + vec_terminate_c_string (chroot_path); + if (VPPCOM_DEBUG > 0) + clib_warning ("[%d] configured heapsize %s, " + "actual heapsize %lld (0x%llx)", + vcm->my_pid, chroot_path, vcl_cfg->heapsize, + vcl_cfg->heapsize); + vec_free (chroot_path); + } + else if (unformat (line_input, "api-prefix %s", &chroot_path)) + { + vec_terminate_c_string (chroot_path); + vl_set_memory_root_path ((char *) chroot_path); + if (VPPCOM_DEBUG > 0) + clib_warning ("[%d] configured api-prefix %s", + vcm->my_pid, chroot_path); + chroot_path = 0; /* Don't vec_free() it! */ + } + else if (unformat (line_input, "uid %d", &uid)) + { + vl_set_memory_uid (uid); + if (VPPCOM_DEBUG > 0) + clib_warning ("[%d] configured uid %d", vcm->my_pid, uid); + } + else if (unformat (line_input, "gid %d", &gid)) + { + vl_set_memory_gid (gid); + if (VPPCOM_DEBUG > 0) + clib_warning ("[%d] configured gid %d", vcm->my_pid, gid); + } + else if (unformat (line_input, "segment-baseva 0x%llx", + &vcl_cfg->segment_baseva)) + { + if (VPPCOM_DEBUG > 0) + clib_warning ("[%d] configured segment_baseva 0x%llx", + vcm->my_pid, vcl_cfg->segment_baseva); + } + else if (unformat (line_input, "segment-size 0x%lx", + &vcl_cfg->segment_size)) + { + if (VPPCOM_DEBUG > 0) + clib_warning ("[%d] configured segment_size 0x%lx (%ld)", + vcm->my_pid, vcl_cfg->segment_size, + vcl_cfg->segment_size); + } + else if (unformat (line_input, "segment-size %ld", + &vcl_cfg->segment_size)) + { + if (VPPCOM_DEBUG > 0) + clib_warning ("[%d] configured segment_size %ld (0x%lx)", + vcm->my_pid, vcl_cfg->segment_size, + vcl_cfg->segment_size); + } + else if (unformat (line_input, "add-segment-size 0x%lx", + &vcl_cfg->add_segment_size)) + { + if (VPPCOM_DEBUG > 0) + clib_warning + ("[%d] configured add_segment_size 0x%lx (%ld)", + vcm->my_pid, vcl_cfg->add_segment_size, + vcl_cfg->add_segment_size); + } + else if (unformat (line_input, "add-segment-size %ld", + &vcl_cfg->add_segment_size)) + { + if (VPPCOM_DEBUG > 0) + clib_warning + ("[%d] configured add_segment_size %ld (0x%lx)", + vcm->my_pid, vcl_cfg->add_segment_size, + vcl_cfg->add_segment_size); + } + else if (unformat (line_input, "preallocated-fifo-pairs %d", + &vcl_cfg->preallocated_fifo_pairs)) + { + if (VPPCOM_DEBUG > 0) + clib_warning ("[%d] configured preallocated_fifo_pairs " + "%d (0x%x)", vcm->my_pid, + vcl_cfg->preallocated_fifo_pairs, + vcl_cfg->preallocated_fifo_pairs); + } + else if (unformat (line_input, "rx-fifo-size 0x%lx", + &vcl_cfg->rx_fifo_size)) + { + if (VPPCOM_DEBUG > 0) + clib_warning ("[%d] configured rx_fifo_size 0x%lx (%ld)", + vcm->my_pid, vcl_cfg->rx_fifo_size, + vcl_cfg->rx_fifo_size); + } + else if (unformat (line_input, "rx-fifo-size %ld", + &vcl_cfg->rx_fifo_size)) + { + if (VPPCOM_DEBUG > 0) + clib_warning ("[%d] configured rx_fifo_size %ld (0x%lx)", + vcm->my_pid, vcl_cfg->rx_fifo_size, + vcl_cfg->rx_fifo_size); + } + else if (unformat (line_input, "tx-fifo-size 0x%lx", + &vcl_cfg->tx_fifo_size)) + { + if (VPPCOM_DEBUG > 0) + clib_warning ("[%d] configured tx_fifo_size 0x%lx (%ld)", + vcm->my_pid, vcl_cfg->tx_fifo_size, + vcl_cfg->tx_fifo_size); + } + else if (unformat (line_input, "tx-fifo-size %ld", + &vcl_cfg->tx_fifo_size)) + { + if (VPPCOM_DEBUG > 0) + clib_warning ("[%d] configured tx_fifo_size %ld (0x%lx)", + vcm->my_pid, vcl_cfg->tx_fifo_size, + vcl_cfg->tx_fifo_size); + } + else if (unformat (line_input, "event-queue-size 0x%lx", + &vcl_cfg->event_queue_size)) + { + if (VPPCOM_DEBUG > 0) + clib_warning ("[%d] configured event_queue_size 0x%lx (%ld)", + vcm->my_pid, vcl_cfg->event_queue_size, + vcl_cfg->event_queue_size); + } + else if (unformat (line_input, "event-queue-size %ld", + &vcl_cfg->event_queue_size)) + { + if (VPPCOM_DEBUG > 0) + clib_warning ("[%d] configured event_queue_size %ld (0x%lx)", + vcm->my_pid, vcl_cfg->event_queue_size, + vcl_cfg->event_queue_size); + } + else if (unformat (line_input, "listen-queue-size 0x%lx", + &vcl_cfg->listen_queue_size)) + { + if (VPPCOM_DEBUG > 0) + clib_warning ("[%d] configured listen_queue_size 0x%lx (%ld)", + vcm->my_pid, vcl_cfg->listen_queue_size, + vcl_cfg->listen_queue_size); + } + else if (unformat (line_input, "listen-queue-size %ld", + &vcl_cfg->listen_queue_size)) + { + if (VPPCOM_DEBUG > 0) + clib_warning ("[%d] configured listen_queue_size %ld (0x%lx)", + vcm->my_pid, vcl_cfg->listen_queue_size, + vcl_cfg->listen_queue_size); + } + else if (unformat (line_input, "app-timeout %f", + &vcl_cfg->app_timeout)) + { + if (VPPCOM_DEBUG > 0) + clib_warning ("[%d] configured app_timeout %f", + vcm->my_pid, vcl_cfg->app_timeout); + } + else if (unformat (line_input, "session-timeout %f", + &vcl_cfg->session_timeout)) + { + if (VPPCOM_DEBUG > 0) + clib_warning ("[%d] configured session_timeout %f", + vcm->my_pid, vcl_cfg->session_timeout); + } + else if (unformat (line_input, "accept-timeout %f", + &vcl_cfg->accept_timeout)) + { + if (VPPCOM_DEBUG > 0) + clib_warning ("[%d] configured accept_timeout %f", + vcm->my_pid, vcl_cfg->accept_timeout); + } + else if (unformat (line_input, "}")) + { + vc_cfg_input = 0; + if (VPPCOM_DEBUG > 0) + clib_warning ("[%d] completed parsing vppcom config!", + vcm->my_pid); + goto input_done; + } + else + { + if (line_input->buffer[line_input->index] != '#') + { + clib_warning ("[%d] Unknown vppcom config option: '%s'", + vcm->my_pid, (char *) + &line_input->buffer[line_input->index]); + } + } + } + } + +input_done: + unformat_free (input); + +file_done: + if (fd >= 0) + close (fd); +} + +/* + * VPPCOM Public API functions + */ +int +vppcom_app_create (char *app_name) +{ + vppcom_main_t *vcm = &vppcom_main; + vppcom_cfg_t *vcl_cfg = &vcm->cfg; + u8 *heap; + mheap_t *h; + int rv; + + if (!vcm->init) + { + char *conf_fname; + + vcm->init = 1; + vcm->my_pid = getpid (); + clib_fifo_validate (vcm->client_session_index_fifo, + vcm->cfg.listen_queue_size); + vppcom_cfg_init (vcl_cfg); + conf_fname = getenv (VPPCOM_CONF_ENV); + if (!conf_fname) + { + conf_fname = VPPCOM_CONF_DEFAULT; + if (VPPCOM_DEBUG > 0) + clib_warning ("[%d] getenv '%s' failed!", vcm->my_pid, + VPPCOM_CONF_ENV); + } + vppcom_cfg_heapsize (conf_fname); + vppcom_cfg_read (conf_fname); + vcm->bind_session_index = ~0; + vcm->main_cpu = os_get_thread_index (); + heap = clib_mem_get_per_cpu_heap (); + h = mheap_header (heap); + + /* make the main heap thread-safe */ + h->flags |= MHEAP_FLAG_THREAD_SAFE; + + vcm->session_index_by_vpp_handles = hash_create (0, sizeof (uword)); + + clib_time_init (&vcm->clib_time); + vppcom_init_error_string_table (); + svm_fifo_segment_init (vcl_cfg->segment_baseva, + 20 /* timeout in secs */ ); + clib_spinlock_init (&vcm->sessions_lockp); + vppcom_api_hookup (); + } + + if (vcm->my_client_index == ~0) + { + vcm->app_state = STATE_APP_START; + rv = vppcom_connect_to_vpp (app_name); + if (rv) + { + clib_warning ("[%s] couldn't connect to VPP.", vcm->my_pid); + return rv; + } + + if (VPPCOM_DEBUG > 0) + clib_warning ("[%d] sending session enable", vcm->my_pid); + + rv = vppcom_app_session_enable (); + if (rv) + { + clib_warning ("[%d] vppcom_app_session_enable() failed!", + vcm->my_pid); + return rv; + } + + if (VPPCOM_DEBUG > 0) + clib_warning ("[%d] sending app attach", vcm->my_pid); + + rv = vppcom_app_attach (); + if (rv) + { + clib_warning ("[%d] vppcom_app_attach() failed!", vcm->my_pid); + return rv; + } + } + + if (VPPCOM_DEBUG > 0) + clib_warning ("[%d] app_name '%s', my_client_index %d (0x%x)", + vcm->my_pid, app_name, vcm->my_client_index, + vcm->my_client_index); + + return VPPCOM_OK; +} + +void +vppcom_app_destroy (void) +{ + vppcom_main_t *vcm = &vppcom_main; + int rv; + + if (vcm->my_client_index == ~0) + return; + + if (VPPCOM_DEBUG > 0) + clib_warning ("[%d] detaching from VPP, my_client_index %d (0x%x)", + vcm->my_pid, vcm->my_client_index, vcm->my_client_index); + + vppcom_app_detach (); + rv = vppcom_wait_for_app_state_change (STATE_APP_ENABLED); + if (PREDICT_FALSE (rv)) + { + if (VPPCOM_DEBUG > 0) + clib_warning ("[%d] application detach timed out, rv = %s (%d)", + vcm->my_pid, vppcom_retval_str (rv), rv); + } + vl_client_disconnect_from_vlib (); + vcm->my_client_index = ~0; + vcm->app_state = STATE_APP_START; +} + +int +vppcom_session_create (u32 vrf, u8 proto, u8 is_nonblocking) +{ + vppcom_main_t *vcm = &vppcom_main; + session_t *session; + u32 session_index; + + clib_spinlock_lock (&vcm->sessions_lockp); + pool_get (vcm->sessions, session); + session_index = session - vcm->sessions; + + session->vrf = vrf; + session->proto = proto; + session->state = STATE_START; + session->is_nonblocking = is_nonblocking; + clib_spinlock_unlock (&vcm->sessions_lockp); + + if (VPPCOM_DEBUG > 0) + clib_warning ("[%d] sid %d", vcm->my_pid, session_index); + + return (int) session_index; +} + +int +vppcom_session_close (uint32_t session_index) +{ + vppcom_main_t *vcm = &vppcom_main; + session_t *session = 0; + int rv; + + clib_spinlock_lock (&vcm->sessions_lockp); + rv = vppcom_session_at_index (session_index, &session); + if (PREDICT_FALSE (rv)) + { + clib_spinlock_unlock (&vcm->sessions_lockp); + if (VPPCOM_DEBUG > 0) + clib_warning ("[%d] invalid session, sid (%d) has been closed!", + vcm->my_pid, session_index); + return rv; + } + clib_spinlock_unlock (&vcm->sessions_lockp); + + if (VPPCOM_DEBUG > 0) + clib_warning ("[%d] sid %d", vcm->my_pid, session_index); + + if (session->is_cut_thru) + { + if (session->is_server) + { + rv = vppcom_session_unbind_cut_thru (session); + if ((VPPCOM_DEBUG > 0) && (rv < 0)) + clib_warning ("[%d] unbind cut-thru (session %d) failed, " + "rv = %s (%d)", + vcm->my_pid, session_index, + vppcom_retval_str (rv), rv); + } + } + else if (session->is_server) + { + rv = vppcom_session_unbind (session_index); + if ((VPPCOM_DEBUG > 0) && (rv < 0)) + clib_warning ("[%d] unbind (session %d) failed, rv = %s (%d)", + vcm->my_pid, session_index, vppcom_retval_str (rv), rv); + } + else + { + rv = vppcom_session_disconnect (session_index); + if ((VPPCOM_DEBUG > 0) && (rv < 0)) + clib_warning ("[%d] disconnect (session %d) failed, rv = %s (%d)", + vcm->my_pid, session_index, vppcom_retval_str (rv), rv); + } + if (rv < 0) + return rv; + + clib_spinlock_lock (&vcm->sessions_lockp); + pool_put_index (vcm->sessions, session_index); + clib_spinlock_unlock (&vcm->sessions_lockp); + return rv; +} + +int +vppcom_session_bind (uint32_t session_index, vppcom_endpt_t * ep) +{ + vppcom_main_t *vcm = &vppcom_main; + session_t *session = 0; + int rv; + ip46_address_t *ip46; + + if (!ep || !ep->ip) + return VPPCOM_EINVAL; + + clib_spinlock_lock (&vcm->sessions_lockp); + rv = vppcom_session_at_index (session_index, &session); + if (PREDICT_FALSE (rv)) + { + clib_spinlock_unlock (&vcm->sessions_lockp); + if (VPPCOM_DEBUG > 0) + clib_warning ("[%d] invalid session, sid (%d) has been closed!", + vcm->my_pid, session_index); + return rv; + } + + if (VPPCOM_DEBUG > 0) + clib_warning ("[%d] sid %d", vcm->my_pid, session_index); + + session->vrf = ep->vrf; + session->is_ip4 = ep->is_ip4; + memset (session->ip, 0, sizeof (session->ip)); + ip46 = (ip46_address_t *) session->ip; + *ip46 = to_ip46 (!ep->is_ip4, ep->ip); + session->port = ep->port; + + clib_spinlock_unlock (&vcm->sessions_lockp); + return VPPCOM_OK; +} + +int +vppcom_session_listen (uint32_t listen_session_index, uint32_t q_len) +{ + vppcom_main_t *vcm = &vppcom_main; + session_t *listen_session = 0; + int rv; + + clib_spinlock_lock (&vcm->sessions_lockp); + rv = vppcom_session_at_index (listen_session_index, &listen_session); + if (PREDICT_FALSE (rv)) + { + clib_spinlock_unlock (&vcm->sessions_lockp); + if (VPPCOM_DEBUG > 0) + clib_warning ("[%d] invalid session, sid (%d) has been closed!", + vcm->my_pid, listen_session_index); + return rv; + } + + if (VPPCOM_DEBUG > 0) + clib_warning ("[%d] sid %d", vcm->my_pid, listen_session_index); + + ASSERT (vcm->bind_session_index == ~0); + vcm->bind_session_index = listen_session_index; + vppcom_send_bind_sock (listen_session); + clib_spinlock_unlock (&vcm->sessions_lockp); + rv = + vppcom_wait_for_session_state_change (listen_session_index, STATE_LISTEN, + vcm->cfg.session_timeout); + if (PREDICT_FALSE (rv)) + { + vcm->bind_session_index = ~0; + if (VPPCOM_DEBUG > 0) + clib_warning ("[%d] server listen timed out, rv = %d (%d)", + vcm->my_pid, vppcom_retval_str (rv), rv); + return rv; + } + + clib_spinlock_lock (&vcm->sessions_lockp); + rv = vppcom_session_at_index (listen_session_index, &listen_session); + if (PREDICT_FALSE (rv)) + { + clib_spinlock_unlock (&vcm->sessions_lockp); + if (VPPCOM_DEBUG > 0) + clib_warning ("[%d] invalid session, sid (%d) has been closed!", + vcm->my_pid, listen_session_index); + return rv; + } + listen_session->is_listen = 1; + clib_spinlock_unlock (&vcm->sessions_lockp); + clib_fifo_validate (vcm->client_session_index_fifo, q_len); + + return VPPCOM_OK; +} + +int +vppcom_session_accept (uint32_t listen_session_index, vppcom_endpt_t * ep, + double wait_for_time) +{ + vppcom_main_t *vcm = &vppcom_main; + session_t *listen_session = 0; + session_t *client_session = 0; + u32 client_session_index; + int rv; + f64 wait_for; + + clib_spinlock_lock (&vcm->sessions_lockp); + rv = vppcom_session_at_index (listen_session_index, &listen_session); + if (PREDICT_FALSE (rv)) + { + clib_spinlock_unlock (&vcm->sessions_lockp); + if (VPPCOM_DEBUG > 0) + clib_warning ("[%d] invalid session, sid (%d) has been closed!", + vcm->my_pid, listen_session_index); + return rv; + } + + if (listen_session->state != STATE_LISTEN) + { + clib_spinlock_unlock (&vcm->sessions_lockp); + if (VPPCOM_DEBUG > 0) + clib_warning ("[%d] session not in listen state, state = %s", + vcm->my_pid, + vppcom_session_state_str (listen_session->state)); + return VPPCOM_EBADFD; + } + wait_for = listen_session->is_nonblocking ? 0 : + (wait_for_time < 0) ? vcm->cfg.accept_timeout : wait_for_time; + + if (VPPCOM_DEBUG > 0) + clib_warning ("[%d] sid %d: %s (%d)", vcm->my_pid, + listen_session_index, + vppcom_session_state_str (listen_session->state), + listen_session->state); + clib_spinlock_unlock (&vcm->sessions_lockp); + + while (1) + { + rv = vppcom_wait_for_client_session_index (wait_for); + if (rv) + { + if ((VPPCOM_DEBUG > 0)) + clib_warning ("[%d] sid %d, accept timed out, rv = %s (%d)", + vcm->my_pid, listen_session_index, + vppcom_retval_str (rv), rv); + if ((wait_for == 0) || (wait_for_time > 0)) + return rv; + } + else + break; + } + + clib_fifo_sub1 (vcm->client_session_index_fifo, client_session_index); + + clib_spinlock_lock (&vcm->sessions_lockp); + rv = vppcom_session_at_index (client_session_index, &client_session); + ASSERT (rv == VPPCOM_OK); + ASSERT (client_session->is_ip4 == listen_session->is_ip4); + + if (VPPCOM_DEBUG > 0) + clib_warning ("[%d] Got a request: client sid %d", vcm->my_pid, + client_session_index); + + ep->vrf = client_session->vrf; + ep->is_cut_thru = client_session->is_cut_thru; + ep->is_ip4 = client_session->is_ip4; + ep->port = client_session->port; + if (client_session->is_ip4) + clib_memcpy (ep->ip, client_session->ip, sizeof (ip4_address_t)); + else + clib_memcpy (ep->ip, client_session->ip, sizeof (ip6_address_t)); + clib_spinlock_unlock (&vcm->sessions_lockp); + return (int) client_session_index; +} + +int +vppcom_session_connect (uint32_t session_index, vppcom_endpt_t * server_ep) +{ + vppcom_main_t *vcm = &vppcom_main; + session_t *session = 0; + int rv; + ip46_address_t *ip46; + + clib_spinlock_lock (&vcm->sessions_lockp); + rv = vppcom_session_at_index (session_index, &session); + if (PREDICT_FALSE (rv)) + { + clib_spinlock_unlock (&vcm->sessions_lockp); + if (VPPCOM_DEBUG > 0) + clib_warning ("[%d] invalid session, sid (%d) has been closed!", + vcm->my_pid, session_index); + return rv; + } + + if (session->state == STATE_CONNECT) + { + clib_spinlock_unlock (&vcm->sessions_lockp); + if (VPPCOM_DEBUG > 0) + clib_warning ("[%d] session, sid (%d) already connected!", + vcm->my_pid, session_index); + return VPPCOM_OK; + } + + session->vrf = server_ep->vrf; + session->is_ip4 = server_ep->is_ip4; + memset (session->ip, 0, sizeof (session->ip)); + ip46 = (ip46_address_t *) session->ip; + *ip46 = to_ip46 (!server_ep->is_ip4, server_ep->ip); + session->port = server_ep->port; + + if (VPPCOM_DEBUG > 0) + { + u8 *ip_str = format (0, "%U", format_ip46_address, + &session->ip, session->is_ip4); + clib_warning ("[%d] connect sid %d to %s server port %d", + vcm->my_pid, session_index, ip_str, + clib_net_to_host_u16 (session->port)); + vec_free (ip_str); + } + + vppcom_send_connect_sock (session, session_index); + clib_spinlock_unlock (&vcm->sessions_lockp); + rv = vppcom_wait_for_session_state_change (session_index, STATE_CONNECT, + vcm->cfg.session_timeout); + if (PREDICT_FALSE (rv)) + { + if (VPPCOM_DEBUG > 0) + clib_warning ("[%d] connect timed out, rv = %s (%d)", + vcm->my_pid, vppcom_retval_str (rv), rv); + return rv; + } + return VPPCOM_OK; +} + +int +vppcom_session_read (uint32_t session_index, void *buf, int n) +{ + vppcom_main_t *vcm = &vppcom_main; + session_t *session = 0; + svm_fifo_t *rx_fifo; + int n_read = 0; + int rv; + int max_dequeue; + char *fifo_str; + + ASSERT (buf); + + clib_spinlock_lock (&vcm->sessions_lockp); + rv = vppcom_session_at_index (session_index, &session); + if (PREDICT_FALSE (rv)) + { + clib_spinlock_unlock (&vcm->sessions_lockp); + if (VPPCOM_DEBUG > 0) + clib_warning ("[%d] invalid session, sid (%d) has been closed!", + vcm->my_pid, session_index); + return rv; + } + + if (session->state == STATE_DISCONNECT) + { + clib_spinlock_unlock (&vcm->sessions_lockp); + if (VPPCOM_DEBUG > 0) + clib_warning ("[%d] sid (%d) has been closed by remote peer!", + vcm->my_pid, session_index); + return VPPCOM_ECONNRESET; + } + + rx_fifo = ((!session->is_cut_thru || session->is_server) ? + session->server_rx_fifo : session->server_tx_fifo); + fifo_str = ((!session->is_cut_thru || session->is_server) ? + "server_rx_fifo" : "server_tx_fifo"); + clib_spinlock_unlock (&vcm->sessions_lockp); + + max_dequeue = (int) svm_fifo_max_dequeue (rx_fifo); + n_read = svm_fifo_dequeue_nowait (rx_fifo, clib_min (n, max_dequeue), buf); + + if (VPPCOM_DEBUG > 2) + clib_warning ("[%d] sid %d, read %d bytes from %s (%p)", vcm->my_pid, + session_index, n_read, fifo_str, rx_fifo); + + return (n_read <= 0) ? VPPCOM_EAGAIN : n_read; +} + +static inline int +vppcom_session_read_ready (session_t * session, u32 session_index) +{ + vppcom_main_t *vcm = &vppcom_main; + svm_fifo_t *rx_fifo; + int ready = 0; + + /* Assumes caller has acquired spinlock: vcm->sessions_lockp */ + if (session->state == STATE_DISCONNECT) + { + if (VPPCOM_DEBUG > 0) + clib_warning ("[%d] sid (%d) has been closed by remote peer!", + vcm->my_pid, session_index); + return VPPCOM_ECONNRESET; + } + + if (session->is_listen) + ready = clib_fifo_elts (vcm->client_session_index_fifo); + else + { + rx_fifo = ((!session->is_cut_thru || session->is_server) ? + session->server_rx_fifo : session->server_tx_fifo); + + ready = svm_fifo_max_dequeue (rx_fifo); + } + + if (VPPCOM_DEBUG > 3) + clib_warning ("[%d] sid %d, peek %s (%p), ready = %d", vcm->my_pid, + session_index, + session->is_server ? "server_rx_fifo" : "server_tx_fifo", + rx_fifo, ready); + return ready; +} + +int +vppcom_session_write (uint32_t session_index, void *buf, int n) +{ + vppcom_main_t *vcm = &vppcom_main; + session_t *session = 0; + svm_fifo_t *tx_fifo; + unix_shared_memory_queue_t *q; + session_fifo_event_t evt; + int rv; + char *fifo_str; + u8 is_nonblocking; + + ASSERT (buf); + + clib_spinlock_lock (&vcm->sessions_lockp); + rv = vppcom_session_at_index (session_index, &session); + if (PREDICT_FALSE (rv)) + { + clib_spinlock_unlock (&vcm->sessions_lockp); + if (VPPCOM_DEBUG > 0) + clib_warning ("[%d] invalid session, sid (%d) has been closed!", + vcm->my_pid, session_index); + return rv; + } + + if (session->state == STATE_DISCONNECT) + { + clib_spinlock_unlock (&vcm->sessions_lockp); + if (VPPCOM_DEBUG > 0) + clib_warning ("[%d] sid (%d) has been closed by remote peer!", + vcm->my_pid, session_index); + return VPPCOM_ECONNRESET; + } + + tx_fifo = ((!session->is_cut_thru || session->is_server) ? + session->server_tx_fifo : session->server_rx_fifo); + fifo_str = ((!session->is_cut_thru || session->is_server) ? + "server_tx_fifo" : "server_rx_fifo"); + + is_nonblocking = session->is_nonblocking; + clib_spinlock_unlock (&vcm->sessions_lockp); + + do + { + rv = svm_fifo_enqueue_nowait (tx_fifo, n, buf); + } + while (!is_nonblocking && (rv <= 0)); + + /* If event wasn't set, add one */ + if (!session->is_cut_thru && (rv > 0) && svm_fifo_set_event (tx_fifo)) + { + int rval; + + /* Fabricate TX event, send to vpp */ + evt.fifo = tx_fifo; + evt.event_type = FIFO_EVENT_APP_TX; + evt.event_id = vcm->tx_event_id++; + + clib_spinlock_lock (&vcm->sessions_lockp); + rval = vppcom_session_at_index (session_index, &session); + if (PREDICT_FALSE (rval)) + { + clib_spinlock_unlock (&vcm->sessions_lockp); + if (VPPCOM_DEBUG > 1) + clib_warning ("[%d] invalid session, sid (%d) has been closed!", + vcm->my_pid, session_index); + return rval; + } + q = session->vpp_event_queue; + clib_spinlock_unlock (&vcm->sessions_lockp); + ASSERT (q); + unix_shared_memory_queue_add (q, (u8 *) & evt, + 0 /* do wait for mutex */ ); + } + + if (VPPCOM_DEBUG > 2) + clib_warning ("[%d] sid %d, wrote %d bytes to %s (%p)", vcm->my_pid, + session_index, rv, fifo_str, tx_fifo); + + return rv; +} + +static inline int +vppcom_session_write_ready (session_t * session, u32 session_index) +{ + vppcom_main_t *vcm = &vppcom_main; + svm_fifo_t *tx_fifo; + char *fifo_str; + int rv; + + /* Assumes caller has acquired spinlock: vcm->sessions_lockp */ + if (session->state == STATE_DISCONNECT) + { + if (VPPCOM_DEBUG > 0) + clib_warning ("[%d] sid (%d) has been closed by remote peer!", + vcm->my_pid, session_index); + return VPPCOM_ECONNRESET; + } + + tx_fifo = ((!session->is_cut_thru || session->is_server) ? + session->server_tx_fifo : session->server_rx_fifo); + fifo_str = ((!session->is_cut_thru || session->is_server) ? + "server_tx_fifo" : "server_rx_fifo"); + + rv = svm_fifo_max_enqueue (tx_fifo); + + if (VPPCOM_DEBUG > 3) + clib_warning ("[%d] sid %d, peek %s (%p), ready = %d", vcm->my_pid, + session_index, fifo_str, tx_fifo, rv); + return rv; +} + +int +vppcom_select (unsigned long n_bits, unsigned long *read_map, + unsigned long *write_map, unsigned long *except_map, + double time_to_wait) +{ + vppcom_main_t *vcm = &vppcom_main; + u32 session_index; + session_t *session = 0; + int rv, bits_set = 0; + f64 timeout = clib_time_now (&vcm->clib_time) + time_to_wait; + u32 minbits = clib_max (n_bits, BITS (uword)); + + ASSERT (sizeof (clib_bitmap_t) == sizeof (long int)); + + if (read_map) + { + clib_bitmap_validate (vcm->rd_bitmap, minbits); + clib_memcpy (vcm->rd_bitmap, read_map, vec_len (vcm->rd_bitmap)); + memset (read_map, 0, vec_len (vcm->rd_bitmap)); + } + if (write_map) + { + clib_bitmap_validate (vcm->wr_bitmap, minbits); + clib_memcpy (vcm->wr_bitmap, write_map, vec_len (vcm->wr_bitmap)); + memset (write_map, 0, vec_len (vcm->wr_bitmap)); + } + if (except_map) + { + clib_bitmap_validate (vcm->ex_bitmap, minbits); + clib_memcpy (vcm->ex_bitmap, except_map, vec_len (vcm->ex_bitmap)); + memset (except_map, 0, vec_len (vcm->ex_bitmap)); + } + + do + { + /* *INDENT-OFF* */ + clib_bitmap_foreach (session_index, vcm->rd_bitmap, + ({ + clib_spinlock_lock (&vcm->sessions_lockp); + rv = vppcom_session_at_index (session_index, &session); + if (rv < 0) + { + clib_spinlock_unlock (&vcm->sessions_lockp); + if (VPPCOM_DEBUG > 1) + clib_warning ("[%d] session %d specified in " + "read_map is closed.", vcm->my_pid, + session_index); + bits_set = VPPCOM_EBADFD; + goto select_done; + } + + rv = vppcom_session_read_ready (session, session_index); + clib_spinlock_unlock (&vcm->sessions_lockp); + if (vcm->ex_bitmap && + clib_bitmap_get (vcm->ex_bitmap, session_index) && (rv < 0)) + { + // TBD: clib_warning + /* coverity[FORWARD_NULL] */ + clib_bitmap_set_no_check (except_map, session_index, 1); + bits_set++; + } + else if (rv > 0) + { + // TBD: clib_warning + /* coverity[FORWARD_NULL] */ + clib_bitmap_set_no_check (read_map, session_index, 1); + bits_set++; + } + })); + + clib_bitmap_foreach (session_index, vcm->wr_bitmap, + ({ + clib_spinlock_lock (&vcm->sessions_lockp); + rv = vppcom_session_at_index (session_index, &session); + if (rv < 0) + { + clib_spinlock_unlock (&vcm->sessions_lockp); + if (VPPCOM_DEBUG > 0) + clib_warning ("[%d] session %d specified in " + "write_map is closed.", vcm->my_pid, + session_index); + bits_set = VPPCOM_EBADFD; + goto select_done; + } + + rv = vppcom_session_write_ready (session, session_index); + clib_spinlock_unlock (&vcm->sessions_lockp); + if (rv > 0 ) + { + // TBD: clib_warning + /* coverity[FORWARD_NULL] */ + clib_bitmap_set_no_check (write_map, session_index, 1); + bits_set++; + } + })); + + clib_bitmap_foreach (session_index, vcm->ex_bitmap, + ({ + clib_spinlock_lock (&vcm->sessions_lockp); + rv = vppcom_session_at_index (session_index, &session); + if (rv < 0) + { + clib_spinlock_unlock (&vcm->sessions_lockp); + if (VPPCOM_DEBUG > 1) + clib_warning ("[%d] session %d specified in " + "except_map is closed.", vcm->my_pid, + session_index); + bits_set = VPPCOM_EBADFD; + goto select_done; + } + + rv = vppcom_session_read_ready (session, session_index); + clib_spinlock_unlock (&vcm->sessions_lockp); + if (rv < 0) + { + // TBD: clib_warning + /* coverity[FORWARD_NULL] */ + clib_bitmap_set_no_check (except_map, session_index, 1); + bits_set++; + } + })); + /* *INDENT-ON* */ + } + while (clib_time_now (&vcm->clib_time) < timeout); + +select_done: + return (bits_set); +} + +/* + * fd.io coding-style-patch-verification: ON + * + * Local Variables: + * eval: (c-set-style "gnu") + * End: + */ diff --git a/src/uri/vppcom.h b/src/uri/vppcom.h new file mode 100644 index 00000000..4b048e03 --- /dev/null +++ b/src/uri/vppcom.h @@ -0,0 +1,152 @@ +/* + * Copyright (c) 2017 Cisco and/or its affiliates. + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at: + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +#ifndef included_vppcom_h +#define included_vppcom_h + +#include <netdb.h> +#include <errno.h> + +/* + * VPPCOM Public API Definitions, Enums, and Data Structures + */ +#define INVALID_SESSION_ID (~0) +#define VPPCOM_VRF_DEFAULT 0 +#define VPPCOM_CONF_ENV "VPPCOM_CONF" +#define VPPCOM_CONF_DEFAULT "/etc/vpp/vppcom.conf" + +typedef enum +{ + VPPCOM_PROTO_TCP = 0, + VPPCOM_PROTO_UDP, +} vppcom_proto_t; + +typedef enum +{ + VPPCOM_IS_IP6 = 0, + VPPCOM_IS_IP4, +} vppcom_is_ip4_t; + +typedef struct vppcom_endpt_t_ +{ + uint32_t vrf; + uint8_t is_cut_thru; + uint8_t is_ip4; + uint8_t *ip; + uint16_t port; +} vppcom_endpt_t; + +typedef enum +{ + VPPCOM_OK = 0, + VPPCOM_EAGAIN = -EAGAIN, + VPPCOM_EINVAL = -EINVAL, + VPPCOM_EBADFD = -EBADFD, + VPPCOM_EAFNOSUPPORT = -EAFNOSUPPORT, + VPPCOM_ECONNRESET = -ECONNRESET, + VPPCOM_ECONNREFUSED = -ECONNREFUSED, + VPPCOM_ETIMEDOUT = -ETIMEDOUT, +} vppcom_error_t; + +/* + * VPPCOM Public API Functions + */ +static inline const char * +vppcom_retval_str (int retval) +{ + char *st; + + switch (retval) + { + case VPPCOM_OK: + st = "VPPCOM_OK"; + break; + + case VPPCOM_EAGAIN: + st = "VPPCOM_EAGAIN"; + break; + + case VPPCOM_EINVAL: + st = "VPPCOM_EINVAL"; + break; + + case VPPCOM_EBADFD: + st = "VPPCOM_EBADFD"; + break; + + case VPPCOM_EAFNOSUPPORT: + st = "VPPCOM_EAFNOSUPPORT"; + break; + + case VPPCOM_ECONNRESET: + st = "VPPCOM_ECONNRESET"; + break; + + case VPPCOM_ECONNREFUSED: + st = "VPPCOM_ECONNREFUSED"; + break; + + case VPPCOM_ETIMEDOUT: + st = "VPPCOM_ETIMEDOUT"; + break; + + default: + st = "UNKNOWN_STATE"; + break; + } + + return st; +} + +static inline int +is_vcom_fd (int fd) +{ +#define VPPCOM_FD_OFFSET (1 << 30) + return (fd >= VPPCOM_FD_OFFSET); +} + +/* TBD: make these constructor/destructor function */ +extern int vppcom_app_create (char *app_name); +extern void vppcom_app_destroy (void); + +extern int vppcom_session_create (uint32_t vrf, uint8_t proto, + uint8_t is_nonblocking); +extern int vppcom_session_close (uint32_t session_index); + +extern int vppcom_session_bind (uint32_t session_index, vppcom_endpt_t * ep); +extern int vppcom_session_listen (uint32_t session_index, uint32_t q_len); +extern int vppcom_session_accept (uint32_t session_index, + vppcom_endpt_t * client_ep, + double wait_for_time); + +extern int vppcom_session_connect (uint32_t session_index, + vppcom_endpt_t * server_ep); +extern int vppcom_session_read (uint32_t session_index, void *buf, int n); +extern int vppcom_session_write (uint32_t session_index, void *buf, int n); + +extern int vppcom_select (unsigned long n_bits, + unsigned long *read_map, + unsigned long *write_map, + unsigned long *except_map, double wait_for_time); + +#endif /* included_vppcom_h */ + +/* + * fd.io coding-style-patch-verification: ON + * + * Local Variables: + * eval: (c-set-style "gnu") + * End: + */ diff --git a/src/uri/vppcom_test.conf b/src/uri/vppcom_test.conf new file mode 100644 index 00000000..e5ac4636 --- /dev/null +++ b/src/uri/vppcom_test.conf @@ -0,0 +1,25 @@ +# Test VPPCOM config file +vppcom { + heapsize 1 + api-prefix daw # this is a comment + uid 1020 this is also a comment. + gid 1020 +# This is yet another comment! + segment-baseva 0x300000000 + segment-size 0x10000000 + segment-size 268435456 + add-segment-size 0x1000000 + add-segment-size 134217728 + preallocated-fifo-pairs 16 + rx-fifo-size 3145728 + rx-fifo-size 0x300000 + tx-fifo-size 3145728 + tx-fifo-size 0x300000 + event-queue-size 1024 + event-queue-size 0x400 + listen-queue-size 32 + listen-queue-size 0x20 + app-timeout 54.3 + session-timeout 66.6 + accept-timeout 0.1 +} |