authorgravatar for 49791153+jumpnbrownweasel@users.noreply.github.comjumpnbrownweasel <49791153+jumpnbrownweasel@users.noreply.github.com> 2022-10-17 16:15:15-07:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2022-10-17 18:15:15-05:00
log71f876295981ecf5dfb8daadc05e7e9c1c7e1cbe
tree8ebedc0c5c72fca68c82297bc70dc9247360baa1
parentce3ffa5e1b30b3ad5fb38639c7e3036d328a598e
signaturebadge-question-mark Signed by PGP key 4AEE18F83AFDEB23

Fix for #13163: DefaultRwLock accumulates write-waiters, eventually fails to write lock (#13180)

* Fix for: DefaultRwLock accumulates write-waiters, eventually fails to write lock #13163 * Comment out debug.print at the end of the last test. * Code formatting * - use equality test after lock/unlock rather than peeking into internals. however, this is still implementation specific and only done for DefaultRwLock. - add num_reads maximum to ensure that reader threads stop if writer threads are starved - use relaxed orderings for the read atomic counter - don't check at the end for non-zero read ops, since the reader threads may only run once if they are starved * More review changes - Monotonic is sufficient for incrementing the reads counter

1 files changed, 127 insertions(+), 1 deletions(-)

lib/std/Thread/RwLock.zig+127-1
...@@ -9,6 +9,7 @@ const RwLock = @This();...@@ -9,6 +9,7 @@ const RwLock = @This();
9const std = @import("../std.zig");9const std = @import("../std.zig");
10const builtin = @import("builtin");10const builtin = @import("builtin");
11const assert = std.debug.assert;11const assert = std.debug.assert;
12const testing = std.testing;
1213
13pub const Impl = if (builtin.single_threaded)14pub const Impl = if (builtin.single_threaded)
14 SingleThreadedRwLock15 SingleThreadedRwLock
...@@ -190,7 +191,7 @@ pub const DefaultRwLock = struct {...@@ -190,7 +191,7 @@ pub const DefaultRwLock = struct {
190 _ = @atomicRmw(usize, &rwl.state, .Add, WRITER, .SeqCst);191 _ = @atomicRmw(usize, &rwl.state, .Add, WRITER, .SeqCst);
191 rwl.mutex.lock();192 rwl.mutex.lock();
192193
193 const state = @atomicRmw(usize, &rwl.state, .Or, IS_WRITING, .SeqCst);194 const state = @atomicRmw(usize, &rwl.state, .Add, IS_WRITING -% WRITER, .SeqCst);
194 if (state & READER_MASK != 0)195 if (state & READER_MASK != 0)
195 rwl.semaphore.wait();196 rwl.semaphore.wait();
196 }197 }
...@@ -247,3 +248,128 @@ pub const DefaultRwLock = struct {...@@ -247,3 +248,128 @@ pub const DefaultRwLock = struct {
247 rwl.semaphore.post();248 rwl.semaphore.post();
248 }249 }
249};250};
251
252test "DefaultRwLock - internal state" {
253 var rwl = DefaultRwLock{};
254
255 // The following failed prior to the fix for Issue #13163,
256 // where the WRITER flag was subtracted by the lock method.
257
258 rwl.lock();
259 rwl.unlock();
260 try testing.expectEqual(rwl, DefaultRwLock{});
261}
262
263test "RwLock - smoke test" {
264 var rwl = RwLock{};
265
266 rwl.lock();
267 try testing.expect(!rwl.tryLock());
268 try testing.expect(!rwl.tryLockShared());
269 rwl.unlock();
270
271 try testing.expect(rwl.tryLock());
272 try testing.expect(!rwl.tryLock());
273 try testing.expect(!rwl.tryLockShared());
274 rwl.unlock();
275
276 rwl.lockShared();
277 try testing.expect(!rwl.tryLock());
278 try testing.expect(rwl.tryLockShared());
279 rwl.unlockShared();
280 rwl.unlockShared();
281
282 try testing.expect(rwl.tryLockShared());
283 try testing.expect(!rwl.tryLock());
284 try testing.expect(rwl.tryLockShared());
285 rwl.unlockShared();
286 rwl.unlockShared();
287
288 rwl.lock();
289 rwl.unlock();
290}
291
292test "RwLock - concurrent access" {
293 if (builtin.single_threaded)
294 return;
295
296 const num_writers: usize = 2;
297 const num_readers: usize = 4;
298 const num_writes: usize = 10000;
299 const num_reads: usize = num_writes * 2;
300
301 const Runner = struct {
302 const Self = @This();
303
304 rwl: RwLock = .{},
305 writes: usize = 0,
306 reads: std.atomic.Atomic(usize) = std.atomic.Atomic(usize).init(0),
307
308 term1: usize = 0,
309 term2: usize = 0,
310 term_sum: usize = 0,
311
312 fn reader(self: *Self) !void {
313 while (true) {
314 self.rwl.lockShared();
315 defer self.rwl.unlockShared();
316
317 if (self.writes >= num_writes or self.reads.load(.Unordered) >= num_reads)
318 break;
319
320 try self.check();
321
322 _ = self.reads.fetchAdd(1, .Monotonic);
323 }
324 }
325
326 fn writer(self: *Self, thread_idx: usize) !void {
327 var prng = std.rand.DefaultPrng.init(thread_idx);
328 var rnd = prng.random();
329
330 while (true) {
331 self.rwl.lock();
332 defer self.rwl.unlock();
333
334 if (self.writes >= num_writes)
335 break;
336
337 try self.check();
338
339 const term1 = rnd.int(usize);
340 self.term1 = term1;
341 try std.Thread.yield();
342
343 const term2 = rnd.int(usize);
344 self.term2 = term2;
345 try std.Thread.yield();
346
347 self.term_sum = term1 +% term2;
348 self.writes += 1;
349 }
350 }
351
352 fn check(self: *const Self) !void {
353 const term_sum = self.term_sum;
354 try std.Thread.yield();
355
356 const term2 = self.term2;
357 try std.Thread.yield();
358
359 const term1 = self.term1;
360 try testing.expectEqual(term_sum, term1 +% term2);
361 }
362 };
363
364 var runner = Runner{};
365 var threads: [num_writers + num_readers]std.Thread = undefined;
366
367 for (threads[0..num_writers]) |*t, i| t.* = try std.Thread.spawn(.{}, Runner.writer, .{ &runner, i });
368 for (threads[num_writers..]) |*t| t.* = try std.Thread.spawn(.{}, Runner.reader, .{&runner});
369
370 for (threads) |t| t.join();
371
372 try testing.expectEqual(num_writes, runner.writes);
373
374 //std.debug.print("reads={}\n", .{ runner.reads.load(.Unordered)});
375}