diff --git a/MergerService/Runners/TaskExecutor.cs b/MergerService/Runners/TaskExecutor.cs index eb1383fb..f164ada3 100644 --- a/MergerService/Runners/TaskExecutor.cs +++ b/MergerService/Runners/TaskExecutor.cs @@ -6,6 +6,7 @@ using MergerService.Controllers; using MergerService.Models.Tasks; using MergerService.Utils; +using System.Collections.Concurrent; using System.Diagnostics; using System.IO.Abstractions; using System.Reflection; @@ -28,6 +29,7 @@ public class TaskExecutor : ITaskExecutor private readonly long _batchMaxBytes; private readonly string _filePath; private readonly bool _shouldValidate; + private readonly int _maxDegreeOfParallelism; private static readonly int DEFAULT_BATCH_SIZE = 1000; public TaskExecutor(IDataFactory dataFactory, ITileMerger tileMerger, ITimeUtils timeUtils, IConfigurationManager configurationManager, @@ -59,6 +61,10 @@ public TaskExecutor(IDataFactory dataFactory, ITileMerger tileMerger, ITimeUtils this._batchMaxSize = DEFAULT_BATCH_SIZE; } + + int numOfThreads = configurationManager.GetConfiguration("GENERAL", "parallel", "numOfThreads"); + // 0 (unset) or negative means "let the runtime decide" per available cores. + this._maxDegreeOfParallelism = numOfThreads > 0 ? numOfThreads : Environment.ProcessorCount; } public void ExecuteTask(MergeTask task, ITaskUtils taskUtils, string? managerCallbackUrl) @@ -123,8 +129,6 @@ public void ExecuteTask(MergeTask task, ITaskUtils taskUtils, string? managerCal long singleTileBatchCount = bounds.Size(); this._metricsProvider.TilesInBatchGauge(singleTileBatchCount); - int tileProgressCount = 0; - // TODO: remove comment and check that the activity is created (When bug will be fixed) // batchActivity.AddTag("size", totalTileCount); @@ -134,9 +138,6 @@ public void ExecuteTask(MergeTask task, ITaskUtils taskUtils, string? managerCal continue; } - List tiles = new List((int)singleTileBatchCount); - long currentBatchBytes = 0; - this._logger.LogInformation($"[{methodName}] Total amount of tiles to merge for current batch: {singleTileBatchCount}"); // Go over the bounds of the current batch @@ -144,25 +145,40 @@ public void ExecuteTask(MergeTask task, ITaskUtils taskUtils, string? managerCal { var batchWorkTimeStopwatch = Stopwatch.StartNew(); + // Enumerate the coords once, then merge in parallel one chunk at a time. Each chunk + // is flushed to the target before the next starts, so memory stays bounded to ~one + // chunk (count-based; the previous byte cap is approximated by batchMaxSize). + var coords = new List((int)singleTileBatchCount); for (int x = bounds.MinX; x <= bounds.MaxX; x++) { for (int y = bounds.MinY; y <= bounds.MaxY; y++) { - this._logger.LogDebug($"[{methodName}] Handle tile z:{bounds.Zoom}, x:{x}, y:{y}"); - Coord coord = new Coord(bounds.Zoom, x, y); - - // Create tile builder list for current coord for all sources - List correspondingTileBuilders = new List(); - // Add target tile - correspondingTileBuilders.Add(() => sources[0].GetCorrespondingTile(coord, shouldUpscale)); - - // Add all sources tiles - this._logger.LogDebug($"[{methodName}] Get tile sources"); + coords.Add(new Coord(bounds.Zoom, x, y)); + } + } + + int chunkSize = this._batchMaxSize; + var parallelOptions = new ParallelOptions { MaxDegreeOfParallelism = this._maxDegreeOfParallelism }; + + foreach (Coord[] chunk in coords.Chunk(chunkSize)) + { + var mergedTiles = new ConcurrentBag(); + + // Merge is the hot path and each coord is independent; the target is only read + // here (writes happen after this parallel phase), and GetCorrespondingTile is + // thread-safe per data source. + Parallel.ForEach(chunk, parallelOptions, coord => + { + var correspondingTileBuilders = new List + { + () => sources[0].GetCorrespondingTile(coord, shouldUpscale) + }; foreach (IData source in sources.Skip(1)) { // TODO: upscale = false - this is a temporary fix till we decide how sources should be upscaled correspondingTileBuilders.Add(() => source.GetCorrespondingTile(coord, false)); } + var tileMergeStopwatch = Stopwatch.StartNew(); Tile? tile = this._tileMerger.MergeTiles(correspondingTileBuilders, coord, strategy, metadata.IsNewTarget); tileMergeStopwatch.Stop(); @@ -170,41 +186,26 @@ public void ExecuteTask(MergeTask task, ITaskUtils taskUtils, string? managerCal if (tile != null) { - tiles.Add(tile); - currentBatchBytes += tile.Size(); - - // Flushes the "tiles" list if it reached the batch size or the batch max bytes - // This is done to prevent memory overflow - if (currentBatchBytes >= this._batchMaxBytes || (this._limitBatchSize && tiles.Count >= this._batchMaxSize)) - { - this.UpdateTargetTiles(target, tiles, task, overallTileProgressCount, totalTileCount, taskUtils); - - tiles.Clear(); - currentBatchBytes = 0; - } + mergedTiles.Add(tile); } + }); - tileProgressCount++; - overallTileProgressCount++; + long progressAfterChunk = Interlocked.Add(ref overallTileProgressCount, chunk.Length); - // Show progress every batchSize - if (overallTileProgressCount % this._batchMaxSize == 0) - { - this._logger.LogInformation( - $"[{methodName}] Job: {task.JobId}, Task: {task.Id}, Tile Count: {overallTileProgressCount} / {totalTileCount}"); - UpdateRelativeProgress(task, overallTileProgressCount, totalTileCount, taskUtils); - } + if (!mergedTiles.IsEmpty) + { + this.UpdateTargetTiles(target, mergedTiles.ToList(), task, progressAfterChunk, totalTileCount, taskUtils); } + + this._logger.LogInformation( + $"[{methodName}] Job: {task.JobId}, Task: {task.Id}, Tile Count: {progressAfterChunk} / {totalTileCount}"); + UpdateRelativeProgress(task, progressAfterChunk, totalTileCount, taskUtils); } + batchWorkTimeStopwatch.Stop(); this._metricsProvider.BatchWorkTimeHistogram(batchWorkTimeStopwatch.Elapsed.TotalSeconds); } - if (tiles.Count > 0) - { - this.UpdateTargetTiles(target, tiles, task, overallTileProgressCount, totalTileCount, taskUtils); - } - this._logger.LogInformation($"[{methodName}] Overall tile Count: {overallTileProgressCount} / {totalTileCount}"); mergeRunTimeStopwatch.Stop(); diff --git a/MergerService/appsettings.json b/MergerService/appsettings.json index 1d96c4c1..ead3a2ba 100644 --- a/MergerService/appsettings.json +++ b/MergerService/appsettings.json @@ -8,6 +8,9 @@ "batchSize": { "limitBatchSize": true, "batchMaxSize": 1000 + }, + "parallel": { + "numOfThreads": 0 } }, "TASK": { diff --git a/MergerServiceUnitTests/Runners/TaskExecutorTest.cs b/MergerServiceUnitTests/Runners/TaskExecutorTest.cs index 5522ac56..81600701 100644 --- a/MergerServiceUnitTests/Runners/TaskExecutorTest.cs +++ b/MergerServiceUnitTests/Runners/TaskExecutorTest.cs @@ -112,19 +112,16 @@ public void WhenGivenSourcesWithOneTile_ShouldWriteAllTilesToTarget(int numberOf public static IEnumerable GetBatchLimitTestParameters() { - // Tests that given a batch limit size, it flushes every time it reaches the limit - // When the limit is 2, and there are total 5 tiles, we expect 3 flushes - // We set the bytes limit to a high value to avoid it being reached by the test + // Merging is chunked by batchMaxSize and each chunk is flushed once. + // limitBatchSize=true, size=2, 5 tiles -> ceil(5/2) = 3 flushes. yield return new object[] { true, 2, 1024 * 1024 * 80, 3, 5 }; - // Tests that given a batch bytes limit, it flushes every time it reaches the limit - // When the limit is one byte more then twice the size of the tile, and there are total 7 tiles, we expect 3 flushes - // (Flush every 3rd tile, and additional flush at the end) - // We disable the size limit to avoid it being used in the test + // limitBatchSize=false -> chunking falls back to the default batch size (1000), + // so a small 7-tile task is written in a single flush. yield return new object[] { - false, 1, (File.ReadAllBytes("tile.jpeg").Length * 2) + 1, 3, 7 + false, 1, (File.ReadAllBytes("tile.jpeg").Length * 2) + 1, 1, 7 }; } @@ -143,7 +140,9 @@ public void WhenConfiguringBatchLimits_ShouldWriteTilesEachTimeAfterReachingBatc Source testSource = new Source($"source", $"source_type"); Coord[] testSourceCoords = new int[totalAmountOfTiles].Select((_, index) => new Coord(1, index, 0)).ToArray(); Tile[] testSourceTiles = testSourceCoords.Select(coord => new Tile(coord, tileBytes)).ToArray(); - TileBounds tileBounds = new TileBounds(1, 0, testSourceCoords.Length, 0, 1); + // Dense single-row bounds: every coord in the bounds has a tile, so chunking by coord equals + // chunking by produced tile. + TileBounds tileBounds = new TileBounds(1, 0, testSourceCoords.Length - 1, 0, 0); Mock sourceDataMock = this._mockRepository.Create(); for (var testSourceCoordIdx = 0; testSourceCoordIdx < testSourceCoords.Length; testSourceCoordIdx++)