From d4bb4266b58dd5ef33040afb6c6f101f2c9895b1 Mon Sep 17 00:00:00 2001 From: Paul David Date: Mon, 11 Apr 2016 14:21:59 -0700 Subject: transport optimize: Added a thread transport offload to share the workload --- host/lib/transport/CMakeLists.txt | 1 + host/lib/transport/zero_copy_recv_offload.cpp | 158 ++++++++++++++++++++++++++ 2 files changed, 159 insertions(+) create mode 100644 host/lib/transport/zero_copy_recv_offload.cpp (limited to 'host/lib/transport') diff --git a/host/lib/transport/CMakeLists.txt b/host/lib/transport/CMakeLists.txt index 79c8a90b7..db21b9f8e 100644 --- a/host/lib/transport/CMakeLists.txt +++ b/host/lib/transport/CMakeLists.txt @@ -122,6 +122,7 @@ LIBUHD_PYTHON_GEN_SOURCE( ) LIBUHD_APPEND_SOURCES( + ${CMAKE_CURRENT_SOURCE_DIR}/zero_copy_recv_offload.cpp ${CMAKE_CURRENT_SOURCE_DIR}/tcp_zero_copy.cpp ${CMAKE_CURRENT_SOURCE_DIR}/buffer_pool.cpp ${CMAKE_CURRENT_SOURCE_DIR}/if_addrs.cpp diff --git a/host/lib/transport/zero_copy_recv_offload.cpp b/host/lib/transport/zero_copy_recv_offload.cpp new file mode 100644 index 000000000..e8b013abc --- /dev/null +++ b/host/lib/transport/zero_copy_recv_offload.cpp @@ -0,0 +1,158 @@ +// +// Copyright 2016 Ettus Research +// +// This program is free software: you can redistribute it and/or modify +// it under the terms of the GNU General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// This program is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU General Public License for more details. +// +// You should have received a copy of the GNU General Public License +// along with this program. If not, see . +// + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +using namespace uhd; +using namespace uhd::transport; + +typedef bounded_buffer bounded_buffer_t; + +/*********************************************************************** + * Zero copy offload transport: + * An intermediate transport that utilizes threading to free + * the main thread from any receive work. + **********************************************************************/ +class zero_copy_recv_offload_impl : public zero_copy_recv_offload { +public: + typedef boost::shared_ptr sptr; + + zero_copy_recv_offload_impl(zero_copy_if::sptr transport, + const double timeout) : + _transport(transport), _timeout(timeout), + _inbox(transport->get_num_recv_frames()), + _recv_done(false) + { + UHD_LOG << "Created threaded transport" << std::endl; + + // Create the receive and send threads to offload + // the system calls onto other threads + _recv_thread = boost::thread( + boost::bind(&zero_copy_recv_offload_impl::enqueue_recv, this) + ); + } + + // Receive thread flags + void set_recv_done() + { + boost::lock_guard guard(_recv_mutex); + _recv_done = true; + } + + bool is_recv_done() + { + boost::lock_guard guard(_recv_mutex); + return _recv_done; + } + + ~zero_copy_recv_offload_impl() + { + // Signal the threads we're finished + set_recv_done(); + + // Wait for them to join + UHD_SAFE_CALL( + _recv_thread.join(); + ) + } + + // The receive thread function is responsible for + // pulling pointers to managed receiver buffers quickly + void enqueue_recv() + { + while (not is_recv_done()) { + managed_recv_buffer::sptr buff = _transport->get_recv_buff(_timeout); + if (not buff) continue; + _inbox.push_with_timed_wait(buff, _timeout); + } + } + + /******************************************************************* + * Receive implementation: + * Pop the receive buffer pointer from the underlying transport + ******************************************************************/ + managed_recv_buffer::sptr get_recv_buff(double timeout) + { + managed_recv_buffer::sptr ptr; + _inbox.pop_with_timed_wait(ptr, timeout); + return ptr; + } + + size_t get_num_recv_frames() const + { + return _transport->get_num_recv_frames(); + } + + size_t get_recv_frame_size() const + { + return _transport->get_recv_frame_size(); + } + + /******************************************************************* + * Send implementation: + * Pass the send buffer pointer from the underlying transport + ******************************************************************/ + managed_send_buffer::sptr get_send_buff(double timeout) + { + return _transport->get_send_buff(timeout); + } + + size_t get_num_send_frames() const + { + return _transport->get_num_send_frames(); + } + + size_t get_send_frame_size() const + { + return _transport->get_send_frame_size(); + } + +private: + // The linked transport + zero_copy_if::sptr _transport; + + const double _timeout; + + // Shared buffers + bounded_buffer_t _inbox; + + // Threading + bool _recv_done; + boost::thread _recv_thread; + boost::mutex _recv_mutex; +}; + +zero_copy_recv_offload::sptr zero_copy_recv_offload::make( + zero_copy_if::sptr transport, + const double timeout) +{ + zero_copy_recv_offload_impl::sptr zero_copy_recv_offload( + new zero_copy_recv_offload_impl(transport, timeout) + ); + + return zero_copy_recv_offload; +} -- cgit v1.2.3