Fixed an intermittent issue where (LL-)HLS requests could be missed

Author: getroot

src/projects/base/publisher/stream.cpp

@@ -465,7 +465,7 @@ namespace pub
 			return nullptr;
 		}
 
-		return _sessions[id];
+		return _sessions.at(id);
 	}
 
 	const std::map<session_id_t, std::shared_ptr<Session>> Stream::GetAllSessions()

src/projects/publishers/hls/hls_stream.cpp

@@ -89,32 +89,48 @@ ov::String HlsStream::GetStreamId() const
 	return ov::String::FormatString("hlsv3/%s", GetUri().CStr());
 }
 
+bool HlsStream::CreateOriginSessionPool()
+{
+	if (_origin_mode == false)
+	{
+		return false;
+	}
+
+	size_t max_pool_size = _worker_count == 0 ? 1 : _worker_count;
+
+	// Create sessions up to _worker_count
+	for (size_t i = 0; i < max_pool_size; i++)
+	{
+		auto session = HlsSession::Create(static_cast<session_id_t>(i),
+											_origin_mode,
+											"",
+											GetApplication(),
+											pub::Stream::GetSharedPtr(),
+											0);
+		logtd("LLHlsStream(%s/%s) - Pre-created origin mode session in pool, session id: %zu", GetApplication()->GetVHostAppName().CStr(), GetName().CStr(), i);
+		AddSession(session);
+	}
+
+	return true;
+}
+
 std::shared_ptr<HlsSession> HlsStream::GetSessionFromPool()
 {
 	// Max session pool size if _worker_count
-	size_t max_pool_size = _worker_count;
+	size_t max_pool_size = _worker_count == 0 ? 1 : _worker_count;
 
 	// Get random index
 	size_t index = ov::Random::GenerateUInt32() % max_pool_size;
 
 	auto session = GetSession(static_cast<session_id_t>(index));
 	if (session == nullptr)
 	{
-		// create
-		session = HlsSession::Create(static_cast<session_id_t>(index),
-										true,
-										"",
-										GetApplication(),
-										pub::Stream::GetSharedPtr(),
-										0);
-		logtd("HlsStream(%s/%s) - Created origin mode session from pool, session id: %zu", GetApplication()->GetVHostAppName().CStr(), GetName().CStr(), index);
-		AddSession(session);
+		return nullptr;
 	}
 
 	return std::static_pointer_cast<HlsSession>(session);
 }
 
-
 bool HlsStream::Start()
 {
 	if (GetState() != State::CREATED)
@@ -127,6 +143,15 @@ bool HlsStream::Start()
 		return false;
 	}
 
+	if (_origin_mode == true)
+	{
+		if (CreateOriginSessionPool() == false)
+		{
+			logte("HLS Stream(%s/%s) - Failed to create origin session pool", GetApplication()->GetVHostAppName().CStr(), GetName().CStr());
+			return false;
+		}
+	}
+
 	auto config = GetApplication()->GetConfig();
 	_ts_config = config.GetPublishers().GetHlsPublisher();
 

src/projects/publishers/hls/hls_stream.h

@@ -149,4 +149,8 @@ class HlsStream final : public pub::Stream, public mpegts::PackagerSink
 	std::shared_mutex _dumps_lock;
 
 	bool _ready_to_play = false; // true if the stream is ready to play, all playlists are ready and have enough segments
+
+
+	bool CreateOriginSessionPool();
+	bool _origin_mode = true;
 };

src/projects/publishers/llhls/llhls_stream.cpp

@@ -44,31 +44,49 @@ ov::String LLHlsStream::GetStreamId() const
 	return ov::String::FormatString("llhls/%s", GetUri().CStr());
 }
 
+bool LLHlsStream::CreateOriginSessionPool()
+{
+	if (_origin_mode == false)
+	{
+		return false;
+	}
+
+	size_t max_pool_size = _worker_count == 0 ? 1 : _worker_count;
+
+	// Create sessions up to _worker_count
+	for (size_t i = 0; i < max_pool_size; i++)
+	{
+		auto session = LLHlsSession::Create(static_cast<session_id_t>(i),
+											_origin_mode,
+											"",
+											GetApplication(),
+											pub::Stream::GetSharedPtr(),
+											0);
+		logtd("LLHlsStream(%s/%s) - Pre-created origin mode session in pool, session id: %zu", GetApplication()->GetVHostAppName().CStr(), GetName().CStr(), i);
+		AddSession(session);
+	}
+
+	return true;
+}
+
 std::shared_ptr<LLHlsSession> LLHlsStream::GetSessionFromPool()
 {
 	// Max session pool size if _worker_count
-	size_t max_pool_size = _worker_count;
+	size_t max_pool_size = _worker_count == 0 ? 1 : _worker_count;
 
 	// Get random index
 	size_t index = ov::Random::GenerateUInt32() % max_pool_size;
 
 	auto session = GetSession(static_cast<session_id_t>(index));
 	if (session == nullptr)
 	{
-		// create
-		session = LLHlsSession::Create(static_cast<session_id_t>(index),
-										_origin_mode,
-										"",
-										GetApplication(),
-										pub::Stream::GetSharedPtr(),
-										0);
-		logtd("LLHlsStream(%s/%s) - Created origin mode session from pool, session id: %zu", GetApplication()->GetVHostAppName().CStr(), GetName().CStr(), index);
-		AddSession(session);
+		return nullptr;
 	}
 
 	return std::static_pointer_cast<LLHlsSession>(session);
 }
 
+
 std::shared_ptr<const pub::Stream::DefaultPlaylistInfo> LLHlsStream::GetDefaultPlaylistInfo() const
 {
 	static auto info = []() -> std::shared_ptr<const pub::Stream::DefaultPlaylistInfo> {
@@ -96,6 +114,15 @@ bool LLHlsStream::Start()
 		return false;
 	}
 
+	if (_origin_mode == true)
+	{
+		if (CreateOriginSessionPool() == false)
+		{
+			logte("LLHlsStream(%s/%s) - Failed to create origin session pool", GetApplication()->GetVHostAppName().CStr(), GetName().CStr());
+			return false;
+		}
+	}
+
 	auto config = GetApplication()->GetConfig();
 
 	auto llhls_config = config.GetPublishers().GetLLHlsPublisher();

src/projects/publishers/llhls/llhls_stream.h

@@ -224,4 +224,6 @@ class LLHlsStream final : public pub::Stream, public bmff::FMp4StorageObserver
 
 	std::map<int32_t, std::shared_ptr<webvtt::Packager>> _vtt_packagers;
 	mutable std::shared_mutex _vtt_packagers_lock;
+
+	bool CreateOriginSessionPool();
 };