Introduced SCTP transport processor

src/processor.cpp

@@ -20,6 +20,8 @@
 
 namespace rtc {
 
+Processor::Processor(size_t limit) : mTasks(limit) {}
+
 Processor::~Processor() { join(); }
 
 void Processor::join() {
@@ -29,15 +31,13 @@ void Processor::join() {
 
 void Processor::schedule() {
 	std::unique_lock lock(mMutex);
-	if (mTasks.empty()) {
+	if (auto next = mTasks.tryPop()) {
+		ThreadPool::Instance().enqueue(std::move(*next));
+	} else {
 		// No more tasks
 		mPending = false;
 		mCondition.notify_all();
-		return;
 	}
-
-	ThreadPool::Instance().enqueue(std::move(mTasks.front()));
-	mTasks.pop();
 }
 
 } // namespace rtc

src/processor.hpp

@@ -22,6 +22,7 @@
 #include "include.hpp"
 #include "init.hpp"
 #include "threadpool.hpp"
+#include "queue.hpp"
 
 #include <condition_variable>
 #include <future>
@@ -34,7 +35,7 @@ namespace rtc {
 // Processed tasks in order by delegating them to the thread pool
 class Processor final {
 public:
-	Processor() = default;
+	Processor(size_t limit = 0);
 	~Processor();
 
 	Processor(const Processor &) = delete;
@@ -52,7 +53,7 @@ protected:
 	// Keep an init token
 	const init_token mInitToken = Init::Token();
 
-	std::queue<std::function<void()>> mTasks;
+	Queue<std::function<void()>> mTasks;
 	bool mPending = false; // true iff a task is pending in the thread pool
 
 	mutable std::mutex mMutex;
@@ -71,7 +72,7 @@ template <class F, class... Args> void Processor::enqueue(F &&f, Args &&... args
 		ThreadPool::Instance().enqueue(std::move(task));
 		mPending = true;
 	} else {
-		mTasks.emplace(std::move(task));
+		mTasks.push(std::move(task));
 	}
 }
 

src/sctptransport.cpp

@@ -88,7 +88,7 @@ void SctpTransport::Cleanup() {
 SctpTransport::SctpTransport(std::shared_ptr<Transport> lower, uint16_t port,
                              message_callback recvCallback, amount_callback bufferedAmountCallback,
                              state_callback stateChangeCallback)
-    : Transport(lower, std::move(stateChangeCallback)), mPort(port),
+    : Transport(lower, std::move(stateChangeCallback)), mPort(port), mProcessor(RECV_QUEUE_LIMIT),
       mSendQueue(0, message_size_func), mBufferedAmountCallback(std::move(bufferedAmountCallback)) {
 	onRecv(recvCallback);
 
@@ -230,6 +230,16 @@ void SctpTransport::close() {
 	}
 }
 
+void SctpTransport::recv(message_ptr message) {
+	// Delegate to processor to release SCTP thread
+	mProcessor.enqueue([this, message = std::move(message)]() { Transport::recv(message); });
+}
+
+void SctpTransport::changeState(State state) {
+	// Delegate to processor to release SCTP thread
+	mProcessor.enqueue([this, state]() { Transport::changeState(state); });
+}
+
 void SctpTransport::connect() {
 	if (!mSock)
 		throw std::logic_error("Attempted SCTP connect with closed socket");

src/sctptransport.hpp

@@ -21,6 +21,7 @@
 
 #include "include.hpp"
 #include "peerconnection.hpp"
+#include "processor.hpp"
 #include "queue.hpp"
 #include "transport.hpp"
 
@@ -35,7 +36,7 @@
 
 namespace rtc {
 
-class SctpTransport : public Transport {
+class SctpTransport final : public Transport {
 public:
 	static void Init();
 	static void Cleanup();
@@ -71,6 +72,9 @@ private:
 		PPID_BINARY_EMPTY = 57
 	};
 
+	void recv(message_ptr message) override;
+	void changeState(State state) override;
+
 	void connect();
 	void shutdown();
 	void close();
@@ -93,6 +97,7 @@ private:
 	const uint16_t mPort;
 	struct socket *mSock;
 
+	Processor mProcessor;
 	std::mutex mSendMutex;
 	Queue<message_ptr> mSendQueue;
 	std::map<uint16_t, size_t> mBufferedAmount;

src/transport.hpp

@@ -67,14 +67,14 @@ public:
 	virtual bool send(message_ptr message) { return outgoing(message); }
 
 protected:
-	void recv(message_ptr message) {
+	virtual void recv(message_ptr message) {
 		try {
 			mRecvCallback(message);
 		} catch (const std::exception &e) {
 			PLOG_WARNING << e.what();
 		}
 	}
-	void changeState(State state) {
+	virtual void changeState(State state) {
 		try {
 			if (mState.exchange(state) != state)
 				mStateChangeCallback(state);