authorgravatar for andrew@ziglang.orgAndrew Kelley <andrew@ziglang.org> 2024-06-21 11:03:00-07:00
committergravatar for andrew@ziglang.orgAndrew Kelley <andrew@ziglang.org> 2024-06-30 13:54:02-07:00
logdc207da184087eec7693fbf17c412c4b67c2080c
treee910cc932ccf099b05ab8f0a01c0f37f74e68980
parent6d3a78b8dd5071dee68b6b6c10f79b7b2f34d5e9

std.Thread.Pool: don't hold the job closure while blocking on IPC read

1 files changed, 49 insertions(+), 42 deletions(-)

lib/std/Thread/Pool.zig+49-42
...@@ -310,58 +310,65 @@ pub fn spawn(pool: *Pool, comptime func: anytype, args: anytype) void {...@@ -310,58 +310,65 @@ pub fn spawn(pool: *Pool, comptime func: anytype, args: anytype) void {
310 pool.cond.signal();310 pool.cond.signal();
311}311}
312312
313fn acquireThreadToken(job_server_options: Options.JobServer, fd_ptr: *std.posix.fd_t) void {
314 if (fd_ptr.* >= 0) return;
315
316 switch (job_server_options) {
317 .abstain => {},
318 .connect, .host => |addr| {
319 const sockfd = std.posix.socket(
320 std.posix.AF.UNIX,
321 std.posix.SOCK.STREAM | std.posix.SOCK.CLOEXEC,
322 0,
323 ) catch |err| {
324 std.log.debug("failed to make socket: {s}", .{@errorName(err)});
325 return;
326 };
327 fd_ptr.* = sockfd;
328
329 std.posix.connect(sockfd, &addr.any, addr.getOsSockLen()) catch |err| {
330 std.log.debug("failed to connect: {s}", .{@errorName(err)});
331 return;
332 };
333
334 var trash_buf: [1]u8 = undefined;
335 _ = std.posix.read(sockfd, &trash_buf) catch |err| {
336 std.log.debug("failed to read: {s}", .{@errorName(err)});
337 return;
338 };
339 },
340 }
341}
342
343fn releaseThreadToken(fd_ptr: *std.posix.fd_t) void {
344 const fd = fd_ptr.*;
345 if (fd >= 0) {
346 std.posix.close(fd);
347 fd_ptr.* = -1;
348 }
349}
350
313fn worker(pool: *Pool) void {351fn worker(pool: *Pool) void {
314 var trash_buf: [1]u8 = undefined;352 var connection: std.posix.fd_t = -1;
315 var connection: ?std.posix.fd_t = null;353 defer releaseThreadToken(&connection);
316 defer if (connection) |fd| std.posix.close(fd);
317354
318 pool.mutex.lock();355 pool.mutex.lock();
319 defer pool.mutex.unlock();356 defer pool.mutex.unlock();
320357
321 while (true) {358 while (true) {
322 while (pool.run_queue.popFirst()) |run_node| {359 const work_available = pool.run_queue.first != null;
323 // Temporarily unlock the mutex in order to execute the run_node.360 if (work_available) {
324 pool.mutex.unlock();361 pool.mutex.unlock();
325 defer pool.mutex.lock();362 defer pool.mutex.lock();
326363 acquireThreadToken(pool.job_server_options, &connection);
327 if (connection == null) switch (pool.job_server_options) {
328 .abstain => {},
329 .connect, .host => |addr| lock: {
330 const sockfd = std.posix.socket(
331 std.posix.AF.UNIX,
332 std.posix.SOCK.STREAM | std.posix.SOCK.CLOEXEC,
333 0,
334 ) catch |err| {
335 std.log.debug("failed to make socket: {s}", .{@errorName(err)});
336 break :lock;
337 };
338 connection = sockfd;
339
340 std.posix.connect(sockfd, &addr.any, addr.getOsSockLen()) catch |err| {
341 std.log.debug("failed to connect: {s}", .{@errorName(err)});
342 break :lock;
343 };
344
345 _ = std.posix.read(sockfd, &trash_buf) catch |err| {
346 std.log.debug("failed to read: {s}", .{@errorName(err)});
347 break :lock;
348 };
349 },
350 };
351
352 const runFn = run_node.data.runFn;
353 runFn(&run_node.data);
354 }364 }
355365 while (pool.run_queue.popFirst()) |run_node| {
356 // Stop executing instead of waiting if the thread pool is no longer running.366 pool.mutex.unlock();
357 if (pool.end_flag)367 defer pool.mutex.lock();
358 break;368 run_node.data.runFn(&run_node.data);
359
360 if (connection) |fd| {
361 std.posix.close(fd);
362 connection = null;
363 }369 }
364370 if (pool.end_flag) return;
371 releaseThreadToken(&connection);
365 pool.cond.wait(&pool.mutex);372 pool.cond.wait(&pool.mutex);
366 }373 }
367}374}