| ... | @@ -27,10 +27,19 @@ const Count = @Int(.unsigned, @divFloor(@bitSizeOf(usize) - 1, 2)); | ... | @@ -27,10 +27,19 @@ const Count = @Int(.unsigned, @divFloor(@bitSizeOf(usize) - 1, 2)); |
| 27 | | 27 | |
| 28 | pub fn tryLock(rl: *RwLock, io: Io) bool { | 28 | pub fn tryLock(rl: *RwLock, io: Io) bool { |
| 29 | if (rl.mutex.tryLock()) { | 29 | if (rl.mutex.tryLock()) { |
| | 30 | // Unlike `lock`, this never registers in `writer_mask`, so holding the mutex does |
| | 31 | // not stop a reader from taking the fast path; the CAS catches one that raced in |
| | 32 | // after `state` was loaded. |
| 30 | const state = @atomicLoad(usize, &rl.state, .seq_cst); | 33 | const state = @atomicLoad(usize, &rl.state, .seq_cst); |
| 31 | if (state & reader_mask == 0) { | 34 | if (state & reader_mask == 0) { |
| 32 | _ = @atomicRmw(usize, &rl.state, .Or, is_writing, .seq_cst); | 35 | _ = @cmpxchgStrong( |
| 33 | return true; | 36 | usize, |
| | 37 | &rl.state, |
| | 38 | state, |
| | 39 | state | is_writing, |
| | 40 | .seq_cst, |
| | 41 | .seq_cst, |
| | 42 | ) orelse return true; |
| 34 | } | 43 | } |
| 35 | | 44 | |
| 36 | rl.mutex.unlock(io); | 45 | rl.mutex.unlock(io); |
| ... | @@ -61,7 +70,12 @@ pub fn lock(rl: *RwLock, io: Io) Io.Cancelable!void { | ... | @@ -61,7 +70,12 @@ pub fn lock(rl: *RwLock, io: Io) Io.Cancelable!void { |
| 61 | if (state & reader_mask != 0) | 70 | if (state & reader_mask != 0) |
| 62 | rl.semaphore.wait(io) catch |err| switch (err) { | 71 | rl.semaphore.wait(io) catch |err| switch (err) { |
| 63 | error.Canceled => { | 72 | error.Canceled => { |
| 64 | rl.unlock(io); | 73 | // Clearing `is_writing` while still holding the mutex means the last reader |
| | 74 | // either saw it set, and posts a permit only we can consume, or did not and |
| | 75 | // never posts. A stale permit would let the next writer in past the readers. |
| | 76 | const prev_state = @atomicRmw(usize, &rl.state, .And, ~is_writing, .seq_cst); |
| | 77 | if (prev_state & reader_mask == 0) rl.semaphore.waitUncancelable(io); |
| | 78 | rl.mutex.unlock(io); |
| 65 | return error.Canceled; | 79 | return error.Canceled; |
| 66 | }, | 80 | }, |
| 67 | }; | 81 | }; |
| ... | @@ -305,6 +319,78 @@ test "lock canceling" { | ... | @@ -305,6 +319,78 @@ test "lock canceling" { |
| 305 | try testing.expectEqual(rl, Io.RwLock.init); | 319 | try testing.expectEqual(rl, Io.RwLock.init); |
| 306 | } | 320 | } |
| 307 | | 321 | |
| | 322 | test "tryLock does not race with readers" { |
| | 323 | if (builtin.single_threaded) return error.SkipZigTest; |
| | 324 | |
| | 325 | const Context = struct { |
| | 326 | rl: Io.RwLock, |
| | 327 | |
| | 328 | fn reader(ctx: *@This(), io: Io) !void { |
| | 329 | while (true) { |
| | 330 | if (ctx.rl.tryLockShared(io)) ctx.rl.unlockShared(io); |
| | 331 | try io.checkCancel(); |
| | 332 | } |
| | 333 | } |
| | 334 | }; |
| | 335 | |
| | 336 | var ctx: Context = .{ .rl = .init }; |
| | 337 | const io = testing.io; |
| | 338 | |
| | 339 | var future = io.concurrent(Context.reader, .{ &ctx, io }) catch |err| switch (err) { |
| | 340 | error.ConcurrencyUnavailable => return error.SkipZigTest, |
| | 341 | }; |
| | 342 | defer future.cancel(io) catch {}; |
| | 343 | |
| | 344 | for (0..1000) |_| { |
| | 345 | if (!ctx.rl.tryLock(io)) continue; |
| | 346 | defer ctx.rl.unlock(io); |
| | 347 | const state = @atomicLoad(usize, &ctx.rl.state, .seq_cst); |
| | 348 | try testing.expectEqual(0, state & reader_mask); |
| | 349 | } |
| | 350 | |
| | 351 | try testing.expectEqual(0, ctx.rl.semaphore.permits); |
| | 352 | } |
| | 353 | |
| | 354 | test "canceled writer does not leak a semaphore permit" { |
| | 355 | if (builtin.single_threaded) return error.SkipZigTest; |
| | 356 | |
| | 357 | const Writer = struct { |
| | 358 | fn lockUnlock(rl: *Io.RwLock, io: Io) Io.Cancelable!void { |
| | 359 | try rl.lock(io); |
| | 360 | rl.unlock(io); |
| | 361 | } |
| | 362 | }; |
| | 363 | |
| | 364 | const io = testing.io; |
| | 365 | |
| | 366 | var rl: Io.RwLock = .init; |
| | 367 | |
| | 368 | for (0..1000) |_| { |
| | 369 | rl.lockSharedUncancelable(io); |
| | 370 | |
| | 371 | var wfuture = io.concurrent(Writer.lockUnlock, .{ &rl, io }) catch |err| switch (err) { |
| | 372 | error.ConcurrencyUnavailable => { |
| | 373 | rl.unlockShared(io); |
| | 374 | return error.SkipZigTest; |
| | 375 | }, |
| | 376 | }; |
| | 377 | var rfuture = io.concurrent(Io.RwLock.unlockShared, .{ &rl, io }) catch |err| switch (err) { |
| | 378 | error.ConcurrencyUnavailable => { |
| | 379 | rl.unlockShared(io); |
| | 380 | wfuture.await(io) catch {}; |
| | 381 | return error.SkipZigTest; |
| | 382 | }, |
| | 383 | }; |
| | 384 | |
| | 385 | // Races the last reader's `post` against the writer's cancelation. |
| | 386 | wfuture.cancel(io) catch {}; |
| | 387 | rfuture.await(io); |
| | 388 | |
| | 389 | try testing.expectEqual(0, rl.state); |
| | 390 | try testing.expectEqual(0, rl.semaphore.permits); |
| | 391 | } |
| | 392 | } |
| | 393 | |
| 308 | fn semaphoreLockCancel(rl: *Io.RwLock, io: Io) !void { | 394 | fn semaphoreLockCancel(rl: *Io.RwLock, io: Io) !void { |
| 309 | try rl.lock(io); //tests semaphore cancelling | 395 | try rl.lock(io); //tests semaphore cancelling |
| 310 | } | 396 | } |