cleaned up locking

client/clientConnection.cpp

@@ -145,13 +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> mlock(mutex_);
+	pendingRequests_.insert(pendingRequest);
-		pendingRequests_.insert(pendingRequest);
-	}
-	std::unique_lock<std::mutex> lck(requestMutex_);
 	send(message);
-	if (pendingRequest->cv.wait_for(lck, std::chrono::milliseconds(timeout)) == std::cv_status::no_timeout)
+	if (pendingRequest->cv.wait_for(mlock, std::chrono::milliseconds(timeout)) == std::cv_status::no_timeout)
 	{
 		response = pendingRequest->response;
 		sumTimeout_ = chronos::msec(0);
@@ -164,10 +161,7 @@ shared_ptr<msg::SerializedMessage> ClientConnection::sendRequest(const msg::Base
 		if (sumTimeout_ > chronos::sec(10))
 			throw SnapException("sum timeout exceeded 10s");
 	}
-	{
+	pendingRequests_.erase(pendingRequest);
-		std::unique_lock<std::mutex> mlock(mutex_);
-		pendingRequests_.erase(pendingRequest);
-	}
 	return response;
 }
 
@@ -198,7 +192,7 @@ void ClientConnection::getNextMessage()
 					req->response->message = baseMessage;
 					req->response->buffer = (char*)malloc(baseMessage.size);
 					memcpy(req->response->buffer, &buffer[0], baseMessage.size);
-					std::unique_lock<std::mutex> lck(requestMutex_);
+					mlock.unlock();
 					req->cv.notify_one();
 					return;
 				}

client/clientConnection.h

@@ -113,7 +113,6 @@ protected:
 	std::atomic<bool> connected_;
 	MessageReceiver* messageReceiver_;
 	mutable std::mutex mutex_;
-	mutable std::mutex requestMutex_;
 	std::set<std::shared_ptr<PendingRequest>> pendingRequests_;
 	uint16_t reqId_;
 	std::string host_;

client/decoder/flacDecoder.cpp

@@ -196,6 +196,15 @@ void error_callback(const FLAC__StreamDecoder *decoder, FLAC__StreamDecoderError
 {
 	(void)decoder, (void)client_data;
 	logS(kLogErr) << "Got error callback: " << FLAC__StreamDecoderErrorStatusString[status] << "\n";
+	/// TODO, see issue #120:
+	// Thu Nov 10 07:26:44 2016 daemon.warn dnsmasq-dhcp[1194]: no address range available for DHCP request via wlan0
+	// Thu Nov 10 07:54:39 2016 daemon.err snapclient[1158]: Got error callback: FLAC__STREAM_DECODER_ERROR_STATUS_LOST_SYNC
+	// Thu Nov 10 07:54:39 2016 daemon.err snapclient[1158]: Got error callback: FLAC__STREAM_DECODER_ERROR_STATUS_LOST_SYNC
+	//
+	// and:
+	// Oct 27 17:37:38 kitchen snapclient[869]: Connected to 192.168.222.10
+	// Oct 27 17:47:13 kitchen snapclient[869]: Got error callback: FLAC__STREAM_DECODER_ERROR_STATUS_UNPARSEABLE_STREAM
+	// Oct 27 17:47:13 kitchen snapclient[869]: Got error callback: FLAC__STREAM_DECODER_ERROR_STATUS_LOST_SYNC
 }
 
 

client/stream.cpp

@@ -76,19 +76,13 @@ void Stream::addChunk(msg::PcmChunk* chunk)
 	while (chunks_.size() * chunk->duration<cs::msec>().count() > 10000)
 		chunks_.pop();
 	chunks_.push(shared_ptr<msg::PcmChunk>(chunk));
-	std::unique_lock<std::mutex> lck(cvMutex_);
-	cv_.notify_one();
 //	logD << "new chunk: " << chunk->duration<cs::msec>().count() << ", Chunks: " << chunks_.size() << "\n";
 }
 
 
 bool Stream::waitForChunk(size_t ms) const
 {
-	if (!chunks_.empty())
+	return chunks_.wait_for(std::chrono::milliseconds(ms));
-		return true;
-
-	std::unique_lock<std::mutex> lck(cvMutex_);
-	return (cv_.wait_for(lck, std::chrono::milliseconds(ms), [this] { return !chunks_.empty(); }));
 }
 
 

client/stream.h

@@ -89,9 +89,6 @@ private:
 	unsigned long playedFrames_;
 	long correctAfterXFrames_;
 	chronos::msec bufferMs_;
-
-	mutable std::condition_variable cv_;
-	mutable std::mutex cvMutex_;
 };
 
 

common/queue.h

@@ -34,9 +34,9 @@ public:
 	{
 		std::unique_lock<std::mutex> mlock(mutex_);
 		while (queue_.empty())
-		{
 			cond_.wait(mlock);
-		}
+
+//		std::lock_guard<std::mutex> lock(mutex_);
 		auto val = queue_.front();
 		queue_.pop();
 		return val;
@@ -48,20 +48,33 @@ public:
 		while (queue_.empty())
 			cond_.wait(mlock);
 
+//		std::lock_guard<std::mutex> lock(mutex_);
 		return queue_.front();
 	}
 
 	void abort_wait()
 	{
-		abort_ = true;
+		{
+			std::lock_guard<std::mutex> mlock(mutex_);
+			abort_ = true;
+		}
 		cond_.notify_one();
 	}
 
+	bool wait_for(std::chrono::milliseconds timeout) const
+	{
+		std::unique_lock<std::mutex> mlock(mutex_);
+		abort_ = false;
+		if (!cond_.wait_for(mlock, timeout, [this] { return (!queue_.empty() || abort_); }))
+			return false;
+		
+		return !queue_.empty() && !abort_;
+	}
+
 	bool try_pop(T& item, std::chrono::microseconds timeout)
 	{
 		std::unique_lock<std::mutex> mlock(mutex_);
 		abort_ = false;
-
 		if (!cond_.wait_for(mlock, timeout, [this] { return (!queue_.empty() || abort_); }))
 			return false;
 		
@@ -83,32 +96,33 @@ public:
 	{
 		std::unique_lock<std::mutex> mlock(mutex_);
 		while (queue_.empty())
-		{
 			cond_.wait(mlock);
-		}
+
 		item = queue_.front();
 		queue_.pop();
 	}
 
 	void push(const T& item)
 	{
-		std::unique_lock<std::mutex> mlock(mutex_);
+		{
-		queue_.push(item);
+			std::lock_guard<std::mutex> mlock(mutex_);
-		mlock.unlock();
+			queue_.push(item);
+		}
 		cond_.notify_one();
 	}
 
 	void push(T&& item)
 	{
-		std::unique_lock<std::mutex> mlock(mutex_);
+		{
-		queue_.push(std::move(item));
+			std::lock_guard<std::mutex> mlock(mutex_);
-		mlock.unlock();
+			queue_.push(std::move(item));
+		}
 		cond_.notify_one();
 	}
 
 	size_t size() const
 	{
-		std::unique_lock<std::mutex> mlock(mutex_);
+		std::lock_guard<std::mutex> mlock(mutex_);
 		return queue_.size();
 	}
 
@@ -123,9 +137,9 @@ public:
 
 private:
 	std::queue<T> queue_;
-	std::atomic<bool> abort_;
+	mutable std::atomic<bool> abort_;
 	mutable std::mutex mutex_;
-	std::condition_variable cond_;
+	mutable std::condition_variable cond_;
 };