diff options
Diffstat (limited to 'host')
-rw-r--r-- | host/include/uhd/device.hpp | 11 | ||||
-rw-r--r-- | host/include/uhd/transport/udp_simple.hpp | 4 | ||||
-rw-r--r-- | host/include/uhd/transport/udp_zero_copy.hpp | 4 | ||||
-rw-r--r-- | host/lib/transport/udp_simple.cpp | 29 | ||||
-rw-r--r-- | host/lib/transport/udp_zero_copy_none.cpp | 39 | ||||
-rw-r--r-- | host/lib/usrp/usrp2/dsp_impl.cpp | 9 | ||||
-rw-r--r-- | host/lib/usrp/usrp2/io_impl.cpp | 14 | ||||
-rw-r--r-- | host/lib/usrp/usrp2/usrp2_impl.cpp | 39 |
8 files changed, 104 insertions, 45 deletions
diff --git a/host/include/uhd/device.hpp b/host/include/uhd/device.hpp index 1a52867c9..47dfa4328 100644 --- a/host/include/uhd/device.hpp +++ b/host/include/uhd/device.hpp @@ -77,6 +77,9 @@ public: * It is up to the caller to call send again on the un-sent * portions of the buffer, until the buffer is exhausted. * + * This is a blocking call and will not return until the number + * of samples returned have been read out of the buffer. + * * \param buff a buffer pointing to some read-only memory * \param metadata data describing the buffer's contents * \param the type of data loaded in the buffer (32fc, 16sc) @@ -101,6 +104,14 @@ public: * The next call to receive, after the remainder becomes exahausted, * will perform an over-the-wire receive as usual. * + * This is a blocking call and will not return until the number + * of samples returned have been written into the buffer. + * However, a call to receive may timeout and return zero samples. + * The timeout duration is decided by the underlying transport layer. + * The caller should assume that the call to receive will not return + * immediately when no packets are available to the transport layer, + * and that the timeout duration is reasonably tuned for performance. + * * \param buff the buffer to fill with IF data * \param metadata data to fill describing the buffer * \param the type of data to fill into the buffer (32fc, 16sc) diff --git a/host/include/uhd/transport/udp_simple.hpp b/host/include/uhd/transport/udp_simple.hpp index 8663128ec..0d8fcc5f0 100644 --- a/host/include/uhd/transport/udp_simple.hpp +++ b/host/include/uhd/transport/udp_simple.hpp @@ -67,9 +67,9 @@ public: /*! * Receive into the provided buffer. - * Returns empty when data is not available. + * Blocks until data is received or a timeout occurs. * \param buff a mutable buffer to receive into - * \return the number of bytes received. + * \return the number of bytes received or zero on timeout */ virtual size_t recv(const boost::asio::mutable_buffer &buff) = 0; }; diff --git a/host/include/uhd/transport/udp_zero_copy.hpp b/host/include/uhd/transport/udp_zero_copy.hpp index 9c3505dd6..1a8d822fd 100644 --- a/host/include/uhd/transport/udp_zero_copy.hpp +++ b/host/include/uhd/transport/udp_zero_copy.hpp @@ -64,9 +64,9 @@ public: /*! * Receive a buffer. + * Blocks until data is received or a timeout occurs. * The memory is managed by the implementation. - * Returns an empty buffer when data is not available. - * \return a smart buffer with memory and size + * \return a smart buffer (empty on timeout) */ virtual smart_buffer::sptr recv(void) = 0; }; diff --git a/host/lib/transport/udp_simple.cpp b/host/lib/transport/udp_simple.cpp index 491cf59db..7004bdfdf 100644 --- a/host/lib/transport/udp_simple.cpp +++ b/host/lib/transport/udp_simple.cpp @@ -16,12 +16,35 @@ // #include <uhd/transport/udp_simple.hpp> +#include <boost/thread.hpp> #include <boost/format.hpp> #include <iostream> 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. + * + * \param socket the asio socket + */ +static void reasonable_recv_timeout( + boost::asio::ip::udp::socket &socket +){ + boost::asio::deadline_timer timer(socket.get_io_service()); + timer.expires_from_now(boost::posix_time::milliseconds(50)); + while (not (socket.available() or timer.expires_from_now().is_negative())){ + boost::this_thread::sleep(boost::posix_time::milliseconds(1)); + } +} + +/*********************************************************************** * UDP connected implementation class **********************************************************************/ class udp_connected_impl : public udp_simple{ @@ -62,7 +85,8 @@ size_t udp_connected_impl::send(const boost::asio::const_buffer &buff){ } size_t udp_connected_impl::recv(const boost::asio::mutable_buffer &buff){ - if (_socket->available() == 0) return 0; + reasonable_recv_timeout(*_socket); + if (not _socket->available()) return 0; return _socket->receive(boost::asio::buffer(buff)); } @@ -112,7 +136,8 @@ size_t udp_broadcast_impl::send(const boost::asio::const_buffer &buff){ } size_t udp_broadcast_impl::recv(const boost::asio::mutable_buffer &buff){ - if (_socket->available() == 0) return 0; + reasonable_recv_timeout(*_socket); + if (not _socket->available()) 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_none.cpp b/host/lib/transport/udp_zero_copy_none.cpp index e95706d94..e29530cf1 100644 --- a/host/lib/transport/udp_zero_copy_none.cpp +++ b/host/lib/transport/udp_zero_copy_none.cpp @@ -16,6 +16,8 @@ // #include <uhd/transport/udp_zero_copy.hpp> +#include <boost/thread.hpp> +#include <boost/format.hpp> using namespace uhd::transport; @@ -67,6 +69,9 @@ public: private: boost::asio::ip::udp::socket *_socket; boost::asio::io_service _io_service; + + size_t get_recv_buff_size(void); + void set_recv_buff_size(size_t); }; udp_zero_copy_impl::udp_zero_copy_impl(const std::string &addr, const std::string &port){ @@ -81,6 +86,18 @@ udp_zero_copy_impl::udp_zero_copy_impl(const std::string &addr, const std::strin _socket = new boost::asio::ip::udp::socket(_io_service); _socket->open(boost::asio::ip::udp::v4()); _socket->connect(receiver_endpoint); + + // set the rx socket buffer size: + // pick a huge size, and deal with whatever we get + set_recv_buff_size(54321e3); //some big number! + size_t current_buff_size = get_recv_buff_size(); + std::cout << boost::format( + "Current rx socket buffer size: %d\n" + ) % current_buff_size; + if (current_buff_size < .1e6) std::cout << boost::format( + "Adjust max rx socket buffer size (linux only):\n" + " sysctl -w net.core.rmem_max=VALUE\n" + ); } udp_zero_copy_impl::~udp_zero_copy_impl(void){ @@ -92,14 +109,21 @@ size_t udp_zero_copy_impl::send(const boost::asio::const_buffer &buff){ } smart_buffer::sptr udp_zero_copy_impl::recv(void){ - size_t available = _socket->available(); + size_t available = 0; + + //implement timeout through polling and sleeping + boost::asio::deadline_timer timer(_socket->get_io_service()); + timer.expires_from_now(boost::posix_time::milliseconds(50)); + while (not ((available = _socket->available()) or timer.expires_from_now().is_negative())){ + boost::this_thread::sleep(boost::posix_time::milliseconds(1)); + } //allocate memory and create buffer uint32_t *buff_mem = new uint32_t[available/sizeof(uint32_t)]; boost::asio::mutable_buffer buff(buff_mem, available); //receive only if data is available - if (available > 0){ + if (available){ _socket->receive(boost::asio::buffer(buff)); } @@ -107,6 +131,17 @@ smart_buffer::sptr udp_zero_copy_impl::recv(void){ return smart_buffer::sptr(new smart_buffer_impl(buff)); } +size_t udp_zero_copy_impl::get_recv_buff_size(void){ + boost::asio::socket_base::receive_buffer_size option; + _socket->get_option(option); + return option.value(); +} + +void udp_zero_copy_impl::set_recv_buff_size(size_t new_size){ + boost::asio::socket_base::receive_buffer_size option(new_size); + _socket->set_option(option); +} + /*********************************************************************** * UDP zero copy make function **********************************************************************/ diff --git a/host/lib/usrp/usrp2/dsp_impl.cpp b/host/lib/usrp/usrp2/dsp_impl.cpp index a32f68872..7831b7667 100644 --- a/host/lib/usrp/usrp2/dsp_impl.cpp +++ b/host/lib/usrp/usrp2/dsp_impl.cpp @@ -21,6 +21,9 @@ using namespace uhd; +static const size_t default_decim = 16; +static const size_t default_interp = 16; + /*********************************************************************** * DDC Helper Methods **********************************************************************/ @@ -37,7 +40,7 @@ static uint32_t calculate_freq_word_and_update_actual_freq(freq_t &freq, freq_t } static uint32_t calculate_iq_scale_word(int16_t i, int16_t q){ - return ((i & 0xffff) << 16) | ((q & 0xffff) << 0); + return (uint16_t(i) << 16) | (uint16_t(q) << 0); } void usrp2_impl::init_ddc_config(void){ @@ -48,7 +51,7 @@ void usrp2_impl::init_ddc_config(void){ ); //initial config and update - _ddc_decim = 64; + _ddc_decim = default_decim; _ddc_freq = 0; update_ddc_config(); @@ -196,7 +199,7 @@ void usrp2_impl::init_duc_config(void){ ); //initial config and update - _duc_interp = 64; + _duc_interp = default_interp; _duc_freq = 0; update_duc_config(); } diff --git a/host/lib/usrp/usrp2/io_impl.cpp b/host/lib/usrp/usrp2/io_impl.cpp index 6273846dc..cc7746720 100644 --- a/host/lib/usrp/usrp2/io_impl.cpp +++ b/host/lib/usrp/usrp2/io_impl.cpp @@ -71,9 +71,9 @@ static inline void host_floats_to_usrp2_items( size_t num_samps ){ unrolled_loop(i, num_samps,{ - int16_t real = host_floats[i].real()*shorts_per_float; - int16_t imag = host_floats[i].imag()*shorts_per_float; - usrp2_items[i] = htonl(((real << 16) & 0xffff) | ((imag << 0) & 0xffff)); + uint16_t real = host_floats[i].real()*shorts_per_float; + uint16_t imag = host_floats[i].imag()*shorts_per_float; + usrp2_items[i] = htonl((real << 16) | (imag << 0)); }); } @@ -84,8 +84,8 @@ static inline void usrp2_items_to_host_floats( ){ unrolled_loop(i, num_samps,{ uint32_t item = ntohl(usrp2_items[i]); - int16_t real = (item >> 16) & 0xffff; - int16_t imag = (item >> 0) & 0xffff; + int16_t real = item >> 16; + int16_t imag = item >> 0; host_floats[i] = fc32_t(real*floats_per_short, imag*floats_per_short); }); } @@ -130,9 +130,7 @@ void usrp2_impl::recv_raw(rx_metadata_t &metadata){ return; //must exit here after setting the buffer } const uint32_t *vrt_hdr = asio::buffer_cast<const uint32_t *>(_rx_smart_buff->get()); - size_t num_header_words32_out; - size_t num_payload_words32_out; - size_t packet_count_out; + size_t num_header_words32_out, num_payload_words32_out, packet_count_out; try{ vrt::unpack( metadata, //output diff --git a/host/lib/usrp/usrp2/usrp2_impl.cpp b/host/lib/usrp/usrp2/usrp2_impl.cpp index 752feb05b..58c82303f 100644 --- a/host/lib/usrp/usrp2/usrp2_impl.cpp +++ b/host/lib/usrp/usrp2/usrp2_impl.cpp @@ -24,6 +24,7 @@ using namespace uhd; using namespace uhd::usrp; using namespace uhd::transport; +namespace asio = boost::asio; /*********************************************************************** * Discovery over the udp transport @@ -43,19 +44,12 @@ uhd::device_addrs_t usrp2::discover(const device_addr_t &hint){ ctrl_data_out.id = htonl(USRP2_CTRL_ID_GIVE_ME_YOUR_IP_ADDR_BRO); udp_transport->send(boost::asio::buffer(&ctrl_data_out, sizeof(ctrl_data_out))); - //loop and recieve until the time is up - size_t num_timeouts = 0; + //loop and recieve until the timeout while(true){ usrp2_ctrl_data_t ctrl_data_in; - size_t len = udp_transport->recv( - boost::asio::buffer(&ctrl_data_in, sizeof(ctrl_data_in)) - ); + size_t len = udp_transport->recv(asio::buffer(&ctrl_data_in, sizeof(ctrl_data_in))); //std::cout << len << "\n"; - if (len < sizeof(usrp2_ctrl_data_t)){ - //sleep a little so we dont burn cpu - if (num_timeouts++ > 50) break; - boost::this_thread::sleep(boost::posix_time::milliseconds(1)); - }else{ + if (len >= sizeof(usrp2_ctrl_data_t)){ //handle the received data switch(ntohl(ctrl_data_in.id)){ case USRP2_CTRL_ID_THIS_IS_MY_IP_ADDR_DUDE: @@ -67,9 +61,11 @@ uhd::device_addrs_t usrp2::discover(const device_addr_t &hint){ new_addr["transport"] = "udp"; new_addr["addr"] = ip_addr.to_string(); usrp2_addrs.push_back(new_addr); - break; + //dont break here, it will exit the while loop + //just continue on to the next loop iteration } } + if (len == 0) break; //timeout } return usrp2_addrs; @@ -164,24 +160,15 @@ usrp2_ctrl_data_t usrp2_impl::ctrl_send_and_recv(const usrp2_ctrl_data_t &out_da out_copy.seq = htonl(++_ctrl_seq_num); _ctrl_transport->send(boost::asio::buffer(&out_copy, sizeof(usrp2_ctrl_data_t))); - //loop and recieve until the time is up - size_t num_timeouts = 0; + //loop until we get the packet or timeout while(true){ usrp2_ctrl_data_t in_data; - size_t len = _ctrl_transport->recv( - boost::asio::buffer(&in_data, sizeof(in_data)) - ); - if (len < sizeof(usrp2_ctrl_data_t)){ - //sleep a little so we dont burn cpu - if (num_timeouts++ > 50) break; - boost::this_thread::sleep(boost::posix_time::milliseconds(1)); - }else{ - //handle the received data - if (ntohl(in_data.seq) == _ctrl_seq_num){ - return in_data; - } - //didnt get seq, continue on... + size_t len = _ctrl_transport->recv(asio::buffer(&in_data, sizeof(in_data))); + if (len >= sizeof(usrp2_ctrl_data_t) and ntohl(in_data.seq) == _ctrl_seq_num){ + return in_data; } + if (len == 0) break; //timeout + //didnt get seq or bad packet, continue looking... } throw std::runtime_error("usrp2 no control response"); } |