mirror of
https://github.com/badaix/snapcast.git
synced 2025-06-02 02:41:49 +02:00
Make controller async, add client Mixer settings
Controller uses the asio io_context and thus uses one thread less. This also fixes a lot of thread sanitizer issues.
This commit is contained in:
badaix
parent
4ce69d4bfb
commit
9bb1c4a041
22 changed files with 838 additions and 711 deletions
|
|
@ -20,7 +20,6 @@
|
|||
#include "common/aixlog.hpp"
|
||||
#include "common/snap_exception.hpp"
|
||||
#include "common/str_compat.hpp"
|
||||
#include "message/factory.hpp"
|
||||
#include "message/hello.hpp"
|
||||
#include <iostream>
|
||||
#include <mutex>
|
||||
|
|
@ -28,10 +27,10 @@
|
|||
|
||||
using namespace std;
|
||||
|
||||
static constexpr auto LOG_TAG = "Connection";
|
||||
|
||||
ClientConnection::ClientConnection(MessageReceiver* receiver, const std::string& host, size_t port)
|
||||
: socket_(io_context_), active_(false), messageReceiver_(receiver), reqId_(1), host_(host), port_(port), readerThread_(nullptr),
|
||||
sumTimeout_(chronos::msec(0))
|
||||
ClientConnection::ClientConnection(boost::asio::io_context& io_context, const ClientSettings::Server& server)
|
||||
: io_context_(io_context), resolver_(io_context_), socket_(io_context_), reqId_(1), server_(server), strand_(io_context_)
|
||||
{
|
||||
base_msg_size_ = base_message_.getSize();
|
||||
buffer_.resize(base_msg_size_);
|
||||
|
|
@ -40,20 +39,7 @@ ClientConnection::ClientConnection(MessageReceiver* receiver, const std::string&
|
|||
|
||||
ClientConnection::~ClientConnection()
|
||||
{
|
||||
stop();
|
||||
}
|
||||
|
||||
|
||||
void ClientConnection::socketRead(void* _to, size_t _bytes)
|
||||
{
|
||||
size_t toRead = _bytes;
|
||||
size_t len = 0;
|
||||
do
|
||||
{
|
||||
len += socket_.read_some(boost::asio::buffer((char*)_to + len, toRead));
|
||||
// cout << "len: " << len << ", error: " << error << endl;
|
||||
toRead = _bytes - len;
|
||||
} while (toRead > 0);
|
||||
disconnect();
|
||||
}
|
||||
|
||||
|
||||
|
|
@ -67,159 +53,194 @@ std::string ClientConnection::getMacAddress()
|
|||
#endif
|
||||
if (mac.empty())
|
||||
mac = "00:00:00:00:00:00";
|
||||
LOG(INFO) << "My MAC: \"" << mac << "\", socket: " << socket_.native_handle() << "\n";
|
||||
LOG(INFO, LOG_TAG) << "My MAC: \"" << mac << "\", socket: " << socket_.native_handle() << "\n";
|
||||
return mac;
|
||||
}
|
||||
|
||||
|
||||
void ClientConnection::start()
|
||||
void ClientConnection::connect(const ResultHandler& handler)
|
||||
{
|
||||
tcp::resolver resolver(io_context_);
|
||||
tcp::resolver::query query(host_, cpt::to_string(port_), boost::asio::ip::resolver_query_base::numeric_service);
|
||||
auto iterator = resolver.resolve(query);
|
||||
LOG(DEBUG) << "Connecting\n";
|
||||
// struct timeval tv;
|
||||
// tv.tv_sec = 5;
|
||||
// tv.tv_usec = 0;
|
||||
// cout << "socket: " << socket->native_handle() << "\n";
|
||||
// setsockopt(socket->native_handle(), SOL_SOCKET, SO_RCVTIMEO, &tv, sizeof(tv));
|
||||
// setsockopt(socket->native_handle(), SOL_SOCKET, SO_SNDTIMEO, &tv, sizeof(tv));
|
||||
socket_.connect(*iterator);
|
||||
LOG(NOTICE) << "Connected to " << socket_.remote_endpoint().address().to_string() << endl;
|
||||
active_ = true;
|
||||
sumTimeout_ = chronos::msec(0);
|
||||
readerThread_ = make_unique<thread>(&ClientConnection::reader, this);
|
||||
}
|
||||
|
||||
|
||||
void ClientConnection::stop()
|
||||
{
|
||||
active_ = false;
|
||||
try
|
||||
tcp::resolver::query query(server_.host, cpt::to_string(server_.port), boost::asio::ip::resolver_query_base::numeric_service);
|
||||
boost::system::error_code ec;
|
||||
LOG(DEBUG, LOG_TAG) << "Resolving host IP\n";
|
||||
auto iterator = resolver_.resolve(query, ec);
|
||||
if (ec)
|
||||
{
|
||||
boost::system::error_code ec;
|
||||
socket_.shutdown(boost::asio::ip::tcp::socket::shutdown_both, ec);
|
||||
LOG(ERROR, LOG_TAG) << "Failed to resolve host '" << server_.host << "', error: " << ec.message() << "\n";
|
||||
handler(ec);
|
||||
return;
|
||||
}
|
||||
|
||||
LOG(DEBUG, LOG_TAG) << "Connecting\n";
|
||||
socket_.connect(*iterator, ec);
|
||||
if (ec)
|
||||
{
|
||||
LOG(ERROR, LOG_TAG) << "Failed to connect to host '" << server_.host << "', error: " << ec.message() << "\n";
|
||||
handler(ec);
|
||||
return;
|
||||
}
|
||||
LOG(NOTICE, LOG_TAG) << "Connected to " << socket_.remote_endpoint().address().to_string() << endl;
|
||||
handler(ec);
|
||||
// getNextMessage();
|
||||
|
||||
#if 0
|
||||
resolver_.async_resolve(query, host_, cpt::to_string(port_), [this, handler](const boost::system::error_code& ec, tcp::resolver::results_type results) {
|
||||
if (ec)
|
||||
LOG(ERROR) << "Error in socket shutdown: " << ec.message() << endl;
|
||||
socket_.close(ec);
|
||||
if (ec)
|
||||
LOG(ERROR) << "Error in socket close: " << ec.message() << endl;
|
||||
if (readerThread_)
|
||||
{
|
||||
LOG(DEBUG) << "joining readerThread\n";
|
||||
readerThread_->join();
|
||||
LOG(ERROR, LOG_TAG) << "Failed to resolve host '" << host_ << "', error: " << ec.message() << "\n";
|
||||
handler(ec);
|
||||
return;
|
||||
}
|
||||
}
|
||||
catch (...)
|
||||
{
|
||||
}
|
||||
readerThread_ = nullptr;
|
||||
LOG(DEBUG) << "readerThread terminated\n";
|
||||
|
||||
resolver_.cancel();
|
||||
socket_.async_connect(*results, [this, handler](const boost::system::error_code& ec) {
|
||||
if (ec)
|
||||
{
|
||||
LOG(ERROR, LOG_TAG) << "Failed to connect to host '" << host_ << "', error: " << ec.message() << "\n";
|
||||
handler(ec);
|
||||
return;
|
||||
}
|
||||
|
||||
LOG(NOTICE, LOG_TAG) << "Connected to " << socket_.remote_endpoint().address().to_string() << endl;
|
||||
handler(ec);
|
||||
getNextMessage();
|
||||
});
|
||||
});
|
||||
#endif
|
||||
}
|
||||
|
||||
|
||||
bool ClientConnection::send(const msg::BaseMessage* message)
|
||||
void ClientConnection::disconnect()
|
||||
{
|
||||
// std::unique_lock<std::mutex> mlock(mutex_);
|
||||
// LOG(DEBUG) << "send: " << message->type << ", size: " << message->getSize() << "\n";
|
||||
std::lock_guard<std::mutex> socketLock(socketMutex_);
|
||||
LOG(DEBUG, LOG_TAG) << "Disconnecting\n";
|
||||
if (!socket_.is_open())
|
||||
return false;
|
||||
// LOG(DEBUG) << "send: " << message->type << ", size: " << message->getSize() << "\n";
|
||||
{
|
||||
LOG(DEBUG, LOG_TAG) << "Not connected\n";
|
||||
return;
|
||||
}
|
||||
boost::system::error_code ec;
|
||||
socket_.shutdown(boost::asio::ip::tcp::socket::shutdown_both, ec);
|
||||
if (ec)
|
||||
LOG(ERROR, LOG_TAG) << "Error in socket shutdown: " << ec.message() << endl;
|
||||
socket_.close(ec);
|
||||
if (ec)
|
||||
LOG(ERROR, LOG_TAG) << "Error in socket close: " << ec.message() << endl;
|
||||
pendingRequests_.clear();
|
||||
LOG(DEBUG, LOG_TAG) << "Disconnected\n";
|
||||
}
|
||||
|
||||
|
||||
void ClientConnection::sendNext()
|
||||
{
|
||||
auto& message = messages_.front();
|
||||
boost::asio::streambuf streambuf;
|
||||
std::ostream stream(&streambuf);
|
||||
tv t;
|
||||
message->sent = t;
|
||||
message->serialize(stream);
|
||||
boost::asio::write(socket_, streambuf);
|
||||
return true;
|
||||
message.msg->sent = t;
|
||||
message.msg->serialize(stream);
|
||||
boost::asio::async_write(socket_, streambuf, boost::asio::bind_executor(strand_, [this](boost::system::error_code ec, std::size_t length) {
|
||||
std::ignore = length;
|
||||
auto handler = messages_.front().handler;
|
||||
messages_.pop_front();
|
||||
if (handler)
|
||||
handler(ec);
|
||||
|
||||
if (!messages_.empty())
|
||||
sendNext();
|
||||
}));
|
||||
}
|
||||
|
||||
|
||||
|
||||
unique_ptr<msg::BaseMessage> ClientConnection::sendRequest(const msg::BaseMessage* message, const chronos::msec& timeout)
|
||||
void ClientConnection::send(const msg::message_ptr& message, const ResultHandler& handler)
|
||||
{
|
||||
unique_ptr<msg::BaseMessage> response(nullptr);
|
||||
if (++reqId_ >= 10000)
|
||||
reqId_ = 1;
|
||||
message->id = reqId_;
|
||||
// LOG(INFO) << "Req: " << message->id << "\n";
|
||||
shared_ptr<PendingRequest> pendingRequest = make_shared<PendingRequest>(reqId_);
|
||||
|
||||
{ // scope for lock
|
||||
std::unique_lock<std::mutex> lock(pendingRequestsMutex_);
|
||||
pendingRequests_.insert(pendingRequest);
|
||||
send(message);
|
||||
}
|
||||
|
||||
if ((response = pendingRequest->waitForResponse(std::chrono::milliseconds(timeout))) != nullptr)
|
||||
{
|
||||
sumTimeout_ = chronos::msec(0);
|
||||
// LOG(INFO) << "Resp: " << pendingRequest->id << "\n";
|
||||
}
|
||||
else
|
||||
{
|
||||
sumTimeout_ += timeout;
|
||||
LOG(WARNING) << "timeout while waiting for response to: " << reqId_ << ", timeout " << sumTimeout_.count() << "\n";
|
||||
if (sumTimeout_ > chronos::sec(10))
|
||||
throw SnapException("sum timeout exceeded 10s");
|
||||
}
|
||||
|
||||
{ // scope for lock
|
||||
std::unique_lock<std::mutex> lock(pendingRequestsMutex_);
|
||||
pendingRequests_.erase(pendingRequest);
|
||||
}
|
||||
return response;
|
||||
}
|
||||
|
||||
|
||||
void ClientConnection::getNextMessage()
|
||||
{
|
||||
socketRead(&buffer_[0], base_msg_size_);
|
||||
base_message_.deserialize(buffer_.data());
|
||||
// LOG(DEBUG) << "getNextMessage: " << base_message_.type << ", size: " << base_message_.size << ", id: " << base_message_.id
|
||||
// << ", refers: " << base_message_.refersTo << "\n";
|
||||
if (base_message_.size > buffer_.size())
|
||||
buffer_.resize(base_message_.size);
|
||||
// {
|
||||
// std::lock_guard<std::mutex> socketLock(socketMutex_);
|
||||
socketRead(buffer_.data(), base_message_.size);
|
||||
tv t;
|
||||
base_message_.received = t;
|
||||
// }
|
||||
|
||||
{ // scope for lock
|
||||
std::unique_lock<std::mutex> lock(pendingRequestsMutex_);
|
||||
for (auto req : pendingRequests_)
|
||||
boost::asio::post(strand_, [this, message, handler]() {
|
||||
messages_.push_back({message, handler});
|
||||
if (messages_.size() > 1)
|
||||
{
|
||||
if (req->id() == base_message_.refersTo)
|
||||
{
|
||||
auto response = msg::factory::createMessage(base_message_, buffer_.data());
|
||||
req->setValue(std::move(response));
|
||||
return;
|
||||
}
|
||||
LOG(DEBUG, LOG_TAG) << "outstanding async_write\n";
|
||||
return;
|
||||
}
|
||||
}
|
||||
|
||||
if (messageReceiver_ != nullptr)
|
||||
messageReceiver_->onMessageReceived(this, base_message_, buffer_.data());
|
||||
sendNext();
|
||||
});
|
||||
}
|
||||
|
||||
|
||||
|
||||
void ClientConnection::reader()
|
||||
void ClientConnection::sendRequest(const msg::message_ptr& message, const chronos::usec& timeout, const MessageHandler<msg::BaseMessage>& handler)
|
||||
{
|
||||
try
|
||||
{
|
||||
while (active_)
|
||||
{
|
||||
getNextMessage();
|
||||
}
|
||||
}
|
||||
catch (...)
|
||||
{
|
||||
if (messageReceiver_ != nullptr)
|
||||
messageReceiver_->onException(this, std::current_exception());
|
||||
}
|
||||
active_ = false;
|
||||
// LOG(INFO, LOG_TAG) << "Req: " << message->id << "\n";
|
||||
boost::asio::post(strand_, [this, message, timeout, handler]() {
|
||||
unique_ptr<msg::BaseMessage> response(nullptr);
|
||||
if (++reqId_ >= 10000)
|
||||
reqId_ = 1;
|
||||
message->id = reqId_;
|
||||
pendingRequests_.insert(make_unique<PendingRequest>(io_context_, strand_, reqId_, timeout, handler));
|
||||
send(message, [handler](const boost::system::error_code& ec) {
|
||||
if (ec)
|
||||
handler(ec, nullptr);
|
||||
});
|
||||
});
|
||||
}
|
||||
|
||||
|
||||
void ClientConnection::getNextMessage(const MessageHandler<msg::BaseMessage>& handler)
|
||||
{
|
||||
boost::asio::async_read(socket_, boost::asio::buffer(buffer_, base_msg_size_),
|
||||
boost::asio::bind_executor(strand_, [this, handler](boost::system::error_code ec, std::size_t length) mutable {
|
||||
if (ec)
|
||||
{
|
||||
LOG(ERROR, LOG_TAG) << "Error reading message header of length " << length << ": " << ec.message() << "\n";
|
||||
if (handler)
|
||||
handler(ec, nullptr);
|
||||
return;
|
||||
}
|
||||
|
||||
base_message_.deserialize(buffer_.data());
|
||||
tv t;
|
||||
base_message_.received = t;
|
||||
// LOG(DEBUG, LOG_TAG) << "getNextMessage: " << base_message_.type << ", size: " << base_message_.size
|
||||
// << ", id: " << base_message_.id << ", refers: " << base_message_.refersTo << "\n";
|
||||
if (base_message_.type > message_type::kLast)
|
||||
{
|
||||
LOG(ERROR, LOG_TAG) << "unknown message type received: " << base_message_.type << ", size: " << base_message_.size << "\n";
|
||||
if (handler)
|
||||
handler(boost::asio::error::invalid_argument, nullptr);
|
||||
return;
|
||||
}
|
||||
else if (base_message_.size > msg::max_size)
|
||||
{
|
||||
LOG(ERROR, LOG_TAG) << "received message of type " << base_message_.type << " to large: " << base_message_.size << "\n";
|
||||
if (handler)
|
||||
handler(boost::asio::error::invalid_argument, nullptr);
|
||||
return;
|
||||
}
|
||||
|
||||
if (base_message_.size > buffer_.size())
|
||||
buffer_.resize(base_message_.size);
|
||||
|
||||
boost::asio::async_read(
|
||||
socket_, boost::asio::buffer(buffer_, base_message_.size),
|
||||
boost::asio::bind_executor(strand_, [this, handler](boost::system::error_code ec, std::size_t length) mutable {
|
||||
if (ec)
|
||||
{
|
||||
LOG(ERROR, LOG_TAG) << "Error reading message body of length " << length << ": " << ec.message() << "\n";
|
||||
if (handler)
|
||||
handler(ec, nullptr);
|
||||
return;
|
||||
}
|
||||
|
||||
auto iter = std::find_if(
|
||||
pendingRequests_.begin(), pendingRequests_.end(),
|
||||
[this](const std::unique_ptr<PendingRequest>& request) { return request->id() == base_message_.refersTo; });
|
||||
auto response = msg::factory::createMessage(base_message_, buffer_.data());
|
||||
if (iter != pendingRequests_.end())
|
||||
{
|
||||
(*iter)->setValue(std::move(response));
|
||||
pendingRequests_.erase(iter);
|
||||
getNextMessage(handler);
|
||||
return;
|
||||
}
|
||||
|
||||
if (handler)
|
||||
handler(ec, std::move(response));
|
||||
}));
|
||||
}));
|
||||
}
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue