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>

host/cmake/Modules/FindLIBERIO.cmake

@@ -0,0 +1,34 @@
+#
+# Copyright 2017 Ettus Research
+#
+# SPDX-License-Identifier: GPL-3.0+
+#
+# - Find liberio
+# Find the liberio includes and client library
+# This module defines
+#  LIBERIO_INCLUDE_DIR, where to find liberio/dma.h
+#  LIBERIO_LIBRARIES, the libraries needed by a liberio client.
+#  LIBERIO_FOUND, If false, do not try to use liberio.
+# also defined, but not for general use are
+#  LIBERIO_LIBRARY, where to find the liberio library.
+
+FIND_PACKAGE(PkgConfig)
+PKG_CHECK_MODULES(PC_LIBERIO QUIET liberio >= 0.3)
+
+FIND_PATH(LIBERIO_INCLUDE_DIR liberio/liberio.h
+	HINTS $ENV{LIBERIO_DIR}/include ${PC_LIBERIO_INCLUDE_DIR}
+	PATH_SUFFIXES liberio
+)
+
+FIND_LIBRARY(LIBERIO_LIBRARY
+	NAMES erio liberio
+	HINTS $ENV{LIBERIO_DIR}/lib ${PC_LIBERIO_LIBDIR} ${PC_LIBERIO_LIBRARY_DIRS}
+)
+
+include(FindPackageHandleStandardArgs)
+FIND_PACKAGE_HANDLE_STANDARD_ARGS(LIBERIO DEFAULT_MSG LIBERIO_LIBRARY LIBERIO_INCLUDE_DIR)
+MARK_AS_ADVANCED(LIBERIO_INCLUDE_DIR LIBERIO_LIBRARY)
+
+set(LIBERIO_LIBRARIES ${LIBERIO_LIBRARY})
+set(LIBERIO_INCLUDE_DIRS ${LIBERIO_INCLUDE_DIR})
+

host/include/config.h.in

@@ -23,6 +23,7 @@
 #cmakedefine UHD_VERSION_ABI   ${UHD_VERSION_ABI}
 #cmakedefine UHD_VERSION_PATCH ${UHD_VERSION_PATCH}
 #cmakedefine ENABLE_USB
+#cmakedefine ENABLE_LIBERIO
 #ifndef UHD_VERSION
 #cmakedefine UHD_VERSION @UHD_VERSION_ADDED@
 #endif

host/lib/CMakeLists.txt

@@ -71,6 +71,7 @@ MESSAGE(STATUS "")
 # Dependencies
 FIND_PACKAGE(USB1)
 FIND_PACKAGE(GPSD)
+FIND_PACKAGE(LIBERIO)
 LIBUHD_REGISTER_COMPONENT("USB" ENABLE_USB ON "ENABLE_LIBUHD;LIBUSB_FOUND" OFF OFF)
 LIBUHD_REGISTER_COMPONENT("GPSD" ENABLE_GPSD OFF "ENABLE_LIBUHD;ENABLE_GPSD;LIBGPS_FOUND" OFF OFF)
 # Devices
@@ -82,6 +83,7 @@ LIBUHD_REGISTER_COMPONENT("USRP1" ENABLE_USRP1 ON "ENABLE_LIBUHD;ENABLE_USB" OFF
 LIBUHD_REGISTER_COMPONENT("USRP2" ENABLE_USRP2 ON "ENABLE_LIBUHD" OFF OFF)
 LIBUHD_REGISTER_COMPONENT("X300" ENABLE_X300 ON "ENABLE_LIBUHD" OFF OFF)
 LIBUHD_REGISTER_COMPONENT("N230" ENABLE_N230 ON "ENABLE_LIBUHD" OFF OFF)
+LIBUHD_REGISTER_COMPONENT("LIBERIO" ENABLE_LIBERIO ON "ENABLE_LIBUHD;LIBERIO_FOUND" OFF OFF)
 LIBUHD_REGISTER_COMPONENT("OctoClock" ENABLE_OCTOCLOCK ON "ENABLE_LIBUHD" OFF OFF)
 
 ########################################################################

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" "" )

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)
+    );
+}
+
+}}

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 */