transport: Added liberio_zero_copy transport implementation

This adds a zero copy transport using the liberio library.

Currently supported API version for liberio is 0.3, this might
still very much break, since the library is still in development.

So far nobody uses it UHD so we might as well merge it.

Signed-off-by: Alex Williams <alex.williams@ni.com>
Signed-off-by: Moritz Fischer <moritz.fischer@ettus.com>

3 files changed, 304 insertions, 0 deletions

diff --git a/host/lib/transport/CMakeLists.txt b/host/lib/transport/CMakeLists.txt
index a6d84cc4a..c8fd16ef3 100644
--- a/host/lib/transport/CMakeLists.txt
+++ b/host/lib/transport/CMakeLists.txt
@@ -138,6 +138,16 @@ IF(ENABLE_X300)
     )
 ENDIF(ENABLE_X300)
 
+IF(ENABLE_LIBERIO)
+    MESSAGE(STATUS "")
+    MESSAGE(STATUS "liberio support enabled.")
+    INCLUDE_DIRECTORIES(${LIBERIO_INCLUDE_DIRS})
+    LIBUHD_APPEND_LIBS(${LIBERIO_LIBRARIES})
+    LIBUHD_APPEND_SOURCES(
+        ${CMAKE_CURRENT_SOURCE_DIR}/liberio_zero_copy.cpp
+    )
+ENDIF(ENABLE_LIBERIO)
+
 # Verbose Debug output for send/recv
 SET( UHD_TXRX_DEBUG_PRINTS OFF CACHE BOOL "Use verbose debug output for send/recv" )
 OPTION( UHD_TXRX_DEBUG_PRINTS "Use verbose debug output for send/recv" "" )
diff --git a/host/lib/transport/liberio_zero_copy.cpp b/host/lib/transport/liberio_zero_copy.cpp
new file mode 100644
index 000000000..666872738
--- /dev/null
+++ b/host/lib/transport/liberio_zero_copy.cpp
@@ -0,0 +1,258 @@
+//
+// Copyright 2017 Ettus Research
+//
+// SPDX-License-Identifier: GPL-3.0+
+//
+
+#include "liberio_zero_copy.hpp"
+#include <uhd/config.hpp>
+#include <uhd/utils/static.hpp>
+#include <uhd/utils/log.hpp>
+#include <liberio/liberio.h>
+#include <boost/make_shared.hpp>
+#include <sys/syslog.h>
+
+namespace uhd { namespace transport {
+
+static const uint64_t USEC = 1000000;
+
+static void liberio_log_cb(int severity, const char *msg, void *)
+{
+    switch (severity) {
+    case LOG_WARNING:
+        UHD_LOG_WARNING("LIBERIO", msg);
+        return;
+    case LOG_NOTICE:
+    case LOG_INFO:
+        UHD_LOG_INFO("LIBERIO", msg);
+        return;
+    default:
+        UHD_LOG_INFO("LIBERIO", msg);
+    };
+}
+
+class liberio_context_holder {
+public:
+    liberio_context_holder(void)
+    {
+        _ctx = liberio_ctx_new();
+        liberio_ctx_register_logger(_ctx, &liberio_log_cb, nullptr);
+    }
+
+    ~liberio_context_holder(void) { liberio_ctx_put(_ctx); }
+
+    liberio_ctx* get(void)
+    {
+        liberio_ctx_get(_ctx);
+        return _ctx;
+    }
+private:
+    liberio_ctx *_ctx;
+};
+
+UHD_SINGLETON_FCN(liberio_context_holder, get_liberio_context_holder);
+
+class liberio_zero_copy_msb : public virtual managed_send_buffer {
+public:
+    liberio_zero_copy_msb(liberio_chan *chan) : _chan(chan), _buf(nullptr) {}
+    ~liberio_zero_copy_msb(void)
+    {
+        liberio_chan_put(_chan);
+    }
+
+    void release(void)
+    {
+        if (_buf) {
+            liberio_buf_set_payload(_buf, 0, _length);
+            liberio_chan_buf_enqueue(_chan, _buf);
+        }
+    }
+
+    sptr get_new(int timeout, size_t &index)
+    {
+        _buf = liberio_chan_buf_dequeue(_chan, timeout * USEC);
+        if (!_buf)
+            return sptr();
+
+        index++;
+
+        return make(this, liberio_buf_get_mem(_buf, 0),
+        liberio_buf_get_len(_buf, 0));
+    }
+
+private:
+    liberio_chan *_chan;
+    liberio_buf *_buf;
+};
+
+class liberio_zero_copy_mrb : public virtual managed_recv_buffer {
+public:
+    liberio_zero_copy_mrb(liberio_chan *chan) : _chan(chan), _buf(nullptr) {}
+    ~liberio_zero_copy_mrb(void)
+    {
+        liberio_chan_put(_chan);
+    }
+
+    void release(void)
+    {
+        if (_buf)
+            liberio_chan_buf_enqueue(_chan, _buf);
+    }
+
+    sptr get_new(int timeout, size_t &index)
+    {
+        _buf = liberio_chan_buf_dequeue(_chan, timeout * USEC);
+        if (!_buf)
+            return sptr();
+
+        index++;
+
+        return make(this, liberio_buf_get_mem(_buf, 0),
+        liberio_buf_get_payload(_buf, 0));
+    }
+
+private:
+    liberio_chan *_chan;
+    liberio_buf *_buf;
+};
+
+class liberio_zero_copy_impl : public liberio_zero_copy {
+public:
+
+    liberio_zero_copy_impl(const std::string &tx_path,
+                           const std::string &rx_path,
+                           const zero_copy_xport_params& xport_params)
+                          : _tx_buf_size(xport_params.send_frame_size),
+                            _rx_buf_size(xport_params.recv_frame_size),
+                            _next_recv_buff_index(0),
+                            _next_send_buff_index(0)
+    {
+        UHD_ASSERT_THROW(xport_params.recv_frame_size > 0);
+        UHD_ASSERT_THROW(xport_params.send_frame_size > 0);
+        UHD_ASSERT_THROW(xport_params.num_send_frames > 0);
+        UHD_ASSERT_THROW(xport_params.num_recv_frames > 0);
+
+        liberio_ctx *ctx = get_liberio_context_holder().get();
+
+        /* we hold a reference, that we'd drop immediately after,
+         * so no requirement to get another one here ...
+         */
+        _tx_chan = liberio_ctx_alloc_chan(ctx, tx_path.c_str(), TX,
+                           USRP_MEMORY_MMAP);
+
+        UHD_ASSERT_THROW(_tx_chan);
+        liberio_chan_stop_streaming(_tx_chan);
+        liberio_chan_request_buffers(_tx_chan, 0);
+        liberio_chan_set_fixed_size(_tx_chan, 0,
+            xport_params.send_frame_size);
+        UHD_ASSERT_THROW(
+            !liberio_chan_request_buffers(
+                _tx_chan, xport_params.num_send_frames
+            )
+        );
+	_num_send_bufs = liberio_chan_get_num_bufs(_tx_chan);
+
+        for (size_t i = 0; i < xport_params.num_send_frames; i++) {
+            liberio_chan_get(_tx_chan);
+            _msb_pool.push_back(
+                boost::make_shared<liberio_zero_copy_msb>(_tx_chan));
+        }
+
+        /* we hold a reference, that we'd drop immediately after,
+         * so no requirement to get another one here ...
+         */
+        _rx_chan = liberio_ctx_alloc_chan(ctx, rx_path.c_str(),
+                          RX, USRP_MEMORY_MMAP);
+        UHD_ASSERT_THROW(_rx_chan);
+
+        /* done with the local reference, the channel keeps its own */
+        liberio_ctx_put(ctx);
+
+        /* stop the channel, free the buffers, set the size, allocate */
+        liberio_chan_stop_streaming(_rx_chan);
+        liberio_chan_request_buffers(_rx_chan, 0);
+        liberio_chan_set_fixed_size(_rx_chan, 0,
+                     xport_params.recv_frame_size);
+        UHD_ASSERT_THROW(!liberio_chan_request_buffers(
+                _rx_chan, xport_params.num_recv_frames));
+	_num_recv_bufs = liberio_chan_get_num_bufs(_rx_chan);
+
+        for (size_t i = 0; i < xport_params.num_recv_frames; i++) {
+            liberio_chan_get(_rx_chan);
+            _mrb_pool.push_back(
+                boost::make_shared<liberio_zero_copy_mrb>(_rx_chan));
+        }
+
+        UHD_ASSERT_THROW(!liberio_chan_enqueue_all(_rx_chan));
+
+        liberio_chan_start_streaming(_rx_chan);
+        liberio_chan_start_streaming(_tx_chan);
+    }
+
+    ~liberio_zero_copy_impl(void)
+    {
+        liberio_chan_put(_tx_chan);
+        liberio_chan_put(_rx_chan);
+    }
+
+    managed_recv_buffer::sptr get_recv_buff(double timeout = 0.1)
+    {
+        if (_next_recv_buff_index == _num_recv_bufs)
+            _next_recv_buff_index = 0;
+        return _mrb_pool[_next_recv_buff_index]->get_new(
+                USEC * timeout, _next_recv_buff_index);
+    }
+
+    size_t get_num_recv_frames(void) const
+    {
+        return liberio_chan_get_num_bufs(_rx_chan);
+    }
+
+    size_t get_recv_frame_size(void) const
+    {
+        return _rx_buf_size;
+    }
+
+    managed_send_buffer::sptr get_send_buff(double timeout = 0.1)
+    {
+        if (_next_send_buff_index == _num_send_bufs)
+            _next_send_buff_index = 0;
+        return _msb_pool[_next_send_buff_index]->get_new(
+            timeout, _next_send_buff_index);
+    }
+
+    size_t get_num_send_frames(void) const
+    {
+        return liberio_chan_get_num_bufs(_tx_chan);
+    }
+
+    size_t get_send_frame_size(void) const
+    {
+        return _tx_buf_size;
+    }
+
+private:
+    liberio_chan *_tx_chan;
+    const size_t _tx_buf_size;
+    size_t _num_send_bufs;
+    liberio_chan *_rx_chan;
+    const size_t _rx_buf_size;
+    size_t _num_recv_bufs;
+
+    std::vector<boost::shared_ptr<liberio_zero_copy_mrb> > _mrb_pool;
+    size_t _next_recv_buff_index;
+    std::vector<boost::shared_ptr<liberio_zero_copy_msb> > _msb_pool;
+    size_t _next_send_buff_index;
+};
+
+liberio_zero_copy::sptr liberio_zero_copy::make(
+    const std::string &tx_path,
+    const std::string &rx_path,
+    const zero_copy_xport_params &default_buff_args)
+{
+    return liberio_zero_copy::sptr(
+        new liberio_zero_copy_impl(tx_path, rx_path, default_buff_args)
+    );
+}
+
+}}
diff --git a/host/lib/transport/liberio_zero_copy.hpp b/host/lib/transport/liberio_zero_copy.hpp
new file mode 100644
index 000000000..bf23b1321
--- /dev/null
+++ b/host/lib/transport/liberio_zero_copy.hpp
@@ -0,0 +1,36 @@
+//
+// Copyright 2017 Ettus Research
+//
+// SPDX-License-Identifier: GPL-3.0+
+//
+
+#ifndef LIBERIO_HPP
+#define LIBERIO_HPP
+
+#include <string>
+#include <vector>
+
+#include <uhd/config.hpp>
+#include <uhd/transport/zero_copy.hpp>
+#include <uhd/types/device_addr.hpp>
+#include <boost/shared_ptr.hpp>
+
+namespace uhd { namespace transport {
+
+/*!
+ * A zero copy transport interface to the liberio DMA library.
+ */
+class liberio_zero_copy : public virtual zero_copy_if {
+public:
+    typedef boost::shared_ptr<liberio_zero_copy> sptr;
+
+    static sptr make(
+        const std::string &tx_path,
+        const std::string &rx_path,
+        const zero_copy_xport_params &default_buff_args
+    );
+};
+
+}}
+
+#endif /* LIBERIO_HPP */