3pp-mirror/snapcast
1
0
Fork 0
mirror of https://github.com/badaix/snapcast.git synced 2025-05-30 09:26:15 +02:00
Code Issues Projects Releases Packages Wiki Activity

protect socket with mutex

Browse source
This commit is contained in:
BadAix 2016-11-13 15:53:48 +01:00
parent 0c5646679e
commit c8fc3ae4b8
6 changed files with 43 additions and 20 deletions
Download patch file Download diff file Expand all files Collapse all files

15
client/clientConnection.cpp
View file

@ -42,7 +42,6 @@ ClientConnection::~ClientConnection()
void ClientConnection::socketRead(void* _to, size_t _bytes)
{
// std::unique_lock<std::mutex> mlock(mutex_);
size_t toRead = _bytes;
size_t len = 0;
do
@ -123,6 +122,7 @@ bool ClientConnection::send(const msg::BaseMessage* message) const
{
// std::unique_lock<std::mutex> mlock(mutex_);
//logD << "send: " << message->type << ", size: " << message->getSize() << "\n";
std::lock_guard<std::mutex> socketLock(socketMutex_);
if (!connected())
return false;
//logD << "send: " << message->type << ", size: " << message->getSize() << "\n";
@ -145,10 +145,10 @@ shared_ptr<msg::SerializedMessage> ClientConnection::sendRequest(const msg::Base
// logO << "Req: " << message->id << "\n";
shared_ptr<PendingRequest> pendingRequest(new PendingRequest(reqId_));
std::unique_lock<std::mutex> mlock(mutex_);
std::unique_lock<std::mutex> lock(pendingRequestsMutex_);
pendingRequests_.insert(pendingRequest);
send(message);
if (pendingRequest->cv.wait_for(mlock, std::chrono::milliseconds(timeout)) == std::cv_status::no_timeout)
if (pendingRequest->cv.wait_for(lock, std::chrono::milliseconds(timeout)) == std::cv_status::no_timeout)
{
response = pendingRequest->response;
sumTimeout_ = chronos::msec(0);
@ -176,12 +176,15 @@ void ClientConnection::getNextMessage()
// logD << "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);
{
std::lock_guard<std::mutex> socketLock(socketMutex_);
socketRead(&buffer[0], baseMessage.size);
}
tv t;
baseMessage.received = t;
{
std::unique_lock<std::mutex> mlock(mutex_);
std::unique_lock<std::mutex> lock(pendingRequestsMutex_);
// logD << "got lock - getNextMessage: " << baseMessage.type << ", size: " << baseMessage.size << ", id: " << baseMessage.id << ", refers: " << baseMessage.refersTo << "\n";
{
for (auto req: pendingRequests_)
@ -192,7 +195,7 @@ void ClientConnection::getNextMessage()
req->response->message = baseMessage;
req->response->buffer = (char*)malloc(baseMessage.size);
memcpy(req->response->buffer, &buffer[0], baseMessage.size);
mlock.unlock();
lock.unlock();
req->cv.notify_one();
return;
}

3
client/clientConnection.h
View file

@ -108,11 +108,12 @@ protected:
void getNextMessage();
asio::io_service io_service_;
mutable std::mutex socketMutex_;
std::shared_ptr<tcp::socket> socket_;
std::atomic<bool> active_;
std::atomic<bool> connected_;
MessageReceiver* messageReceiver_;
mutable std::mutex mutex_;
mutable std::mutex pendingRequestsMutex_;
std::set<std::shared_ptr<PendingRequest>> pendingRequests_;
uint16_t reqId_;
std::string host_;

26
server/controlSession.cpp
View file

@ -41,7 +41,10 @@ ControlSession::~ControlSession()
void ControlSession::start()
{
active_ = true;
{
std::lock_guard<std::mutex> activeLock(activeMutex_);
active_ = true;
}
readerThread_ = new thread(&ControlSession::reader, this);
writerThread_ = new thread(&ControlSession::writer, this);
}
@ -49,13 +52,20 @@ void ControlSession::start()
void ControlSession::stop()
{
std::unique_lock<std::mutex> mlock(mutex_);
active_ = false;
{
std::lock_guard<std::mutex> activeLock(activeMutex_);
if (!active_)
return;
active_ = false;
}
try
{
std::error_code ec;
if (socket_)
{
std::lock_guard<std::mutex> socketLock(socketMutex_);
socket_->shutdown(asio::ip::tcp::socket::shutdown_both, ec);
if (ec) logE << "Error in socket shutdown: " << ec.message() << "\n";
socket_->close(ec);
@ -95,9 +105,12 @@ void ControlSession::sendAsync(const std::string& message)
bool ControlSession::send(const std::string& message) const
{
// logO << "send: " << message->type << ", size: " << message->size << ", id: " << message->id << ", refers: " << message->refersTo << "\n";
std::unique_lock<std::mutex> mlock(mutex_);
if (!socket_ || !active_)
return false;
std::lock_guard<std::mutex> socketLock(socketMutex_);
{
std::lock_guard<std::mutex> activeLock(activeMutex_);
if (!socket_ || !active_)
return false;
}
asio::streambuf streambuf;
std::ostream request_stream(&streambuf);
request_stream << message << "\r\n";
@ -109,7 +122,6 @@ bool ControlSession::send(const std::string& message) const
void ControlSession::reader()
{
active_ = true;
try
{
std::stringstream message;

3
server/controlSession.h
View file

@ -76,7 +76,8 @@ protected:
void writer();
std::atomic<bool> active_;
mutable std::mutex mutex_;
mutable std::mutex activeMutex_;
mutable std::mutex socketMutex_;
std::thread* readerThread_;
std::thread* writerThread_;
std::shared_ptr<tcp::socket> socket_;

13
server/streamSession.cpp
View file

@ -55,7 +55,7 @@ const PcmStreamPtr StreamSession::pcmStream() const
void StreamSession::start()
{
{
std::lock_guard<std::mutex> mlock(mutex_);
std::lock_guard<std::mutex> activeLock(activeMutex_);
active_ = true;
}
readerThread_.reset(new thread(&StreamSession::reader, this));
@ -66,7 +66,7 @@ void StreamSession::start()
void StreamSession::stop()
{
{
std::lock_guard<std::mutex> mlock(mutex_);
std::lock_guard<std::mutex> activeLock(activeMutex_);
if (!active_)
return;
@ -78,6 +78,7 @@ void StreamSession::stop()
std::error_code ec;
if (socket_)
{
std::lock_guard<std::mutex> socketLock(socketMutex_);
socket_->shutdown(asio::ip::tcp::socket::shutdown_both, ec);
if (ec) logE << "Error in socket shutdown: " << ec.message() << "\n";
socket_->close(ec);
@ -144,8 +145,9 @@ bool StreamSession::send(const msg::BaseMessage* message) const
{
//TODO on exception: set active = false
// logO << "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> mlock(mutex_);
std::lock_guard<std::mutex> activeLock(activeMutex_);
if (!socket_ || !active_)
return false;
}
@ -176,7 +178,10 @@ void StreamSession::getNextMessage()
// logO << "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);
{
std::lock_guard<std::mutex> socketLock(socketMutex_);
socketRead(&buffer[0], baseMessage.size);
}
tv t;
baseMessage.received = t;

3
server/streamSession.h
View file

@ -89,11 +89,12 @@ protected:
void reader();
void writer();
mutable std::mutex mutex_;
mutable std::mutex activeMutex_;
std::atomic<bool> active_;
std::unique_ptr<std::thread> readerThread_;
std::unique_ptr<std::thread> writerThread_;
mutable std::mutex socketMutex_;
std::shared_ptr<tcp::socket> socket_;
MessageReceiver* messageReceiver_;
Queue<std::shared_ptr<const msg::BaseMessage>> messages_;