Cannot reuse rx_streamer after set_time_next_pps

[mmatthebi]

Issue Description

I modified rx_timed_samples to receive multiple bursts in a loop. In each loop, I waited for a PPS signal to arrive and then reset the time zero. I cannot reuse the streamer after the second reuse attempt.

Setup Details

I compile attached source code below in UHD4.2.0.0 on a USRP X410 with

$ g++ rx_timed_samples.cpp -o rx_timed_samples -lboost_system -lboost_program_options -luhd -lpthread

and run it with

$ ./rx_timed_samples --args="addr=localhost" --rate 128e6 --nsamps 6000 --channels="0,1" 

Expected Behavior

I expect all receptions to be successful without timeout.

Actual Behaviour

This is the output of the command:

# ./rx_timed_samples --args="addr=localhost" --rate 128e6 --nsamps 6000 --channels="0,1" 
                                 
Creating the usrp device with: addr=localhost...
[INFO] [UHD] linux; GNU C++ version 9.2.0; Boost_107100; UHD_4.2.0.0-0-g46a70d85
[INFO] [MPMD] Initializing 1 device(s) in parallel with args: mgmt_addr=127.0.0.1,type=x4xx,product=x410,serial=323F754,name=NE-LAB-X
410-01,fpga=X4_200,claimed=False,addr=localhost               
[INFO] [MPM.PeriphManager] init() called with device args `fpga=X4_200,mgmt_addr=127.0.0.1,name=NE-LAB-X410-01,product=x410,clock_sou
rce=internal,time_source=internal'.           
Using Device: Single USRP:
  Device: X400-Series Device
  Mboard 0: x410
  RX Channel: 0
    RX DSP: 0                                                     
    RX Dboard: A                                                  
    RX Subdev: 0                                                  
  RX Channel: 1                                                   
    RX DSP: 1
    RX Dboard: A
    RX Subdev: 1
  RX Channel: 2
    RX DSP: 2                                                     
    RX Dboard: B                                                  
    RX Subdev: 0                                                  
  RX Channel: 3                                                   
    RX DSP: 3
    RX Dboard: B
    RX Subdev: 1
  TX Channel: 0                                           
TX DSP: 0
    TX Dboard: A
    TX Subdev: 0
  TX Channel: 1
    TX DSP: 1
    TX Dboard: A
    TX Subdev: 1
  TX Channel: 2
    TX DSP: 2
    TX Dboard: B
    TX Subdev: 0
  TX Channel: 3
    TX DSP: 3
    TX Dboard: B
    TX Subdev: 1

Setting RX Rate: 128.000000 Msps...
[WARNING] [MULTI_USRP] Could not set RX rate to 128.000 MHz. Actual rate is 122.880 MHz
[WARNING] [MULTI_USRP] Could not set RX rate to 128.000 MHz. Actual rate is 122.880 MHz
[WARNING] [MULTI_USRP] Could not set RX rate to 128.000 MHz. Actual rate is 122.880 MHz
[WARNING] [MULTI_USRP] Could not set RX rate to 128.000 MHz. Actual rate is 122.880 MHz
Actual RX Rate: 122.880000 Msps...

Setting device timestamp to 0...
[WARNING] [0/Radio#0] Attempting to set tick rate to 0. Skipping.

Begin streaming 6000 samples, 1.500000 seconds in the future...
Received packet: 2000 samples, 1 full secs, 0.500000 frac secs
Received packet: 2000 samples, 1 full secs, 0.500016 frac secs
Received packet: 2000 samples, 1 full secs, 0.500033 frac secs

Done!


Begin streaming 6000 samples, 1.500000 seconds in the future...
Received packet: 2000 samples, 1 full secs, 0.500016 frac secs
Received packet: 2000 samples, 1 full secs, 0.500033 frac secs
Receive timeout before all samples received...

Done!


Begin streaming 6000 samples, 1.500000 seconds in the future...
Received packet: 2000 samples, 1 full secs, 0.500016 frac secs
Received packet: 2000 samples, 1 full secs, 0.500033 frac secs
Receive timeout before all samples received...

Done!


Begin streaming 6000 samples, 1.500000 seconds in the future...
Received packet: 2000 samples, 1 full secs, 0.500016 frac secs
Received packet: 2000 samples, 1 full secs, 0.500033 frac secs
Receive timeout before all samples received...

Done!


Begin streaming 6000 samples, 1.500000 seconds in the future...
Received packet: 2000 samples, 1 full secs, 0.500016 frac secs
Received packet: 2000 samples, 1 full secs, 0.500033 frac secs
Receive timeout before all samples received...

Done!


Begin streaming 6000 samples, 1.500000 seconds in the future...
Received packet: 2000 samples, 1 full secs, 0.500016 frac secs
Received packet: 2000 samples, 1 full secs, 0.500033 frac secs
Receive timeout before all samples received...

Done!


Begin streaming 6000 samples, 1.500000 seconds in the future...
Received packet: 2000 samples, 1 full secs, 0.500016 frac secs
Received packet: 2000 samples, 1 full secs, 0.500033 frac secs
Receive timeout before all samples received...

Done!


Begin streaming 6000 samples, 1.500000 seconds in the future...
Received packet: 2000 samples, 1 full secs, 0.500016 frac secs
Received packet: 2000 samples, 1 full secs, 0.500033 frac secs
Receive timeout before all samples received...

Done!


Begin streaming 6000 samples, 1.500000 seconds in the future...
Received packet: 2000 samples, 1 full secs, 0.500016 frac secs
Received packet: 2000 samples, 1 full secs, 0.500033 frac secs
Receive timeout before all samples received...

Done!


Begin streaming 6000 samples, 1.500000 seconds in the future...
Received packet: 2000 samples, 1 full secs, 0.500016 frac secs
Received packet: 2000 samples, 1 full secs, 0.500033 frac secs
Receive timeout before all samples received...

Done!

As visible, only the first loop iteration works as expected. All subsequent loops fail and they return one packet of 2000 samples too few. Further investigations show, that it is the first 2000 samples which are somewhere lost in the abyss.

Steps to reproduce the problem

Used source code:

//
// Copyright 2010-2011 Ettus Research LLC     
// Copyright 2018 Ettus Research, a National Instruments Company                                                                     
//                           
// SPDX-License-Identifier: GPL-3.0-or-later                     
//                                                                                                                                   

#include <uhd/usrp/multi_usrp.hpp>                                                                                                   
#include <uhd/utils/safe_main.hpp>                                                                                                   
#include <uhd/utils/thread.hpp>
#include <boost/algorithm/string.hpp>                         
#include <boost/format.hpp>                                       
#include <boost/program_options.hpp>
#include <complex>                                                                                                                   
#include <iostream>
                                                                  
namespace po = boost::program_options;

int UHD_SAFE_MAIN(int argc, char* argv[])
{                                                                                                                                    
    // variables to be set by po
    std::string args;                                                                                                                
    std::string wire;                                             
    double seconds_in_future;                                                                                                        
    size_t total_num_samps;
    double rate;
    std::string channel_list;
                                                                  
    // setup the program options                                                                                                     
    po::options_description desc("Allowed options");                                                                                 
    // clang-format off                                           
    desc.add_options()                                            
        ("help", "help message")
        ("args", po::value<std::string>(&args)->default_value(""), "single uhd device address args")
        ("wire", po::value<std::string>(&wire)->default_value(""), "the over the wire type, sc16, sc8, etc")
        ("secs", po::value<double>(&seconds_in_future)->default_value(1.5), "number of seconds in the future to receive")
        ("nsamps", po::value<size_t>(&total_num_samps)->default_value(10000), "total number of samples to receive")
        ("rate", po::value<double>(&rate)->default_value(100e6/16), "rate of incoming samples")
        ("dilv", "specify to disable inner-loop verbose")
        ("channels", po::value<std::string>(&channel_list)->default_value("0"), "which channel(s) to use (specify \"0\", \"1\", \"0,1
\", etc)")                                                        
    ;
    // clang-format on  
    po::variables_map vm;
    po::store(po::parse_command_line(argc, argv, desc), vm);
po::notify(vm);

    // print the help message
    if (vm.count("help")) {
        std::cout << boost::format("UHD RX Timed Samples %s") % desc << std::endl;
        return ~0;
    }

    bool verbose = vm.count("dilv") == 0;

    // create a usrp device
    std::cout << std::endl;
    std::cout << boost::format("Creating the usrp device with: %s...") % args
              << std::endl;
    uhd::usrp::multi_usrp::sptr usrp = uhd::usrp::multi_usrp::make(args);
    std::cout << boost::format("Using Device: %s") % usrp->get_pp_string() << std::endl;

    // detect which channels to use
    std::vector<std::string> channel_strings;
    std::vector<size_t> channel_nums;
    boost::split(channel_strings, channel_list, boost::is_any_of("\"',"));
    for (size_t ch = 0; ch < channel_strings.size(); ch++) {
        size_t chan = std::stoi(channel_strings[ch]);
        if (chan >= usrp->get_tx_num_channels() or chan >= usrp->get_rx_num_channels()) {
            throw std::runtime_error("Invalid channel(s) specified.");
        } else
            channel_nums.push_back(std::stoi(channel_strings[ch]));
    }

    // set the rx sample rate
    std::cout << boost::format("Setting RX Rate: %f Msps...") % (rate / 1e6) << std::endl;
    usrp->set_rx_rate(rate);
    std::cout << boost::format("Actual RX Rate: %f Msps...") % (usrp->get_rx_rate() / 1e6)
              << std::endl
              << std::endl;

    std::cout << boost::format("Setting device timestamp to 0...") << std::endl;

    // create a receive streamer
    uhd::stream_args_t stream_args("fc32", wire); // complex floats
    stream_args.channels             = channel_nums;
    uhd::rx_streamer::sptr rx_stream = usrp->get_rx_stream(stream_args);

    for (int i = 0; i < 10; i++) {
    //usrp->set_time_now(uhd::time_spec_t(0.0));
    usrp->set_time_next_pps(uhd::time_spec_t(0.0));
    std::this_thread::sleep_for(std::chrono::milliseconds(1100));

    // setup streaming
    std::cout << std::endl;
    std::cout << boost::format("Begin streaming %u samples, %f seconds in the future...")
                     % total_num_samps % seconds_in_future
              << std::endl;
    uhd::stream_cmd_t stream_cmd(uhd::stream_cmd_t::STREAM_MODE_NUM_SAMPS_AND_DONE);
    stream_cmd.num_samps  = total_num_samps;
    stream_cmd.stream_now = false;
    stream_cmd.time_spec  = uhd::time_spec_t(seconds_in_future);
    rx_stream->issue_stream_cmd(stream_cmd);

    // meta-data will be filled in by recv()
    uhd::rx_metadata_t md;

    // allocate buffer to receive with samples
    std::vector<std::complex<float>> buff(rx_stream->get_max_num_samps());
    std::vector<void*> buffs;
    for (size_t ch = 0; ch < rx_stream->get_num_channels(); ch++)
        buffs.push_back(&buff.front()); // same buffer for each channel

    // the first call to recv() will block this many seconds before receiving
    double timeout = seconds_in_future + 0.1; // timeout (delay before receive + padding)

    size_t num_acc_samps = 0; // number of accumulated samples
    while (num_acc_samps < total_num_samps) {
        // receive a single packet
        size_t num_rx_samps = rx_stream->recv(buffs, buff.size(), md, timeout, true);

        // use a small timeout for subsequent packets
        timeout = 0.1;

        // handle the error code
        if (md.error_code == uhd::rx_metadata_t::ERROR_CODE_TIMEOUT)
            break;
        if (md.error_code != uhd::rx_metadata_t::ERROR_CODE_NONE) {
            throw std::runtime_error(
                str(boost::format("Receiver error %s") % md.strerror()));
        }

        if (verbose)
            std::cout << boost::format(
                             "Received packet: %u samples, %u full secs, %f frac secs")
                             % num_rx_samps % md.time_spec.get_full_secs()
                             % md.time_spec.get_frac_secs()
                      << std::endl;

        num_acc_samps += num_rx_samps;
    }

    if (num_acc_samps < total_num_samps)
        std::cerr << "Receive timeout before all samples received..." << std::endl;

    // finished
    std::cout << std::endl << "Done!" << std::endl << std::endl;
    }
    return EXIT_SUCCESS;
}