Add environment variable to set max concurrent backup compression tasks

Agent/Phantom.Agent.Services/AgentServiceConfiguration.cs

@@ -0,0 +1,3 @@
+namespace Phantom.Agent.Services; 
+
+public readonly record struct AgentServiceConfiguration(int MaxConcurrentCompressionTasks);

Agent/Phantom.Agent.Services/AgentServices.cs

@@ -18,10 +18,10 @@ public sealed class AgentServices {
 	internal JavaRuntimeRepository JavaRuntimeRepository { get; }
 	internal InstanceSessionManager InstanceSessionManager { get; }
 
-	public AgentServices(AgentInfo agentInfo, AgentFolders agentFolders) {
+	public AgentServices(AgentInfo agentInfo, AgentFolders agentFolders, AgentServiceConfiguration serviceConfiguration) {
 		this.AgentFolders = agentFolders;
 		this.TaskManager = new TaskManager(PhantomLogger.Create<TaskManager, AgentServices>());
-		this.BackupManager = new BackupManager(agentFolders);
+		this.BackupManager = new BackupManager(agentFolders, serviceConfiguration.MaxConcurrentCompressionTasks);
 		this.JavaRuntimeRepository = new JavaRuntimeRepository();
 		this.InstanceSessionManager = new InstanceSessionManager(agentInfo, agentFolders, JavaRuntimeRepository, TaskManager, BackupManager);
 	}
@@ -40,6 +40,8 @@ public sealed class AgentServices {
 		
 		await TaskManager.Stop();
 		
+		BackupManager.Dispose();
+		
 		Logger.Information("Services stopped.");
 	}
 }

Agent/Phantom.Agent.Services/Backups/BackupManager.cs

@@ -5,15 +5,21 @@ using Serilog;
 
 namespace Phantom.Agent.Services.Backups;
 
-sealed class BackupManager {
+sealed class BackupManager : IDisposable {
 	private readonly string destinationBasePath;
 	private readonly string temporaryBasePath;
+	private readonly SemaphoreSlim compressionSemaphore;
 
-	public BackupManager(AgentFolders agentFolders) {
+	public BackupManager(AgentFolders agentFolders, int maxConcurrentCompressionTasks) {
 		this.destinationBasePath = agentFolders.BackupsFolderPath;
 		this.temporaryBasePath = Path.Combine(agentFolders.TemporaryFolderPath, "backups");
+		this.compressionSemaphore = new SemaphoreSlim(maxConcurrentCompressionTasks, maxConcurrentCompressionTasks);
 	}
 
+	public void Dispose() {
+		compressionSemaphore.Dispose();
+	}
+	
 	public async Task<BackupCreationResult> CreateBackup(string loggerName, InstanceProcess process, CancellationToken cancellationToken) {
 		try {
 			if (!await process.BackupSemaphore.Wait(TimeSpan.FromSeconds(1), cancellationToken)) {
@@ -26,23 +32,21 @@ sealed class BackupManager {
 		}
 
 		try {
-			return await new BackupCreator(destinationBasePath, temporaryBasePath, loggerName, process, cancellationToken).CreateBackup();
+			return await new BackupCreator(this, loggerName, process, cancellationToken).CreateBackup();
 		} finally {
 			process.BackupSemaphore.Release();
 		}
 	}
 
 	private sealed class BackupCreator {
-		private readonly string destinationBasePath;
-		private readonly string temporaryBasePath;
+		private readonly BackupManager manager;
 		private readonly string loggerName;
 		private readonly ILogger logger;
 		private readonly InstanceProcess process;
 		private readonly CancellationToken cancellationToken;
 
-		public BackupCreator(string destinationBasePath, string temporaryBasePath, string loggerName, InstanceProcess process, CancellationToken cancellationToken) {
-			this.destinationBasePath = destinationBasePath;
-			this.temporaryBasePath = temporaryBasePath;
+		public BackupCreator(BackupManager manager, string loggerName, InstanceProcess process, CancellationToken cancellationToken) {
+			this.manager = manager;
 			this.loggerName = loggerName;
 			this.logger = PhantomLogger.Create<BackupManager>(loggerName);
 			this.process = process;
@@ -72,7 +76,7 @@ sealed class BackupManager {
 			try {
 				await dispatcher.DisableAutomaticSaving();
 				await dispatcher.SaveAllChunks();
-				return await new BackupArchiver(destinationBasePath, temporaryBasePath, loggerName, process.InstanceProperties, cancellationToken).ArchiveWorld(resultBuilder);
+				return await new BackupArchiver(manager.destinationBasePath, manager.temporaryBasePath, loggerName, process.InstanceProperties, cancellationToken).ArchiveWorld(resultBuilder);
 			} catch (OperationCanceledException) {
 				resultBuilder.Kind = BackupCreationResultKind.BackupCancelled;
 				logger.Warning("Backup creation was cancelled.");
@@ -92,11 +96,21 @@ sealed class BackupManager {
 				}
 			}
 		}
-
+		
 		private async Task CompressWorldArchive(string filePath, BackupCreationResult.Builder resultBuilder) {
-			var compressedFilePath = await BackupCompressor.Compress(filePath, cancellationToken);
-			if (compressedFilePath == null) {
-				resultBuilder.Warnings |= BackupCreationWarnings.CouldNotCompressWorldArchive;
+			if (!await manager.compressionSemaphore.WaitAsync(TimeSpan.FromSeconds(1), cancellationToken)) {
+				logger.Information("Too many compression tasks running, waiting for one of them to complete...");
+				await manager.compressionSemaphore.WaitAsync(cancellationToken);
+			}
+
+			logger.Information("Compressing backup...");
+			try {
+				var compressedFilePath = await BackupCompressor.Compress(filePath, cancellationToken);
+				if (compressedFilePath == null) {
+					resultBuilder.Warnings |= BackupCreationWarnings.CouldNotCompressWorldArchive;
+				}
+			} finally {
+				manager.compressionSemaphore.Release();
 			}
 		}
 

Agent/Phantom.Agent/Program.cs

@@ -23,7 +23,7 @@ try {
 	PhantomLogger.Root.InformationHeading("Initializing Phantom Panel agent...");
 	PhantomLogger.Root.Information("Agent version: {Version}", fullVersion);
 
-	var (serverHost, serverPort, javaSearchPath, agentKeyToken, agentKeyFilePath, agentName, maxInstances, maxMemory, allowedServerPorts, allowedRconPorts) = Variables.LoadOrExit();
+	var (serverHost, serverPort, javaSearchPath, agentKeyToken, agentKeyFilePath, agentName, maxInstances, maxMemory, allowedServerPorts, allowedRconPorts, maxConcurrentBackupCompressionTasks) = Variables.LoadOrExit();
 
 	var agentKey = await AgentKey.Load(agentKeyToken, agentKeyFilePath);
 	if (agentKey == null) {
@@ -42,7 +42,7 @@ try {
 
 	var (serverCertificate, agentToken) = agentKey.Value;
 	var agentInfo = new AgentInfo(agentGuid.Value, agentName, ProtocolVersion, fullVersion, maxInstances, maxMemory, allowedServerPorts, allowedRconPorts);
-	var agentServices = new AgentServices(agentInfo, folders);
+	var agentServices = new AgentServices(agentInfo, folders, new AgentServiceConfiguration(maxConcurrentBackupCompressionTasks));
 
 	MessageListener MessageListenerFactory(RpcServerConnection connection) {
 		return new MessageListener(connection, agentServices, shutdownCancellationTokenSource);

Agent/Phantom.Agent/Variables.cs

@@ -15,7 +15,8 @@ sealed record Variables(
 	ushort MaxInstances,
 	RamAllocationUnits MaxMemory,
 	AllowedPorts AllowedServerPorts,
-	AllowedPorts AllowedRconPorts
+	AllowedPorts AllowedRconPorts,
+	ushort MaxConcurrentBackupCompressionTasks
 ) {
 	private static Variables LoadOrThrow() {
 		var (agentKeyToken, agentKeyFilePath) = EnvironmentVariables.GetEitherString("AGENT_KEY", "AGENT_KEY_FILE").Require;
@@ -31,7 +32,8 @@ sealed record Variables(
 			(ushort) EnvironmentVariables.GetInteger("MAX_INSTANCES", min: 1, max: 10000).Require,
 			EnvironmentVariables.GetString("MAX_MEMORY").MapParse(RamAllocationUnits.FromString).Require,
 			EnvironmentVariables.GetString("ALLOWED_SERVER_PORTS").MapParse(AllowedPorts.FromString).Require,
-			EnvironmentVariables.GetString("ALLOWED_RCON_PORTS").MapParse(AllowedPorts.FromString).Require
+			EnvironmentVariables.GetString("ALLOWED_RCON_PORTS").MapParse(AllowedPorts.FromString).Require,
+			(ushort) EnvironmentVariables.GetInteger("MAX_CONCURRENT_BACKUP_COMPRESSION_TASKS", min: 1, max: 10000).WithDefault(1)
 		);
 	}
 

README.md

@@ -92,6 +92,7 @@ Use volumes to persist either the whole `/data` folder, or just `/data/data` if
 * **Agent Configuration**
   - `MAX_INSTANCES` is the number of instances that can be created.
   - `MAX_MEMORY` is the maximum amount of RAM that can be distributed among all instances. Use a positive integer with an optional suffix 'M' for MB, or 'G' for GB. Examples: `4096M`, `16G`
+  - `MAX_CONCURRENT_BACKUP_COMPRESSION_TASKS` is how many backup compression tasks can run at the same time. Limiting concurrent compression tasks limits memory usage of compression, but it increases time between backups because the next backup is only scheduled once the current one completes. Default: `1`
 * **Minecraft Configuration**
   - `JAVA_SEARCH_PATH` is a path to a folder which will be searched for Java runtime installations. Linux default: `/usr/lib/jvm`
   - `ALLOWED_SERVER_PORTS` is a comma-separated list of ports and port ranges that can be used as Minecraft Server ports. Example: `25565,25900,26000-27000`