Merge pull request #509 from paullouisageneau/fix-init

Refactor Init as a singleton

src/global.cpp

@@ -56,7 +56,7 @@ void plogInit(plog::Severity severity, plog::IAppender *appender) {
 	}
 }
 
-}
+} // namespace
 
 namespace rtc {
 
@@ -102,10 +102,10 @@ void InitLogger(plog::Severity severity, plog::IAppender *appender) {
 	plogInit(severity, appender);
 }
 
-void Preload() { Init::Preload(); }
-void Cleanup() { Init::Cleanup(); }
+void Preload() { Init::Instance().preload(); }
+void Cleanup() { Init::Instance().cleanup(); }
 
-void SetSctpSettings(SctpSettings s) { Init::SetSctpSettings(std::move(s)); }
+void SetSctpSettings(SctpSettings s) { Init::Instance().setSctpSettings(std::move(s)); }
 
 } // namespace rtc
 

src/impl/init.cpp

@@ -39,9 +39,62 @@
 
 namespace rtc {
 
-namespace {
+struct Init::TokenPayload {
+	TokenPayload() { Init::Instance().doInit(); }
 
-void doInit() {
+	~TokenPayload() {
+		std::thread t([]() { Init::Instance().doCleanup(); });
+		t.detach();
+	}
+};
+
+Init &Init::Instance() {
+	static Init *instance = new Init;
+	return *instance;
+}
+
+Init::Init() {}
+
+Init::~Init() {}
+
+init_token Init::token() {
+	std::lock_guard lock(mMutex);
+	if (auto locked = mWeak.lock())
+		return locked;
+
+	mGlobal = std::make_shared<TokenPayload>();
+	mWeak = *mGlobal;
+	return *mGlobal;
+}
+
+void Init::preload() {
+	std::lock_guard lock(mMutex);
+	if (!mGlobal) {
+		mGlobal = std::make_shared<TokenPayload>();
+		mWeak = *mGlobal;
+	}
+}
+
+void Init::cleanup() {
+	std::lock_guard lock(mMutex);
+	mGlobal.reset();
+}
+
+void Init::setSctpSettings(SctpSettings s) {
+	std::lock_guard lock(mMutex);
+	if (mGlobal)
+		impl::SctpTransport::SetSettings(s);
+
+	mCurrentSctpSettings = std::move(s); // store for next init
+}
+
+void Init::doInit() {
+	// mMutex needs to be locked
+
+	if (std::exchange(mInitialized, true))
+		return;
+
+	PLOG_DEBUG << "Global initialization";
 
 #ifdef _WIN32
 	WSADATA wsaData;
@@ -58,6 +111,7 @@ void doInit() {
 #endif
 
 	impl::SctpTransport::Init();
+	impl::SctpTransport::SetSettings(mCurrentSctpSettings);
 	impl::DtlsTransport::Init();
 #if RTC_ENABLE_WEBSOCKET
 	impl::TlsTransport::Init();
@@ -67,7 +121,16 @@ void doInit() {
 #endif
 }
 
-void doCleanup() {
+void Init::doCleanup() {
+	std::lock_guard lock(mMutex);
+	if (mGlobal)
+		return;
+
+	if (!std::exchange(mInitialized, false))
+		return;
+
+	PLOG_DEBUG << "Global cleanup";
+
 	impl::ThreadPool::Instance().join();
 
 	impl::SctpTransport::Cleanup();
@@ -84,67 +147,4 @@ void doCleanup() {
 #endif
 }
 
-} // namespace
-
-weak_ptr<void> Init::Weak;
-shared_ptr<void> *Init::Global = nullptr;
-bool Init::Initialized = false;
-SctpSettings Init::CurrentSctpSettings = {};
-std::recursive_mutex Init::Mutex;
-
-init_token Init::Token() {
-	std::unique_lock lock(Mutex);
-	if (auto token = Weak.lock())
-		return token;
-
-	delete Global;
-	Global = new shared_ptr<void>(new Init());
-	Weak = *Global;
-	return *Global;
-}
-
-void Init::Preload() {
-	std::unique_lock lock(Mutex);
-	auto token = Token();
-	if (!Global)
-		Global = new shared_ptr<void>(token);
-}
-
-void Init::Cleanup() {
-	std::unique_lock lock(Mutex);
-	delete Global;
-	Global = nullptr;
-}
-
-void Init::SetSctpSettings(SctpSettings s) {
-	auto token = Token();
-	std::unique_lock lock(Mutex);
-	impl::SctpTransport::SetSettings(s);
-	CurrentSctpSettings = std::move(s); // store for next init
-}
-
-Init::Init() {
-	// Mutex is locked by Token() here
-	if (!std::exchange(Initialized, true)) {
-		PLOG_DEBUG << "Global initialization";
-		doInit();
-		impl::SctpTransport::SetSettings(CurrentSctpSettings);
-	}
-}
-
-Init::~Init() {
-	std::thread t([]() {
-		// We need to lock Mutex ourselves
-		std::unique_lock lock(Mutex);
-		if (Global)
-			return;
-
-		if (std::exchange(Initialized, false)) {
-			PLOG_DEBUG << "Global cleanup";
-			doCleanup();
-		}
-	});
-	t.detach();
-}
-
 } // namespace rtc

src/impl/init.hpp

@@ -31,21 +31,32 @@ using init_token = shared_ptr<void>;
 
 class Init {
 public:
-	static init_token Token();
-	static void Preload();
-	static void Cleanup();
-	static void SetSctpSettings(SctpSettings s);
+	static Init &Instance();
 
-	~Init();
+	Init(const Init &) = delete;
+	Init &operator=(const Init &) = delete;
+	Init(Init &&) = delete;
+	Init &operator=(Init &&) = delete;
+
+	init_token token();
+	void preload();
+	void cleanup();
+	void setSctpSettings(SctpSettings s);
 
 private:
 	Init();
+	~Init();
+
+	void doInit();
+	void doCleanup();
+
+	std::optional<shared_ptr<void>> mGlobal;
+	weak_ptr<void> mWeak;
+	bool mInitialized = false;
+	SctpSettings mCurrentSctpSettings = {};
+	std::mutex mMutex;
 
-	static weak_ptr<void> Weak;
-	static shared_ptr<void> *Global;
-	static bool Initialized;
-	static SctpSettings CurrentSctpSettings;
-	static std::recursive_mutex Mutex;
+	struct TokenPayload;
 };
 
 } // namespace rtc

src/impl/peerconnection.hpp

@@ -116,7 +116,7 @@ struct PeerConnection : std::enable_shared_from_this<PeerConnection> {
 	synchronized_callback<shared_ptr<rtc::Track>> trackCallback;
 
 private:
-	const init_token mInitToken = Init::Token();
+	const init_token mInitToken = Init::Instance().token();
 	const future_certificate_ptr mCertificate;
 	const unique_ptr<Processor> mProcessor;
 

src/impl/processor.hpp

@@ -51,7 +51,7 @@ protected:
 	void schedule();
 
 	// Keep an init token
-	const init_token mInitToken = Init::Token();
+	const init_token mInitToken = Init::Instance().token();
 
 	Queue<std::function<void()>> mTasks;
 	bool mPending = false; // true iff a task is pending in the thread pool

src/impl/threadpool.hpp

@@ -66,7 +66,7 @@ public:
 	template <class F, class... Args>
 	auto schedule(clock::time_point time, F &&f, Args &&...args) -> invoke_future_t<F, Args...>;
 
-protected:
+private:
 	ThreadPool();
 	~ThreadPool();
 
@@ -115,7 +115,7 @@ auto ThreadPool::schedule(clock::time_point time, F &&f, Args &&...args)
 	});
 	std::future<R> result = task->get_future();
 
-	mTasks.push({time, [task = std::move(task), token = Init::Token()]() { return (*task)(); }});
+	mTasks.push({time, [task = std::move(task), token = Init::Instance().token()]() { return (*task)(); }});
 	mTasksCondition.notify_one();
 	return result;
 }

src/impl/websocket.hpp

@@ -75,7 +75,7 @@ struct WebSocket final : public Channel, public std::enable_shared_from_this<Web
 	std::atomic<State> state = State::Closed;
 
 private:
-	const init_token mInitToken = Init::Token();
+	const init_token mInitToken = Init::Instance().token();
 
 	const certificate_ptr mCertificate;
 	bool mIsSecure;

src/impl/websocketserver.hpp

@@ -50,7 +50,7 @@ struct WebSocketServer final : public std::enable_shared_from_this<WebSocketServ
 	synchronized_callback<shared_ptr<rtc::WebSocket>> clientCallback;
 
 private:
-	const init_token mInitToken = Init::Token();
+	const init_token mInitToken = Init::Instance().token();
 
 	void runLoop();