zen hub port reuse (#850)

- Feature: Added `--allow-port-probing` option to control whether zenserver searches for a free port on startup (default: true, automatically false when --dedicated is set)
- Feature: Added new hub options for controlling provisioned storage server instances:
  - `--hub-instance-http` - HTTP server implementation for instances (asio/httpsys)
  - `--hub-instance-http-threads` - Number of HTTP connection threads per instance
  - `--hub-instance-corelimit` - Limit CPU concurrency per instance
- Improvement: Hub now manages a deterministic port pool for provisioned instances allowing reuse of unused ports

1 files changed, 47 insertions, 9 deletions

diff --git a/src/zenserver/hub/hub.cpp b/src/zenserver/hub/hub.cpp
index c9720b32d..3c9f40eaa 100644
--- a/src/zenserver/hub/hub.cpp
+++ b/src/zenserver/hub/hub.cpp
@@ -22,6 +22,8 @@ ZEN_THIRD_PARTY_INCLUDES_END
 #	include <zencore/workthreadpool.h>
 #endif
 
+#include <numeric>
+
 namespace zen {
 
 ///////////////////////////////////////////////////////////////////////////
@@ -137,10 +139,10 @@ Hub::Hub(const Configuration&		   Config,
 	m_HydrationTempPath = m_RunEnvironment.CreateChildDir("hydration_temp");
 	ZEN_INFO("using hydration temp path: '{}'", m_HydrationTempPath);
 
-	// This is necessary to ensure the hub assigns a distinct port range.
-	// We need to do this primarily because otherwise automated tests will
-	// fail as the test runner will create processes in the default range.
-	m_RunEnvironment.SetNextPortNumber(m_Config.BasePortNumber);
+	ZEN_ASSERT(uint64_t(Config.BasePortNumber) + Config.InstanceLimit <= std::numeric_limits<uint16_t>::max());
+
+	m_FreePorts.resize(Config.InstanceLimit);
+	std::iota(m_FreePorts.begin(), m_FreePorts.end(), Config.BasePortNumber);
 
 #if ZEN_PLATFORM_WINDOWS
 	if (m_Config.UseJobObject)
@@ -199,6 +201,15 @@ Hub::Provision(std::string_view ModuleId, HubProvisionedInstanceInfo& OutInfo, s
 	bool				   IsNewInstance = false;
 	{
 		RwLock::ExclusiveLockScope _(m_Lock);
+		uint16_t				   AllocatedPort		= 0;
+		auto					   RestoreAllocatedPort = MakeGuard([this, &AllocatedPort]() {
+			  if (AllocatedPort != 0)
+			  {
+				  m_FreePorts.push_back(AllocatedPort);
+				  AllocatedPort = 0;
+			  }
+		  });
+
 		if (auto It = m_Instances.find(std::string(ModuleId)); It == m_Instances.end())
 		{
 			std::string Reason;
@@ -211,9 +222,18 @@ Hub::Provision(std::string_view ModuleId, HubProvisionedInstanceInfo& OutInfo, s
 				return false;
 			}
 
-			IsNewInstance = true;
-			auto NewInstance =
-				std::make_unique<StorageServerInstance>(m_RunEnvironment, ModuleId, m_FileHydrationPath, m_HydrationTempPath);
+			AllocatedPort = m_FreePorts.front();
+			m_FreePorts.pop_front();
+
+			IsNewInstance	 = true;
+			auto NewInstance = std::make_unique<StorageServerInstance>(
+				m_RunEnvironment,
+				StorageServerInstance::Configuration{.BasePort			= AllocatedPort,
+													 .HydrationTempPath = m_HydrationTempPath,
+													 .FileHydrationPath = m_FileHydrationPath,
+													 .HttpThreadCount	= m_Config.InstanceHttpThreadCount,
+													 .CoreLimit			= m_Config.InstanceCoreLimit},
+				ModuleId);
 #if ZEN_PLATFORM_WINDOWS
 			if (m_JobObject.IsValid())
 			{
@@ -222,6 +242,7 @@ Hub::Provision(std::string_view ModuleId, HubProvisionedInstanceInfo& OutInfo, s
 #endif
 			Instance = NewInstance.get();
 			m_Instances.emplace(std::string(ModuleId), std::move(NewInstance));
+			AllocatedPort = 0;
 
 			ZEN_INFO("Created new storage server instance for module '{}'", ModuleId);
 		}
@@ -258,7 +279,13 @@ Hub::Provision(std::string_view ModuleId, HubProvisionedInstanceInfo& OutInfo, s
 		{
 			// Clean up
 			RwLock::ExclusiveLockScope _(m_Lock);
-			m_Instances.erase(std::string(ModuleId));
+			if (auto It = m_Instances.find(std::string(ModuleId)); It != m_Instances.end())
+			{
+				ZEN_ASSERT(It->second != nullptr);
+				uint16_t BasePort = It->second->GetBasePort();
+				m_FreePorts.push_back(BasePort);
+				m_Instances.erase(It);
+			}
 		}
 		return false;
 	}
@@ -337,6 +364,7 @@ Hub::Deprovision(const std::string& ModuleId, std::string& OutReason)
 	auto _ = MakeGuard([&] {
 		RwLock::ExclusiveLockScope _(m_Lock);
 		m_DeprovisioningModules.erase(ModuleId);
+		m_FreePorts.push_back(BasePort);
 	});
 
 	Instance->Deprovision();
@@ -413,7 +441,17 @@ Hub::CanProvisionInstance(std::string_view ModuleId, std::string& OutReason)
 
 	if (gsl::narrow_cast<int>(m_Instances.size()) >= m_Config.InstanceLimit)
 	{
-		OutReason = fmt::format("instance limit exceeded ({})", m_Config.InstanceLimit);
+		OutReason = fmt::format("instance limit ({}) exceeded", m_Config.InstanceLimit);
+
+		return false;
+	}
+
+	// Since deprovisioning happens outside the lock and we don't add the port back until the instance is full shut down we might be under
+	// the instance limit but all ports may be in use
+	if (m_FreePorts.empty())
+	{
+		OutReason = fmt::format("no free ports available, deprovisioning of instances might be in flight ({})",
+								m_Config.InstanceLimit - m_Instances.size());
 
 		return false;
 	}