Skip to content

perf(service): parallelize intra-task tile merge (MAPCO-11325) #260

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
shimoncohen wants to merge 1 commit into
base: logic-1-gpkgclient-perf
Choose a base branch
from
+53 −50
Open

perf(service): parallelize intra-task tile merge (MAPCO-11325) #260

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
83 changes: 42 additions & 41 deletions MergerService/Runners/TaskExecutor.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -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,
Expand Down Expand Up @@ -59,6 +61,10 @@ public TaskExecutor(IDataFactory dataFactory, ITileMerger tileMerger, ITimeUtils

this._batchMaxSize = DEFAULT_BATCH_SIZE;
}

int numOfThreads = configurationManager.GetConfiguration<int>("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)
Expand Down Expand Up @@ -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);

Expand All @@ -134,77 +138,74 @@ public void ExecuteTask(MergeTask task, ITaskUtils taskUtils, string? managerCal
continue;
}

List<Tile> tiles = new List<Tile>((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
using (this._activitySource.StartActivity($"[{methodName}] merging tiles"))
{
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<Coord>((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<CorrespondingTileBuilder> correspondingTileBuilders = new List<CorrespondingTileBuilder>();
// 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<Tile>();

// 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<CorrespondingTileBuilder>
{
() => 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();
this._metricsProvider.MergeTimePerTileHistogram(tileMergeStopwatch.Elapsed.TotalSeconds, metadata.TargetFormat);

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();
Expand Down
3 changes: 3 additions & 0 deletions MergerService/appsettings.json
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,9 @@
"batchSize": {
"limitBatchSize": true,
"batchMaxSize": 1000
},
"parallel": {
"numOfThreads": 0
}
},
"TASK": {
Expand Down
17 changes: 8 additions & 9 deletions MergerServiceUnitTests/Runners/TaskExecutorTest.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -112,19 +112,16 @@ public void WhenGivenSourcesWithOneTile_ShouldWriteAllTilesToTarget(int numberOf

public static IEnumerable<object[]> 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
};
}

Expand All @@ -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<IData> sourceDataMock = this._mockRepository.Create<IData>();

for (var testSourceCoordIdx = 0; testSourceCoordIdx < testSourceCoords.Length; testSourceCoordIdx++)
Expand Down
Loading