Class: Teek::BackgroundRactor4x::BackgroundWork

Inherits:
Object
  • Object
show all
Defined in:
lib/teek/background_ractor4x.rb

Overview

Ractor-based background work using Ruby 4.x Ractor::Port for streaming and Ractor.shareable_proc for blocks.

Examples:

Block form

work = BackgroundWork.new(app, urls) do |task, data|
  data.each { |url| task.yield(fetch(url)) }
end
work.on_progress { |r| update_ui(r) }.on_done { puts "Done!" }

Worker class form

work = BackgroundWork.new(app, data, worker: MyWorker)
work.on_progress { |r| update_ui(r) }

Defined Under Namespace

Classes: TaskContext

Instance Method Summary collapse

Constructor Details

#initialize(app, data, worker: nil) {|task, data| ... } ⇒ BackgroundWork

Returns a new instance of BackgroundWork.

Parameters:

  • app (Teek::App) —

    the application instance (for after scheduling)

  • data (Object) —

    data passed to the worker block/class

  • worker (Class, nil) (defaults to: nil) —

    optional worker class (must respond to #call(task, data))

Yields:

  • (task, data) —

    block executed inside a Ractor

Yield Parameters:

  • task (TaskContext) —

    context for yielding results and checking messages

  • data (Object) —

    the data passed to the constructor



63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
 
# File 'lib/teek/background_ractor4x.rb', line 63

def initialize(app, data, worker: nil, &block)
  @app = app
  @data = data
  @work_block = block || (worker && proc { |t, d| worker.new.call(t, d) })
  @callbacks = { progress: nil, done: nil, message: nil }
  @started = false
  @done = false
  @paused = false

  # Communication
  @output_queue = Thread::Queue.new
  @control_port = nil  # Set by worker, received back
  @pending_messages = []  # Queued until control_port ready
  @worker_ractor = nil
  @bridge_thread = nil
end

Instance Method Details

#close ⇒ self

Force-close the Ractor and all associated resources.

Returns:

  • (self) 


154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
 
# File 'lib/teek/background_ractor4x.rb', line 154

def close
  @done = true
  # Send stop to let the worker terminate itself — Ruby 4.x doesn't
  # allow closing a Ractor from outside. The message thread will
  # raise StopIteration on the Ractor's main thread, which triggers
  # the rescue block that sends [:done] to the output port and exits
  # the Ractor cleanly.
  begin
    @control_port&.send(:stop)
  rescue Ractor::ClosedError
    # Already closed
  end
  @control_port = nil
  # Wait for the bridge thread to receive [:done] and exit. Without
  # this, the zombie bridge thread blocks subsequent operations on
  # Windows (Ractor::Port#receive holds the GVL).
  @bridge_thread&.join(2)
  self
end

#done? ⇒ Boolean

Returns whether the worker has finished.

Returns:

  • (Boolean) —

    whether the worker has finished



202
203
204
 
# File 'lib/teek/background_ractor4x.rb', line 202

def done?
  @done
end

#on_done { ... } ⇒ self

Register a callback for when the worker finishes. Auto-starts the task if not already started.

Yields:

  • called on the main thread when the worker completes

Returns:

  • (self) 


94
95
96
97
98
 
# File 'lib/teek/background_ractor4x.rb', line 94

def on_done(&block)
  @callbacks[:done] = block
  maybe_start
  self
end

#on_message {|msg| ... } ⇒ self

Register a callback for custom messages sent by the worker via Teek::BackgroundRactor4x::BackgroundWork::TaskContext#send_message.

Yields:

  • (msg) —

    called on the main thread with the message

Returns:

  • (self) 


104
105
106
107
 
# File 'lib/teek/background_ractor4x.rb', line 104

def on_message(&block)
  @callbacks[:message] = block
  self
end

#on_progress {|value| ... } ⇒ self

Register a callback for progress updates from the worker. Auto-starts the task if not already started.

Yields:

  • (value) —

    called on the main thread each time the worker yields a result

Returns:

  • (self) 


84
85
86
87
88
 
# File 'lib/teek/background_ractor4x.rb', line 84

def on_progress(&block)
  @callbacks[:progress] = block
  maybe_start
  self
end

#pause ⇒ self

Pause the worker. The worker will block on the next Teek::BackgroundRactor4x::BackgroundWork::TaskContext#yield or Teek::BackgroundRactor4x::BackgroundWork::TaskContext#check_pause until #resume is called.

Returns:

  • (self) 


130
131
132
133
134
 
# File 'lib/teek/background_ractor4x.rb', line 130

def pause
  @paused = true
  send_message(:pause)
  self
end

#paused? ⇒ Boolean

Returns whether the worker is paused.

Returns:

  • (Boolean) —

    whether the worker is paused



207
208
209
 
# File 'lib/teek/background_ractor4x.rb', line 207

def paused?
  @paused
end

#resume ⇒ self

Resume a paused worker.

Returns:

  • (self) 


138
139
140
141
142
 
# File 'lib/teek/background_ractor4x.rb', line 138

def resume
  @paused = false
  send_message(:resume)
  self
end

#send_message(msg) ⇒ self

Send a message to the worker. The worker can receive it via Teek::BackgroundRactor4x::BackgroundWork::TaskContext#check_message or Teek::BackgroundRactor4x::BackgroundWork::TaskContext#wait_message. Messages are queued if the worker's control port isn't ready yet.

Parameters:

  • msg (Object) —

    any Ractor-shareable value

Returns:

  • (self) 


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

def send_message(msg)
  if @control_port
    begin
      @control_port.send(msg)
    rescue Ractor::ClosedError
      # Port already closed, task is done - ignore
    end
  else
    @pending_messages << msg
  end
  self
end

#start ⇒ self

Explicitly start the background work. Called automatically by #on_progress and #on_done; only needed when using #on_message alone.

Returns:

  • (self) 


178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
 
# File 'lib/teek/background_ractor4x.rb', line 178

def start
  return self if @started
  @started = true

  # Wrap in isolated proc for Ractor sharing. The block can only access
  # its parameters (task, data), not outer-scope variables.
  isolation_error = false
  begin
    shareable_block = Ractor.shareable_proc(&@work_block)
  rescue Ractor::IsolationError
    isolation_error = true
  end
  if isolation_error
    raise Ractor::IsolationError,
      "Background work block must not reference outside variables (including `app`). " \
      "Use t.yield() to send results to on_progress, which runs on the main thread."
  end

  start_ractor(shareable_block)
  start_polling
  self
end

#stop ⇒ self

Request the worker to stop. Raises StopIteration inside the worker on the next Teek::BackgroundRactor4x::BackgroundWork::TaskContext#check_message or Teek::BackgroundRactor4x::BackgroundWork::TaskContext#yield.

Returns:

  • (self) 


147
148
149
150
 
# File 'lib/teek/background_ractor4x.rb', line 147

def stop
  send_message(:stop)
  self
end