aboutsummaryrefslogtreecommitdiffstats
path: root/host/lib/transport/liberio_zero_copy.cpp
diff options
context:
space:
mode:
Diffstat (limited to 'host/lib/transport/liberio_zero_copy.cpp')
-rw-r--r--host/lib/transport/liberio_zero_copy.cpp258
1 files changed, 258 insertions, 0 deletions
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)
+ );
+}
+
+}}