transport: Implement eov indications for Rx and Tx streams

3 files changed, 384 insertions, 0 deletions

diff --git a/host/tests/rx_streamer_test.cpp b/host/tests/rx_streamer_test.cpp
index 1b1311908..953b82028 100644
--- a/host/tests/rx_streamer_test.cpp
+++ b/host/tests/rx_streamer_test.cpp
@@ -18,6 +18,7 @@ namespace uhd { namespace transport {
 struct mock_header_t
 {
     bool eob             = false;
+    bool eov             = false;
     bool has_tsf         = false;
     uint64_t tsf         = 0;
     size_t payload_bytes = 0;
@@ -39,6 +40,7 @@ public:
     struct packet_info_t
     {
         bool eob             = false;
+        bool eov             = false;
         bool has_tsf         = false;
         uint64_t tsf         = 0;
         size_t payload_bytes = 0;
@@ -55,6 +57,7 @@ public:
 
         packet_info_t info;
         info.eob           = header.eob;
+        info.eov           = header.eov;
         info.has_tsf       = header.has_tsf;
         info.tsf           = header.tsf;
         info.payload_bytes = header.payload_bytes;
@@ -742,3 +745,109 @@ BOOST_AUTO_TEST_CASE(test_recv_alignment_error)
     BOOST_CHECK_EQUAL(num_samps_ret, 0);
     BOOST_CHECK_EQUAL(metadata.error_code, uhd::rx_metadata_t::ERROR_CODE_ALIGNMENT);
 }
+
+BOOST_AUTO_TEST_CASE(test_recv_one_channel_one_eov)
+{
+    const size_t NUM_PACKETS = 5;
+    const std::string format("fc64");
+
+    auto recv_links = make_links(1);
+    auto streamer   = make_rx_streamer(recv_links, format);
+
+    const size_t spp       = streamer->get_max_num_samps();
+    const size_t num_samps = spp * NUM_PACKETS;
+    std::vector<std::complex<double>> buff(num_samps);
+
+    for (size_t i = 0; i < NUM_PACKETS; i++) {
+        mock_header_t header;
+        header.eob     = false;
+        header.has_tsf = true;
+        header.tsf     = i;
+
+        for (size_t j = 0; j < NUM_PACKETS; j++) {
+            header.eov = (i == j);
+            push_back_recv_packet(recv_links[0], header, spp);
+        }
+
+        uhd::rx_metadata_t metadata;
+        // Create a vector with storage for two EOVs even though we expect
+        // only one, since filling the EOV vector results in an early
+        // termination of `recv()` (which we don't want here).
+        std::vector<size_t> eov_positions(2);
+        metadata.eov_positions = eov_positions.data();
+        metadata.eov_positions_size = eov_positions.size();
+
+        std::cout << "receiving packet " << i << std::endl;
+
+        size_t num_samps_ret =
+            streamer->recv(buff.data(), buff.size(), metadata, 1.0, false);
+
+        BOOST_CHECK_EQUAL(num_samps_ret, num_samps);
+        BOOST_CHECK_EQUAL(metadata.eov_positions, eov_positions.data());
+        BOOST_CHECK_EQUAL(metadata.eov_positions_size, eov_positions.size());
+        BOOST_CHECK_EQUAL(metadata.eov_positions_count, 1);
+        BOOST_CHECK_EQUAL(eov_positions[0], (i + 1) * spp);
+    }
+}
+
+BOOST_AUTO_TEST_CASE(test_recv_two_channel_aggregate_eov)
+{
+    const size_t NUM_PACKETS = 20;
+    const std::string format("fc64");
+
+    // This vector defines which packets in each channel's mock link will
+    // signal EOV in their packet headers.
+    //
+    // For example, for a vector with 3 values, [3, 5, 8]:
+    //   Link 0 packets with EOV: 3rd, 6th, 9th, 12th, 15th, ...
+    //   Link 1 packets with EOV: 5th, 10th, 15th, 20th, ...
+    //   Link 2 packets with EOV: 8th, 16th, 24th, 32nd, ...
+    const std::vector<size_t> eov_every_nth_packet{3, 5};
+
+    const size_t num_chans = eov_every_nth_packet.size();
+    auto recv_links = make_links(num_chans);
+    auto streamer   = make_rx_streamer(recv_links, format);
+
+    const size_t spp       = streamer->get_max_num_samps();
+    const size_t num_samps = spp * NUM_PACKETS;
+
+    std::vector<std::vector<std::complex<double>>> buffer(num_chans);
+    std::vector<void*> buffers;
+    for (size_t i = 0; i < num_chans; i++) {
+        buffer[i].resize(num_samps);
+        buffers.push_back(&buffer[i].front());
+    }
+
+    mock_header_t header;
+    std::vector<size_t> expected_eov_offsets;
+    for (size_t i = 0; i < NUM_PACKETS; i++) {
+        bool eov = false;
+        for (size_t ch = 0; ch < num_chans; ch++) {
+            header.eob     = false;
+            header.has_tsf = false;
+            header.eov = ((i + 1) % eov_every_nth_packet[ch]) == 0;
+
+            push_back_recv_packet(recv_links[ch], header, spp);
+
+            eov |= header.eov;
+        }
+        if(eov) {
+            expected_eov_offsets.push_back(spp * (i + 1));
+        }
+    }
+
+    uhd::rx_metadata_t metadata;
+
+    std::vector<size_t> eov_positions(expected_eov_offsets.size() + 1);
+    metadata.eov_positions = eov_positions.data();
+    metadata.eov_positions_size = eov_positions.size();
+
+    size_t num_samps_ret =
+        streamer->recv(buffers, num_samps, metadata, 1.0, false);
+
+    BOOST_CHECK_EQUAL(num_samps_ret, num_samps);
+    BOOST_CHECK_EQUAL(metadata.eov_positions_count, expected_eov_offsets.size());
+    for(size_t i = 0; i < metadata.eov_positions_count; i++) {
+        BOOST_CHECK_EQUAL(expected_eov_offsets[i], metadata.eov_positions[i]);
+    }
+}
diff --git a/host/tests/streamer_benchmark.cpp b/host/tests/streamer_benchmark.cpp
index 02aa102a0..0c11441ed 100644
--- a/host/tests/streamer_benchmark.cpp
+++ b/host/tests/streamer_benchmark.cpp
@@ -47,6 +47,7 @@ public:
     struct packet_info_t
     {
         bool eob             = false;
+        bool eov             = false;
         bool has_tsf         = false;
         uint64_t tsf         = 0;
         size_t payload_bytes = 0;
@@ -108,6 +109,7 @@ public:
     struct packet_info_t
     {
         bool eob             = false;
+        bool eov             = false;
         bool has_tsf         = false;
         uint64_t tsf         = 0;
         size_t payload_bytes = 0;
diff --git a/host/tests/tx_streamer_test.cpp b/host/tests/tx_streamer_test.cpp
index 09c8357df..2288bd029 100644
--- a/host/tests/tx_streamer_test.cpp
+++ b/host/tests/tx_streamer_test.cpp
@@ -26,6 +26,7 @@ public:
     struct packet_info_t
     {
         bool eob             = false;
+        bool eov             = false;
         bool has_tsf         = false;
         uint64_t tsf         = 0;
         size_t payload_bytes = 0;
@@ -158,6 +159,34 @@ pop_send_packet(mock_send_link::sptr send_link)
     return std::make_tuple(info, data, packet_samps, packet.first);
 }
 
+//! Generates a non-biased random number in the range (low, high).
+static size_t generate_rand(size_t low, size_t high)
+{
+    return (std::rand() % (high - low + 1)) + low;
+}
+
+/*!
+ * Generates a vector of legal random EOV positions.
+ * `eovs` is a vector already sized to the desired number of EOV positions.
+ * The range [1, num_samps) will be broken into N=`eov.size()`
+ * non-overlapping adjacent ranges, and a random value within each range will
+ * be generated and stored in the vector.
+ */
+static void generate_random_eov_positions(
+    std::vector<size_t>& eovs, const size_t num_samps)
+{
+    UHD_ASSERT_THROW(!eovs.empty());
+
+    const size_t num_eovs   = eovs.size();
+    const size_t range_size = (num_samps - 1) / num_eovs;
+    size_t low              = 1;
+    for (size_t i = 0; i < num_eovs; i++) {
+        const size_t high = low + range_size - 1;
+        eovs[i]           = generate_rand(low, high);
+        low               = high + 1;
+    }
+}
+
 /*!
  * Tests
  */
@@ -216,6 +245,250 @@ BOOST_AUTO_TEST_CASE(test_send_one_channel_one_packet)
     }
 }
 
+BOOST_AUTO_TEST_CASE(test_send_one_channel_eov_lte_spp)
+{
+    const size_t NUM_PKTS_TO_TEST = 30;
+    const std::string format("fc32");
+
+    auto send_links = make_links(1);
+    auto streamer   = make_tx_streamer(send_links, format);
+
+    // Allocate metadata
+    uhd::tx_metadata_t metadata;
+    metadata.has_time_spec = true;
+    metadata.time_spec     = uhd::time_spec_t(0.0);
+
+    // Allocate buffer
+    const size_t num_samps = streamer->get_max_num_samps();
+    std::vector<std::complex<float>> buff(num_samps);
+
+    // Send buffer and check resultant packets
+    size_t num_accum_samps = 0;
+
+    for (size_t i = 0; i < NUM_PKTS_TO_TEST; i++) {
+        std::cout << "sending packet " << i << std::endl;
+
+        // Vary number of EOVs for each send
+        const size_t num_eovs = (i % 10) + 1;
+        std::vector<size_t> eovs(num_eovs);
+        generate_random_eov_positions(eovs, num_samps);
+
+        metadata.eov_positions      = eovs.data();
+        metadata.eov_positions_size = eovs.size();
+
+        const size_t num_sent = streamer->send(&buff.front(), num_samps, metadata, 1.0);
+        BOOST_CHECK_EQUAL(num_sent, num_samps);
+        metadata.time_spec += uhd::time_spec_t(0, num_sent, SAMP_RATE);
+
+        mock_tx_data_xport::packet_info_t info;
+        std::complex<uint16_t>* data;
+        size_t packet_samps;
+        boost::shared_array<uint8_t> frame_buff;
+
+        // Check number of packets written (should be # of EOVs plus one)
+        size_t num_packets_written = send_links[0]->get_num_packets();
+        BOOST_CHECK_EQUAL(num_packets_written, eovs.size() + 1);
+
+        // Pop each packets and check size of each relative to EOV positions
+        size_t total_samps_popped = 0;
+        size_t eov_index          = 0;
+        size_t last_eov_position  = 0;
+        while (total_samps_popped < num_samps) {
+            std::tie(info, data, packet_samps, frame_buff) =
+                pop_send_packet(send_links[0]);
+
+            // All but the last packet should have an EOV indication
+            if (eov_index < eovs.size()) {
+                BOOST_CHECK_EQUAL(eovs[eov_index] - last_eov_position, packet_samps);
+                BOOST_CHECK(info.eov);
+                last_eov_position = eovs[eov_index];
+            } else {
+                BOOST_CHECK_EQUAL(num_samps - last_eov_position, packet_samps);
+                BOOST_CHECK(not info.eov);
+            }
+
+            // Verify correctness of TSF data
+            BOOST_CHECK(info.has_tsf);
+            BOOST_CHECK_EQUAL(info.tsf, num_accum_samps * TICK_RATE / SAMP_RATE);
+
+            eov_index++;
+            num_accum_samps += packet_samps;
+            total_samps_popped += packet_samps;
+        }
+    }
+}
+
+BOOST_AUTO_TEST_CASE(test_send_one_channel_eov_gt_spp)
+{
+    const size_t NUM_PKTS_TO_TEST = 30;
+    const std::string format("fc32");
+
+    auto send_links = make_links(1);
+    auto streamer   = make_tx_streamer(send_links, format);
+
+    // Allocate metadata
+    uhd::tx_metadata_t metadata;
+    metadata.has_time_spec = true;
+    metadata.time_spec     = uhd::time_spec_t(0.0);
+
+    // Allocate buffer
+    const size_t spp       = streamer->get_max_num_samps();
+    const size_t num_samps = spp * 50;
+    std::vector<std::complex<float>> buff(num_samps);
+
+    // Send buffer and check resultant packets
+    size_t num_accum_samps = 0;
+
+    for (size_t i = 0; i < NUM_PKTS_TO_TEST; i++) {
+        std::cout << "sending packet " << i << std::endl;
+
+        // Vary number of EOVs for each send
+        const size_t num_eovs = (i % 5) + 1;
+        std::vector<size_t> eovs(num_eovs);
+        generate_random_eov_positions(eovs, num_samps);
+
+        metadata.eov_positions      = eovs.data();
+        metadata.eov_positions_size = eovs.size();
+
+        const size_t num_sent = streamer->send(&buff.front(), num_samps, metadata, 1.0);
+        BOOST_CHECK_EQUAL(num_sent, num_samps);
+        metadata.time_spec += uhd::time_spec_t(0, num_sent, SAMP_RATE);
+
+        mock_tx_data_xport::packet_info_t info;
+        std::complex<uint16_t>* data;
+        size_t packet_samps;
+        boost::shared_array<uint8_t> frame_buff;
+
+        size_t total_samps_popped   = 0;
+        size_t eov_index            = 0;
+        size_t last_eov_position    = 0;
+        size_t distance_to_next_eov = eovs[eov_index] - last_eov_position;
+        size_t distance_to_end      = num_samps;
+        while (total_samps_popped < num_samps) {
+            std::tie(info, data, packet_samps, frame_buff) =
+                pop_send_packet(send_links[0]);
+
+            if (distance_to_next_eov <= spp) {
+                // Next EOV is within a single SPP: ensure packet is EOV
+                BOOST_CHECK_EQUAL(distance_to_next_eov, packet_samps);
+                BOOST_CHECK(info.eov);
+                last_eov_position = eovs[eov_index];
+                eov_index++;
+                if (eov_index < eovs.size()) {
+                    // NOTE: Addition of `packet_samps` is to compensate for
+                    // its subtraction below
+                    distance_to_next_eov =
+                        eovs[eov_index] - last_eov_position + packet_samps;
+                } else {
+                    // No more EOVs
+                    distance_to_next_eov = std::numeric_limits<size_t>::max();
+                }
+            } else if (distance_to_end <= spp) {
+                // End of data within a single SPP
+                BOOST_CHECK_EQUAL(distance_to_end, packet_samps);
+                BOOST_CHECK(not info.eov);
+            } else {
+                BOOST_CHECK(not info.eov);
+            }
+
+            // Verify correctness of TSF data
+            BOOST_CHECK(info.has_tsf);
+            BOOST_CHECK_EQUAL(info.tsf, num_accum_samps * TICK_RATE / SAMP_RATE);
+
+            distance_to_end -= packet_samps;
+            distance_to_next_eov -= packet_samps;
+            total_samps_popped += packet_samps;
+            num_accum_samps += packet_samps;
+        }
+    }
+}
+
+BOOST_AUTO_TEST_CASE(test_send_one_channel_eov_corner_case)
+{
+    const std::string format("fc32");
+
+    auto send_links = make_links(1);
+    auto streamer   = make_tx_streamer(send_links, format);
+
+    // Allocate metadata
+    uhd::tx_metadata_t metadata;
+    metadata.has_time_spec = true;
+    metadata.time_spec     = uhd::time_spec_t(0.0);
+
+    // Allocate buffer
+    const size_t spp       = streamer->get_max_num_samps();
+    const size_t num_samps = spp * 2;
+    std::vector<std::complex<float>> buff(num_samps);
+
+    // Mark every sample as an EOV :D
+    std::vector<size_t> eovs(num_samps);
+    for (size_t i = 0; i < num_samps; i++) {
+        eovs[i] = i + 1;
+    }
+
+    metadata.eov_positions      = eovs.data();
+    metadata.eov_positions_size = eovs.size();
+
+    const size_t num_sent = streamer->send(&buff.front(), num_samps, metadata, 1.0);
+    BOOST_CHECK_EQUAL(num_sent, num_samps);
+
+    mock_tx_data_xport::packet_info_t info;
+    std::complex<uint16_t>* data;
+    size_t packet_samps;
+    boost::shared_array<uint8_t> frame_buff;
+
+    // Check all packets for EOV
+    BOOST_CHECK_EQUAL(send_links[0]->get_num_packets(), num_samps);
+
+    for (size_t i = 0; i < num_samps; i++) {
+        std::tie(info, data, packet_samps, frame_buff) = pop_send_packet(send_links[0]);
+        BOOST_CHECK_EQUAL(packet_samps, 1);
+        BOOST_CHECK(info.eov);
+    }
+}
+
+BOOST_AUTO_TEST_CASE(test_send_one_channel_eov_error_cases)
+{
+    const std::string format("fc32");
+
+    auto send_links = make_links(1);
+    auto streamer   = make_tx_streamer(send_links, format);
+
+    uhd::tx_metadata_t metadata;
+    metadata.has_time_spec = true;
+    metadata.time_spec     = uhd::time_spec_t(0.0);
+
+    const size_t spp       = streamer->get_max_num_samps();
+    const size_t num_samps = spp;
+    std::vector<std::complex<float>> buff(num_samps);
+
+    // Error case: EOV of 0
+    size_t eov                  = 0;
+    metadata.eov_positions      = &eov;
+    metadata.eov_positions_size = 1;
+
+    BOOST_CHECK_THROW(
+        streamer->send(&buff.front(), num_samps, metadata, 1.0), uhd::value_error);
+
+    // Error case: Adjacent EOV values that are the same
+    std::vector<size_t> eovs{100, 100};
+    metadata.eov_positions      = eovs.data();
+    metadata.eov_positions_size = eovs.size();
+
+    BOOST_CHECK_THROW(
+        streamer->send(&buff.front(), num_samps, metadata, 1.0), uhd::value_error);
+
+    // Error case: EOV values not monotonically increasing
+    eovs = {50, 25};
+
+    BOOST_CHECK_THROW(
+        streamer->send(&buff.front(), num_samps, metadata, 1.0), uhd::value_error);
+
+    // Error case: EOV values greater than nsamps_per_buff
+    BOOST_CHECK_THROW(
+        streamer->send(&buff.front(), 1, metadata, 1.0), uhd::value_error);
+}
+
 BOOST_AUTO_TEST_CASE(test_send_one_channel_multi_packet)
 {
     const size_t NUM_BUFFS_TO_TEST = 5;