3pp-mirror/snapcast
1
0
Fork 0
mirror of https://github.com/badaix/snapcast.git synced 2025-05-10 23:56:43 +02:00
Code Issues Projects Releases Packages Wiki Activity

Use strand executors

Browse source
This commit is contained in:
badaix 2021-09-18 11:11:34 +02:00
parent e6872593ee
commit 12aeb5859c
15 changed files with 38 additions and 38 deletions
Download patch file Download diff file Expand all files Collapse all files

2
server/stream_session.cpp
View file

@ -28,7 +28,7 @@ using namespace streamreader;
static constexpr auto LOG_TAG = "StreamSession";
StreamSession::StreamSession(net::any_io_executor executor, StreamMessageReceiver* receiver)
StreamSession::StreamSession(const net::any_io_executor& executor, StreamMessageReceiver* receiver)
: messageReceiver_(receiver), pcmStream_(nullptr), strand_(net::make_strand(executor))
{
base_msg_size_ = baseMessage_.getSize();

2
server/stream_session.hpp
View file

@ -120,7 +120,7 @@ class StreamSession : public std::enable_shared_from_this<StreamSession>
{
public:
/// ctor. Received message from the client are passed to StreamMessageReceiver
StreamSession(net::any_io_executor strand, StreamMessageReceiver* receiver);
StreamSession(const net::any_io_executor& executor, StreamMessageReceiver* receiver);
virtual ~StreamSession() = default;
virtual std::string getIP() = 0;

2
server/streamreader/airplay_stream.cpp
View file

@ -229,7 +229,7 @@ void AirplayStream::pipeReadLine()
try
{
int fd = open(pipePath_.c_str(), O_RDONLY | O_NONBLOCK);
pipe_fd_ = std::make_unique<boost::asio::posix::stream_descriptor>(ioc_, fd);
pipe_fd_ = std::make_unique<boost::asio::posix::stream_descriptor>(strand_, fd);
LOG(INFO, LOG_TAG) << "Metadata pipe opened: " << pipePath_ << "\n";
}
catch (const std::exception& e)

2
server/streamreader/alsa_stream.cpp
View file

@ -66,7 +66,7 @@ void wait(boost::asio::steady_timer& timer, const std::chrono::duration<Rep, Per
AlsaStream::AlsaStream(PcmListener* pcmListener, boost::asio::io_context& ioc, const ServerSettings& server_settings, const StreamUri& uri)
: PcmStream(pcmListener, ioc, server_settings, uri), handle_(nullptr), read_timer_(ioc), silence_(0ms)
: PcmStream(pcmListener, ioc, server_settings, uri), handle_(nullptr), read_timer_(strand_), silence_(0ms)
{
device_ = uri_.getQuery("device", "hw:0");
send_silence_ = (uri_.getQuery("send_silence", "false") == "true");

2
server/streamreader/asio_stream.hpp
View file

@ -83,7 +83,7 @@ void AsioStream<ReadStream>::wait(Timer& timer, const std::chrono::duration<Rep,
template <typename ReadStream>
AsioStream<ReadStream>::AsioStream(PcmListener* pcmListener, boost::asio::io_context& ioc, const ServerSettings& server_settings, const StreamUri& uri)
: PcmStream(pcmListener, ioc, server_settings, uri), read_timer_(ioc), state_timer_(ioc)
: PcmStream(pcmListener, ioc, server_settings, uri), read_timer_(strand_), state_timer_(strand_)
{
chunk_ = std::make_unique<msg::PcmChunk>(sampleFormat_, chunk_ms_);
LOG(DEBUG, "AsioStream") << "Chunk duration: " << chunk_->durationMs() << " ms, frames: " << chunk_->getFrameCount() << ", size: " << chunk_->payloadSize

2
server/streamreader/file_stream.cpp
View file

@ -53,7 +53,7 @@ void FileStream::do_connect()
{
LOG(DEBUG, LOG_TAG) << "connect\n";
int fd = open(uri_.path.c_str(), O_RDONLY | O_NONBLOCK);
stream_ = std::make_unique<boost::asio::posix::stream_descriptor>(ioc_, fd);
stream_ = std::make_unique<boost::asio::posix::stream_descriptor>(strand_, fd);
on_connect();
}

6
server/streamreader/pcm_stream.cpp
View file

@ -42,8 +42,8 @@ static constexpr auto LOG_TAG = "PcmStream";
PcmStream::PcmStream(PcmListener* pcmListener, boost::asio::io_context& ioc, const ServerSettings& server_settings, const StreamUri& uri)
: active_(false), pcmListeners_{pcmListener}, uri_(uri), chunk_ms_(20), state_(ReaderState::kIdle), ioc_(ioc), server_settings_(server_settings),
req_id_(0), property_timer_(ioc)
: active_(false), strand_(net::make_strand(ioc.get_executor())), pcmListeners_{pcmListener}, uri_(uri), chunk_ms_(20), state_(ReaderState::kIdle),
ioc_(ioc), server_settings_(server_settings), req_id_(0), property_timer_(strand_)
{
encoder::EncoderFactory encoderFactory;
if (uri_.query.find(kUriCodec) == uri_.query.end())
@ -61,7 +61,7 @@ PcmStream::PcmStream(PcmListener* pcmListener, boost::asio::io_context& ioc, con
if (uri_.query.find(kControlScript) != uri_.query.end())
{
stream_ctrl_ = std::make_unique<ScriptStreamControl>(ioc, uri_.query[kControlScript]);
stream_ctrl_ = std::make_unique<ScriptStreamControl>(strand_, uri_.query[kControlScript]);
}
if (uri_.query.find(kUriChunkMs) != uri_.query.end())

5
server/streamreader/pcm_stream.hpp
View file

@ -41,6 +41,8 @@
namespace bp = boost::process;
namespace net = boost::asio;
using json = nlohmann::json;
@ -178,6 +180,7 @@ protected:
/// Send request to stream control script
void sendRequest(const std::string& method, const jsonrpcpp::Parameter& params, ResultHandler handler);
net::strand<net::any_io_executor> strand_;
std::chrono::time_point<std::chrono::steady_clock> tvEncodedChunk_;
std::vector<PcmListener*> pcmListeners_;
StreamUri uri_;
@ -185,7 +188,7 @@ protected:
size_t chunk_ms_;
std::unique_ptr<encoder::Encoder> encoder_;
std::string name_;
ReaderState state_;
std::atomic<ReaderState> state_;
Metatags metadata_;
Properties properties_;
boost::asio::io_context& ioc_;

2
server/streamreader/pipe_stream.cpp
View file

@ -62,7 +62,7 @@ void PipeStream::do_connect()
pipe_size = fcntl(fd, F_GETPIPE_SZ);
#endif
LOG(TRACE, LOG_TAG) << "Stream: " << name_ << ", connect to pipe: " << uri_.path << ", fd: " << fd << ", pipe size: " << pipe_size << "\n";
stream_ = std::make_unique<boost::asio::posix::stream_descriptor>(ioc_, fd);
stream_ = std::make_unique<boost::asio::posix::stream_descriptor>(strand_, fd);
on_connect();
}

6
server/streamreader/process_stream.cpp
View file

@ -113,12 +113,12 @@ void ProcessStream::do_connect()
fcntl(pipe_stdout_.native_source(), F_SETFL, flags | O_NONBLOCK);
process_ = bp::child(path_ + exe_ + " " + params_, bp::std_out > pipe_stdout_, bp::std_err > pipe_stderr_, bp::start_dir = path_);
stream_ = make_unique<stream_descriptor>(ioc_, pipe_stdout_.native_source());
stream_stderr_ = make_unique<stream_descriptor>(ioc_, pipe_stderr_.native_source());
stream_ = make_unique<stream_descriptor>(strand_, pipe_stdout_.native_source());
stream_stderr_ = make_unique<stream_descriptor>(strand_, pipe_stderr_.native_source());
on_connect();
if (wd_timeout_sec_ > 0)
{
watchdog_ = make_unique<Watchdog>(ioc_, this);
watchdog_ = make_unique<Watchdog>(strand_, this);
watchdog_->start(std::chrono::seconds(wd_timeout_sec_));
}
else

16
server/streamreader/stream_control.cpp
View file

@ -36,7 +36,7 @@ namespace streamreader
static constexpr auto LOG_TAG = "Script";
StreamControl::StreamControl(net::io_context& ioc) : ioc_(ioc), strand_(net::make_strand(ioc.get_executor()))
StreamControl::StreamControl(const net::any_io_executor& executor) : executor_(executor)
{
}
@ -61,7 +61,7 @@ void StreamControl::start(const std::string& stream_id, const ServerSettings& se
void StreamControl::command(const jsonrpcpp::Request& request, const OnResponse& response_handler)
{
// use strand to serialize commands sent from different threads
net::post(strand_, [this, request, response_handler]() {
net::post(executor_, [this, request, response_handler]() {
if (response_handler)
request_callbacks_[request.id()] = response_handler;
@ -134,7 +134,7 @@ void StreamControl::onLog(std::string message)
ScriptStreamControl::ScriptStreamControl(net::io_context& ioc, const std::string& script) : StreamControl(ioc), script_(script)
ScriptStreamControl::ScriptStreamControl(const net::any_io_executor& executor, const std::string& script) : StreamControl(executor), script_(script)
{
// auto fileExists = [](const std::string& filename) {
// struct stat buffer;
@ -158,22 +158,20 @@ void ScriptStreamControl::doStart(const std::string& stream_id, const ServerSett
{
process_ = bp::child(
script_ + params.str(), bp::std_out > pipe_stdout_, bp::std_err > pipe_stderr_, bp::std_in < in_,
bp::on_exit =
[](int exit, const std::error_code& ec_in) {
bp::on_exit = [](int exit, const std::error_code& ec_in) {
auto severity = AixLog::Severity::debug;
if (exit != 0)
severity = AixLog::Severity::error;
LOG(severity, LOG_TAG) << "Exit code: " << exit << ", message: " << ec_in.message() << "\n";
},
ioc_);
});
}
catch (const std::exception& e)
{
throw SnapException("Failed to start control script: '" + script_ + "', exception: " + e.what());
}
stream_stdout_ = make_unique<boost::asio::posix::stream_descriptor>(ioc_, pipe_stdout_.native_source());
stream_stderr_ = make_unique<boost::asio::posix::stream_descriptor>(ioc_, pipe_stderr_.native_source());
stream_stdout_ = make_unique<boost::asio::posix::stream_descriptor>(executor_, pipe_stdout_.native_source());
stream_stderr_ = make_unique<boost::asio::posix::stream_descriptor>(executor_, pipe_stderr_.native_source());
stdoutReadLine();
stderrReadLine();
}

11
server/streamreader/stream_control.hpp
View file

@ -28,9 +28,7 @@
#include <map>
#include <string>
#include <boost/asio/io_context.hpp>
#include <boost/asio/read_until.hpp>
#include <boost/asio/strand.hpp>
#include <boost/asio.hpp>
#include "jsonrpcpp.hpp"
#include "server_settings.hpp"
@ -54,7 +52,7 @@ public:
using OnResponse = std::function<void(const jsonrpcpp::Response& response)>;
using OnLog = std::function<void(std::string message)>;
StreamControl(net::io_context& ioc);
StreamControl(const net::any_io_executor& executor);
virtual ~StreamControl();
void start(const std::string& stream_id, const ServerSettings& server_setttings, const OnNotification& notification_handler,
@ -70,7 +68,7 @@ protected:
void onReceive(const std::string& json);
void onLog(std::string message);
net::io_context& ioc_;
net::any_io_executor executor_;
private:
OnRequest request_handler_;
@ -78,14 +76,13 @@ private:
OnLog log_handler_;
std::map<jsonrpcpp::Id, OnResponse> request_callbacks_;
net::strand<net::any_io_executor> strand_;
};
class ScriptStreamControl : public StreamControl
{
public:
ScriptStreamControl(net::io_context& ioc, const std::string& script);
ScriptStreamControl(const net::any_io_executor& executor, const std::string& script);
virtual ~ScriptStreamControl() = default;
void stop() override;

4
server/streamreader/tcp_stream.cpp
View file

@ -61,7 +61,7 @@ TcpStream::TcpStream(PcmListener* pcmListener, boost::asio::io_context& ioc, con
LOG(INFO, LOG_TAG) << "TcpStream host: " << host_ << ", port: " << port_ << ", is server: " << is_server_ << "\n";
if (is_server_)
acceptor_ = make_unique<tcp::acceptor>(ioc_, tcp::endpoint(boost::asio::ip::address::from_string(host_), port_));
acceptor_ = make_unique<tcp::acceptor>(strand_, tcp::endpoint(boost::asio::ip::address::from_string(host_), port_));
}
@ -87,7 +87,7 @@ void TcpStream::do_connect()
}
else
{
stream_ = make_unique<tcp::socket>(ioc_);
stream_ = make_unique<tcp::socket>(strand_);
boost::asio::ip::tcp::endpoint endpoint(boost::asio::ip::address::from_string(host_), port_);
stream_->async_connect(endpoint, [this](const boost::system::error_code& ec) {
if (!ec)

2
server/streamreader/watchdog.cpp
View file

@ -29,7 +29,7 @@ using namespace std;
namespace streamreader
{
Watchdog::Watchdog(boost::asio::io_context& ioc, WatchdogListener* listener) : timer_(ioc), listener_(listener)
Watchdog::Watchdog(const net::any_io_executor& executor, WatchdogListener* listener) : timer_(executor), listener_(listener)
{
}

4
server/streamreader/watchdog.hpp
View file

@ -22,6 +22,8 @@
#include <boost/asio.hpp>
#include <memory>
namespace net = boost::asio;
namespace streamreader
{
@ -39,7 +41,7 @@ public:
class Watchdog
{
public:
Watchdog(boost::asio::io_context& ioc, WatchdogListener* listener = nullptr);
Watchdog(const net::any_io_executor& executor, WatchdogListener* listener = nullptr);
virtual ~Watchdog();
void start(const std::chrono::milliseconds& timeout);