usrp-e: implemented mmap with new zero_copy timeout work, added much debug verbosity

3 files changed, 107 insertions, 108 deletions

diff --git a/host/lib/usrp/usrp_e/io_impl.cpp b/host/lib/usrp/usrp_e/io_impl.cpp
index 0b5fa054b..1e577a4df 100644
--- a/host/lib/usrp/usrp_e/io_impl.cpp
+++ b/host/lib/usrp/usrp_e/io_impl.cpp
@@ -37,7 +37,7 @@ zero_copy_if::sptr usrp_e_make_mmap_zero_copy(usrp_e_iface::sptr iface);
  **********************************************************************/
 static const size_t tx_async_report_sid = 1;
 static const int underflow_flags = async_metadata_t::EVENT_CODE_UNDERFLOW | async_metadata_t::EVENT_CODE_UNDERFLOW_IN_PACKET;
-static const double recv_timeout_ms = 100;
+static const bool recv_debug = true;
 
 /***********************************************************************
  * io impl details (internal to this file)
@@ -66,10 +66,10 @@ struct usrp_e_impl::io_impl{
         recv_pirate_crew.join_all();
     }
 
-    bool get_recv_buffs(vrt_packet_handler::managed_recv_buffs_t &buffs, size_t timeout_ms){
+    bool get_recv_buffs(vrt_packet_handler::managed_recv_buffs_t &buffs, double timeout){
         UHD_ASSERT_THROW(buffs.size() == 1);
         boost::this_thread::disable_interruption di; //disable because the wait can throw
-        return recv_pirate_booty->pop_with_timed_wait(buffs.front(), boost::posix_time::milliseconds(timeout_ms));
+        return recv_pirate_booty->pop_with_timed_wait(buffs.front(), timeout);
     }
 
     //a pirate's life is the life for me!
@@ -94,9 +94,17 @@ void usrp_e_impl::io_impl::recv_pirate_loop(
     //size_t next_packet_seq = 0;
 
     while(recv_pirate_crew_raiding){
-        managed_recv_buffer::sptr buff = this->data_xport->get_recv_buff(recv_timeout_ms);
+        managed_recv_buffer::sptr buff = this->data_xport->get_recv_buff();
         if (not buff.get()) continue; //ignore timeout/error buffers
 
+        if (recv_debug){
+            std::cout << "len " << buff->size() << std::endl;
+            for (size_t i = 0; i < 9; i++){
+                std::cout << boost::format("    0x%08x") % buff->cast<const boost::uint32_t *>()[i] << std::endl;
+            }
+            std::cout << std::endl << std::endl;
+        }
+
         try{
             //extract the vrt header packet info
             vrt::if_packet_info_t if_packet_info;
@@ -181,20 +189,18 @@ void usrp_e_impl::handle_overrun(size_t){
  * Data Send
  **********************************************************************/
 bool get_send_buffs(
-    zero_copy_if::sptr trans,
+    zero_copy_if::sptr trans, double timeout,
     vrt_packet_handler::managed_send_buffs_t &buffs
 ){
     UHD_ASSERT_THROW(buffs.size() == 1);
-    buffs[0] = trans->get_send_buff();
+    buffs[0] = trans->get_send_buff(timeout);
     return buffs[0].get() != NULL;
 }
 
 size_t usrp_e_impl::send(
-    const std::vector<const void *> &buffs,
-    size_t num_samps,
-    const tx_metadata_t &metadata,
-    const io_type_t &io_type,
-    send_mode_t send_mode
+    const std::vector<const void *> &buffs, size_t num_samps,
+    const tx_metadata_t &metadata, const io_type_t &io_type,
+    send_mode_t send_mode, double timeout
 ){
     otw_type_t send_otw_type;
     send_otw_type.width = 16;
@@ -208,7 +214,7 @@ size_t usrp_e_impl::send(
         io_type, send_otw_type,                    //input and output types to convert
         MASTER_CLOCK_RATE,                         //master clock tick rate
         uhd::transport::vrt::if_hdr_pack_le,
-        boost::bind(&get_send_buffs, _io_impl->data_xport, _1),
+        boost::bind(&get_send_buffs, _io_impl->data_xport, timeout, _1),
         get_max_send_samps_per_packet()
     );
 }
@@ -217,12 +223,9 @@ size_t usrp_e_impl::send(
  * Data Recv
  **********************************************************************/
 size_t usrp_e_impl::recv(
-    const std::vector<void *> &buffs,
-    size_t num_samps,
-    rx_metadata_t &metadata,
-    const io_type_t &io_type,
-    recv_mode_t recv_mode,
-    size_t timeout_ms
+    const std::vector<void *> &buffs, size_t num_samps,
+    rx_metadata_t &metadata, const io_type_t &io_type,
+    recv_mode_t recv_mode, double timeout
 ){
     otw_type_t recv_otw_type;
     recv_otw_type.width = 16;
@@ -236,7 +239,7 @@ size_t usrp_e_impl::recv(
         io_type, recv_otw_type,                    //input and output types to convert
         MASTER_CLOCK_RATE,                         //master clock tick rate
         uhd::transport::vrt::if_hdr_unpack_le,
-        boost::bind(&usrp_e_impl::io_impl::get_recv_buffs, _io_impl.get(), _1, timeout_ms),
+        boost::bind(&usrp_e_impl::io_impl::get_recv_buffs, _io_impl.get(), _1, timeout),
         boost::bind(&usrp_e_impl::handle_overrun, this, _1)
     );
 }
@@ -245,11 +248,8 @@ size_t usrp_e_impl::recv(
  * Async Recv
  **********************************************************************/
 bool usrp_e_impl::recv_async_msg(
-    async_metadata_t &async_metadata,
-    size_t timeout_ms
+    async_metadata_t &async_metadata, double timeout
 ){
     boost::this_thread::disable_interruption di; //disable because the wait can throw
-    return _io_impl->async_msg_fifo->pop_with_timed_wait(
-        async_metadata, boost::posix_time::milliseconds(timeout_ms)
-    );
+    return _io_impl->async_msg_fifo->pop_with_timed_wait(async_metadata, timeout);
 }
diff --git a/host/lib/usrp/usrp_e/usrp_e_impl.hpp b/host/lib/usrp/usrp_e/usrp_e_impl.hpp
index 2457e27cc..49912050e 100644
--- a/host/lib/usrp/usrp_e/usrp_e_impl.hpp
+++ b/host/lib/usrp/usrp_e/usrp_e_impl.hpp
@@ -82,9 +82,9 @@ public:
     ~usrp_e_impl(void);
 
     //the io interface
-    size_t send(const std::vector<const void *> &, size_t, const uhd::tx_metadata_t &, const uhd::io_type_t &, send_mode_t);
-    size_t recv(const std::vector<void *> &, size_t, uhd::rx_metadata_t &, const uhd::io_type_t &, recv_mode_t, size_t);
-    bool recv_async_msg(uhd::async_metadata_t &, size_t);
+    size_t send(const std::vector<const void *> &, size_t, const uhd::tx_metadata_t &, const uhd::io_type_t &, send_mode_t, double);
+    size_t recv(const std::vector<void *> &, size_t, uhd::rx_metadata_t &, const uhd::io_type_t &, recv_mode_t, double);
+    bool recv_async_msg(uhd::async_metadata_t &, double);
     size_t get_max_send_samps_per_packet(void) const{return 503;}
     size_t get_max_recv_samps_per_packet(void) const{return 503;}
 
diff --git a/host/lib/usrp/usrp_e/usrp_e_mmap_zero_copy.cpp b/host/lib/usrp/usrp_e/usrp_e_mmap_zero_copy.cpp
index 0910caa6f..e62205a3f 100644
--- a/host/lib/usrp/usrp_e/usrp_e_mmap_zero_copy.cpp
+++ b/host/lib/usrp/usrp_e/usrp_e_mmap_zero_copy.cpp
@@ -15,86 +15,27 @@
 // along with this program.  If not, see <http://www.gnu.org/licenses/>.
 //
 
+#include "usrp_e_iface.hpp"
 #include <uhd/transport/zero_copy.hpp>
 #include <uhd/utils/assert.hpp>
 #include <linux/usrp_e.h>
 #include <sys/mman.h> //mmap
 #include <unistd.h> //getpagesize
 #include <poll.h> //poll
-#include <boost/cstdint.hpp>
-#include "usrp_e_iface.hpp"
+#include <boost/bind.hpp>
+#include <boost/enable_shared_from_this.hpp>
+#include <iostream>
 
 using namespace uhd;
 using namespace uhd::transport;
 
-static const bool debug_verbose = false;
-
-/***********************************************************************
- * The managed receive buffer implementation
- **********************************************************************/
-class usrp_e_mmap_managed_recv_buffer : public managed_recv_buffer{
-public:
-    usrp_e_mmap_managed_recv_buffer(
-        const void *mem, size_t len, ring_buffer_info *info
-    ):
-        _buff(mem, len), _info(info)
-    {
-        /* NOP */
-    }
-
-    ~usrp_e_mmap_managed_recv_buffer(void){
-        _info->flags = RB_KERNEL;
-    }
-
-private:
-    const boost::asio::const_buffer &get(void) const{
-        return _buff;
-    }
-
-    const boost::asio::const_buffer _buff;
-    ring_buffer_info *_info;
-};
-
-/***********************************************************************
- * The managed send buffer implementation
- **********************************************************************/
-class usrp_e_mmap_managed_send_buffer : public managed_send_buffer{
-public:
-    usrp_e_mmap_managed_send_buffer(
-        void *mem, size_t len, ring_buffer_info *info, int fd
-    ):
-        _buff(mem, len), _info(info), _fd(fd), _commited(false)
-    {
-        /* NOP */
-    }
-
-    ~usrp_e_mmap_managed_send_buffer(void){
-        if (not _commited) this->commit(0);
-    }
-
-    ssize_t commit(size_t num_bytes){
-        _commited = true;
-        _info->len = num_bytes;
-        _info->flags = RB_USER;
-        ssize_t ret = ::write(_fd, NULL, 0);
-        return (ret < 0)? ret : num_bytes;
-    }
-
-private:
-    const boost::asio::mutable_buffer &get(void) const{
-        return _buff;
-    }
-
-    const boost::asio::mutable_buffer _buff;
-    ring_buffer_info *_info;
-    int _fd;
-    bool _commited;
-};
+static const bool fp_verbose = true; //fast-path verbose
+static const bool sp_verbose = true; //slow-path verbose
 
 /***********************************************************************
  * The zero copy interface implementation
  **********************************************************************/
-class usrp_e_mmap_zero_copy_impl : public zero_copy_if{
+class usrp_e_mmap_zero_copy_impl : public zero_copy_if, public boost::enable_shared_from_this<usrp_e_mmap_zero_copy_impl> {
 public:
     usrp_e_mmap_zero_copy_impl(usrp_e_iface::sptr iface):
         _fd(iface->get_file_descriptor()), _recv_index(0), _send_index(0)
@@ -105,15 +46,26 @@ public:
         _frame_size = page_size/2;
 
         //calculate the memory size
-        size_t map_size =
+        _map_size =
             (_rb_size.num_pages_rx_flags + _rb_size.num_pages_tx_flags) * page_size +
             (_rb_size.num_rx_frames + _rb_size.num_tx_frames) * _frame_size;
 
+        //print sizes summary
+        if (sp_verbose){
+            std::cout << "page_size:          " << page_size                   << std::endl;
+            std::cout << "frame_size:         " << _frame_size                 << std::endl;
+            std::cout << "num_pages_rx_flags: " << _rb_size.num_pages_rx_flags << std::endl;
+            std::cout << "num_rx_frames:      " << _rb_size.num_rx_frames      << std::endl;
+            std::cout << "num_pages_tx_flags: " << _rb_size.num_pages_tx_flags << std::endl;
+            std::cout << "num_tx_frames:      " << _rb_size.num_tx_frames      << std::endl;
+            std::cout << "map_size:           " << _map_size                   << std::endl;
+        }
+
         //call mmap to get the memory
-        void *ring_buffer = mmap(
-            NULL, map_size, PROT_READ | PROT_WRITE, MAP_SHARED, _fd, 0
+        _mapped_mem = ::mmap(
+            NULL, _map_size, PROT_READ | PROT_WRITE, MAP_SHARED, _fd, 0
         );
-        UHD_ASSERT_THROW(ring_buffer != MAP_FAILED);
+        UHD_ASSERT_THROW(_mapped_mem != MAP_FAILED);
 
         //calculate the memory offsets for info and buffers
         size_t recv_info_off = 0;
@@ -121,16 +73,31 @@ public:
         size_t send_info_off = recv_buff_off + (_rb_size.num_rx_frames * _frame_size);
         size_t send_buff_off = send_info_off + (_rb_size.num_pages_tx_flags * page_size);
 
+        //print offset summary
+        if (sp_verbose){
+            std::cout << "recv_info_off: " << recv_info_off << std::endl;
+            std::cout << "recv_buff_off: " << recv_buff_off << std::endl;
+            std::cout << "send_info_off: " << send_info_off << std::endl;
+            std::cout << "send_buff_off: " << send_buff_off << std::endl;
+        }
+
         //set the internal pointers for info and buffers
         typedef ring_buffer_info (*rbi_pta)[];
-        boost::uint8_t *rb_ptr = reinterpret_cast<boost::uint8_t *>(ring_buffer);
+        char *rb_ptr = reinterpret_cast<char *>(_mapped_mem);
         _recv_info = reinterpret_cast<rbi_pta>(rb_ptr + recv_info_off);
         _recv_buff = rb_ptr + recv_buff_off;
         _send_info = reinterpret_cast<rbi_pta>(rb_ptr + send_info_off);
         _send_buff = rb_ptr + send_buff_off;
     }
 
-    managed_recv_buffer::sptr get_recv_buff(size_t timeout_ms){
+    ~usrp_e_mmap_zero_copy_impl(void){
+        if (sp_verbose) std::cout << "cleanup: munmap" << std::endl;
+        ::munmap(_mapped_mem, _map_size);
+    }
+
+    managed_recv_buffer::sptr get_recv_buff(double timeout){
+        if (fp_verbose) std::cout << "get_recv_buff: " << _recv_index << std::endl;
+
         //grab pointers to the info and buffer
         ring_buffer_info *info = (*_recv_info) + _recv_index;
         void *mem = _recv_buff + _frame_size*_recv_index;
@@ -140,7 +107,8 @@ public:
             pollfd pfd;
             pfd.fd = _fd;
             pfd.events = POLLIN;
-            ssize_t poll_ret = poll(&pfd, 1, timeout_ms);
+            ssize_t poll_ret = ::poll(&pfd, 1, size_t(timeout*1e3));
+            if (fp_verbose) std::cout << "  POLLIN: " << poll_ret << std::endl;
             if (poll_ret <= 0) return managed_recv_buffer::sptr();
         }
 
@@ -148,8 +116,10 @@ public:
         if (++_recv_index == size_t(_rb_size.num_rx_frames)) _recv_index = 0;
 
         //return the managed buffer for this frame
-        return managed_recv_buffer::sptr(
-            new usrp_e_mmap_managed_recv_buffer(mem, info->len, info)
+        if (fp_verbose) std::cout << "  make_recv_buff: " << info->len << std::endl;
+        return managed_recv_buffer::make_safe(
+            boost::asio::const_buffer(mem, info->len),
+            boost::bind(&usrp_e_mmap_zero_copy_impl::release, shared_from_this(), info)
         );
     }
 
@@ -157,7 +127,9 @@ public:
         return _rb_size.num_rx_frames;
     }
 
-    managed_send_buffer::sptr get_send_buff(void){
+    managed_send_buffer::sptr get_send_buff(double timeout){
+        if (fp_verbose) std::cout << "get_send_buff: " << _send_index << std::endl;
+
         //grab pointers to the info and buffer
         ring_buffer_info *info = (*_send_info) + _send_index;
         void *mem = _send_buff + _frame_size*_send_index;
@@ -167,7 +139,8 @@ public:
             pollfd pfd;
             pfd.fd = _fd;
             pfd.events = POLLOUT;
-            ssize_t poll_ret = poll(&pfd, 1, -1 /*forever*/);
+            ssize_t poll_ret = ::poll(&pfd, 1, size_t(timeout*1e3));
+            if (fp_verbose) std::cout << "  POLLOUT: " << poll_ret << std::endl;
             if (poll_ret <= 0) return managed_send_buffer::sptr();
         }
 
@@ -175,8 +148,10 @@ public:
         if (++_send_index == size_t(_rb_size.num_tx_frames)) _send_index = 0;
 
         //return the managed buffer for this frame
-        return managed_send_buffer::sptr(
-            new usrp_e_mmap_managed_send_buffer(mem, _frame_size, info, _fd)
+        if (fp_verbose) std::cout << "  make_send_buff: " << _frame_size << std::endl;
+        return managed_send_buffer::make_safe(
+            boost::asio::mutable_buffer(mem, _frame_size),
+            boost::bind(&usrp_e_mmap_zero_copy_impl::commit, shared_from_this(), info, _1)
         );
     }
 
@@ -185,11 +160,35 @@ public:
     }
 
 private:
+
+    void release(ring_buffer_info *info){
+        if (fp_verbose) std::cout << "recv buff: release" << std::endl;
+        info->flags = RB_KERNEL;
+    }
+
+    void commit(ring_buffer_info *info, size_t len){
+        if (fp_verbose) std::cout << "send buff: commit " << len << std::endl;
+        info->len = len;
+        info->flags = RB_USER;
+        if (::write(_fd, NULL, 0) < 0){
+            std::cerr << UHD_THROW_SITE_INFO("write error") << std::endl;
+        }
+    }
+
     int _fd;
+
+    //the mapped memory itself
+    void *_mapped_mem;
+
+    //mapped memory sizes
     usrp_e_ring_buffer_size_t _rb_size;
-    size_t _frame_size;
+    size_t _frame_size, _map_size;
+
+    //pointers to sections in the mapped memory
     ring_buffer_info (*_recv_info)[], (*_send_info)[];
-    boost::uint8_t *_recv_buff, *_send_buff;
+    char *_recv_buff, *_send_buff;
+
+    //indexes into sub-sections of mapped memory
     size_t _recv_index, _send_index;
 };