stream server uses async send and receive

client/browseZeroConf/browse_bonjour.cpp

@@ -192,10 +192,10 @@ bool BrowseBonjour::browse(const string& serviceName, mDNSResult& result, int /*
     {
         DNSServiceHandle service(new DNSServiceRef(NULL));
         for (auto& reply : replyCollection)
-            CHECKED(
-                DNSServiceResolve(service.get(), 0, 0, reply.name.c_str(), reply.regtype.c_str(), reply.domain.c_str(),
-                                  [](DNSServiceRef /*service*/, DNSServiceFlags /*flags*/, uint32_t /*interfaceIndex*/, DNSServiceErrorType errorCode, const char* /*fullName*/,
-                                     const char* hosttarget, uint16_t port, uint16_t /*txtLen*/, const unsigned char* /*txtRecord*/, void* context) {
+            CHECKED(DNSServiceResolve(service.get(), 0, 0, reply.name.c_str(), reply.regtype.c_str(), reply.domain.c_str(),
+                                      [](DNSServiceRef /*service*/, DNSServiceFlags /*flags*/, uint32_t /*interfaceIndex*/, DNSServiceErrorType errorCode,
+                                         const char* /*fullName*/, const char* hosttarget, uint16_t port, uint16_t /*txtLen*/,
+                                         const unsigned char* /*txtRecord*/, void* context) {
                                           auto resultCollection = static_cast<deque<mDNSResolve>*>(context);
 
                                           CHECKED(errorCode);

server/control_server.cpp

@@ -32,7 +32,7 @@ using namespace std;
 using json = nlohmann::json;
 
 
-ControlServer::ControlServer(boost::asio::io_context* io_context, const ServerSettings::TcpSettings& tcp_settings,
+ControlServer::ControlServer(boost::asio::io_context& io_context, const ServerSettings::TcpSettings& tcp_settings,
                              const ServerSettings::HttpSettings& http_settings, ControlMessageReceiver* controlMessageReceiver)
     : io_context_(io_context), tcp_settings_(tcp_settings), http_settings_(http_settings), controlMessageReceiver_(controlMessageReceiver)
 {
@@ -145,7 +145,7 @@ void ControlServer::start()
             {
                 LOG(INFO) << "Creating TCP acceptor for address: " << address << ", port: " << tcp_settings_.port << "\n";
                 acceptor_tcp_.emplace_back(
-                    make_unique<tcp::acceptor>(*io_context_, tcp::endpoint(boost::asio::ip::address::from_string(address), tcp_settings_.port)));
+                    make_unique<tcp::acceptor>(io_context_, tcp::endpoint(boost::asio::ip::address::from_string(address), tcp_settings_.port)));
             }
             catch (const boost::system::system_error& e)
             {
@@ -161,7 +161,7 @@ void ControlServer::start()
             {
                 LOG(INFO) << "Creating HTTP acceptor for address: " << address << ", port: " << http_settings_.port << "\n";
                 acceptor_http_.emplace_back(
-                    make_unique<tcp::acceptor>(*io_context_, tcp::endpoint(boost::asio::ip::address::from_string(address), http_settings_.port)));
+                    make_unique<tcp::acceptor>(io_context_, tcp::endpoint(boost::asio::ip::address::from_string(address), http_settings_.port)));
             }
             catch (const boost::system::system_error& e)
             {

server/control_server.hpp

@@ -45,7 +45,7 @@ using acceptor_ptr = std::unique_ptr<tcp::acceptor>;
 class ControlServer : public ControlMessageReceiver
 {
 public:
-    ControlServer(boost::asio::io_context* io_context, const ServerSettings::TcpSettings& tcp_settings, const ServerSettings::HttpSettings& http_settings,
+    ControlServer(boost::asio::io_context& io_context, const ServerSettings::TcpSettings& tcp_settings, const ServerSettings::HttpSettings& http_settings,
                   ControlMessageReceiver* controlMessageReceiver = nullptr);
     virtual ~ControlServer();
 
@@ -71,7 +71,7 @@ private:
     std::vector<acceptor_ptr> acceptor_tcp_;
     std::vector<acceptor_ptr> acceptor_http_;
 
-    boost::asio::io_context* io_context_;
+    boost::asio::io_context& io_context_;
     ServerSettings::TcpSettings tcp_settings_;
     ServerSettings::HttpSettings http_settings_;
     ControlMessageReceiver* controlMessageReceiver_;

server/snapserver.cpp

@@ -265,11 +265,10 @@ int main(int argc, char* argv[])
         }
 
         boost::asio::io_context io_context;
-        std::unique_ptr<StreamServer> streamServer(new StreamServer(&io_context, settings));
+        std::unique_ptr<StreamServer> streamServer(new StreamServer(io_context, settings));
         streamServer->start();
 
-        auto func = [](boost::asio::io_context* ioservice) -> void { ioservice->run(); };
-        std::thread t(func, &io_context);
+        std::thread t([&] { io_context.run(); });
 
         while (!g_terminated)
             chronos::sleep(100);

server/stream_server.cpp

@@ -29,7 +29,7 @@ using namespace std;
 using json = nlohmann::json;
 
 
-StreamServer::StreamServer(boost::asio::io_context* io_context, const ServerSettings& serverSettings) : io_context_(io_context), settings_(serverSettings)
+StreamServer::StreamServer(boost::asio::io_context& io_context, const ServerSettings& serverSettings) : io_context_(io_context), settings_(serverSettings)
 {
 }
 
@@ -76,8 +76,15 @@ void StreamServer::onChunkRead(const PcmStream* pcmStream, msg::PcmChunk* chunk,
     //	LOG(INFO) << "onChunkRead (" << pcmStream->getName() << "): " << duration << "ms\n";
     bool isDefaultStream(pcmStream == streamManager_->getDefaultStream().get());
 
-    msg::message_ptr shared_message(chunk);
     std::lock_guard<std::recursive_mutex> mlock(sessionsMutex_);
+    unique_ptr<msg::PcmChunk> chunk_ptr(chunk);
+
+    std::ostringstream oss;
+    tv t;
+    chunk_ptr->sent = t;
+    chunk_ptr->serialize(oss);
+    shared_const_buffer buffer(oss.str());
+
     for (auto s : sessions_)
     {
         if (!settings_.stream.sendAudioToMutedClients)
@@ -95,9 +102,9 @@ void StreamServer::onChunkRead(const PcmStream* pcmStream, msg::PcmChunk* chunk,
         }
 
         if (!s->pcmStream() && isDefaultStream) //->getName() == "default")
-            s->sendAsync(shared_message);
+            s->sendAsync(buffer);
         else if (s->pcmStream().get() == pcmStream)
-            s->sendAsync(shared_message);
+            s->sendAsync(buffer);
     }
 }
 
@@ -789,7 +796,7 @@ void StreamServer::handleAccept(tcp::socket socket)
         socket.set_option(tcp::no_delay(true));
 
         SLOG(NOTICE) << "StreamServer::NewConnection: " << socket.remote_endpoint().address().to_string() << endl;
-        shared_ptr<StreamSession> session = make_shared<StreamSession>(this, std::move(socket));
+        shared_ptr<StreamSession> session = make_shared<StreamSession>(io_context_, this, std::move(socket));
 
         session->setBufferMs(settings_.stream.bufferMs);
         session->start();
@@ -828,7 +835,7 @@ void StreamServer::start()
             {
                 LOG(INFO) << "Creating stream acceptor for address: " << address << ", port: " << settings_.stream.port << "\n";
                 acceptor_.emplace_back(
-                    make_unique<tcp::acceptor>(*io_context_, tcp::endpoint(boost::asio::ip::address::from_string(address), settings_.stream.port)));
+                    make_unique<tcp::acceptor>(io_context_, tcp::endpoint(boost::asio::ip::address::from_string(address), settings_.stream.port)));
             }
             catch (const boost::system::system_error& e)
             {

server/stream_server.hpp

@@ -55,7 +55,7 @@ using session_ptr = std::shared_ptr<StreamSession>;
 class StreamServer : public MessageReceiver, ControlMessageReceiver, PcmListener
 {
 public:
-    StreamServer(boost::asio::io_context* io_context, const ServerSettings& serverSettings);
+    StreamServer(boost::asio::io_context& io_context, const ServerSettings& serverSettings);
     virtual ~StreamServer();
 
     void start();
@@ -85,7 +85,7 @@ private:
     void ProcessRequest(const jsonrpcpp::request_ptr request, jsonrpcpp::entity_ptr& response, jsonrpcpp::notification_ptr& notification) const;
     mutable std::recursive_mutex sessionsMutex_;
     std::set<session_ptr> sessions_;
-    boost::asio::io_context* io_context_;
+    boost::asio::io_context& io_context_;
     std::vector<acceptor_ptr> acceptor_;
 
     ServerSettings settings_;

server/stream_session.cpp

@@ -26,10 +26,11 @@
 using namespace std;
 
 
-
-StreamSession::StreamSession(MessageReceiver* receiver, tcp::socket&& socket)
-    : active_(false), readerThread_(nullptr), writerThread_(nullptr), socket_(std::move(socket)), messageReceiver_(receiver), pcmStream_(nullptr)
+StreamSession::StreamSession(boost::asio::io_context& ioc, MessageReceiver* receiver, tcp::socket&& socket)
+    : buffer_pos_(0), socket_(std::move(socket)), messageReceiver_(receiver), pcmStream_(nullptr), strand_(ioc)
 {
+    base_msg_size_ = baseMessage_.getSize();
+    buffer_.resize(base_msg_size_);
 }
 
 
@@ -39,6 +40,57 @@ StreamSession::~StreamSession()
 }
 
 
+void StreamSession::read_next()
+{
+    auto self(shared_from_this());
+    boost::asio::async_read(socket_, boost::asio::buffer(buffer_, base_msg_size_),
+                            boost::asio::bind_executor(strand_, [this, self](boost::system::error_code ec, std::size_t length) mutable {
+                                if (ec)
+                                {
+                                    LOG(ERROR) << "Error reading message header: " << ec.message() << "\n";
+                                    messageReceiver_->onDisconnect(this);
+                                    return;
+                                }
+
+                                baseMessage_.deserialize(buffer_.data());
+                                LOG(DEBUG) << "getNextMessage: " << baseMessage_.type << ", size: " << baseMessage_.size << ", id: " << baseMessage_.id
+                                           << ", refers: " << baseMessage_.refersTo << "\n";
+                                if (baseMessage_.type > message_type::kLast)
+                                {
+                                    stringstream ss;
+                                    ss << "unknown message type received: " << baseMessage_.type << ", size: " << baseMessage_.size;
+                                    throw std::runtime_error(ss.str().c_str());
+                                }
+                                else if (baseMessage_.size > msg::max_size)
+                                {
+                                    stringstream ss;
+                                    ss << "received message of type " << baseMessage_.type << " to large: " << baseMessage_.size;
+                                    throw std::runtime_error(ss.str().c_str());
+                                }
+
+                                if (baseMessage_.size > buffer_.size())
+                                    buffer_.resize(baseMessage_.size);
+
+                                boost::asio::async_read(
+                                    socket_, boost::asio::buffer(buffer_, baseMessage_.size),
+                                    boost::asio::bind_executor(strand_, [this, self](boost::system::error_code ec, std::size_t length) mutable {
+                                        if (ec)
+                                        {
+                                            LOG(ERROR) << "Error reading message body: " << ec.message() << "\n";
+                                            messageReceiver_->onDisconnect(this);
+                                            return;
+                                        }
+
+                                        tv t;
+                                        baseMessage_.received = t;
+                                        if (messageReceiver_ != nullptr)
+                                            messageReceiver_->onMessageReceived(this, baseMessage_, buffer_.data());
+                                        read_next();
+                                    }));
+                            }));
+}
+
+
 void StreamSession::setPcmStream(PcmStreamPtr pcmStream)
 {
     pcmStream_ = pcmStream;
@@ -53,88 +105,69 @@ const PcmStreamPtr StreamSession::pcmStream() const
 
 void StreamSession::start()
 {
-    {
-        std::lock_guard<std::mutex> activeLock(activeMutex_);
-        active_ = true;
-    }
-    readerThread_.reset(new thread(&StreamSession::reader, this));
-    writerThread_.reset(new thread(&StreamSession::writer, this));
+    strand_.post([this]() { read_next(); });
 }
 
 
 void StreamSession::stop()
 {
-    {
-        std::lock_guard<std::mutex> activeLock(activeMutex_);
-        if (!active_)
-            return;
-
-        active_ = false;
-    }
-
-    try
-    {
+    LOG(DEBUG) << "StreamSession::stop\n";
     boost::system::error_code ec;
-        {
-            std::lock_guard<std::mutex> socketLock(socketMutex_);
     socket_.shutdown(boost::asio::ip::tcp::socket::shutdown_both, ec);
     if (ec)
         LOG(ERROR) << "Error in socket shutdown: " << ec.message() << "\n";
     socket_.close(ec);
     if (ec)
         LOG(ERROR) << "Error in socket close: " << ec.message() << "\n";
-        }
-        if (readerThread_ && readerThread_->joinable())
-        {
-            LOG(DEBUG) << "StreamSession joining readerThread\n";
-            readerThread_->join();
-        }
-        if (writerThread_ && writerThread_->joinable())
-        {
-            LOG(DEBUG) << "StreamSession joining writerThread\n";
-            messages_.abort_wait();
-            writerThread_->join();
-        }
-    }
-    catch (...)
-    {
-    }
-
-    readerThread_ = nullptr;
-    writerThread_ = nullptr;
     LOG(DEBUG) << "StreamSession stopped\n";
 }
 
 
-void StreamSession::socketRead(void* _to, size_t _bytes)
+void StreamSession::send_next()
 {
-    size_t read = 0;
-    do
+    auto self(shared_from_this());
+    auto buffer = messages_.front();
+
+    boost::asio::async_write(socket_, buffer, boost::asio::bind_executor(strand_, [this, self, buffer](boost::system::error_code ec, std::size_t length) {
+                                 // boost::asio::async_write(socket_, *buffer, [this, self](boost::system::error_code ec, std::size_t length) {
+                                 messages_.pop_front();
+                                 if (ec)
                                  {
-        read += socket_.read_some(boost::asio::buffer((char*)_to + read, _bytes - read));
-    } while (active_ && (read < _bytes));
+                                     LOG(ERROR) << "StreamSession write error: " << ec.message() << "\n";
+                                     messageReceiver_->onDisconnect(this);
+                                     return;
+                                 }
+                                 if (!messages_.empty())
+                                     send_next();
+                             }));
 }
 
 
-void StreamSession::sendAsync(const msg::message_ptr& message, bool sendNow)
+void StreamSession::sendAsync(shared_const_buffer const_buf)
+{
+    strand_.post([this, const_buf]() {
+        messages_.push_back(const_buf);
+        if (messages_.size() > 1)
+        {
+            LOG(DEBUG) << "outstanding async_write\n";
+            return;
+        }
+        send_next();
+    });
+}
+
+
+void StreamSession::sendAsync(msg::message_ptr message, bool sendNow)
 {
     if (!message)
         return;
 
-    // the writer will take care about old messages
-    while (messages_.size() > 2000) // chunk->getDuration() > 10000)
-        messages_.pop();
+    tv t;
+    message->sent = t;
+    std::ostringstream oss;
+    message->serialize(oss);
 
-    if (sendNow)
-        messages_.push_front(message);
-    else
-        messages_.push(message);
-}
-
-
-bool StreamSession::active() const
-{
-    return active_;
+    sendAsync(shared_const_buffer(oss.str()));
 }
 
 
@@ -144,115 +177,13 @@ void StreamSession::setBufferMs(size_t bufferMs)
 }
 
 
-bool StreamSession::send(const msg::message_ptr& message)
+bool StreamSession::send(msg::message_ptr message)
 {
-    // TODO on exception: set active = false
-    //	LOG(INFO) << "send: " << message->type << ", size: " << message->getSize() << ", id: " << message->id << ", refers: " << message->refersTo << "\n";
-    std::lock_guard<std::mutex> socketLock(socketMutex_);
-    {
-        std::lock_guard<std::mutex> activeLock(activeMutex_);
-        if (!active_)
-            return false;
-    }
-    boost::asio::streambuf streambuf;
-    std::ostream stream(&streambuf);
-    tv t;
-    message->sent = t;
-    message->serialize(stream);
-    boost::asio::write(socket_, streambuf);
-    //	LOG(INFO) << "done: " << message->type << ", size: " << message->size << ", id: " << message->id << ", refers: " << message->refersTo << "\n";
-    return true;
-}
-
-
-void StreamSession::getNextMessage()
-{
-    msg::BaseMessage baseMessage;
-    size_t baseMsgSize = baseMessage.getSize();
-    vector<char> buffer(baseMsgSize);
-    socketRead(&buffer[0], baseMsgSize);
-    baseMessage.deserialize(&buffer[0]);
-
-    if (baseMessage.type > message_type::kLast)
-    {
-        stringstream ss;
-        ss << "unknown message type received: " << baseMessage.type << ", size: " << baseMessage.size;
-        throw std::runtime_error(ss.str().c_str());
-    }
-    else if (baseMessage.size > msg::max_size)
-    {
-        stringstream ss;
-        ss << "received message of type " << baseMessage.type << " to large: " << baseMessage.size;
-        throw std::runtime_error(ss.str().c_str());
-    }
-
-    //	LOG(INFO) << "getNextMessage: " << baseMessage.type << ", size: " << baseMessage.size << ", id: " << baseMessage.id << ", refers: " <<
-    // baseMessage.refersTo << "\n";
-    if (baseMessage.size > buffer.size())
-        buffer.resize(baseMessage.size);
-
-    socketRead(&buffer[0], baseMessage.size);
-    tv t;
-    baseMessage.received = t;
-
-    if (active_ && (messageReceiver_ != nullptr))
-        messageReceiver_->onMessageReceived(this, baseMessage, &buffer[0]);
-}
-
-
-void StreamSession::reader()
-{
-    try
-    {
-        while (active_)
-        {
-            getNextMessage();
-        }
-    }
-    catch (const std::exception& e)
-    {
-        SLOG(ERROR) << "Exception in StreamSession::reader(): " << e.what() << endl;
-    }
-
-    if (active_ && (messageReceiver_ != nullptr))
-        messageReceiver_->onDisconnect(this);
-}
-
-
-void StreamSession::writer()
-{
-    try
-    {
-        boost::asio::streambuf streambuf;
-        std::ostream stream(&streambuf);
-        shared_ptr<msg::BaseMessage> message;
-        while (active_)
-        {
-            if (messages_.try_pop(message, std::chrono::milliseconds(500)))
-            {
-                if (bufferMs_ > 0)
-                {
-                    const msg::WireChunk* wireChunk = dynamic_cast<const msg::WireChunk*>(message.get());
-                    if (wireChunk != nullptr)
-                    {
-                        chronos::time_point_clk now = chronos::clk::now();
-                        size_t age = 0;
-                        if (now > wireChunk->start())
-                            age = std::chrono::duration_cast<chronos::msec>(now - wireChunk->start()).count();
-                        // LOG(DEBUG) << "PCM chunk. Age: " << age << ", buffer: " << bufferMs_ << ", age > buffer: " << (age > bufferMs_) << "\n";
-                        if (age > bufferMs_)
-                            continue;
-                    }
-                }
-                send(message);
-            }
-        }
-    }
-    catch (const std::exception& e)
-    {
-        SLOG(ERROR) << "Exception in StreamSession::writer(): " << e.what() << endl;
-    }
-
-    if (active_ && (messageReceiver_ != nullptr))
-        messageReceiver_->onDisconnect(this);
+    sendAsync(message);
+    // // TODO on exception: set active = false
+    // //	LOG(INFO) << "send: " << message->type << ", size: " << message->getSize() << ", id: " << message->id << ", refers: " << message->refersTo <<
+    // "\n"; boost::asio::streambuf streambuf; std::ostream stream(&streambuf); tv t; message->sent = t; message->serialize(stream); boost::asio::write(socket_,
+    // streambuf);
+    // //	LOG(INFO) << "done: " << message->type << ", size: " << message->size << ", id: " << message->id << ", refers: " << message->refersTo << "\n";
+    // return true;
 }

server/stream_session.hpp

@@ -25,11 +25,14 @@
 #include <atomic>
 #include <boost/asio.hpp>
 #include <condition_variable>
+#include <deque>
 #include <memory>
 #include <mutex>
 #include <set>
+#include <sstream>
 #include <string>
 #include <thread>
+#include <vector>
 
 
 using boost::asio::ip::tcp;
@@ -47,28 +50,58 @@ public:
 };
 
 
+// A reference-counted non-modifiable buffer class.
+class shared_const_buffer
+{
+public:
+    // Construct from a std::string.
+    explicit shared_const_buffer(const std::string& data) : data_(new std::vector<char>(data.begin(), data.end())), buffer_(boost::asio::buffer(*data_))
+    {
+    }
+
+    // Implement the ConstBufferSequence requirements.
+    typedef boost::asio::const_buffer value_type;
+    typedef const boost::asio::const_buffer* const_iterator;
+    const boost::asio::const_buffer* begin() const
+    {
+        return &buffer_;
+    }
+    const boost::asio::const_buffer* end() const
+    {
+        return &buffer_ + 1;
+    }
+
+private:
+    std::shared_ptr<std::vector<char>> data_;
+    boost::asio::const_buffer buffer_;
+};
+
+
 /// Endpoint for a connected client.
 /**
  * Endpoint for a connected client.
  * Messages are sent to the client with the "send" method.
  * Received messages from the client are passed to the MessageReceiver callback
  */
-class StreamSession
+class StreamSession : public std::enable_shared_from_this<StreamSession>
 {
 public:
     /// ctor. Received message from the client are passed to MessageReceiver
-    StreamSession(MessageReceiver* receiver, tcp::socket&& socket);
+    StreamSession(boost::asio::io_context& ioc, MessageReceiver* receiver, tcp::socket&& socket);
     ~StreamSession();
     void start();
     void stop();
 
     /// Sends a message to the client (synchronous)
-    bool send(const msg::message_ptr& message);
+    bool send(msg::message_ptr message);
 
     /// Sends a message to the client (asynchronous)
-    void sendAsync(const msg::message_ptr& message, bool sendNow = false);
+    void sendAsync(msg::message_ptr message, bool sendNow = false);
 
-    bool active() const;
+    /// Sends a message to the client (asynchronous)
+    // void sendAsync(std::shared_ptr<boost::asio::streambuf> sb);
+
+    void sendAsync(shared_const_buffer const_buf);
 
     /// Max playout latency. No need to send PCM data that is older than bufferMs
     void setBufferMs(size_t bufferMs);
@@ -84,22 +117,19 @@ public:
     const PcmStreamPtr pcmStream() const;
 
 protected:
-    void socketRead(void* _to, size_t _bytes);
-    void getNextMessage();
-    void reader();
-    void writer();
+    void read_next();
+    void send_next();
 
-    mutable std::mutex activeMutex_;
-    std::atomic<bool> active_;
-
-    std::unique_ptr<std::thread> readerThread_;
-    std::unique_ptr<std::thread> writerThread_;
-    mutable std::mutex socketMutex_;
+    msg::BaseMessage baseMessage_;
+    std::vector<char> buffer_;
+    size_t buffer_pos_;
+    size_t base_msg_size_;
     tcp::socket socket_;
     MessageReceiver* messageReceiver_;
-    Queue<std::shared_ptr<msg::BaseMessage>> messages_;
     size_t bufferMs_;
     PcmStreamPtr pcmStream_;
+    boost::asio::io_context::strand strand_;
+    std::deque<shared_const_buffer> messages_;
 };