Class: Temporalio::Internal::Worker::WorkflowInstance::Scheduler
- Inherits:
-
Object
- Object
- Temporalio::Internal::Worker::WorkflowInstance::Scheduler
- Defined in:
- lib/temporalio/internal/worker/workflow_instance/scheduler.rb
Overview
Deterministic Fiber::Scheduler implementation.
Instance Method Summary collapse
-
#block(_blocker, timeout = nil) ⇒ Object
Fiber::Scheduler methods.
- #close ⇒ Object
- #context ⇒ Object
- #fiber(&block) ⇒ Object
-
#initialize(instance) ⇒ Scheduler
constructor
A new instance of Scheduler.
- #io_wait(io, events, timeout) ⇒ Object
- #kernel_sleep(duration = nil) ⇒ Object
- #process_wait(pid, flags) ⇒ Object
- #run_until_all_yielded ⇒ Object
- #stack_trace ⇒ Object
- #timeout_after(duration, exception_class, *exception_arguments) ⇒ Object
- #unblock(_blocker, fiber) ⇒ Object
- #wait_condition(cancellation:, &block) ⇒ Object
Constructor Details
#initialize(instance) ⇒ Scheduler
Returns a new instance of Scheduler.
16 17 18 19 20 21 22 |
# File 'lib/temporalio/internal/worker/workflow_instance/scheduler.rb', line 16 def initialize(instance) @instance = instance @fibers = [] @ready = [] @wait_conditions = {} @wait_condition_counter = 0 end |
Instance Method Details
#block(_blocker, timeout = nil) ⇒ Object
Fiber::Scheduler methods
Note, we do not implement many methods here such as io_read and such. While it might seem to make sense to implement them and raise, we actually want to default to the blocking behavior of them not being present. This is so advanced things like logging still work inside of workflows. So we only implement the bare minimum.
108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 |
# File 'lib/temporalio/internal/worker/workflow_instance/scheduler.rb', line 108 def block(_blocker, timeout = nil) # TODO(cretz): Make the blocker visible in the stack trace? # We just yield because unblock will resume this. We will just wrap in timeout if needed. if timeout begin Workflow.timeout(timeout) { Fiber.yield } true rescue Timeout::Error false end else Fiber.yield true end end |
#close ⇒ Object
125 126 127 |
# File 'lib/temporalio/internal/worker/workflow_instance/scheduler.rb', line 125 def close # Nothing to do here, lifetime of scheduler is controlled by the instance end |
#context ⇒ Object
24 25 26 |
# File 'lib/temporalio/internal/worker/workflow_instance/scheduler.rb', line 24 def context @instance.context end |
#fiber(&block) ⇒ Object
129 130 131 132 133 134 135 136 137 138 139 140 141 142 |
# File 'lib/temporalio/internal/worker/workflow_instance/scheduler.rb', line 129 def fiber(&block) if @instance.context_frozen raise Workflow::InvalidWorkflowStateError, 'Cannot schedule fibers in this context' end fiber = Fiber.new do block.call # steep:ignore ensure @fibers.delete(Fiber.current) end @fibers << fiber @ready << fiber fiber end |
#io_wait(io, events, timeout) ⇒ Object
144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 |
# File 'lib/temporalio/internal/worker/workflow_instance/scheduler.rb', line 144 def io_wait(io, events, timeout) # Do not allow if IO disabled unless @instance.io_enabled raise Workflow::NondeterminismError, 'Cannot perform IO from inside a workflow. If this is known to be safe, ' \ 'the code can be run in a Temporalio::Workflow::Unsafe.durable_scheduler_disabled ' \ 'or Temporalio::Workflow::Unsafe.io_enabled block.' end # Use regular Ruby behavior of blocking this thread. There is no Ruby implementation of io_wait we can just # delegate to at this time (or default scheduler or anything like that), so we had to implement this # ourselves. readers = events.nobits?(IO::READABLE) ? nil : [io] writers = events.nobits?(IO::WRITABLE) ? nil : [io] priority = events.nobits?(IO::PRIORITY) ? nil : [io] ready = IO.select(readers, writers, priority, timeout) # steep:ignore result = 0 unless ready.nil? result |= IO::READABLE if ready[0]&.include?(io) result |= IO::WRITABLE if ready[1]&.include?(io) result |= IO::PRIORITY if ready[2]&.include?(io) end result end |
#kernel_sleep(duration = nil) ⇒ Object
170 171 172 |
# File 'lib/temporalio/internal/worker/workflow_instance/scheduler.rb', line 170 def kernel_sleep(duration = nil) Workflow.sleep(duration) end |
#process_wait(pid, flags) ⇒ Object
174 175 176 |
# File 'lib/temporalio/internal/worker/workflow_instance/scheduler.rb', line 174 def process_wait(pid, flags) raise NotImplementedError, 'Cannot wait on other processes in workflows' end |
#run_until_all_yielded ⇒ Object
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/temporalio/internal/worker/workflow_instance/scheduler.rb', line 28 def run_until_all_yielded loop do # Run all fibers until all yielded while (fiber = @ready.shift) fiber.resume end # Find the _first_ resolvable wait condition and if there, resolve it, and loop again, otherwise return. # It is important that we both let fibers get all settled _before_ this and only allow a _single_ wait # condition to be satisfied before looping. This allows wait condition users to trust that the line of # code after the wait condition still has the condition satisfied. # @type var cond_fiber: Fiber? cond_fiber = nil cond_result = nil @wait_conditions.each do |seq, cond| # Evaluate condition or skip if not true next unless (cond_result = cond.first.call) # There have been reports of this fiber being completed already, so we make sure not to process if it # has, but we still delete it deleted_cond = @wait_conditions.delete(seq) next unless deleted_cond&.last&.alive? cond_fiber = deleted_cond.last break end return unless cond_fiber cond_fiber.resume(cond_result) end end |
#stack_trace ⇒ Object
89 90 91 92 93 94 95 96 |
# File 'lib/temporalio/internal/worker/workflow_instance/scheduler.rb', line 89 def stack_trace # Collect backtraces of known fibers, separating with a blank line. We make sure to remove any lines that # reference Temporal paths, and we remove any empty backtraces. dir_path = @instance.illegal_call_tracing_disabled { File.dirname(Temporalio._root_file_path) } @fibers.map do |fiber| fiber.backtrace.reject { |s| s.start_with?(dir_path) }.join("\n") end.reject(&:empty?).join("\n\n") end |
#timeout_after(duration, exception_class, *exception_arguments) ⇒ Object
178 179 180 |
# File 'lib/temporalio/internal/worker/workflow_instance/scheduler.rb', line 178 def timeout_after(duration, exception_class, *exception_arguments, &) context.timeout(duration, exception_class, *exception_arguments, summary: 'Timeout timer', &) end |
#unblock(_blocker, fiber) ⇒ Object
182 183 184 |
# File 'lib/temporalio/internal/worker/workflow_instance/scheduler.rb', line 182 def unblock(_blocker, fiber) @ready << fiber end |
#wait_condition(cancellation:, &block) ⇒ Object
60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 |
# File 'lib/temporalio/internal/worker/workflow_instance/scheduler.rb', line 60 def wait_condition(cancellation:, &block) raise Workflow::InvalidWorkflowStateError, 'Cannot wait in this context' if @instance.context_frozen if cancellation&.canceled? raise Error::CanceledError, cancellation.canceled_reason || 'Wait condition canceled before started' end seq = (@wait_condition_counter += 1) @wait_conditions[seq] = [block, Fiber.current] # Add a cancellation callback cancel_callback_key = cancellation&.add_cancel_callback do # Only if the condition is still present cond = @wait_conditions.delete(seq) if cond&.last&.alive? cond&.last&.raise(Error::CanceledError.new(cancellation&.canceled_reason || 'Wait condition canceled')) end end # This blocks until a resume is called on this fiber result = Fiber.yield # Remove cancellation callback (only needed on success) cancellation&.remove_cancel_callback(cancel_callback_key) if cancel_callback_key result end |