transport: Implement a single-threaded I/O service

The inline_io_service connects transports to links without any
worker threads. Send operations go directly to the link, and recv
will perform the I/O as part of the get_recv_buffer() call.

The inline_io_service also supports muxed links natively. The receive
mux is entirely inline. There is no separate thread for the
inline_io_service, and that continues here. A queue is created for
each client of the mux, and packets are processed as they come in. If
a packet is to go up to a different client, the packet is queued up
for later. When that client attempts to recv(), the queue is checked
first, and the attempts to receive from the link happen ONLY if no
packet was found.

Also add mock transport to test I/O service APIs. Tests I/O service
construction and some basic packet transmision. One case will also
uses a single link that is shared between the send and recv transports.
That link is muxed between two compatible but different transports.

1 files changed, 121 insertions, 0 deletions

diff --git a/host/lib/include/uhdlib/transport/inline_io_service.hpp b/host/lib/include/uhdlib/transport/inline_io_service.hpp
new file mode 100644
index 000000000..f10e7018d
--- /dev/null
+++ b/host/lib/include/uhdlib/transport/inline_io_service.hpp
@@ -0,0 +1,121 @@
+//
+// Copyright 2019 Ettus Research, a National Instruments brand
+//
+// SPDX-License-Identifier: GPL-3.0-or-later
+//
+
+#ifndef INCLUDED_UHDLIB_TRANSPORT_INLINE_IO_SERVICE_HPP
+#define INCLUDED_UHDLIB_TRANSPORT_INLINE_IO_SERVICE_HPP
+
+#include <uhdlib/transport/io_service.hpp>
+#include <unordered_map>
+#include <list>
+
+namespace uhd { namespace transport {
+
+class inline_recv_mux;
+class inline_recv_cb;
+
+/*!
+ * Single-threaded I/O service
+ * Note this is not an appropriate io_service to use with polling-mode drivers,
+ * since such drivers require a thread to poll them and not block (i.e.
+ * timeouts are not allowed at the link interface)
+ */
+class inline_io_service : public virtual io_service,
+                          public std::enable_shared_from_this<inline_io_service>
+{
+public:
+    using sptr = std::shared_ptr<inline_io_service>;
+    static sptr make(void)
+    {
+        return sptr(new inline_io_service());
+    }
+
+    ~inline_io_service();
+
+    void attach_recv_link(recv_link_if::sptr link);
+    void attach_send_link(send_link_if::sptr link);
+
+    recv_io_if::sptr make_recv_client(recv_link_if::sptr data_link,
+        size_t num_recv_frames,
+        recv_callback_t cb,
+        send_link_if::sptr fc_link,
+        size_t num_send_frames,
+        recv_io_if::fc_callback_t fc_cb);
+
+    send_io_if::sptr make_send_client(send_link_if::sptr send_link,
+        size_t num_send_frames,
+        send_io_if::send_callback_t send_cb,
+        recv_link_if::sptr recv_link,
+        size_t num_recv_frames,
+        recv_callback_t recv_cb);
+
+private:
+    friend class inline_recv_io;
+    friend class inline_send_io;
+
+    inline_io_service()                         = default;
+    inline_io_service(const inline_io_service&) = delete;
+
+    /*!
+     * Senders are free to mux a send_link, but the total reserved send_frames
+     * must be less than or equal to the link's capacity
+     *
+     * \param link the link used for sending data
+     * \param num_frames number of frames to reserve for this connection
+     */
+    void connect_sender(send_link_if* link, size_t num_frames);
+
+    /*!
+     * Disconnect the sender and free resources
+     *
+     * \param link the link that was used for sending data
+     * \param num_frames number of frames to release (same as reservation)
+     */
+    void disconnect_sender(send_link_if* link, size_t num_frames);
+
+    /*!
+     * Connect a receiver to the link and reserve resources
+     * \param link the recv link to use for getting data
+     * \param cb a callback for processing received data
+     * \param num_frames the number of frames to reserve for this receiver
+     */
+    void connect_receiver(recv_link_if* link, inline_recv_cb* cb, size_t num_frames);
+
+    /*!
+     * Disconnect the receiver from the provided link and free resources
+     * \param link the recv link that was used for reception
+     * \param cb the callback to disassociate
+     * \param num_frames the number of frames that was reserved for the cb
+     */
+    void disconnect_receiver(recv_link_if* link, inline_recv_cb* cb, size_t num_frames);
+
+    /*
+     * Function to perform recv operations on a link, which is potentially
+     * muxed. Packets are forwarded to the appropriate mux or callback.
+     *
+     * \param recv_io_cb the callback+interface initiating the operation
+     * \param recv_link link to perform receive on
+     * \param timeout_ms timeout to wait for a buffer on the link
+     * \return a frame_buff uptr with either a buffer with data or no buffer
+     */
+    frame_buff::uptr recv(
+        inline_recv_cb* recv_io_cb, recv_link_if* recv_link, int32_t timeout_ms);
+
+    /* Track whether link is muxed, the callback, and buffer reservations */
+    std::unordered_map<recv_link_if*,
+        std::tuple<inline_recv_mux*, inline_recv_cb*, size_t>>
+        _recv_tbl;
+
+    /* Track how many send_frames have been reserved for each link */
+    std::unordered_map<send_link_if*, size_t> _send_tbl;
+
+    /* Shared ptr kept to avoid untimely release */
+    std::list<send_link_if::sptr> _send_links;
+    std::list<recv_link_if::sptr> _recv_links;
+};
+
+}} // namespace uhd::transport
+
+#endif /* INCLUDED_UHDLIB_TRANSPORT_INLINE_IO_SERVICE_HPP */