Class: Synctest::Bubble

Inherits:
Object
  • Object
show all
Defined in:
lib/synctest/bubble.rb

Overview

Coordinates the threads, clock, and synchronization objects in one run.

Defined Under Namespace

Classes: Blocker, ThreadState

Constant Summary collapse

NANOSECONDS = 
1_000_000_000
TIMEOUT = 
Object.new.freeze
CLOSED = 
Object.new.freeze

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(start_at:, timeout:) ⇒ Bubble

Returns a new instance of Bubble.



26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
 
# File 'lib/synctest/bubble.rb', line 26

def initialize(start_at:, timeout:)
  unless start_at.is_a?(Time)
    raise ArgumentError, "start_at must be a Time"
  end
  if timeout && (!timeout.is_a?(Numeric) || timeout <= 0)
    raise ArgumentError, "timeout must be a positive number or nil"
  end

  @start_at = start_at.dup.freeze
  @timeout = timeout&.to_f
  @now_ns = 0

  @lock = Mutex.new
  @condition = ConditionVariable.new
  @states = {}.compare_by_identity
  @children = {}.compare_by_identity
  @resource_waiters = {}.compare_by_identity
  @mutex_owners = {}.compare_by_identity
  @monitor_owners = {}.compare_by_identity
  @grants = {}.compare_by_identity
  @queue_grants = {}.compare_by_identity

  @pending_children = 0
  @root = nil
  @root_active = false
  @active_waiter = nil
  @background_errors = {}.compare_by_identity
  @observed_error_threads = {}.compare_by_identity
  @fatal = nil
  @aborting = false
  @active = false
  @progress_sequence = 0
end

Instance Attribute Details

#start_at ⇒ Object (readonly)

Returns the value of attribute start_at.



24
25
26
 
# File 'lib/synctest/bubble.rb', line 24

def start_at
  @start_at
end

Instance Method Details

#active? ⇒ Boolean

Returns:

  • (Boolean) 


60
61
62
 
# File 'lib/synctest/bubble.rb', line 60

def active?
  @active
end

#attach_child(thread) ⇒ Object 



200
201
202
203
204
205
206
207
208
209
 
# File 'lib/synctest/bubble.rb', line 200

def attach_child(thread)
  thread.instance_variable_set(Synctest::OBJECT_BUBBLE_IVAR, self)
  with_lock do
    @pending_children -= 1
    @children[thread] = true
    @states[thread] = state(:running)
    signal_locked
    reconcile_locked
  end
end

#await_child_attachment(thread) ⇒ Object

Thread.new returns before the new thread is guaranteed to have executed any Ruby code. Do not hand the Thread object to user code until its reservation has become a real member, or an immediate Thread#kill could strand a pending reservation forever.



215
216
217
218
219
220
221
222
223
224
225
226
 
# File 'lib/synctest/bubble.rb', line 215

def await_child_attachment(thread)
  with_lock do
    until @states.key?(thread)
      unless thread.alive?
        @pending_children -= 1
        raise Error, "child thread exited before joining its Synctest bubble"
      end

      Raw::CONDITION_WAIT.bind_call(@condition, @lock, 0.01)
    end
  end
end

#cancel_child_reservation ⇒ Object 



192
193
194
195
196
197
198
 
# File 'lib/synctest/bubble.rb', line 192

def cancel_child_reservation
  with_lock do
    @pending_children -= 1
    signal_locked
    reconcile_locked
  end
end

#child_finished(thread, error) ⇒ Object 



228
229
230
231
232
233
234
235
236
237
238
239
 
# File 'lib/synctest/bubble.rb', line 228

def child_finished(thread, error)
  with_lock do
    @background_errors[thread] = error if error && !@aborting
    current = @states[thread]
    @states[thread] = state(:exited)
    remove_blocker_locked(current.blocker) if current&.blocker
    release_owned_resources_locked(thread)
    wake_resource_locked(thread, kinds: [:join], result: thread, all: true)
    signal_locked
    reconcile_locked unless @fatal
  end
end

#condition_signal(condition, broadcast:) ⇒ Object 



364
365
366
367
368
369
370
371
372
373
374
375
 
# File 'lib/synctest/bubble.rb', line 364

def condition_signal(condition, broadcast:)
  with_lock do
    wake_resource_locked(
      condition,
      kinds: [:condition],
      result: true,
      all: broadcast
    )
    reconcile_locked
  end
  condition
end

#condition_wait(condition, mutex, timeout) ⇒ Object

ConditionVariable ---------------------------------------------------



341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
 
# File 'lib/synctest/bubble.rb', line 341

def condition_wait(condition, mutex, timeout)
  blocker = nil
  released = false

  with_lock do
    release_mutex_locked(mutex, Thread.current)
    released = true
    deadline = timeout.nil? ? nil : @now_ns + duration_to_nanoseconds(timeout)
    blocker = register_blocker_locked(
      thread: Thread.current,
      kind: :condition,
      resource: condition,
      deadline_ns: deadline,
      description: timeout.nil? ? "ConditionVariable#wait" : "ConditionVariable#wait until #{format_deadline(deadline)}"
    )
  end

  result = await(blocker)
  result.equal?(TIMEOUT) ? nil : 0
ensure
  mutex_lock(mutex) if released && !aborting?
end

#diagnostics ⇒ Object 



604
605
606
 
# File 'lib/synctest/bubble.rb', line 604

def diagnostics
  with_lock { diagnostics_locked }
end

#duration_to_nanoseconds(duration) ⇒ Object

Diagnostics and utilities ------------------------------------------



596
597
598
 
# File 'lib/synctest/bubble.rb', line 596

def duration_to_nanoseconds(duration)
  interval_to_nanoseconds(duration, negative_as_zero: false)
end

#interrupt_thread(thread) ⇒ Object 



148
149
150
151
152
153
154
155
156
 
# File 'lib/synctest/bubble.rb', line 148

def interrupt_thread(thread)
  with_lock do
    current = @states[thread]
    if current && %i[blocked waiting].include?(current.status)
      @states[thread] = state(:runnable, current.blocker)
      signal_locked
    end
  end
end

#join(thread, timeout) ⇒ Object

Raises:

  • (ThreadError) 


241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
 
# File 'lib/synctest/bubble.rb', line 241

def join(thread, timeout)
  raise ThreadError, "Target thread must not be current thread" if thread.equal?(Thread.current)

  deadline = timeout.nil? ? nil : with_lock { @now_ns } + timeout_to_nanoseconds(timeout)
  blocker = with_lock do
    unless thread_finished_locked?(thread)
      register_blocker_locked(
        thread: Thread.current,
        kind: :join,
        resource: thread,
        deadline_ns: deadline,
        description: "join #{thread_label(thread)}"
      )
    end
  end

  result = blocker && await(blocker)
  return nil if result.equal?(TIMEOUT)

  raise_join_error(thread)
  thread
end

#monitor_enter(monitor) ⇒ Object

Monitor -------------------------------------------------------------



379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
 
# File 'lib/synctest/bubble.rb', line 379

def monitor_enter(monitor)
  thread = Thread.current
  with_lock do
    loop do
      owner, depth = @monitor_owners[monitor]
      if owner.equal?(thread)
        Raw::MONITOR_ENTER.bind_call(monitor)
        @monitor_owners[monitor] = [thread, depth + 1]
        return monitor
      end

      granted = @grants[monitor]
      if (!granted || granted.equal?(thread)) && Raw::MONITOR_TRY_ENTER.bind_call(monitor)
        @grants.delete(monitor)
        @monitor_owners[monitor] = [thread, 1]
        return monitor
      end

      blocker = register_blocker_locked(
        thread: thread,
        kind: :monitor_lock,
        resource: monitor,
        description: "Monitor#enter"
      )
      await_locked(blocker)
    end
  end
end

#monitor_exit(monitor) ⇒ Object 



428
429
430
431
432
433
434
 
# File 'lib/synctest/bubble.rb', line 428

def monitor_exit(monitor)
  with_lock do
    release_monitor_locked(monitor, Thread.current)
    reconcile_locked
  end
  monitor
end

#monitor_owned?(monitor) ⇒ Boolean

Returns:

  • (Boolean) 


436
437
438
 
# File 'lib/synctest/bubble.rb', line 436

def monitor_owned?(monitor)
  with_lock { @monitor_owners[monitor]&.first.equal?(Thread.current) }
end

#monitor_try_enter(monitor) ⇒ Object 



408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
 
# File 'lib/synctest/bubble.rb', line 408

def monitor_try_enter(monitor)
  thread = Thread.current
  with_lock do
    owner, depth = @monitor_owners[monitor]
    if owner.equal?(thread)
      Raw::MONITOR_ENTER.bind_call(monitor)
      @monitor_owners[monitor] = [thread, depth + 1]
      return true
    end

    granted = @grants[monitor]
    return false if granted && !granted.equal?(thread)
    return false unless Raw::MONITOR_TRY_ENTER.bind_call(monitor)

    @grants.delete(monitor)
    @monitor_owners[monitor] = [thread, 1]
    true
  end
end

#monitor_wait(monitor, condition, timeout) ⇒ Object 



440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
 
# File 'lib/synctest/bubble.rb', line 440

def monitor_wait(monitor, condition, timeout)
  depth = nil
  blocker = nil

  with_lock do
    owner, depth = @monitor_owners[monitor]
    raise ThreadError, "current thread not owner" unless owner.equal?(Thread.current)

    depth.times { Raw::MONITOR_EXIT.bind_call(monitor) }
    @monitor_owners.delete(monitor)
    grant_next_locked(monitor, :monitor_lock)

    deadline = timeout.nil? ? nil : @now_ns + duration_to_nanoseconds(timeout)
    blocker = register_blocker_locked(
      thread: Thread.current,
      kind: :condition,
      resource: condition,
      deadline_ns: deadline,
      description: timeout.nil? ? "Monitor condition wait" : "Monitor condition wait until #{format_deadline(deadline)}"
    )
  end

  result = await(blocker)
  !result.equal?(TIMEOUT)
ensure
  depth&.times { monitor_enter(monitor) } unless aborting?
end

#monotonic_time(unit = :float_second) ⇒ Object 



104
105
106
107
 
# File 'lib/synctest/bubble.rb', line 104

def monotonic_time(unit = :float_second)
  nanoseconds = with_lock { @now_ns }
  convert_nanoseconds(nanoseconds, unit)
end

#mutex_lock(mutex) ⇒ Object

Mutex ---------------------------------------------------------------



266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
 
# File 'lib/synctest/bubble.rb', line 266

def mutex_lock(mutex)
  thread = Thread.current
  with_lock do
    loop do
      owner = @mutex_owners[mutex]
      raise ThreadError, "deadlock; recursive locking" if owner.equal?(thread)

      granted = @grants[mutex]
      if (!granted || granted.equal?(thread)) && Raw::MUTEX_TRY_LOCK.bind_call(mutex)
        @grants.delete(mutex)
        @mutex_owners[mutex] = thread
        return mutex
      end

      blocker = register_blocker_locked(
        thread: thread,
        kind: :mutex_lock,
        resource: mutex,
        description: "Mutex#lock"
      )
      await_locked(blocker)
    end
  end
end

#mutex_owned?(mutex) ⇒ Boolean

Returns:

  • (Boolean) 


314
315
316
 
# File 'lib/synctest/bubble.rb', line 314

def mutex_owned?(mutex)
  with_lock { @mutex_owners[mutex].equal?(Thread.current) }
end

#mutex_sleep(mutex, timeout) ⇒ Object 



318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
 
# File 'lib/synctest/bubble.rb', line 318

def mutex_sleep(mutex, timeout)
  released = false
  start_ns = with_lock do
    release_mutex_locked(mutex, Thread.current)
    released = true
    @now_ns
  end

  deadline = timeout.nil? ? nil : start_ns + duration_to_nanoseconds(timeout)
  blocker = register_blocker(
    kind: :mutex_sleep,
    resource: Thread.current,
    deadline_ns: deadline,
    description: timeout.nil? ? "Mutex#sleep" : "Mutex#sleep until #{format_deadline(deadline)}"
  )
  await(blocker)
  nil
ensure
  mutex_lock(mutex) if released && !aborting?
end

#mutex_try_lock(mutex) ⇒ Object 



291
292
293
294
295
296
297
298
299
300
301
302
303
304
 
# File 'lib/synctest/bubble.rb', line 291

def mutex_try_lock(mutex)
  thread = Thread.current
  with_lock do
    return false if @mutex_owners[mutex].equal?(thread)

    granted = @grants[mutex]
    return false if granted && !granted.equal?(thread)
    return false unless Raw::MUTEX_TRY_LOCK.bind_call(mutex)

    @grants.delete(mutex)
    @mutex_owners[mutex] = thread
    true
  end
end

#mutex_unlock(mutex) ⇒ Object 



306
307
308
309
310
311
312
 
# File 'lib/synctest/bubble.rb', line 306

def mutex_unlock(mutex)
  with_lock do
    release_mutex_locked(mutex, Thread.current)
    reconcile_locked
  end
  mutex
end

#queue_clear(queue) ⇒ Object 



568
569
570
571
572
573
574
575
576
577
 
# File 'lib/synctest/bubble.rb', line 568

def queue_clear(queue)
  with_lock do
    raw_queue_clear(queue)
    @queue_grants.delete(queue)
    fill_sized_queue_locked(queue) if queue.is_a?(SizedQueue)
    settle_queue_waiters_locked(queue)
    reconcile_locked
  end
  queue
end

#queue_close(queue) ⇒ Object 



552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
 
# File 'lib/synctest/bubble.rb', line 552

def queue_close(queue)
  with_lock do
    raw_queue_close(queue)
    waiters_for_locked(queue).dup.each do |waiter|
      case waiter.kind
      when :queue_pop
        settle_queue_waiters_locked(queue)
      when :queue_push
        wake_blocker_locked(waiter, error: ClosedQueueError.new("queue closed"))
      end
    end
    reconcile_locked
  end
  queue
end

#queue_num_waiting(queue, native_count) ⇒ Object 



588
589
590
591
592
 
# File 'lib/synctest/bubble.rb', line 588

def queue_num_waiting(queue, native_count)
  with_lock do
    native_count + waiters_for_locked(queue).count { |waiter| !waiter.ready }
  end
end

#queue_pop(queue, non_block, timeout) ⇒ Object

Queue ---------------------------------------------------------------



470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
 
# File 'lib/synctest/bubble.rb', line 470

def queue_pop(queue, non_block, timeout)
  if non_block && !timeout.nil?
    raise ArgumentError, "can't set a timeout if non_block is enabled"
  end

  with_lock do
    loop do
      grant = queue_grants_for_locked(queue)
      begin
        item = raw_queue_pop(queue, true)
        if grant&.thread&.equal?(Thread.current)
          consume_queue_grant_locked(queue, grant) if grant
        elsif grant && raw_queue_length(queue).zero?
          @queue_grants.delete(queue)
          settle_queue_waiters_locked(queue)
        end
        fill_sized_queue_locked(queue) if queue.is_a?(SizedQueue)
        reconcile_locked
        return item
      rescue ThreadError
        consume_queue_grant_locked(queue, grant) if grant&.thread&.equal?(Thread.current)
        return nil if Raw::QUEUE_CLOSED.bind_call(queue)
        raise if non_block
      end

      deadline = timeout.nil? ? nil : @now_ns + timeout_to_nanoseconds(timeout)
      blocker = register_blocker_locked(
        thread: Thread.current,
        kind: :queue_pop,
        resource: queue,
        deadline_ns: deadline,
        description: timeout.nil? ? "Queue#pop" : "Queue#pop until #{format_deadline(deadline)}"
      )
      result = await_locked(blocker)
      return nil if result.equal?(TIMEOUT) || result.equal?(CLOSED)
    end
  end
end

#queue_push(queue, item) ⇒ Object 



509
510
511
512
513
514
515
516
517
518
 
# File 'lib/synctest/bubble.rb', line 509

def queue_push(queue, item)
  with_lock do
    raise ClosedQueueError, "queue closed" if Raw::QUEUE_CLOSED.bind_call(queue)

    Raw::QUEUE_PUSH.bind_call(queue, item)
    grant_queued_item_locked(queue)
    reconcile_locked
  end
  queue
end

#realtime(unit = :float_second) ⇒ Object 



109
110
111
112
 
# File 'lib/synctest/bubble.rb', line 109

def realtime(unit = :float_second)
  nanoseconds = (@start_at.to_r * NANOSECONDS).to_i + with_lock { @now_ns }
  convert_nanoseconds(nanoseconds, unit)
end

#reserve_child ⇒ Object

Thread lifecycle -----------------------------------------------------



183
184
185
186
187
188
189
190
 
# File 'lib/synctest/bubble.rb', line 183

def reserve_child
  with_lock do
    raise Error, "cannot create a thread in a finished Synctest bubble" unless @active && !@aborting

    @pending_children += 1
    signal_locked
  end
end

#run ⇒ Object 



64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
 
# File 'lib/synctest/bubble.rb', line 64

def run
  @root = Thread.current
  @active = true
  with_lock do
    @root_active = true
    @states[@root] = state(:running)
  end
  Synctest.set_current_bubble(self)

  result = nil
  error = nil

  begin
    result = yield
  rescue Exception => exception # rubocop:disable Lint/RescueException
    error = exception
  ensure
    root_returned

    begin
      error ? abort_and_join : finish
    rescue Exception => finish_error # rubocop:disable Lint/RescueException
      error = finish_error if error.nil? || finish_error.is_a?(StalledError)
    ensure
      @active = false
      Synctest.set_current_bubble(nil)
      release_references
    end
  end

  raise error if error

  result
end

#sized_queue_push(queue, item, non_block, timeout) ⇒ Object 



520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
 
# File 'lib/synctest/bubble.rb', line 520

def sized_queue_push(queue, item, non_block, timeout)
  if non_block && !timeout.nil?
    raise ArgumentError, "can't set a timeout if non_block is enabled"
  end

  blocker = with_lock do
    raise ClosedQueueError, "queue closed" if Raw::QUEUE_CLOSED.bind_call(queue)

    begin
      Raw::SIZED_QUEUE_PUSH.bind_call(queue, item, true)
      grant_queued_item_locked(queue)
      reconcile_locked
      return queue
    rescue ThreadError
      raise if non_block
    end

    deadline = timeout.nil? ? nil : @now_ns + timeout_to_nanoseconds(timeout)
    register_blocker_locked(
      thread: Thread.current,
      kind: :queue_push,
      resource: queue,
      deadline_ns: deadline,
      description: timeout.nil? ? "SizedQueue#push" : "SizedQueue#push until #{format_deadline(deadline)}",
      result: item
    )
  end

  result = await(blocker)
  result.equal?(TIMEOUT) ? nil : queue
end

#sized_queue_set_max(queue, maximum) ⇒ Object 



579
580
581
582
583
584
585
586
 
# File 'lib/synctest/bubble.rb', line 579

def sized_queue_set_max(queue, maximum)
  with_lock do
    result = Raw::SIZED_QUEUE_SET_MAX.bind_call(queue, maximum)
    fill_sized_queue_locked(queue)
    reconcile_locked
    result
  end
end

#sleep(duration = nil) ⇒ Object 



114
115
116
117
118
119
120
121
122
123
124
125
 
# File 'lib/synctest/bubble.rb', line 114

def sleep(duration = nil)
  start_ns = with_lock { @now_ns }
  deadline = duration.nil? ? nil : start_ns + duration_to_nanoseconds(duration)
  blocker = register_blocker(
    kind: :sleep,
    resource: Thread.current,
    deadline_ns: deadline,
    description: duration.nil? ? "sleep forever" : "sleep until #{format_deadline(deadline)}"
  )
  await(blocker)
  Rational(with_lock { @now_ns } - start_ns, NANOSECONDS).round
end

#stop ⇒ Object 



127
128
129
130
131
132
133
134
135
 
# File 'lib/synctest/bubble.rb', line 127

def stop
  blocker = register_blocker(
    kind: :stop,
    resource: Thread.current,
    description: "Thread.stop"
  )
  await(blocker)
  nil
end

#timeout_to_nanoseconds(timeout) ⇒ Object 



600
601
602
 
# File 'lib/synctest/bubble.rb', line 600

def timeout_to_nanoseconds(timeout)
  interval_to_nanoseconds(timeout, negative_as_zero: true)
end

#wait ⇒ Object 



158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
 
# File 'lib/synctest/bubble.rb', line 158

def wait
  thread = Thread.current
  blocker = with_lock do
    if @active_waiter
      raise ConcurrentWaitError, "only one Synctest.wait may be active in a bubble"
    end

    @active_waiter = thread
    register_blocker_locked(
      thread: thread,
      kind: :wait,
      description: "Synctest.wait",
      status: :waiting
    )
  end

  await(blocker)
  raise_background_error
  nil
ensure
  with_lock { @active_waiter = nil if @active_waiter.equal?(thread) } if thread
end

#wake_thread(thread) ⇒ Object 



137
138
139
140
141
142
143
144
145
146
 
# File 'lib/synctest/bubble.rb', line 137

def wake_thread(thread)
  with_lock do
    blocker = @states[thread]&.blocker
    return false unless blocker && %i[sleep stop mutex_sleep].include?(blocker.kind)

    wake_blocker_locked(blocker, result: true)
    reconcile_locked
    true
  end
end

#wall_time ⇒ Object 



99
100
101
102
 
# File 'lib/synctest/bubble.rb', line 99

def wall_time
  nanoseconds = with_lock { @now_ns }
  @start_at + Rational(nanoseconds, NANOSECONDS)
end