diff --git a/client/browseZeroConf/browse_bonjour.cpp b/client/browseZeroConf/browse_bonjour.cpp index 4bf99d97..8d7bd133 100644 --- a/client/browseZeroConf/browse_bonjour.cpp +++ b/client/browseZeroConf/browse_bonjour.cpp @@ -192,16 +192,16 @@ 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) { - auto resultCollection = static_cast*>(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*>(context); - CHECKED(errorCode); - resultCollection->push_back(mDNSResolve{string(hosttarget), ntohs(port)}); - }, - &resolveCollection)); + CHECKED(errorCode); + resultCollection->push_back(mDNSResolve{string(hosttarget), ntohs(port)}); + }, + &resolveCollection)); runService(service); } diff --git a/server/control_server.cpp b/server/control_server.cpp index daaab35e..743a977c 100644 --- a/server/control_server.cpp +++ b/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(*io_context_, tcp::endpoint(boost::asio::ip::address::from_string(address), tcp_settings_.port))); + make_unique(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(*io_context_, tcp::endpoint(boost::asio::ip::address::from_string(address), http_settings_.port))); + make_unique(io_context_, tcp::endpoint(boost::asio::ip::address::from_string(address), http_settings_.port))); } catch (const boost::system::system_error& e) { diff --git a/server/control_server.hpp b/server/control_server.hpp index caf526c6..f8bc773f 100644 --- a/server/control_server.hpp +++ b/server/control_server.hpp @@ -45,7 +45,7 @@ using acceptor_ptr = std::unique_ptr; 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_tcp_; std::vector acceptor_http_; - boost::asio::io_context* io_context_; + boost::asio::io_context& io_context_; ServerSettings::TcpSettings tcp_settings_; ServerSettings::HttpSettings http_settings_; ControlMessageReceiver* controlMessageReceiver_; diff --git a/server/snapserver.cpp b/server/snapserver.cpp index 56037b86..5898a887 100644 --- a/server/snapserver.cpp +++ b/server/snapserver.cpp @@ -265,11 +265,10 @@ int main(int argc, char* argv[]) } boost::asio::io_context io_context; - std::unique_ptr streamServer(new StreamServer(&io_context, settings)); + std::unique_ptr 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); diff --git a/server/stream_server.cpp b/server/stream_server.cpp index 5a52f6e7..c9779f13 100644 --- a/server/stream_server.cpp +++ b/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 mlock(sessionsMutex_); + unique_ptr 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 session = make_shared(this, std::move(socket)); + shared_ptr session = make_shared(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(*io_context_, tcp::endpoint(boost::asio::ip::address::from_string(address), settings_.stream.port))); + make_unique(io_context_, tcp::endpoint(boost::asio::ip::address::from_string(address), settings_.stream.port))); } catch (const boost::system::system_error& e) { diff --git a/server/stream_server.hpp b/server/stream_server.hpp index 7bd0bc67..82c0fa04 100644 --- a/server/stream_server.hpp +++ b/server/stream_server.hpp @@ -55,7 +55,7 @@ using session_ptr = std::shared_ptr; 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 sessions_; - boost::asio::io_context* io_context_; + boost::asio::io_context& io_context_; std::vector acceptor_; ServerSettings settings_; diff --git a/server/stream_session.cpp b/server/stream_session.cpp index 7c9623ee..4d3d8f4a 100644 --- a/server/stream_session.cpp +++ b/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 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 activeLock(activeMutex_); - if (!active_) - return; - - active_ = false; - } - - try - { - boost::system::error_code ec; - { - std::lock_guard 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::stop\n"; + boost::system::error_code ec; + 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"; LOG(DEBUG) << "StreamSession stopped\n"; } -void StreamSession::socketRead(void* _to, size_t _bytes) +void StreamSession::send_next() { - size_t read = 0; - do - { - read += socket_.read_some(boost::asio::buffer((char*)_to + read, _bytes - read)); - } while (active_ && (read < _bytes)); + 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) + { + 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 socketLock(socketMutex_); - { - std::lock_guard 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 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 message; - while (active_) - { - if (messages_.try_pop(message, std::chrono::milliseconds(500))) - { - if (bufferMs_ > 0) - { - const msg::WireChunk* wireChunk = dynamic_cast(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(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; } diff --git a/server/stream_session.hpp b/server/stream_session.hpp index 837f7770..b1e26e81 100644 --- a/server/stream_session.hpp +++ b/server/stream_session.hpp @@ -25,11 +25,14 @@ #include #include #include +#include #include #include #include +#include #include #include +#include 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(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> 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 { 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 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 active_; - - std::unique_ptr readerThread_; - std::unique_ptr writerThread_; - mutable std::mutex socketMutex_; + msg::BaseMessage baseMessage_; + std::vector buffer_; + size_t buffer_pos_; + size_t base_msg_size_; tcp::socket socket_; MessageReceiver* messageReceiver_; - Queue> messages_; size_t bufferMs_; PcmStreamPtr pcmStream_; + boost::asio::io_context::strand strand_; + std::deque messages_; };