diff options
Diffstat (limited to 'host')
| -rw-r--r-- | host/docs/transport.rst | 11 | ||||
| -rw-r--r-- | host/docs/usrp2.rst | 15 | ||||
| -rw-r--r-- | host/examples/test_async_messages.cpp | 104 | ||||
| -rw-r--r-- | host/include/uhd/transport/udp_simple.hpp | 4 | ||||
| -rw-r--r-- | host/include/uhd/types/metadata.hpp | 4 | ||||
| -rw-r--r-- | host/lib/transport/udp_simple.cpp | 44 | ||||
| -rw-r--r-- | host/lib/transport/udp_zero_copy_asio.cpp | 31 | ||||
| -rw-r--r-- | host/lib/transport/vrt_packet_handler.hpp | 16 | ||||
| -rw-r--r-- | host/lib/usrp/usrp2/fw_common.h | 4 | ||||
| -rw-r--r-- | host/lib/usrp/usrp2/io_impl.cpp | 169 | ||||
| -rw-r--r-- | host/lib/usrp/usrp2/mboard_impl.cpp | 40 | ||||
| -rw-r--r-- | host/lib/usrp/usrp2/usrp2_iface.cpp | 14 | ||||
| -rw-r--r-- | host/lib/usrp/usrp2/usrp2_impl.cpp | 18 | ||||
| -rw-r--r-- | host/lib/usrp/usrp2/usrp2_impl.hpp | 16 | ||||
| -rw-r--r-- | host/lib/usrp/usrp2/usrp2_regs.hpp | 5 | 
15 files changed, 350 insertions, 145 deletions
| diff --git a/host/docs/transport.rst b/host/docs/transport.rst index 432db4bb5..2f730f8e4 100644 --- a/host/docs/transport.rst +++ b/host/docs/transport.rst @@ -40,6 +40,17 @@ The following parameters can be used to alter the transport's default behavior:  as the asynchronous send implementation is currently disabled.  ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ +Flow control parameters +^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ +The host-based flow control expects periodic update packets from the device. +These update packets inform the host of the last packet consumed by the device, +which allows the host to determine throttling conditions for the transmission of packets. +The following mechanisms affect the transmission of periodic update packets: + +* **ups_per_fifo:** The number of update packets for each FIFO's worth of bytes sent into the device +* **ups_per_sec:** The number of update packets per second + +^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^  Resize socket buffers  ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^  It may be useful increase the size of the socket buffers to diff --git a/host/docs/usrp2.rst b/host/docs/usrp2.rst index 1ebab388a..8fa666a49 100644 --- a/host/docs/usrp2.rst +++ b/host/docs/usrp2.rst @@ -39,12 +39,10 @@ Use the card burner tool (windows)  ------------------------------------------------------------------------  Setup networking  ------------------------------------------------------------------------ -The USRP2 only supports gigabit ethernet, and -will not work with a 10/100 Mbps interface. -Because the USRP2 uses gigabit ethernet pause frames for flow control, -you cannot use multiple USRP2s with a switch or a hub. -It is recommended that each USRP2 be plugged directly into its own -dedicated gigabit ethernet interface on the host computer. +The USRP2 only supports gigabit ethernet, +and will not work with a 10/100 Mbps interface. +However, a 10/100 Mbps interface can be connected indirectly +to a USRP2 through a gigabit ethernet switch.  ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^  Setup the host interface @@ -63,8 +61,9 @@ It is recommended that you change or disable your firewall settings.  ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^  Multiple device configuration  ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ -As described above, you will need one ethernet interface per USRP2. -Each ethernet interface should have its own subnet, +For maximum throughput, one ethernet interface per USRP2 is recommended, +although multiple devices may be connected via a gigabit ethernet switch. +In any case, each ethernet interface should have its own subnet,  and the corresponding USRP2 device should be assigned an address in that subnet.  Example: diff --git a/host/examples/test_async_messages.cpp b/host/examples/test_async_messages.cpp index e4a996ef5..61db7ec04 100644 --- a/host/examples/test_async_messages.cpp +++ b/host/examples/test_async_messages.cpp @@ -19,21 +19,25 @@  #include <uhd/utils/safe_main.hpp>  #include <uhd/utils/static.hpp>  #include <uhd/usrp/single_usrp.hpp> +#include <boost/assign/list_of.hpp>  #include <boost/program_options.hpp> +#include <boost/foreach.hpp> +#include <boost/bind.hpp>  #include <boost/format.hpp> +#include <cstdlib>  #include <complex>  #include <iostream>  namespace po = boost::program_options;  /*! - * Test that no messages are received: + * Test the eob ack message:   *    Send a burst of many samples that will fragment internally. - *    We expect to not get any async messages. + *    We expect to get an eob ack async message.   */ -void test_no_async_message(uhd::usrp::single_usrp::sptr sdev){ +bool test_eob_ack_message(uhd::usrp::single_usrp::sptr sdev){      uhd::device::sptr dev = sdev->get_device(); -    std::cout << "Test no async message... " << std::flush; +    std::cout << "Test eob ack message... " << std::flush;      uhd::tx_metadata_t md;      md.start_of_burst = true; @@ -50,19 +54,28 @@ void test_no_async_message(uhd::usrp::single_usrp::sptr sdev){      );      uhd::async_metadata_t async_md; -    if (dev->recv_async_msg(async_md)){ +    if (not dev->recv_async_msg(async_md)){          std::cout << boost::format(              "failed:\n" -            "    Got unexpected event code 0x%x.\n" -        ) % async_md.event_code << std::endl; -        //clear the async messages -        while (dev->recv_async_msg(async_md, 0)){}; +            "    Async message recv timed out.\n" +        ) << std::endl; +        return false;      } -    else{ + +    switch(async_md.event_code){ +    case uhd::async_metadata_t::EVENT_CODE_EOB_ACK:          std::cout << boost::format(              "success:\n" -            "    Did not get an async message.\n" +            "    Got event code eob ack message.\n"          ) << std::endl; +        return true; + +    default: +        std::cout << boost::format( +            "failed:\n" +            "    Got unexpected event code 0x%x.\n" +        ) % async_md.event_code << std::endl; +        return false;      }  } @@ -71,7 +84,7 @@ void test_no_async_message(uhd::usrp::single_usrp::sptr sdev){   *    Send a start of burst packet with no following end of burst.   *    We expect to get an underflow(within a burst) async message.   */ -void test_underflow_message(uhd::usrp::single_usrp::sptr sdev){ +bool test_underflow_message(uhd::usrp::single_usrp::sptr sdev){      uhd::device::sptr dev = sdev->get_device();      std::cout << "Test underflow message... " << std::flush; @@ -80,18 +93,19 @@ void test_underflow_message(uhd::usrp::single_usrp::sptr sdev){      md.end_of_burst   = false;      md.has_time_spec  = false; -    dev->send(NULL, 0, md, +    dev->send( +        NULL, 0, md,          uhd::io_type_t::COMPLEX_FLOAT32,          uhd::device::SEND_MODE_FULL_BUFF      );      uhd::async_metadata_t async_md; -    if (not dev->recv_async_msg(async_md)){ +    if (not dev->recv_async_msg(async_md, 1)){          std::cout << boost::format(              "failed:\n"              "    Async message recv timed out.\n"          ) << std::endl; -        return; +        return false;      }      switch(async_md.event_code){ @@ -100,13 +114,14 @@ void test_underflow_message(uhd::usrp::single_usrp::sptr sdev){              "success:\n"              "    Got event code underflow message.\n"          ) << std::endl; -        break; +        return true;      default:          std::cout << boost::format(              "failed:\n"              "    Got unexpected event code 0x%x.\n"          ) % async_md.event_code << std::endl; +        return false;      }  } @@ -115,7 +130,7 @@ void test_underflow_message(uhd::usrp::single_usrp::sptr sdev){   *    Send a burst packet that occurs at a time in the past.   *    We expect to get a time error async message.   */ -void test_time_error_message(uhd::usrp::single_usrp::sptr sdev){ +bool test_time_error_message(uhd::usrp::single_usrp::sptr sdev){      uhd::device::sptr dev = sdev->get_device();      std::cout << "Test time error message... " << std::flush; @@ -127,7 +142,8 @@ void test_time_error_message(uhd::usrp::single_usrp::sptr sdev){      sdev->set_time_now(uhd::time_spec_t(200.0)); //time at 200s -    dev->send(NULL, 0, md, +    dev->send( +        NULL, 0, md,          uhd::io_type_t::COMPLEX_FLOAT32,          uhd::device::SEND_MODE_FULL_BUFF      ); @@ -138,7 +154,7 @@ void test_time_error_message(uhd::usrp::single_usrp::sptr sdev){              "failed:\n"              "    Async message recv timed out.\n"          ) << std::endl; -        return; +        return false;      }      switch(async_md.event_code){ @@ -147,29 +163,38 @@ void test_time_error_message(uhd::usrp::single_usrp::sptr sdev){              "success:\n"              "    Got event code time error message.\n"          ) << std::endl; -        break; +        return true;      default:          std::cout << boost::format(              "failed:\n"              "    Got unexpected event code 0x%x.\n"          ) % async_md.event_code << std::endl; +        return false;      }  } +void flush_async_md(uhd::usrp::single_usrp::sptr sdev){ +    uhd::device::sptr dev = sdev->get_device(); +    uhd::async_metadata_t async_md; +    while (dev->recv_async_msg(async_md, 1.0)){} +} +  int UHD_SAFE_MAIN(int argc, char *argv[]){      uhd::set_thread_priority_safe();      //variables to be set by po      std::string args;      double rate; +    size_t ntests;      //setup the program options      po::options_description desc("Allowed options");      desc.add_options()          ("help", "help message") -        ("args", po::value<std::string>(&args)->default_value(""), "single uhd device address args") -        ("rate", po::value<double>(&rate)->default_value(1.5e6), "rate of outgoing samples") +        ("args",   po::value<std::string>(&args)->default_value(""), "single uhd device address args") +        ("rate",   po::value<double>(&rate)->default_value(1.5e6),   "rate of outgoing samples") +        ("ntests", po::value<size_t>(&ntests)->default_value(10),    "number of tests to run")      ;      po::variables_map vm;      po::store(po::parse_command_line(argc, argv, desc), vm); @@ -195,9 +220,38 @@ int UHD_SAFE_MAIN(int argc, char *argv[]){      //------------------------------------------------------------------      // begin asyc messages test      //------------------------------------------------------------------ -    test_no_async_message(sdev); -    test_underflow_message(sdev); -    test_time_error_message(sdev); +    static const uhd::dict<std::string, boost::function<bool(uhd::usrp::single_usrp::sptr)> > +        tests = boost::assign::map_list_of +        ("Test EOB ACK   ",    &test_eob_ack_message) +        ("Test Underflow ",  &test_underflow_message) +        ("Test Time Error", &test_time_error_message) +    ; + +    //init result counts +    uhd::dict<std::string, size_t> failures, successes; +    BOOST_FOREACH(const std::string &key, tests.keys()){ +        failures[key] = 0; +        successes[key] = 0; +    } + +    //run the tests, pick at random +    for (size_t n = 0; n < ntests; n++){ +        std::string key = tests.keys()[std::rand() % tests.size()]; +        bool pass = tests[key](sdev); +        flush_async_md(sdev); + +        //store result +        if (pass) successes[key]++; +        else      failures[key]++; +    } + +    //print the result summary +    std::cout << std::endl << "Summary:" << std::endl << std::endl; +    BOOST_FOREACH(const std::string &key, tests.keys()){ +        std::cout << boost::format( +            "%s   ->   %3d successes, %3d failures" +        ) % key % successes[key] % failures[key] << std::endl; +    }      //finished      std::cout << std::endl << "Done!" << std::endl << std::endl; diff --git a/host/include/uhd/transport/udp_simple.hpp b/host/include/uhd/transport/udp_simple.hpp index c84393ecf..83f895ba9 100644 --- a/host/include/uhd/transport/udp_simple.hpp +++ b/host/include/uhd/transport/udp_simple.hpp @@ -73,10 +73,10 @@ public:       * Receive into the provided buffer.       * Blocks until data is received or a timeout occurs.       * \param buff a mutable buffer to receive into -     * \param timeout_ms the timeout in milliseconds +     * \param timeout the timeout in seconds       * \return the number of bytes received or zero on timeout       */ -    virtual size_t recv(const boost::asio::mutable_buffer &buff, size_t timeout_ms) = 0; +    virtual size_t recv(const boost::asio::mutable_buffer &buff, double timeout = 0.1) = 0;  };  }} //namespace diff --git a/host/include/uhd/types/metadata.hpp b/host/include/uhd/types/metadata.hpp index 65952941c..96c4ad0d3 100644 --- a/host/include/uhd/types/metadata.hpp +++ b/host/include/uhd/types/metadata.hpp @@ -130,7 +130,7 @@ namespace uhd{          /*!           * Event codes: -         * - success: a packet was successfully transmitted +         * - eob ack: an eob packet was successfully transmitted           * - underflow: an internal send buffer has emptied           * - sequence error: packet loss between host and device           * - time error: packet had time that was late (or too early) @@ -138,7 +138,7 @@ namespace uhd{           * - sequence error in burst: packet loss within a burst           */          enum event_code_t { -            EVENT_CODE_SUCCESS    = 0x1, +            EVENT_CODE_EOB_ACK    = 0x1,              EVENT_CODE_UNDERFLOW  = 0x2,              EVENT_CODE_SEQ_ERROR  = 0x4,              EVENT_CODE_TIME_ERROR = 0x8, diff --git a/host/lib/transport/udp_simple.cpp b/host/lib/transport/udp_simple.cpp index 89750f99d..5829b462b 100644 --- a/host/lib/transport/udp_simple.cpp +++ b/host/lib/transport/udp_simple.cpp @@ -27,23 +27,25 @@ using namespace uhd::transport;   * Helper Functions   **********************************************************************/  /*! - * A receive timeout for a socket: - * - * It seems that asio cannot have timeouts with synchronous io. - * However, we can implement a polling loop that will timeout. - * This is okay bacause this is the slow-path implementation. - * + * Wait for available data or timeout.   * \param socket the asio socket - * \param timeout_ms the timeout in milliseconds + * \param timeout the timeout in seconds + * \return false for timeout, true for data   */ -static void reasonable_recv_timeout( -    boost::asio::ip::udp::socket &socket, size_t timeout_ms +static bool wait_available( +    boost::asio::ip::udp::socket &socket, double timeout  ){ -    boost::asio::deadline_timer timer(socket.get_io_service()); -    timer.expires_from_now(boost::posix_time::milliseconds(timeout_ms)); -    while (not (socket.available() or timer.expires_from_now().is_negative())){ -        boost::this_thread::sleep(boost::posix_time::milliseconds(1)); -    } +    //setup timeval for timeout +    timeval tv; +    tv.tv_sec = 0; +    tv.tv_usec = long(timeout*1e6); + +    //setup rset for timeout +    fd_set rset; +    FD_ZERO(&rset); +    FD_SET(socket.native(), &rset); + +    return ::select(socket.native()+1, &rset, NULL, NULL, &tv) > 0;  }  /*********************************************************************** @@ -57,7 +59,7 @@ public:      //send/recv      size_t send(const boost::asio::const_buffer &); -    size_t recv(const boost::asio::mutable_buffer &, size_t); +    size_t recv(const boost::asio::mutable_buffer &, double);  private:      boost::asio::ip::udp::socket   *_socket; @@ -86,9 +88,8 @@ size_t udp_connected_impl::send(const boost::asio::const_buffer &buff){      return _socket->send(boost::asio::buffer(buff));  } -size_t udp_connected_impl::recv(const boost::asio::mutable_buffer &buff, size_t timeout_ms){ -    reasonable_recv_timeout(*_socket, timeout_ms); -    if (not _socket->available()) return 0; +size_t udp_connected_impl::recv(const boost::asio::mutable_buffer &buff, double timeout){ +    if (not wait_available(*_socket, timeout)) return 0;      return _socket->receive(boost::asio::buffer(buff));  } @@ -103,7 +104,7 @@ public:      //send/recv      size_t send(const boost::asio::const_buffer &); -    size_t recv(const boost::asio::mutable_buffer &, size_t); +    size_t recv(const boost::asio::mutable_buffer &, double);  private:      boost::asio::ip::udp::socket   *_socket; @@ -137,9 +138,8 @@ size_t udp_broadcast_impl::send(const boost::asio::const_buffer &buff){      return _socket->send_to(boost::asio::buffer(buff), _receiver_endpoint);  } -size_t udp_broadcast_impl::recv(const boost::asio::mutable_buffer &buff, size_t timeout_ms){ -    reasonable_recv_timeout(*_socket, timeout_ms); -    if (not _socket->available()) return 0; +size_t udp_broadcast_impl::recv(const boost::asio::mutable_buffer &buff, double timeout){ +    if (not wait_available(*_socket, timeout)) return 0;      boost::asio::ip::udp::endpoint sender_endpoint;      return _socket->receive_from(boost::asio::buffer(buff), sender_endpoint);  } diff --git a/host/lib/transport/udp_zero_copy_asio.cpp b/host/lib/transport/udp_zero_copy_asio.cpp index d84aeefdd..938ae4473 100644 --- a/host/lib/transport/udp_zero_copy_asio.cpp +++ b/host/lib/transport/udp_zero_copy_asio.cpp @@ -59,16 +59,23 @@ static const size_t DEFAULT_NUM_RECV_FRAMES = 32;  #else  static const size_t DEFAULT_NUM_RECV_FRAMES = MIN_RECV_SOCK_BUFF_SIZE/udp_simple::mtu;  #endif +  //The non-async send only ever requires a single frame  //because the buffer will be committed before a new get.  #ifdef USE_ASIO_ASYNC_SEND  static const size_t DEFAULT_NUM_SEND_FRAMES = 32;  #else -static const size_t DEFAULT_NUM_SEND_FRAMES = MIN_SEND_SOCK_BUFF_SIZE/udp_simple::mtu;; +static const size_t DEFAULT_NUM_SEND_FRAMES = MIN_SEND_SOCK_BUFF_SIZE/udp_simple::mtu;  #endif -//a single concurrent thread for io_service seems to be the fastest +//The number of service threads to spawn for async ASIO: +//A single concurrent thread for io_service seems to be the fastest. +//Threads are disabled when no async implementations are enabled. +#if defined(USE_ASIO_ASYNC_RECV) || defined(USE_ASIO_ASYNC_SEND)  static const size_t CONCURRENCY_HINT = 1; +#else +static const size_t CONCURRENCY_HINT = 0; +#endif  /***********************************************************************   * Zero Copy UDP implementation with ASIO: @@ -86,11 +93,12 @@ public:          const std::string &port,          const device_addr_t &hints      ): -        _io_service(hints.cast<size_t>("concurrency_hint", CONCURRENCY_HINT)),          _recv_frame_size(size_t(hints.cast<double>("recv_frame_size", udp_simple::mtu))),          _num_recv_frames(size_t(hints.cast<double>("num_recv_frames", DEFAULT_NUM_RECV_FRAMES))),          _send_frame_size(size_t(hints.cast<double>("send_frame_size", udp_simple::mtu))), -        _num_send_frames(size_t(hints.cast<double>("num_send_frames", DEFAULT_NUM_SEND_FRAMES))) +        _num_send_frames(size_t(hints.cast<double>("num_send_frames", DEFAULT_NUM_SEND_FRAMES))), +        _concurrency_hint(hints.cast<size_t>("concurrency_hint", CONCURRENCY_HINT)), +        _io_service(_concurrency_hint)      {          //std::cout << boost::format("Creating udp transport for %s %s") % addr % port << std::endl; @@ -129,7 +137,7 @@ public:          //spawn the service threads that will run the io service          _work = new asio::io_service::work(_io_service); //new work to delete later -        for (size_t i = 0; i < CONCURRENCY_HINT; i++) _thread_group.create_thread( +        for (size_t i = 0; i < _concurrency_hint; i++) _thread_group.create_thread(              boost::bind(&udp_zero_copy_asio_impl::service, this)          );      } @@ -292,12 +300,6 @@ public:      size_t get_send_frame_size(void) const {return _send_frame_size;}  private: -    //asio guts -> socket and service -    asio::ip::udp::socket   *_socket; -    asio::io_service        _io_service; -    asio::io_service::work  *_work; -    int                     _sock_fd; -      //memory management -> buffers and fifos      boost::thread_group _thread_group;      boost::shared_array<char> _send_buffer, _recv_buffer; @@ -305,6 +307,13 @@ private:      pending_buffs_type::sptr _pending_recv_buffs, _pending_send_buffs;      const size_t _recv_frame_size, _num_recv_frames;      const size_t _send_frame_size, _num_send_frames; + +    //asio guts -> socket and service +    size_t                  _concurrency_hint; +    asio::io_service        _io_service; +    asio::ip::udp::socket   *_socket; +    asio::io_service::work  *_work; +    int                     _sock_fd;  };  /*********************************************************************** diff --git a/host/lib/transport/vrt_packet_handler.hpp b/host/lib/transport/vrt_packet_handler.hpp index 939517411..278bcfeaa 100644 --- a/host/lib/transport/vrt_packet_handler.hpp +++ b/host/lib/transport/vrt_packet_handler.hpp @@ -318,7 +318,7 @@ template <typename T> UHD_INLINE T get_context_code(      ){          //load the rest of the if_packet_info in here          if_packet_info.num_payload_words32 = (num_samps*chans_per_otw_buff*otw_type.get_sample_size())/sizeof(boost::uint32_t); -        if_packet_info.packet_count = state.next_packet_seq++; +        if_packet_info.packet_count = state.next_packet_seq;          //get send buffers for each channel          managed_send_buffs_t send_buffs(buffs.size()/chans_per_otw_buff); @@ -345,6 +345,7 @@ template <typename T> UHD_INLINE T get_context_code(              size_t num_bytes_total = (vrt_header_offset_words32+if_packet_info.num_packet_words32)*sizeof(boost::uint32_t);              send_buffs[i]->commit(num_bytes_total);          } +        state.next_packet_seq++; //increment sequence after commits          return num_samps;      } @@ -387,10 +388,19 @@ template <typename T> UHD_INLINE T get_context_code(              if_packet_info.sob = metadata.start_of_burst;              if_packet_info.eob = metadata.end_of_burst; +            //TODO remove this code when sample counts of zero are supported by hardware +            std::vector<const void *> buffs_(buffs); +            size_t total_num_samps_(total_num_samps); +            if (total_num_samps == 0){ +                static const boost::uint64_t zeros = 0; //max size of a host sample +                buffs_ = std::vector<const void *>(buffs.size(), &zeros); +                total_num_samps_ = 1; +            } +              return _send1(                  state, -                buffs, 0, -                std::min(total_num_samps, max_samples_per_packet), +                buffs_, 0, +                std::min(total_num_samps_, max_samples_per_packet),                  if_packet_info,                  io_type, otw_type,                  vrt_packer, diff --git a/host/lib/usrp/usrp2/fw_common.h b/host/lib/usrp/usrp2/fw_common.h index e812e1221..2cd3ee595 100644 --- a/host/lib/usrp/usrp2/fw_common.h +++ b/host/lib/usrp/usrp2/fw_common.h @@ -33,8 +33,8 @@ extern "C" {  #endif  //fpga and firmware compatibility numbers -#define USRP2_FPGA_COMPAT_NUM 2 -#define USRP2_FW_COMPAT_NUM 6 +#define USRP2_FPGA_COMPAT_NUM 3 +#define USRP2_FW_COMPAT_NUM 7  //used to differentiate control packets over data port  #define USRP2_INVALID_VRT_HEADER 0 diff --git a/host/lib/usrp/usrp2/io_impl.cpp b/host/lib/usrp/usrp2/io_impl.cpp index bbe9c273f..f25b73f80 100644 --- a/host/lib/usrp/usrp2/io_impl.cpp +++ b/host/lib/usrp/usrp2/io_impl.cpp @@ -18,11 +18,11 @@  #include "../../transport/vrt_packet_handler.hpp"  #include "usrp2_impl.hpp"  #include "usrp2_regs.hpp" +#include <uhd/utils/byteswap.hpp>  #include <uhd/utils/thread_priority.hpp>  #include <uhd/transport/convert_types.hpp>  #include <uhd/transport/alignment_buffer.hpp>  #include <boost/format.hpp> -#include <boost/asio.hpp> //htonl and ntohl  #include <boost/bind.hpp>  #include <boost/thread.hpp>  #include <iostream> @@ -32,7 +32,73 @@ using namespace uhd::usrp;  using namespace uhd::transport;  namespace asio = boost::asio; -static const int underflow_flags = async_metadata_t::EVENT_CODE_UNDERFLOW | async_metadata_t::EVENT_CODE_UNDERFLOW_IN_PACKET; +/*********************************************************************** + * constants + **********************************************************************/ +static const int underflow_flags = 0 +    | async_metadata_t::EVENT_CODE_UNDERFLOW +    | async_metadata_t::EVENT_CODE_UNDERFLOW_IN_PACKET +; + +static const size_t vrt_send_header_offset_words32 = 1; + +/*********************************************************************** + * flow control monitor for a single tx channel + *  - the pirate thread calls update + *  - the get send buffer calls check + **********************************************************************/ +class flow_control_monitor{ +public: +    typedef boost::uint32_t seq_type; +    typedef boost::shared_ptr<flow_control_monitor> sptr; + +    /*! +     * Make a new flow control monitor. +     * \param max_seqs_out num seqs before throttling +     */ +    flow_control_monitor(seq_type max_seqs_out){ +        _last_seq_out = 0; +        _last_seq_ack = 0; +        _max_seqs_out = max_seqs_out; +    } + +    /*! +     * Check the flow control condition. +     * \param seq the sequence to go out +     * \param timeout the timeout in seconds +     * \return false on timeout +     */ +    UHD_INLINE bool check_fc_condition(seq_type seq, double timeout){ +        boost::this_thread::disable_interruption di; //disable because the wait can throw +        boost::unique_lock<boost::mutex> lock(_fc_mutex); +        _last_seq_out = seq; +        return _fc_cond.timed_wait( +            lock, +            boost::posix_time::microseconds(long(timeout*1e6)), +            boost::bind(&flow_control_monitor::ready, this) +        ); +    } + +    /*! +     * Update the flow control condition. +     * \param seq the last sequence number to be ACK'd +     */ +    UHD_INLINE void update_fc_condition(seq_type seq){ +        boost::unique_lock<boost::mutex> lock(_fc_mutex); +        _last_seq_ack = seq; +        lock.unlock(); +        _fc_cond.notify_one(); +    } + +private: +    bool ready(void){ +        return seq_type(_last_seq_out -_last_seq_ack) < _max_seqs_out; +    } + +    boost::mutex _fc_mutex; +    boost::condition _fc_cond; +    seq_type _last_seq_out, _last_seq_ack, _max_seqs_out; +};  /***********************************************************************   * io impl details (internal to this file) @@ -44,12 +110,14 @@ static const int underflow_flags = async_metadata_t::EVENT_CODE_UNDERFLOW | asyn  struct usrp2_impl::io_impl{      typedef alignment_buffer<managed_recv_buffer::sptr, time_spec_t> alignment_buffer_type; -    io_impl(size_t num_frames, size_t width): +    io_impl(size_t num_recv_frames, size_t send_frame_size, size_t width):          packet_handler_recv_state(width), -        recv_pirate_booty(alignment_buffer_type::make(num_frames-3, width)), +        recv_pirate_booty(alignment_buffer_type::make(num_recv_frames-3, width)),          async_msg_fifo(bounded_buffer<async_metadata_t>::make(100/*messages deep*/))      { -        /* NOP */ +        for (size_t i = 0; i < width; i++) fc_mons.push_back( +            flow_control_monitor::sptr(new flow_control_monitor(usrp2_impl::sram_bytes/send_frame_size)) +        );      }      ~io_impl(void){ @@ -63,6 +131,29 @@ struct usrp2_impl::io_impl{          return recv_pirate_booty->pop_elems_with_timed_wait(buffs, timeout);      } +    bool get_send_buffs( +        const std::vector<zero_copy_if::sptr> &trans, +        vrt_packet_handler::managed_send_buffs_t &buffs, +        double timeout +    ){ +        UHD_ASSERT_THROW(trans.size() == buffs.size()); + +        //calculate the flow control word +        const boost::uint32_t fc_word32 = packet_handler_send_state.next_packet_seq; + +        //grab a managed buffer for each index +        for (size_t i = 0; i < buffs.size(); i++){ +            if (not fc_mons[i]->check_fc_condition(fc_word32, timeout)) return false; +            buffs[i] = trans[i]->get_send_buff(timeout); +            if (not buffs[i].get()) return false; +            buffs[i]->cast<boost::uint32_t *>()[0] = uhd::htonx(fc_word32); +        } +        return true; +    } + +    //flow control monitors +    std::vector<flow_control_monitor::sptr> fc_mons; +      //state management for the vrt packet handler code      vrt_packet_handler::recv_state packet_handler_recv_state;      vrt_packet_handler::send_state packet_handler_send_state; @@ -112,8 +203,16 @@ void usrp2_impl::io_impl::recv_pirate_loop(                  );                  metadata.event_code = vrt_packet_handler::get_context_code<async_metadata_t::event_code_t>(vrt_hdr, if_packet_info); +                //catch the flow control packets and react +                if (metadata.event_code == 0){ +                    boost::uint32_t fc_word32 = (vrt_hdr + if_packet_info.num_header_words32)[1]; +                    this->fc_mons[index]->update_fc_condition(uhd::ntohx(fc_word32)); +                    continue; +                } +                  //print the famous U, and push the metadata into the message queue                  if (metadata.event_code & underflow_flags) std::cerr << "U" << std::flush; +                //else std::cout << "metadata.event_code " << metadata.event_code << std::endl;                  async_msg_fifo->push_with_pop_on_full(metadata);                  continue;              } @@ -142,23 +241,39 @@ void usrp2_impl::io_impl::recv_pirate_loop(  /***********************************************************************   * Helper Functions   **********************************************************************/ +#include <uhd/usrp/mboard_props.hpp> //TODO remove when hack below is fixed +  void usrp2_impl::io_init(void){ -    //send a small data packet so the usrp2 knows the udp source port -    BOOST_FOREACH(zero_copy_if::sptr data_transport, _data_transports){ -        managed_send_buffer::sptr send_buff = data_transport->get_send_buff(); -        static const boost::uint32_t data = htonl(USRP2_INVALID_VRT_HEADER); -        std::memcpy(send_buff->cast<void*>(), &data, sizeof(data)); -        send_buff->commit(sizeof(data)); -        //drain the recv buffers (may have junk) -        while (data_transport->get_recv_buff().get()){}; -    } -    //the number of recv frames is the number for the first transport      //the assumption is that all data transports should be identical -    size_t num_frames = _data_transports.front()->get_num_recv_frames(); +    const size_t num_recv_frames = _data_transports.front()->get_num_recv_frames(); +    const size_t send_frame_size = _data_transports.front()->get_send_frame_size();      //create new io impl -    _io_impl = UHD_PIMPL_MAKE(io_impl, (num_frames, _data_transports.size())); +    _io_impl = UHD_PIMPL_MAKE(io_impl, (num_recv_frames, send_frame_size, _data_transports.size())); + +    //TODO temporary fix for weird power up state, remove when FPGA fixed +    { +        //send an initial packet to all transports +        tx_metadata_t md; md.end_of_burst = true; +        this->send( +            std::vector<const void *>(_data_transports.size(), NULL), 0, md, +            io_type_t::COMPLEX_FLOAT32, device::SEND_MODE_ONE_PACKET, 0 +        ); + +        //issue a stream command to each motherboard +        BOOST_FOREACH(usrp2_mboard_impl::sptr mboard, _mboards){ +            (*mboard)[MBOARD_PROP_STREAM_CMD] = stream_cmd_t(stream_cmd_t::STREAM_MODE_STOP_CONTINUOUS); +        } + +        //wait +        boost::this_thread::sleep(boost::posix_time::milliseconds(100)); + +        //flush all transport receive queues (no timeout) +        BOOST_FOREACH(zero_copy_if::sptr xport, _data_transports){ +            while(xport->get_recv_buff(0).get() != NULL){}; +        } +    }      //create a new pirate thread for each zc if (yarr!!)      for (size_t i = 0; i < _data_transports.size(); i++){ @@ -183,23 +298,10 @@ bool usrp2_impl::recv_async_msg(  /***********************************************************************   * Send Data   **********************************************************************/ -static bool get_send_buffs( -    const std::vector<udp_zero_copy::sptr> &trans, -    vrt_packet_handler::managed_send_buffs_t &buffs, -    double timeout -){ -    UHD_ASSERT_THROW(trans.size() == buffs.size()); -    bool good = true; -    for (size_t i = 0; i < buffs.size(); i++){ -        buffs[i] = trans[i]->get_send_buff(timeout); -        good = good and (buffs[i].get() != NULL); -    } -    return good; -} -  size_t usrp2_impl::get_max_send_samps_per_packet(void) const{      static const size_t hdr_size = 0          + vrt::max_if_hdr_words32*sizeof(boost::uint32_t) +        + vrt_send_header_offset_words32*sizeof(boost::uint32_t)          - sizeof(vrt::if_packet_info_t().cid) //no class id ever used      ;      const size_t bpp = _data_transports.front()->get_send_frame_size() - hdr_size; @@ -218,8 +320,9 @@ size_t usrp2_impl::send(          io_type, _tx_otw_type,                     //input and output types to convert          _mboards.front()->get_master_clock_freq(), //master clock tick rate          uhd::transport::vrt::if_hdr_pack_be, -        boost::bind(&get_send_buffs, _data_transports, _1, timeout), -        get_max_send_samps_per_packet() +        boost::bind(&usrp2_impl::io_impl::get_send_buffs, _io_impl.get(), _data_transports, _1, timeout), +        get_max_send_samps_per_packet(), +        vrt_send_header_offset_words32      );  } diff --git a/host/lib/usrp/usrp2/mboard_impl.cpp b/host/lib/usrp/usrp2/mboard_impl.cpp index a0e6adfad..8f3ae5c1b 100644 --- a/host/lib/usrp/usrp2/mboard_impl.cpp +++ b/host/lib/usrp/usrp2/mboard_impl.cpp @@ -21,6 +21,7 @@  #include <uhd/usrp/dsp_utils.hpp>  #include <uhd/usrp/mboard_props.hpp>  #include <uhd/utils/assert.hpp> +#include <uhd/utils/byteswap.hpp>  #include <uhd/utils/algorithm.hpp>  #include <uhd/types/mac_addr.hpp>  #include <uhd/types/dict.hpp> @@ -38,11 +39,24 @@ using namespace uhd::usrp;  usrp2_mboard_impl::usrp2_mboard_impl(      size_t index,      transport::udp_simple::sptr ctrl_transport, -    size_t recv_frame_size +    transport::zero_copy_if::sptr data_transport, +    size_t recv_samps_per_packet, +    const device_addr_t &flow_control_hints  ):      _index(index), -    _recv_frame_size(recv_frame_size) +    _recv_samps_per_packet(recv_samps_per_packet)  { +    //Send a small data packet so the usrp2 knows the udp source port. +    //This setup must happen before further initialization occurs +    //or the async update packets will cause ICMP destination unreachable. +    transport::managed_send_buffer::sptr send_buff = data_transport->get_send_buff(); +    static const boost::uint32_t data[2] = { +        uhd::htonx(boost::uint32_t(0 /* don't care seq num */)), +        uhd::htonx(boost::uint32_t(USRP2_INVALID_VRT_HEADER)) +    }; +    std::memcpy(send_buff->cast<void*>(), &data, sizeof(data)); +    send_buff->commit(sizeof(data)); +      //make a new interface for usrp2 stuff      _iface = usrp2_iface::make(ctrl_transport); @@ -69,13 +83,8 @@ usrp2_mboard_impl::usrp2_mboard_impl(          _allowed_decim_and_interp_rates.push_back(i);      } -    //Issue a stop streaming command (in case it was left running). -    //Since this command is issued before the networking is setup, -    //most if not all junk packets will never make it to the socket. -    this->issue_ddc_stream_cmd(stream_cmd_t::STREAM_MODE_STOP_CONTINUOUS); -      //init the rx control registers -    _iface->poke32(U2_REG_RX_CTRL_NSAMPS_PER_PKT, _recv_frame_size); +    _iface->poke32(U2_REG_RX_CTRL_NSAMPS_PER_PKT, _recv_samps_per_packet);      _iface->poke32(U2_REG_RX_CTRL_NCHANNELS, 1);      _iface->poke32(U2_REG_RX_CTRL_CLEAR_OVERRUN, 1); //reset      _iface->poke32(U2_REG_RX_CTRL_VRT_HEADER, 0 @@ -94,6 +103,16 @@ usrp2_mboard_impl::usrp2_mboard_impl(      _iface->poke32(U2_REG_TX_CTRL_REPORT_SID, 1);  //sid 1 (different from rx)      _iface->poke32(U2_REG_TX_CTRL_POLICY, U2_FLAG_TX_CTRL_POLICY_NEXT_PACKET); +    //setting the cycles per update +    const double ups_per_sec = flow_control_hints.cast<double>("ups_per_sec", 100); +    const size_t cycles_per_up = size_t(_clock_ctrl->get_master_clock_rate()/ups_per_sec); +    _iface->poke32(U2_REG_TX_CTRL_CYCLES_PER_UP, U2_FLAG_TX_CTRL_UP_ENB | cycles_per_up); + +    //setting the packets per update +    const double ups_per_fifo = flow_control_hints.cast<double>("ups_per_fifo", 8); +    const size_t packets_per_up = size_t(usrp2_impl::sram_bytes/ups_per_fifo/data_transport->get_send_frame_size()); +    _iface->poke32(U2_REG_TX_CTRL_PACKETS_PER_UP, U2_FLAG_TX_CTRL_UP_ENB | packets_per_up); +      //init the ddc      init_ddc_config(); @@ -115,7 +134,8 @@ usrp2_mboard_impl::usrp2_mboard_impl(  }  usrp2_mboard_impl::~usrp2_mboard_impl(void){ -    /* NOP */ +    _iface->poke32(U2_REG_TX_CTRL_CYCLES_PER_UP, 0); +    _iface->poke32(U2_REG_TX_CTRL_PACKETS_PER_UP, 0);  }  /*********************************************************************** @@ -178,7 +198,7 @@ void usrp2_mboard_impl::set_time_spec(const time_spec_t &time_spec, bool now){  void usrp2_mboard_impl::issue_ddc_stream_cmd(const stream_cmd_t &stream_cmd){      _iface->poke32(U2_REG_RX_CTRL_STREAM_CMD, dsp_type1::calc_stream_cmd_word( -        stream_cmd, _recv_frame_size +        stream_cmd, _recv_samps_per_packet      ));      _iface->poke32(U2_REG_RX_CTRL_TIME_SECS,  boost::uint32_t(stream_cmd.time_spec.get_full_secs()));      _iface->poke32(U2_REG_RX_CTRL_TIME_TICKS, stream_cmd.time_spec.get_tick_count(get_master_clock_freq())); diff --git a/host/lib/usrp/usrp2/usrp2_iface.cpp b/host/lib/usrp/usrp2/usrp2_iface.cpp index 2d450bfc6..55c42567e 100644 --- a/host/lib/usrp/usrp2/usrp2_iface.cpp +++ b/host/lib/usrp/usrp2/usrp2_iface.cpp @@ -30,18 +30,6 @@  using namespace uhd;  using namespace uhd::transport; -/*! - * FIXME: large timeout, ethernet pause frames... - * - * Use a large timeout to work-around the fact that - * flow-control may throttle outgoing control packets - * due to its use of ethernet pause frames. - * - * This will be fixed when host-based flow control is implemented, - * along with larger incoming send buffers using the on-board SRAM. - */ -static const size_t CONTROL_TIMEOUT_MS = 3000; //3 seconds -  class usrp2_iface_impl : public usrp2_iface{  public:  /*********************************************************************** @@ -187,7 +175,7 @@ public:          boost::uint8_t usrp2_ctrl_data_in_mem[udp_simple::mtu]; //allocate max bytes for recv          const usrp2_ctrl_data_t *ctrl_data_in = reinterpret_cast<const usrp2_ctrl_data_t *>(usrp2_ctrl_data_in_mem);          while(true){ -            size_t len = _ctrl_transport->recv(boost::asio::buffer(usrp2_ctrl_data_in_mem), CONTROL_TIMEOUT_MS); +            size_t len = _ctrl_transport->recv(boost::asio::buffer(usrp2_ctrl_data_in_mem));              if(len >= sizeof(boost::uint32_t) and ntohl(ctrl_data_in->proto_ver) != USRP2_FW_COMPAT_NUM){                  throw std::runtime_error(str(boost::format(                      "Expected protocol compatibility number %d, but got %d:\n" diff --git a/host/lib/usrp/usrp2/usrp2_impl.cpp b/host/lib/usrp/usrp2/usrp2_impl.cpp index a680708ad..afc69f703 100644 --- a/host/lib/usrp/usrp2/usrp2_impl.cpp +++ b/host/lib/usrp/usrp2/usrp2_impl.cpp @@ -17,7 +17,7 @@  #include "usrp2_impl.hpp"  #include <uhd/transport/if_addrs.hpp> -#include <uhd/transport/udp_simple.hpp> +#include <uhd/transport/udp_zero_copy.hpp>  #include <uhd/usrp/device_props.hpp>  #include <uhd/utils/assert.hpp>  #include <uhd/utils/static.hpp> @@ -35,9 +35,6 @@ using namespace uhd::usrp;  using namespace uhd::transport;  namespace asio = boost::asio; -//! wait this long for a control response when discovering devices -static const size_t DISCOVERY_TIMEOUT_MS = 100; -  /***********************************************************************   * Helper Functions   **********************************************************************/ @@ -99,7 +96,7 @@ static uhd::device_addrs_t usrp2_find(const device_addr_t &hint){      boost::uint8_t usrp2_ctrl_data_in_mem[udp_simple::mtu]; //allocate max bytes for recv      const usrp2_ctrl_data_t *ctrl_data_in = reinterpret_cast<const usrp2_ctrl_data_t *>(usrp2_ctrl_data_in_mem);      while(true){ -        size_t len = udp_transport->recv(asio::buffer(usrp2_ctrl_data_in_mem), DISCOVERY_TIMEOUT_MS); +        size_t len = udp_transport->recv(asio::buffer(usrp2_ctrl_data_in_mem));          //std::cout << len << "\n";          if (len > offsetof(usrp2_ctrl_data_t, data)){              //handle the received data @@ -128,7 +125,7 @@ static device::sptr usrp2_make(const device_addr_t &device_addr){      //create a ctrl and data transport for each address      std::vector<udp_simple::sptr> ctrl_transports; -    std::vector<udp_zero_copy::sptr> data_transports; +    std::vector<zero_copy_if::sptr> data_transports;      BOOST_FOREACH(const std::string &addr, std::split_string(device_addr["addr"])){          ctrl_transports.push_back(udp_simple::make_connected( @@ -141,7 +138,7 @@ static device::sptr usrp2_make(const device_addr_t &device_addr){      //create the usrp2 implementation guts      return device::sptr( -        new usrp2_impl(ctrl_transports, data_transports) +        new usrp2_impl(ctrl_transports, data_transports, device_addr)      );  } @@ -154,7 +151,8 @@ UHD_STATIC_BLOCK(register_usrp2_device){   **********************************************************************/  usrp2_impl::usrp2_impl(      std::vector<udp_simple::sptr> ctrl_transports, -    std::vector<udp_zero_copy::sptr> data_transports +    std::vector<zero_copy_if::sptr> data_transports, +    const device_addr_t &flow_control_hints  ):      _data_transports(data_transports)  { @@ -173,7 +171,9 @@ usrp2_impl::usrp2_impl(      //create a new mboard handler for each control transport      for(size_t i = 0; i < ctrl_transports.size(); i++){          _mboards.push_back(usrp2_mboard_impl::sptr(new usrp2_mboard_impl( -            i, ctrl_transports[i], this->get_max_recv_samps_per_packet() +            i, ctrl_transports[i], data_transports[i], +            this->get_max_recv_samps_per_packet(), +            flow_control_hints          )));          //use an empty name when there is only one mboard          std::string name = (ctrl_transports.size() > 1)? boost::lexical_cast<std::string>(i) : ""; diff --git a/host/lib/usrp/usrp2/usrp2_impl.hpp b/host/lib/usrp/usrp2/usrp2_impl.hpp index 558726a2b..2531bd6cb 100644 --- a/host/lib/usrp/usrp2/usrp2_impl.hpp +++ b/host/lib/usrp/usrp2/usrp2_impl.hpp @@ -33,7 +33,7 @@  #include <boost/function.hpp>  #include <uhd/transport/vrt_if_packet.hpp>  #include <uhd/transport/udp_simple.hpp> //mtu -#include <uhd/transport/udp_zero_copy.hpp> +#include <uhd/transport/zero_copy.hpp>  #include <uhd/usrp/dboard_manager.hpp>  #include <uhd/usrp/subdev_spec.hpp> @@ -84,7 +84,9 @@ public:      usrp2_mboard_impl(          size_t index,          uhd::transport::udp_simple::sptr, -        size_t recv_frame_size +        uhd::transport::zero_copy_if::sptr, +        size_t recv_samps_per_packet, +        const uhd::device_addr_t &flow_control_hints      );      ~usrp2_mboard_impl(void); @@ -95,7 +97,7 @@ public:  private:      size_t _index;      int _rev_hi, _rev_lo; -    const size_t _recv_frame_size; +    const size_t _recv_samps_per_packet;      //properties for this mboard      void get(const wax::obj &, wax::obj &); @@ -171,14 +173,18 @@ private:   */  class usrp2_impl : public uhd::device{  public: +    static const size_t sram_bytes = size_t(1 << 20); +      /*!       * Create a new usrp2 impl base.       * \param ctrl_transports the udp transports for control       * \param data_transports the udp transports for data +     * \param flow_control_hints optional flow control params       */      usrp2_impl(          std::vector<uhd::transport::udp_simple::sptr> ctrl_transports, -        std::vector<uhd::transport::udp_zero_copy::sptr> data_transports +        std::vector<uhd::transport::zero_copy_if::sptr> data_transports, +        const uhd::device_addr_t &flow_control_hints      );      ~usrp2_impl(void); @@ -208,7 +214,7 @@ private:      uhd::dict<std::string, usrp2_mboard_impl::sptr> _mboard_dict;      //io impl methods and members -    std::vector<uhd::transport::udp_zero_copy::sptr> _data_transports; +    std::vector<uhd::transport::zero_copy_if::sptr> _data_transports;      uhd::otw_type_t _rx_otw_type, _tx_otw_type;      UHD_PIMPL_DECL(io_impl) _io_impl;      void io_init(void); diff --git a/host/lib/usrp/usrp2/usrp2_regs.hpp b/host/lib/usrp/usrp2/usrp2_regs.hpp index 064ad4e95..c3a4d22de 100644 --- a/host/lib/usrp/usrp2/usrp2_regs.hpp +++ b/host/lib/usrp/usrp2/usrp2_regs.hpp @@ -193,9 +193,14 @@  #define U2_REG_TX_CTRL_CLEAR_STATE       _SR_ADDR(SR_TX_CTRL + 1)  #define U2_REG_TX_CTRL_REPORT_SID        _SR_ADDR(SR_TX_CTRL + 2)  #define U2_REG_TX_CTRL_POLICY            _SR_ADDR(SR_TX_CTRL + 3) +#define U2_REG_TX_CTRL_CYCLES_PER_UP     _SR_ADDR(SR_TX_CTRL + 4) +#define U2_REG_TX_CTRL_PACKETS_PER_UP    _SR_ADDR(SR_TX_CTRL + 5)  #define U2_FLAG_TX_CTRL_POLICY_WAIT          (0x1 << 0)  #define U2_FLAG_TX_CTRL_POLICY_NEXT_PACKET   (0x1 << 1)  #define U2_FLAG_TX_CTRL_POLICY_NEXT_BURST    (0x1 << 2) +//enable flag for registers: cycles and packets per update packet +#define U2_FLAG_TX_CTRL_UP_ENB              (1ul << 31) +  #endif /* INCLUDED_USRP2_REGS_HPP */ | 
