authorgravatar for kbutcher6200@gmail.comkprotty <kbutcher6200@gmail.com> 2021-06-26 09:03:53-05:00
committergravatar for kbutcher6200@gmail.comkprotty <kbutcher6200@gmail.com> 2021-06-30 21:49:00-05:00
logfd4a607bb2f1a1cbf8b8c1fd5d35f5f775e79114
tree65a2b16c47c2784562d3a93eeefdcc04e883b88c
parent18bcb2e990853f3e32cb0d72beb962390c9e8714

std.Thread: fix futex test + thread errors

2 files changed, 90 insertions(+), 98 deletions(-)

lib/std/Thread.zig+2
......@@ -385,6 +385,7 @@ const PosixThreadImpl = struct {
385385 };
386386
387387 const args_ptr = try allocator.create(Args);
388 args_ptr.* = args;
388389 errdefer allocator.destroy(args_ptr);
389390
390391 var attr: c.pthread_attr_t = undefined;
......@@ -523,6 +524,7 @@ const LinuxThreadImpl = struct {
523524 error.PermissionDenied => unreachable,
524525 else => |e| return e,
525526 };
527 assert(mapped.len >= map_bytes);
526528 errdefer os.munmap(mapped);
527529
528530 // map everything but the guard page as read/write
lib/std/Thread/Futex.zig+88-98
......@@ -391,70 +391,74 @@ test "Futex - wait/wake" {
391391}
392392
393393test "Futex - Signal" {
394 if (!single_threaded) {
394 if (single_threaded) {
395395 return;
396396 }
397397
398 try (struct {
398 const Paddle = struct {
399399 value: Atomic(u32) = Atomic(u32).init(0),
400 current: u32 = 0,
400401
401 const Self = @This();
402
403 fn send(self: *Self, value: u32) void {
404 self.value.store(value, .Release);
405 Futex.wake(&self.value, 1);
406 }
407
408 fn recv(self: *Self, expected: u32) void {
409 while (true) {
410 const value = self.value.load(.Acquire);
411 if (value == expected) break;
412 Futex.wait(&self.value, value, null) catch unreachable;
413 }
414 }
402 fn run(self: *@This(), hit_to: *@This()) !void {
403 var iterations: usize = 4;
404 while (iterations > 0) : (iterations -= 1) {
415405
416 const start_value = 1;
406 var value: u32 = undefined;
407 while (true) {
408 value = self.value.load(.Acquire);
409 if (value != self.current) break;
410 Futex.wait(&self.value, self.current, null) catch unreachable;
411 }
417412
418 fn runThread(rx: *Self, tx: *Self) void {
419 var iterations: u32 = start_value;
420 while (iterations < 10) : (iterations += 1) {
421 rx.recv(iterations);
422 tx.send(iterations);
413 try testing.expectEqual(value, self.current + 1);
414 self.current = value;
415
416 _ = hit_to.value.fetchAdd(1, .Release);
417 Futex.wake(&hit_to.value, 1);
423418 }
424419 }
420 };
425421
426 fn run() !void {
427 var ping = Self{};
428 var pong = Self{};
422 var ping = Paddle{};
423 var pong = Paddle{};
429424
430 const t1 = try std.Thread.spawn(.{}, runThread, .{ &ping, &pong });
431 defer t1.join();
425 const t1 = try std.Thread.spawn(.{}, Paddle.run, .{&ping, &pong});
426 defer t1.join();
432427
433 const t2 = try std.Thread.spawn(.{}, runThread, .{ &pong, &ping });
434 defer t2.join();
428 const t2 = try std.Thread.spawn(.{}, Paddle.run, .{&pong, &ping});
429 defer t2.join();
435430
436 ping.send(start_value);
437 }
438 }).run();
431 _ = ping.value.fetchAdd(1, .Release);
432 Futex.wake(&ping.value, 1);
439433}
440434
441435test "Futex - Broadcast" {
442 if (!single_threaded) {
436 if (single_threaded) {
443437 return;
444438 }
445439
446 try (struct {
447 threads: [10]std.Thread = undefined,
440 const Context = struct {
441 threads: [4]std.Thread = undefined,
448442 broadcast: Atomic(u32) = Atomic(u32).init(0),
449443 notified: Atomic(usize) = Atomic(usize).init(0),
450444
451 const Self = @This();
452
453445 const BROADCAST_EMPTY = 0;
454446 const BROADCAST_SENT = 1;
455447 const BROADCAST_RECEIVED = 2;
456448
457 fn runReceiver(self: *Self) void {
449 fn runSender(self: *@This()) !void {
450 self.broadcast.store(BROADCAST_SENT, .Monotonic);
451 Futex.wake(&self.broadcast, @intCast(u32, self.threads.len));
452
453 while (true) {
454 const broadcast = self.broadcast.load(.Acquire);
455 if (broadcast == BROADCAST_RECEIVED) break;
456 try testing.expectEqual(broadcast, BROADCAST_SENT);
457 Futex.wait(&self.broadcast, broadcast, null) catch unreachable;
458 }
459 }
460
461 fn runReceiver(self: *@This()) void {
458462 while (true) {
459463 const broadcast = self.broadcast.load(.Acquire);
460464 if (broadcast == BROADCAST_SENT) break;
......@@ -468,66 +472,55 @@ test "Futex - Broadcast" {
468472 Futex.wake(&self.broadcast, 1);
469473 }
470474 }
475 };
471476
472 fn run() !void {
473 var self = Self{};
474
475 for (self.threads) |*thread|
476 thread.* = try std.Thread.spawn(runReceiver, &self);
477 defer for (self.threads) |thread|
478 thread.join();
477 var ctx = Context{};
478 for (ctx.threads) |*thread|
479 thread.* = try std.Thread.spawn(.{}, Context.runReceiver, .{&ctx});
480 defer for (ctx.threads) |thread|
481 thread.join();
479482
480 std.time.sleep(16 * std.time.ns_per_ms);
481 self.broadcast.store(BROADCAST_SENT, .Monotonic);
482 Futex.wake(&self.broadcast, @intCast(u32, self.threads.len));
483 // Try to wait for the threads to start before running runSender().
484 // NOTE: not actually needed for correctness.
485 std.time.sleep(16 * std.time.ns_per_ms);
486 try ctx.runSender();
483487
484 while (true) {
485 const broadcast = self.broadcast.load(.Acquire);
486 if (broadcast == BROADCAST_RECEIVED) break;
487 try testing.expectEqual(broadcast, BROADCAST_SENT);
488 Futex.wait(&self.broadcast, broadcast, null) catch unreachable;
489 }
490
491 const notified = self.notified.load(.Monotonic);
492 try testing.expectEqual(notified, self.threads.len);
493 }
494 }).run();
488 const notified = ctx.notified.load(.Monotonic);
489 try testing.expectEqual(notified, ctx.threads.len);
495490}
496491
497492test "Futex - Chain" {
498 if (!single_threaded) {
493 if (single_threaded) {
499494 return;
500495 }
501496
502 try (struct {
503 completed: Signal = .{},
504 threads: [10]struct {
505 thread: std.Thread,
506 signal: Signal,
507 } = undefined,
508
509 const Signal = struct {
510 state: Atomic(u32) = Atomic(u32).init(0),
497 const Signal = struct {
498 value: Atomic(u32) = Atomic(u32).init(0),
511499
512 fn wait(self: *Signal) void {
513 while (true) {
514 const value = self.value.load(.Acquire);
515 if (value == 1) break;
516 assert(value == 0);
517 Futex.wait(&self.value, 0, null) catch unreachable;
518 }
500 fn wait(self: *@This()) void {
501 while (true) {
502 const value = self.value.load(.Acquire);
503 if (value == 1) break;
504 assert(value == 0);
505 Futex.wait(&self.value, 0, null) catch unreachable;
519506 }
507 }
520508
521 fn notify(self: *Signal) void {
522 assert(self.value.load(.Unordered) == 0);
523 self.value.store(1, .Release);
524 Futex.wake(&self.value, 1);
525 }
526 };
509 fn notify(self: *@This()) void {
510 assert(self.value.load(.Unordered) == 0);
511 self.value.store(1, .Release);
512 Futex.wake(&self.value, 1);
513 }
514 };
527515
528 const Self = @This();
516 const Context = struct {
517 completed: Signal = .{},
518 threads: [4]struct {
519 thread: std.Thread,
520 signal: Signal,
521 } = undefined,
529522
530 fn runThread(self: *Self, index: usize) void {
523 fn run(self: *@This(), index: usize) void {
531524 const this_signal = &self.threads[index].signal;
532525
533526 var next_signal = &self.completed;
......@@ -538,21 +531,18 @@ test "Futex - Chain" {
538531 this_signal.wait();
539532 next_signal.notify();
540533 }
534 };
541535
542 fn run() !void {
543 var self = Self{};
544
545 for (self.threads) |*entry, index| {
546 entry.signal = .{};
547 entry.thread = try std.Thread.spawn(.{}, runThread, .{&self, index});
548 }
536 var ctx = Context{};
537 for (ctx.threads) |*entry, index| {
538 entry.signal = .{};
539 entry.thread = try std.Thread.spawn(.{}, Context.run, .{&ctx, index});
540 }
549541
550 self.threads[0].signal.notify();
551 self.completed.wait();
542 ctx.threads[0].signal.notify();
543 ctx.completed.wait();
552544
553 for (self.threads) |entry| {
554 entry.thread.join();
555 }
556 }
557 }).run();
545 for (ctx.threads) |entry| {
546 entry.thread.join();
547 }
558548}