Changed Queue to non-blocking

Author: paullouisageneau

src/impl/datachannel.cpp

@@ -108,7 +108,7 @@ void DataChannel::remoteClose() {
 }
 
 optional<message_variant> DataChannel::receive() {
-	auto next = mRecvQueue.tryPop();
+	auto next = mRecvQueue.pop();
 	return next ? std::make_optional(to_variant(std::move(**next))) : nullopt;
 }
 

src/impl/dtlstransport.cpp

@@ -329,7 +329,7 @@ int DtlsTransport::TimeoutCallback(gnutls_transport_ptr_t ptr, unsigned int ms)
 	DtlsTransport *t = static_cast<DtlsTransport *>(ptr);
 	try {
 		bool isReadable = t->mIncomingQueue.wait(
-		    ms != GNUTLS_INDEFINITE_TIMEOUT ? std::make_optional(milliseconds(ms)) : nullopt);
+		    ms != GNUTLS_INDEFINITE_TIMEOUT ? std::make_optional(milliseconds(ms)) : nullopt); // TODO
 		return isReadable ? 1 : 0;
 
 	} catch (const std::exception &e) {
@@ -541,7 +541,7 @@ void DtlsTransport::runRecvLoop() {
 		byte buffer[bufferSize];
 		while (mIncomingQueue.running()) {
 			// Process pending messages
-			while (auto next = mIncomingQueue.tryPop()) {
+			while (auto next = mIncomingQueue.pop()) {
 				message_ptr message = std::move(*next);
 				if (demuxMessage(message))
 					continue;
@@ -599,7 +599,7 @@ void DtlsTransport::runRecvLoop() {
 				}
 			}
 
-			mIncomingQueue.wait(duration);
+			mIncomingQueue.wait(duration); // TODO
 		}
 	} catch (const std::exception &e) {
 		PLOG_ERROR << "DTLS recv: " << e.what();

src/impl/peerconnection.cpp

@@ -1110,7 +1110,7 @@ void PeerConnection::triggerTrack(weak_ptr<Track> weakTrack) {
 
 void PeerConnection::triggerPendingDataChannels() {
 	while (dataChannelCallback) {
-		auto next = mPendingDataChannels.tryPop();
+		auto next = mPendingDataChannels.pop();
 		if (!next)
 			break;
 
@@ -1128,7 +1128,7 @@ void PeerConnection::triggerPendingDataChannels() {
 
 void PeerConnection::triggerPendingTracks() {
 	while (trackCallback) {
-		auto next = mPendingTracks.tryPop();
+		auto next = mPendingTracks.pop();
 		if (!next)
 			break;
 

src/impl/processor.cpp

@@ -21,7 +21,7 @@ void Processor::join() {
 
 void Processor::schedule() {
 	std::unique_lock lock(mMutex);
-	if (auto next = mTasks.tryPop()) {
+	if (auto next = mTasks.pop()) {
 		ThreadPool::Instance().enqueue(std::move(*next));
 	} else {
 		// No more tasks

src/impl/queue.hpp

@@ -34,19 +34,14 @@ template <typename T> class Queue {
 	size_t amount() const; // amount
 	void push(T element);
 	optional<T> pop();
-	optional<T> tryPop();
 	optional<T> peek();
 	optional<T> exchange(T element);
-	bool wait(const optional<std::chrono::milliseconds> &duration = nullopt);
 
 private:
-	void pushImpl(T element);
-	optional<T> popImpl();
-
 	const size_t mLimit;
 	size_t mAmount;
 	std::queue<T> mQueue;
-	std::condition_variable mPopCondition, mPushCondition;
+	std::condition_variable mPushCondition;
 	amount_function mAmountFunction;
 	bool mStopping = false;
 
@@ -66,7 +61,6 @@ template <typename T> Queue<T>::~Queue() { stop(); }
 template <typename T> void Queue<T>::stop() {
 	std::lock_guard lock(mMutex);
 	mStopping = true;
-	mPopCondition.notify_all();
 	mPushCondition.notify_all();
 }
 
@@ -98,18 +92,22 @@ template <typename T> size_t Queue<T>::amount() const {
 template <typename T> void Queue<T>::push(T element) {
 	std::unique_lock lock(mMutex);
 	mPushCondition.wait(lock, [this]() { return !mLimit || mQueue.size() < mLimit || mStopping; });
-	pushImpl(std::move(element));
+	if (mStopping)
+		return;
+
+	mAmount += mAmountFunction(element);
+	mQueue.emplace(std::move(element));
 }
 
 template <typename T> optional<T> Queue<T>::pop() {
 	std::unique_lock lock(mMutex);
-	mPopCondition.wait(lock, [this]() { return !mQueue.empty() || mStopping; });
-	return popImpl();
-}
+	if (mQueue.empty())
+		return nullopt;
 
-template <typename T> optional<T> Queue<T>::tryPop() {
-	std::unique_lock lock(mMutex);
-	return popImpl();
+	mAmount -= mAmountFunction(mQueue.front());
+	optional<T> element{std::move(mQueue.front())};
+	mQueue.pop();
+	return element;
 }
 
 template <typename T> optional<T> Queue<T>::peek() {
@@ -126,36 +124,6 @@ template <typename T> optional<T> Queue<T>::exchange(T element) {
 	return std::make_optional(std::move(element));
 }
 
-template <typename T> bool Queue<T>::wait(const optional<std::chrono::milliseconds> &duration) {
-	std::unique_lock lock(mMutex);
-	if (duration) {
-		return mPopCondition.wait_for(lock, *duration,
-		                              [this]() { return !mQueue.empty() || mStopping; });
-	} else {
-		mPopCondition.wait(lock, [this]() { return !mQueue.empty() || mStopping; });
-		return true;
-	}
-}
-
-template <typename T> void Queue<T>::pushImpl(T element) {
-	if (mStopping)
-		return;
-
-	mAmount += mAmountFunction(element);
-	mQueue.emplace(std::move(element));
-	mPopCondition.notify_one();
-}
-
-template <typename T> optional<T> Queue<T>::popImpl() {
-	if (mQueue.empty())
-		return nullopt;
-
-	mAmount -= mAmountFunction(mQueue.front());
-	optional<T> element{std::move(mQueue.front())};
-	mQueue.pop();
-	return element;
-}
-
 } // namespace rtc::impl
 
 #endif

src/impl/track.cpp

@@ -63,7 +63,7 @@ void Track::close() {
 }
 
 optional<message_variant> Track::receive() {
-	if (auto next = mRecvQueue.tryPop()) {
+	if (auto next = mRecvQueue.pop()) {
 		message_ptr message = *next;
 		if (message->type == Message::Control)
 			return to_variant(**next); // The same message may be frowarded into multiple Tracks

src/impl/websocket.cpp

@@ -133,7 +133,7 @@ bool WebSocket::isClosed() const { return state == State::Closed; }
 size_t WebSocket::maxMessageSize() const { return DEFAULT_MAX_MESSAGE_SIZE; }
 
 optional<message_variant> WebSocket::receive() {
-	while (auto next = mRecvQueue.tryPop()) {
+	while (auto next = mRecvQueue.pop()) {
 		message_ptr message = *next;
 		if (message->type != Message::Control)
 			return to_variant(std::move(*message));
@@ -147,7 +147,7 @@ optional<message_variant> WebSocket::peek() {
 		if (message->type != Message::Control)
 			return to_variant(std::move(*message));
 
-		mRecvQueue.tryPop();
+		mRecvQueue.pop();
 	}
 	return nullopt;
 }