Class: Prosody::AsyncTaskProcessor

Inherits:
Object
  • Object
show all
Defined in:
lib/prosody/processor.rb,
lib/prosody/native_stubs.rb,
sig/processor.rbs

Overview

Internal processor for executing tasks asynchronously. This class is used internally by the native code.

Instance Method Summary collapse

Constructor Details

#initialize(logger = Prosody.logger) ⇒ AsyncTaskProcessor

Creates a new processor with the given logger

Parameters:

  • logger (defaults to: Prosody.logger) —

    logger for diagnostic messages (defaults to Prosody.logger)

  • (Logger) 


112
113
114
115
116
117
 
# File 'lib/prosody/processor.rb', line 112

def initialize(logger = Prosody.logger)
  @logger = logger
  @command_queue = Queue.new
  @processing_thread = nil
  @tracer = nil
end

Instance Method Details

#handle_execute(command, barrier) ⇒ void

This method returns an undefined value.

Handles execution of a task with proper context propagation and error handling. The barrier belongs to the external async gem and is intentionally kept untyped rather than publishing a partial signature for that dependency.

Parameters:



209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
 
# File 'lib/prosody/processor.rb', line 209

def handle_execute(command, barrier)
  task_id = command.task_id
  carrier = command.carrier
  event_context = command.event_context
  token = command.token
  callback = command.callback
  task_block = command.block

  # Extract parent context from the incoming carrier for distributed tracing
  parent_ctx = OpenTelemetry.propagation.extract(carrier)

  # Create the dispatch span as a child of the extracted context, then
  # capture the resulting context so it can be explicitly restored inside
  # the worker fiber. The span is owned by run_with_cancellation, which
  # finishes it in ensure.
  dispatch_ctx = OpenTelemetry::Context.with_current(parent_ctx) do
    span = @tracer.start_span("async_dispatch", kind: :consumer)
    OpenTelemetry::Trace.with_span(span) { OpenTelemetry::Context.current }
  end

  @logger.debug("Executing task #{task_id}")

  begin
    barrier.async do
      run_with_cancellation(task_id, token, task_block, callback, dispatch_ctx, event_context)
    end
  rescue => e
    # If we failed to enqueue, finish the span here since run_with_cancellation
    # will never take ownership.
    OpenTelemetry::Trace.current_span(dispatch_ctx).finish
    raise e
  end
end

#process_commands ⇒ void

This method returns an undefined value.

Main processing loop for the async thread



179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
 
# File 'lib/prosody/processor.rb', line 179

def process_commands
  Async do
    # Barrier tracks all running tasks for clean shutdown
    barrier = Async::Barrier.new

    loop do
      command = @command_queue.pop

      case command
      when Commands::Execute
        handle_execute(command, barrier)
      when Commands::Shutdown
        @logger.debug("Received shutdown command")
        # Wait for all tasks to complete before shutting down
        barrier.wait
        break
      else
        @logger.warn("Unknown command type: #{command.class}")
      end
    end
  end
rescue => e
  @logger.error("Error in process_commands: #{e.message}")
  @logger.error(e.backtrace.join("\n"))
end

#run_with_cancellation(task_id, token, task_block, callback, dispatch_ctx, event_context) ⇒ Object

Executes a task with proper cancellation support.

Spawns a worker task and a cancellation watcher within a barrier. When cancellation is signaled, the worker receives Async::Stop (similar to Python's asyncio.CancelledError). The worker can catch Async::Stop for cleanup. The barrier ensures both tasks are cleaned up when the block exits, even if an unexpected error occurs.

Parameters:

  • task_id (String) —

    The task identifier for logging

  • token (CancellationToken) —

    The token to monitor for cancellation

  • task_block (Proc) —

    The work to execute

  • callback (Proc) —

    The callback to notify of completion or error

  • dispatch_ctx (OpenTelemetry::Context) —

    Context with async_dispatch span active



256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
 
# File 'lib/prosody/processor.rb', line 256

def run_with_cancellation(task_id, token, task_block, callback, dispatch_ctx, event_context)
  span = OpenTelemetry::Trace.current_span(dispatch_ctx)
  # Use a barrier to ensure both tasks are cleaned up when block exits
  barrier = Async::Barrier.new

  # Spawn worker task - handles its own result reporting
  worker_task = barrier.async do |task|
    task.annotate("Worker for task #{task_id}")
    # Async fibers do not inherit fiber-local OTel context from their parent,
    # so we must explicitly restore dispatch_ctx so user spans are children
    # of async_dispatch.
    OpenTelemetry::Context.with_current(dispatch_ctx) do
      result = task_block.call
      if callback.call(true, result)
        @logger.debug("Task #{task_id} completed successfully")
      end
    rescue Async::Stop
      # Task was cancelled - report via callback
      if callback.call(false, RuntimeError.new("Task cancelled"))
        @logger.debug("Task #{task_id} was cancelled")
      end
    rescue => e
      Prosody::SentryIntegration.capture_exception(e, event_context.merge(task_id: task_id))
      if callback.call(false, e)
        @logger.error("Error executing task #{task_id}: #{e.message}")
        span.record_exception(e)
        span.status = OpenTelemetry::Trace::Status.error(e.to_s)
      end
    ensure
      # Always signal the cancellation watcher to stop waiting
      token.cancel
    end
  end

  # Spawn cancellation watcher - bridges the thread-boundary cancellation signal
  # into the fiber scheduler. The CancellationToken is a one-shot channel: the
  # Rust bridge pushes a signal from its thread, and this fiber pops it and
  # translates it into Async::Stop on the worker. We can't call worker_task.stop
  # directly from the Rust thread since Async task control must happen on the
  # scheduler thread.
  barrier.async do |task|
    task.annotate("Cancellation watcher for task #{task_id}")
    begin
      token.wait
      worker_task.stop
    rescue => e
      @logger.debug("Cancellation watcher error: #{e.message}")
      span.record_exception(e)
      span.status = OpenTelemetry::Trace::Status.error(e.to_s)
    end
  end

  # Wait for worker to complete (normally, via Async::Stop, or with error)
  worker_task.wait
ensure
  # Stop any remaining tasks (primarily the cancellation watcher).
  # Finish the span last so it covers the full execution, and is guaranteed
  # to close even if barrier.stop raises.
  begin
    barrier&.stop
  ensure
    span.finish
  end
end

#running? ⇒ Boolean

Checks if the processor thread is running.

Returns:

  • (Boolean) —

    true if the processor is running, false otherwise



171
172
173
 
# File 'lib/prosody/processor.rb', line 171

def running?
  @processing_thread&.alive?
end

#start ⇒ Object

Starts the processor by launching a dedicated thread Does nothing if the processor is already running

Returns:

  • the Thread if started, or nil if already running



124
125
126
127
128
129
130
131
132
133
134
135
136
137
 
# File 'lib/prosody/processor.rb', line 124

def start
  return if running?

  @logger.debug("Starting async task processor")
  @processing_thread = Thread.new do
    # Initialize the tracer in the processing thread to keep
    # OpenTelemetry context within the same thread
    @tracer = OpenTelemetry.tracer_provider.tracer(
      "Prosody::AsyncTaskProcessor",
      Prosody::VERSION
    )
    process_commands
  end
end

#stop ⇒ void

This method returns an undefined value.

Gracefully stops the processor Does nothing if the processor is already stopped



143
144
145
146
147
148
 
# File 'lib/prosody/processor.rb', line 143

def stop
  return unless running?

  @logger.debug("Stopping async task processor")
  @command_queue.push(Commands::Shutdown.new)
end

#submit(task_id, carrier, event_context, callback) { ... } ⇒ Object

Submits a task for asynchronous execution

Parameters:

  • task_id —

    unique identifier for the task

  • carrier —

    OpenTelemetry context carrier for tracing

  • callback —

    called with (success, result) when task completes

Yields:

  • the block to execute asynchronously

Returns:

  • token that can be used to cancel the task



158
159
160
161
162
163
164
 
# File 'lib/prosody/processor.rb', line 158

def submit(task_id, carrier, event_context, callback, &task_block)
  token = CancellationToken.new
  @command_queue.push(
    Commands::Execute.new(task_id, carrier, event_context.transform_keys(&:to_sym), task_block, callback, token)
  )
  token
end