Skip to content

Add support tasks which can be used to share work with other threads that are e.g. waiting for a mutex #3159

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

Draft
IntegratedQuantum wants to merge 2 commits into
base: master
Choose a base branch
from
+147 −23
Draft

Add support tasks which can be used to share work with other threads that are e.g. waiting for a mutex #3159

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
6ba1a30
Add support tasks which can be used to share work with other threads …
IntegratedQuantum Jun 3, 2026
928cd9c
More generic interface and parallelize more parts of the NoiseBasedVo…
IntegratedQuantum Jun 4, 2026
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
96 changes: 74 additions & 22 deletions src/server/terrain/climategen/NoiseBasedVoronoi.zig
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,8 +28,11 @@ pub fn init(parameters: ZonElement) void {
pub fn generateMapFragment(map: *ClimateMapFragment, worldSeed: u64) void {
var seed: u64 = worldSeed;

const generator = GenerationStructure.init(main.stackAllocator, map.pos.wx, map.pos.wy, ClimateMapFragment.mapSize, ClimateMapFragment.mapSize, terrain.biomes.byTypeBiomes, seed);
defer generator.deinit(main.stackAllocator);
const arena = main.globalAllocator.createArena();
defer main.globalAllocator.destroyArena(arena);

const generator = GenerationStructure.init(arena, map.pos.wx, map.pos.wy, ClimateMapFragment.mapSize, ClimateMapFragment.mapSize, terrain.biomes.byTypeBiomes, seed);
defer generator.deinit(arena);

generator.toMap(map, ClimateMapFragment.mapSize, ClimateMapFragment.mapSize, worldSeed);

Expand Down Expand Up @@ -156,12 +159,44 @@ const Chunk = struct {
}
};

const ChunkGenerationContext = struct {
chunks: Array2D(*Chunk),
resultAllocator: NeverFailingAllocator,
wx: i32,
wy: i32,
tree: *TreeNode,
worldSeed: u64,
offset: [2]u31 = undefined,
neighborOffsets: []const [2]i32 = undefined,

pub fn run(self: *ChunkGenerationContext, taskIndex: usize) void {
const x = self.offset[0] + @as(u31, @intCast(taskIndex))/(@as(u31, @intCast(self.chunks.width))/2)*2;
const y = self.offset[1] + @as(u31, @intCast(taskIndex))%(@as(u31, @intCast(self.chunks.width))/2)*2;
var neighbors: [8]*const Chunk = undefined;
var j: usize = 0;
for (self.neighborOffsets) |neighborOffset| {
const nx = x + neighborOffset[0];
if (nx < 0 or @as(usize, @intCast(nx)) >= self.chunks.width) continue;
const ny = y + neighborOffset[1];
if (ny < 0 or @as(usize, @intCast(ny)) >= self.chunks.height) continue;
neighbors[j] = self.chunks.get(@intCast(nx), @intCast(ny));
j += 1;
}
self.chunks.ptr(x, y).* = Chunk.init(self.resultAllocator, self.tree, self.worldSeed, self.wx +% x*chunkSize -% 4*chunkSize, self.wy +% y*chunkSize -% 4*chunkSize, neighbors[0..j]);
}
};

const GenerationStructure = struct {
chunks: Array2D(*Chunk) = undefined, // Implemented as slices into the original array!

pub fn init(allocator: NeverFailingAllocator, wx: i32, wy: i32, width: u31, height: u31, tree: *TreeNode, worldSeed: u64) GenerationStructure {
const self: GenerationStructure = .{
var generationContext: ChunkGenerationContext = .{
.chunks = Array2D(*Chunk).init(allocator, 8 + @divExact(width, chunkSize), 8 + @divExact(height, chunkSize)),
.resultAllocator = allocator,
.wx = wx,
.wy = wy,
.tree = tree,
.worldSeed = worldSeed,
};
// Generate chunks in an interleaved pattern:
const offset: [4][2]u31 = .{
Expand All @@ -177,25 +212,16 @@ const GenerationStructure = struct {
&.{.{0, -1}, .{0, 1}, .{-1, 0}, .{1, 0}, .{-1, -1}, .{1, -1}, .{-1, 1}, .{1, 1}},
};
for (0..4) |i| {
var x: u31 = offset[i][0];
while (x < self.chunks.width) : (x += 2) {
var y: u31 = offset[i][1];
while (y < self.chunks.height) : (y += 2) {
var neighbors: [8]*const Chunk = undefined;
var j: usize = 0;
for (neighborOffsets[i]) |neighborOffset| {
const nx = x + neighborOffset[0];
if (nx < 0 or @as(usize, @intCast(nx)) >= self.chunks.width) continue;
const ny = y + neighborOffset[1];
if (ny < 0 or @as(usize, @intCast(ny)) >= self.chunks.height) continue;
neighbors[j] = self.chunks.get(@intCast(nx), @intCast(ny));
j += 1;
}
self.chunks.ptr(x, y).* = Chunk.init(allocator, tree, worldSeed, wx +% x*chunkSize -% 4*chunkSize, wy +% y*chunkSize -% 4*chunkSize, neighbors[0..j]);
}
}
generationContext.offset = offset[i];
generationContext.neighborOffsets = neighborOffsets[i];
const taskCount = ((generationContext.chunks.width - offset[i][0] + 1)/2)*((generationContext.chunks.height - offset[i][1] + 1)/2);
const supportTask = main.utils.ThreadPool.GenericSupportTask(*ChunkGenerationContext).init(&generationContext, taskCount);
supportTask.runAndSchedule();
supportTask.waitForCompletionAndDeinit();
}
return self;
return .{
.chunks = generationContext.chunks,
};
}

pub fn deinit(self: GenerationStructure, allocator: NeverFailingAllocator) void {
Expand Down Expand Up @@ -481,6 +507,23 @@ const GenerationStructure = struct {
}
}

const MapPlacementContext = struct {
preMap: *[preMapSize][preMapSize]BiomeSample,
wx: i32,
wy: i32,
worldSeed: u64,
biomeCandidates: []*BiomePoint,

const taskCountPerAxis = 16;
const taskSize = @divExact(preMapSize, taskCountPerAxis);

pub fn run(self: *const MapPlacementContext, taskIndex: usize) void {
const relX = @as(i32, @intCast(taskIndex%taskCountPerAxis*taskSize)) - margin;
const relY = @as(i32, @intCast(taskIndex/taskCountPerAxis*taskSize)) - margin;
fillRecursively(self.wx, self.wy, self.preMap, self.biomeCandidates, self.worldSeed, relX, relY, taskSize, taskSize);
}
};

pub fn toMap(self: GenerationStructure, map: *ClimateMapFragment, width: u31, height: u31, worldSeed: u64) void {
var preMap: [preMapSize][preMapSize]BiomeSample = undefined;
var allCandidates: main.List(*BiomePoint) = .initCapacity(main.stackAllocator, 1024);
Expand All @@ -490,7 +533,16 @@ const GenerationStructure = struct {
allCandidates.append(main.stackAllocator, candidate);
}
}
fillRecursively(map.pos.wx, map.pos.wy, &preMap, allCandidates.items, worldSeed, -margin, -margin, preMapSize, preMapSize);
var mapPlacementContext: MapPlacementContext = .{
.preMap = &preMap,
.wx = map.pos.wx,
.wy = map.pos.wy,
.worldSeed = worldSeed,
.biomeCandidates = allCandidates.items,
};
const supportTask = main.utils.ThreadPool.GenericSupportTask(*const MapPlacementContext).init(&mapPlacementContext, MapPlacementContext.taskCountPerAxis*MapPlacementContext.taskCountPerAxis);
supportTask.runAndSchedule();
supportTask.waitForCompletionAndDeinit();
addTransitionBiomes(&preMap);
for (0..ClimateMapFragment.mapEntrysSize) |_x| {
@memcpy(&map.map[_x], preMap[_x + margin][margin..][0..ClimateMapFragment.mapEntrysSize]);
Expand Down
74 changes: 73 additions & 1 deletion src/utils.zig
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -781,6 +781,56 @@ pub const ThreadPool = struct { // MARK: ThreadPool
clean: *const fn (*anyopaque) void,
taskType: TaskType = .misc,
};
/// State data must assume that other threads access it after it is finished!
/// State must be freed externally!
pub const SupportTask = struct {
self: *anyopaque,
run: *const fn (*anyopaque) void,
};
pub fn GenericSupportTask(T: type) type {
return struct {
data: T,

remainingTasks: std.atomic.Value(isize),
remainingIncompleteTasks: std.atomic.Value(isize),

pub fn init(data: T, taskCount: isize) *@This() {
const self = main.globalAllocator.create(@This());
self.* = .{
.data = data,
.remainingTasks = .init(taskCount),
.remainingIncompleteTasks = .init(taskCount),
};
return self;
}

fn privateDeinit(self: *@This()) void {
main.globalAllocator.destroy(self);
}

pub fn runAndSchedule(self: *@This()) void {
if (self.remainingTasks.load(.monotonic) > 0) {
main.threadPool.addSupportTask(.{
.run = main.meta.castFunctionSelfToAnyopaque(runAndSchedule),
.self = self,
});
}

while (true) {
const taskIndex = self.remainingTasks.fetchSub(1, .monotonic) - 1;
if (taskIndex < 0) return;
defer _ = self.remainingIncompleteTasks.fetchSub(1, .release);

self.data.run(@intCast(taskIndex));
}
}

pub fn waitForCompletionAndDeinit(self: *@This()) void {
while (self.remainingIncompleteTasks.load(.acquire) != 0) {}
main.heap.GarbageCollection.deferredFree(.{.ptr = self, .freeFunction = main.meta.castFunctionSelfToAnyopaque(privateDeinit)});
}
};
}
pub const Performance = struct {
mutex: main.utils.Mutex = .{},
tasks: [taskTypes]u32 = undefined,
Expand Down Expand Up @@ -818,6 +868,7 @@ pub const ThreadPool = struct { // MARK: ThreadPool
semaphore: main.utils.Semaphore = .{},
allocator: NeverFailingAllocator,
running: Atomic(bool) = .init(true),
supportTasks: ConcurrentQueue(SupportTask),

performance: Performance,

Expand All @@ -830,6 +881,7 @@ pub const ThreadPool = struct { // MARK: ThreadPool
.currentTasks = allocator.alloc(Atomic(?*const VTable), threadCount),
.loadList = .init(allocator),
.playerJobQueue = .init(allocator, 1024),
.supportTasks = .init(allocator, 1024),
.performance = .{},
.allocator = allocator,
};
Expand Down Expand Up @@ -864,6 +916,8 @@ pub const ThreadPool = struct { // MARK: ThreadPool

self.playerJobQueue.deinit();

self.supportTasks.deinit();

self.allocator.free(self.currentTasks);
self.allocator.free(self.threads);
self.allocator.destroy(self);
Expand Down Expand Up @@ -918,12 +972,20 @@ pub const ThreadPool = struct { // MARK: ThreadPool
return null;
}

pub fn getSupportTask(self: *ThreadPool) ?SupportTask {
return self.supportTasks.popFront();
}

fn run(self: *ThreadPool, id: usize) void {
main.initThreadLocals();
defer main.deinitThreadLocals();

var lastUpdate = main.timestamp();
outer: while (self.running.load(.monotonic)) {
while (self.supportTasks.popFront()) |task| {
task.run(task.self);
}

main.heap.GarbageCollection.syncPoint();

self.semaphore.timedWait(.fromMilliseconds(10)) catch continue :outer;
Expand Down Expand Up @@ -975,6 +1037,10 @@ pub const ThreadPool = struct { // MARK: ThreadPool
_ = self.trueQueueSize.fetchAdd(1, .monotonic);
}

pub fn addSupportTask(self: *ThreadPool, task: SupportTask) void {
self.supportTasks.pushBack(task);
}

pub fn addPlayer(self: *ThreadPool, player: *main.server.User) void {
player.increaseRefCount();
self.playerJobQueue.pushBack(player);
Expand Down Expand Up @@ -1450,7 +1516,13 @@ pub fn Cache(comptime T: type, comptime numberOfBuckets: u32, comptime bucketSiz

pub fn findOrCreate(self: *@This(), compareAndHash: anytype, comptime initFunction: fn (@TypeOf(compareAndHash)) *T, comptime postGetFunction: ?fn (*T) void) *T {
const index: u32 = compareAndHash.hashCode() & hashMask;
self.buckets[index].mutex.lock();
while (!self.buckets[index].mutex.tryLock()) {
const task = main.threadPool.getSupportTask() orelse {
self.buckets[index].mutex.lock();
break;
};
task.run(task.self);
}
defer self.buckets[index].mutex.unlock();
const result = self.buckets[index].findOrCreate(compareAndHash, initFunction);
if (postGetFunction) |fun| fun(result);
Expand Down
Loading