On Mon, 18 Sept 2023 at 11:14, Tushar Vyavahare
<tushar.vyavahare@xxxxxxxxx> wrote:
>
> Improve the receive_pkt() function to enable it to receive packets from
> multiple sockets. Define a sock_num variable to iterate through all the
> sockets in the Rx path. Add nb_valid_entries to check that all the
> expected number of packets are received.
>
> Revise the function __receive_pkts() to only inspect the receive ring
> once, handle any received packets, and promptly return. Implement a bitmap
> to store the value of MAX_SOCKETS.
>
> Signed-off-by: Tushar Vyavahare <tushar.vyavahare@xxxxxxxxx>
> ---
> tools/testing/selftests/bpf/xskxceiver.c | 276 ++++++++++++++---------
> tools/testing/selftests/bpf/xskxceiver.h | 2 +
> 2 files changed, 171 insertions(+), 107 deletions(-)
>
> diff --git a/tools/testing/selftests/bpf/xskxceiver.c b/tools/testing/selftests/bpf/xskxceiver.c
> index 9f241f503eed..cf3a723cc827 100644
> --- a/tools/testing/selftests/bpf/xskxceiver.c
> +++ b/tools/testing/selftests/bpf/xskxceiver.c
> @@ -80,6 +80,7 @@
> #include <linux/if_ether.h>
> #include <linux/mman.h>
> #include <linux/netdev.h>
> +#include <linux/bitmap.h>
> #include <arpa/inet.h>
> #include <net/if.h>
> #include <locale.h>
> @@ -540,8 +541,10 @@ static int test_spec_set_mtu(struct test_spec *test, int mtu)
>
> static void pkt_stream_reset(struct pkt_stream *pkt_stream)
> {
> - if (pkt_stream)
> + if (pkt_stream) {
> pkt_stream->current_pkt_nb = 0;
> + pkt_stream->nb_rx_pkts = 0;
> + }
> }
>
> static struct pkt *pkt_stream_get_next_tx_pkt(struct pkt_stream *pkt_stream)
> @@ -641,14 +644,16 @@ static u32 pkt_nb_frags(u32 frame_size, struct pkt_stream *pkt_stream, struct pk
> return nb_frags;
> }
>
> -static void pkt_set(struct xsk_umem_info *umem, struct pkt *pkt, int offset, u32 len)
> +static void pkt_set(struct pkt_stream *pkt_stream, struct pkt *pkt, int offset, u32 len)
> {
> pkt->offset = offset;
> pkt->len = len;
> - if (len > MAX_ETH_JUMBO_SIZE)
> + if (len > MAX_ETH_JUMBO_SIZE) {
> pkt->valid = false;
> - else
> + } else {
> pkt->valid = true;
> + pkt_stream->nb_valid_entries++;
> + }
> }
>
> static u32 pkt_get_buffer_len(struct xsk_umem_info *umem, u32 len)
> @@ -670,7 +675,7 @@ static struct pkt_stream *pkt_stream_generate(struct xsk_umem_info *umem, u32 nb
> for (i = 0; i < nb_pkts; i++) {
> struct pkt *pkt = &pkt_stream->pkts[i];
>
> - pkt_set(umem, pkt, 0, pkt_len);
> + pkt_set(pkt_stream, pkt, 0, pkt_len);
> pkt->pkt_nb = i;
> }
>
> @@ -702,7 +707,7 @@ static void __pkt_stream_replace_half(struct ifobject *ifobj, u32 pkt_len,
>
> pkt_stream = pkt_stream_clone(umem, ifobj->xsk->pkt_stream);
> for (i = 1; i < ifobj->xsk->pkt_stream->nb_pkts; i += 2)
> - pkt_set(umem, &pkt_stream->pkts[i], offset, pkt_len);
> + pkt_set(pkt_stream, &pkt_stream->pkts[i], offset, pkt_len);
>
> ifobj->xsk->pkt_stream = pkt_stream;
> }
> @@ -724,6 +729,8 @@ static void pkt_stream_receive_half(struct test_spec *test)
> pkt_stream = test->ifobj_rx->xsk->pkt_stream;
> for (i = 1; i < pkt_stream->nb_pkts; i += 2)
> pkt_stream->pkts[i].valid = false;
> +
> + pkt_stream->nb_valid_entries /= 2;
> }
>
> static u64 pkt_get_addr(struct pkt *pkt, struct xsk_umem_info *umem)
> @@ -797,6 +804,10 @@ static struct pkt_stream *__pkt_stream_generate_custom(struct ifobject *ifobj, s
>
> if (pkt->valid && pkt->len > pkt_stream->max_pkt_len)
> pkt_stream->max_pkt_len = pkt->len;
> +
> + if (pkt->valid)
> + pkt_stream->nb_valid_entries++;
> +
> pkt_nb++;
> }
>
> @@ -1018,133 +1029,179 @@ static int complete_pkts(struct xsk_socket_info *xsk, int batch_size)
> return TEST_PASS;
> }
>
> -static int receive_pkts(struct test_spec *test, struct pollfd *fds)
> +static int __receive_pkts(struct test_spec *test, struct xsk_socket_info *xsk)
> {
> - struct timeval tv_end, tv_now, tv_timeout = {THREAD_TMOUT, 0};
> - struct pkt_stream *pkt_stream = test->ifobj_rx->xsk->pkt_stream;
> - struct xsk_socket_info *xsk = test->ifobj_rx->xsk;
> + u32 frags_processed = 0, nb_frags = 0, pkt_len = 0;
> u32 idx_rx = 0, idx_fq = 0, rcvd, pkts_sent = 0;
> + struct pkt_stream *pkt_stream = xsk->pkt_stream;
> struct ifobject *ifobj = test->ifobj_rx;
> struct xsk_umem_info *umem = xsk->umem;
> + struct pollfd fds = { };
> struct pkt *pkt;
> + u64 first_addr;
> int ret;
>
> - ret = gettimeofday(&tv_now, NULL);
> - if (ret)
> - exit_with_error(errno);
> - timeradd(&tv_now, &tv_timeout, &tv_end);
> -
> - pkt = pkt_stream_get_next_rx_pkt(pkt_stream, &pkts_sent);
> - while (pkt) {
> - u32 frags_processed = 0, nb_frags = 0, pkt_len = 0;
> - u64 first_addr;
> + fds.fd = xsk_socket__fd(xsk->xsk);
> + fds.events = POLLIN;
>
> - ret = gettimeofday(&tv_now, NULL);
> - if (ret)
> - exit_with_error(errno);
> - if (timercmp(&tv_now, &tv_end, >)) {
> - ksft_print_msg("ERROR: [%s] Receive loop timed out\n", __func__);
> - return TEST_FAILURE;
> - }
> + ret = kick_rx(xsk);
> + if (ret)
> + return TEST_FAILURE;
>
> - ret = kick_rx(xsk);
> - if (ret)
> + if (ifobj->use_poll) {
> + ret = poll(&fds, 1, POLL_TMOUT);
> + if (ret < 0)
> return TEST_FAILURE;
>
> - if (ifobj->use_poll) {
> - ret = poll(fds, 1, POLL_TMOUT);
> - if (ret < 0)
> - return TEST_FAILURE;
> -
> - if (!ret) {
> - if (!is_umem_valid(test->ifobj_tx))
> - return TEST_PASS;
> -
> - ksft_print_msg("ERROR: [%s] Poll timed out\n", __func__);
> - return TEST_FAILURE;
> - }
> + if (!ret) {
> + if (!is_umem_valid(test->ifobj_tx))
> + return TEST_PASS;
>
> - if (!(fds->revents & POLLIN))
> - continue;
> + ksft_print_msg("ERROR: [%s] Poll timed out\n", __func__);
> + return TEST_CONTINUE;
> }
>
> - rcvd = xsk_ring_cons__peek(&xsk->rx, BATCH_SIZE, &idx_rx);
> - if (!rcvd)
> - continue;
> + if (!(fds.revents & POLLIN))
> + return TEST_CONTINUE;
> + }
>
> - if (ifobj->use_fill_ring) {
> - ret = xsk_ring_prod__reserve(&umem->fq, rcvd, &idx_fq);
> - while (ret != rcvd) {
> - if (xsk_ring_prod__needs_wakeup(&umem->fq)) {
> - ret = poll(fds, 1, POLL_TMOUT);
> - if (ret < 0)
> - return TEST_FAILURE;
> - }
> - ret = xsk_ring_prod__reserve(&umem->fq, rcvd, &idx_fq);
> + rcvd = xsk_ring_cons__peek(&xsk->rx, BATCH_SIZE, &idx_rx);
> + if (!rcvd)
> + return TEST_CONTINUE;
> +
> + if (ifobj->use_fill_ring) {
> + ret = xsk_ring_prod__reserve(&umem->fq, rcvd, &idx_fq);
> + while (ret != rcvd) {
> + if (xsk_ring_prod__needs_wakeup(&umem->fq)) {
> + ret = poll(&fds, 1, POLL_TMOUT);
> + if (ret < 0)
> + return TEST_FAILURE;
> }
> + ret = xsk_ring_prod__reserve(&umem->fq, rcvd, &idx_fq);
> }
> + }
>
> - while (frags_processed < rcvd) {
> - const struct xdp_desc *desc = xsk_ring_cons__rx_desc(&xsk->rx, idx_rx++);
> - u64 addr = desc->addr, orig;
> + while (frags_processed < rcvd) {
> + const struct xdp_desc *desc = xsk_ring_cons__rx_desc(&xsk->rx, idx_rx++);
> + u64 addr = desc->addr, orig;
>
> - orig = xsk_umem__extract_addr(addr);
> - addr = xsk_umem__add_offset_to_addr(addr);
> + orig = xsk_umem__extract_addr(addr);
> + addr = xsk_umem__add_offset_to_addr(addr);
>
> + if (!nb_frags) {
> + pkt = pkt_stream_get_next_rx_pkt(pkt_stream, &pkts_sent);
> if (!pkt) {
> ksft_print_msg("[%s] received too many packets addr: %lx len %u\n",
> __func__, addr, desc->len);
> return TEST_FAILURE;
> }
> + }
>
> - print_verbose("Rx: addr: %lx len: %u options: %u pkt_nb: %u valid: %u\n",
> - addr, desc->len, desc->options, pkt->pkt_nb, pkt->valid);
> + print_verbose("Rx: addr: %lx len: %u options: %u pkt_nb: %u valid: %u\n",
> + addr, desc->len, desc->options, pkt->pkt_nb, pkt->valid);
>
> - if (!is_frag_valid(umem, addr, desc->len, pkt->pkt_nb, pkt_len) ||
> - !is_offset_correct(umem, pkt, addr) ||
> - (ifobj->use_metadata && !is_metadata_correct(pkt, umem->buffer, addr)))
> - return TEST_FAILURE;
> + if (!is_frag_valid(umem, addr, desc->len, pkt->pkt_nb, pkt_len) ||
> + !is_offset_correct(umem, pkt, addr) || (ifobj->use_metadata &&
> + !is_metadata_correct(pkt, umem->buffer, addr)))
> + return TEST_FAILURE;
>
> - if (!nb_frags++)
> - first_addr = addr;
> - frags_processed++;
> - pkt_len += desc->len;
> - if (ifobj->use_fill_ring)
> - *xsk_ring_prod__fill_addr(&umem->fq, idx_fq++) = orig;
> + if (!nb_frags++)
> + first_addr = addr;
> + frags_processed++;
> + pkt_len += desc->len;
> + if (ifobj->use_fill_ring)
> + *xsk_ring_prod__fill_addr(&umem->fq, idx_fq++) = orig;
>
> - if (pkt_continues(desc->options))
> - continue;
> + if (pkt_continues(desc->options))
> + continue;
>
> - /* The complete packet has been received */
> - if (!is_pkt_valid(pkt, umem->buffer, first_addr, pkt_len) ||
> - !is_offset_correct(umem, pkt, addr))
> - return TEST_FAILURE;
> + /* The complete packet has been received */
> + if (!is_pkt_valid(pkt, umem->buffer, first_addr, pkt_len) ||
> + !is_offset_correct(umem, pkt, addr))
> + return TEST_FAILURE;
>
> - pkt = pkt_stream_get_next_rx_pkt(pkt_stream, &pkts_sent);
> - nb_frags = 0;
> - pkt_len = 0;
> - }
> + pkt_stream->nb_rx_pkts++;
> + nb_frags = 0;
> + pkt_len = 0;
> + }
>
> - if (nb_frags) {
> - /* In the middle of a packet. Start over from beginning of packet. */
> - idx_rx -= nb_frags;
> - xsk_ring_cons__cancel(&xsk->rx, nb_frags);
> - if (ifobj->use_fill_ring) {
> - idx_fq -= nb_frags;
> - xsk_ring_prod__cancel(&umem->fq, nb_frags);
> - }
> - frags_processed -= nb_frags;
> + if (nb_frags) {
> + /* In the middle of a packet. Start over from beginning of packet. */
> + idx_rx -= nb_frags;
> + xsk_ring_cons__cancel(&xsk->rx, nb_frags);
> + if (ifobj->use_fill_ring) {
> + idx_fq -= nb_frags;
> + xsk_ring_prod__cancel(&umem->fq, nb_frags);
> }
> + frags_processed -= nb_frags;
> + }
>
> - if (ifobj->use_fill_ring)
> - xsk_ring_prod__submit(&umem->fq, frags_processed);
> - if (ifobj->release_rx)
> - xsk_ring_cons__release(&xsk->rx, frags_processed);
> + if (ifobj->use_fill_ring)
> + xsk_ring_prod__submit(&umem->fq, frags_processed);
> + if (ifobj->release_rx)
> + xsk_ring_cons__release(&xsk->rx, frags_processed);
> +
> + pthread_mutex_lock(&pacing_mutex);
> + pkts_in_flight -= pkts_sent;
> + pthread_mutex_unlock(&pacing_mutex);
> + pkts_sent = 0;
This patch looks much bigger than it is. You have only removed the
while loop and therefore had to change the indentation of the whole
function. The change looks good.
> +
> +return TEST_CONTINUE;
> +}
> +
> +bool all_packets_received(struct test_spec *test, struct xsk_socket_info *xsk, u32 sock_num,
> + unsigned long *bitmap)
> +{
> + struct pkt_stream *pkt_stream = xsk->pkt_stream;
>
> - pthread_mutex_lock(&pacing_mutex);
> - pkts_in_flight -= pkts_sent;
> - pthread_mutex_unlock(&pacing_mutex);
> - pkts_sent = 0;
> + if (!pkt_stream) {
> + __test_and_set_bit((1 << sock_num), bitmap);
> + return false;
> + }
> +
> + if (pkt_stream->nb_rx_pkts == pkt_stream->nb_valid_entries) {
> + __test_and_set_bit((1 << sock_num), bitmap);
> + if (test_bit(test->nb_sockets, bitmap))
> + return true;
> + }
> +
> + return false;
> +}
> +
> +static int receive_pkts(struct test_spec *test)
> +{
> + struct timeval tv_end, tv_now, tv_timeout = {THREAD_TMOUT, 0};
> + u32 sock_num = 0;
> + int res, ret;
> +
> + DECLARE_BITMAP(bitmap, MAX_SOCKETS);
This is a declaration that should be bunched with the declarations above.
> +
> + ret = gettimeofday(&tv_now, NULL);
> + if (ret)
> + exit_with_error(errno);
> +
> + timeradd(&tv_now, &tv_timeout, &tv_end);
> +
> + while (1) {
> + sock_num = (sock_num + 1) % test->nb_sockets;
> +
> + struct xsk_socket_info *xsk = &test->ifobj_rx->xsk_arr[sock_num];
> +
> + if ((all_packets_received(test, xsk, sock_num, bitmap)))
> + break;
> +
> + res = __receive_pkts(test, xsk);
> + if (!(res == TEST_PASS || res == TEST_CONTINUE))
> + return res;
> +
> + ret = gettimeofday(&tv_now, NULL);
> + if (ret)
> + exit_with_error(errno);
> +
> + if (timercmp(&tv_now, &tv_end, >)) {
> + ksft_print_msg("ERROR: [%s] Receive loop timed out\n", __func__);
> + return TEST_FAILURE;
> + }
> }
>
> return TEST_PASS;
> @@ -1577,7 +1634,6 @@ static void *worker_testapp_validate_rx(void *arg)
> {
> struct test_spec *test = (struct test_spec *)arg;
> struct ifobject *ifobject = test->ifobj_rx;
> - struct pollfd fds = { };
> int err;
>
> if (test->current_step == 1) {
> @@ -1592,12 +1648,9 @@ static void *worker_testapp_validate_rx(void *arg)
> }
> }
>
> - fds.fd = xsk_socket__fd(ifobject->xsk->xsk);
> - fds.events = POLLIN;
> -
> pthread_barrier_wait(&barr);
>
> - err = receive_pkts(test, &fds);
> + err = receive_pkts(test);
>
> if (!err && ifobject->validation_func)
> err = ifobject->validation_func(ifobject);
> @@ -1734,9 +1787,15 @@ static int __testapp_validate_traffic(struct test_spec *test, struct ifobject *i
> pthread_join(t0, NULL);
>
> if (test->total_steps == test->current_step || test->fail) {
> + u32 i;
> +
> if (ifobj2)
> - xsk_socket__delete(ifobj2->xsk->xsk);
> - xsk_socket__delete(ifobj1->xsk->xsk);
> + for (i = 0; i < test->nb_sockets; i++)
> + xsk_socket__delete(ifobj2->xsk_arr[i].xsk);
> +
> + for (i = 0; i < test->nb_sockets; i++)
> + xsk_socket__delete(ifobj1->xsk_arr[i].xsk);
> +
> testapp_clean_xsk_umem(ifobj1);
> if (ifobj2 && !ifobj2->shared_umem)
> testapp_clean_xsk_umem(ifobj2);
> @@ -1812,8 +1871,6 @@ static int swap_xsk_resources(struct ifobject *ifobj_tx, struct ifobject *ifobj_
> {
> int ret;
>
> - xsk_socket__delete(ifobj_tx->xsk->xsk);
> - xsk_socket__delete(ifobj_rx->xsk->xsk);
> ifobj_tx->xsk = &ifobj_tx->xsk_arr[1];
> ifobj_rx->xsk = &ifobj_rx->xsk_arr[1];
>
> @@ -1831,6 +1888,10 @@ static int testapp_xdp_prog_cleanup(struct test_spec *test)
> if (testapp_validate_traffic(test))
> return TEST_FAILURE;
>
> + test->ifobj_tx->xsk_arr[0].pkt_stream = NULL;
> + test->ifobj_rx->xsk_arr[0].pkt_stream = NULL;
> + test->ifobj_tx->xsk_arr[1].pkt_stream = test->tx_pkt_stream_default;
> + test->ifobj_rx->xsk_arr[1].pkt_stream = test->rx_pkt_stream_default;´
Would it make more sense if this was part of the function below?
> if (swap_xsk_resources(test->ifobj_tx, test->ifobj_rx))
> return TEST_FAILURE;
> return testapp_validate_traffic(test);
> @@ -1861,6 +1922,7 @@ static int testapp_stats_tx_invalid_descs(struct test_spec *test)
> {
> pkt_stream_replace_half(test, XSK_UMEM__INVALID_FRAME_SIZE, 0);
> test->ifobj_tx->validation_func = validate_tx_invalid_descs;
> + test->ifobj_rx->xsk->pkt_stream->nb_valid_entries /= 2;
Should be part of pkt_stream_replace_half()
> return testapp_validate_traffic(test);
> }
>
> diff --git a/tools/testing/selftests/bpf/xskxceiver.h b/tools/testing/selftests/bpf/xskxceiver.h
> index e003f9ca4692..aa6cccb862bc 100644
> --- a/tools/testing/selftests/bpf/xskxceiver.h
> +++ b/tools/testing/selftests/bpf/xskxceiver.h
> @@ -108,6 +108,8 @@ struct pkt_stream {
> u32 current_pkt_nb;
> struct pkt *pkts;
> u32 max_pkt_len;
> + u32 nb_rx_pkts;
> + u32 nb_valid_entries;
> bool verbatim;
> };
>
> --
> 2.34.1
>
>
