Class: OMQ::Ractor

Inherits:
Object
  • Object
show all
Defined in:
lib/omq/ractor.rb

Overview

Bridges OMQ sockets into a Ruby Ractor for true parallel processing.

Sockets stay in the main Ractor (Async context). Bridge fibers/threads shuttle messages to/from a worker Ractor. Per-connection serialization converts between Ruby objects and ZMQ byte frames transparently: inproc uses Ractor.make_shareable, ipc/tcp use Marshal.

Examples:

Simple pipeline

Async do
  pull = OMQ::PULL.new
  pull.bind("tcp://127.0.0.1:5555")
  push = OMQ::PUSH.new
  push.connect("tcp://127.0.0.1:5556")

  worker = OMQ::Ractor.new(pull, push) do |omq|
    pull_p, push_p = omq.sockets
    loop do
      msg = pull_p.receive
      push_p << transform(msg)
    end
  end

  worker.join
end

Defined Under Namespace

Modules: TransparentDelegator Classes: Context, MarshalConnection, SerializeCache, ShareableConnection, SocketClosedError, SocketProxy, SocketSet, TopicMarshalConnection, TopicShareableConnection

Constant Summary collapse

HANDSHAKE_TIMEOUT = 
1
TOPIC_SOCKET_TYPES =

Socket types that use topic/group-based routing. These get topic-aware connection wrappers that preserve the first frame (topic/group) as a plain string for matching.

%i[PUB SUB XPUB XSUB RADIO DISH].freeze

Instance Method Summary collapse

Constructor Details

#initialize(*sockets, serialize: true, data: nil) {|Context| ... } ⇒ Ractor

Creates a new OMQ::Ractor that bridges the given sockets into a worker Ractor.

Parameters:

  • sockets (Array<Socket>) —

    sockets to bridge

  • serialize (Boolean) (defaults to: true) —

    whether to auto-serialize per connection (default: true)

  • data (Object, nil) (defaults to: nil) —

    optional shareable data accessible as omq.data inside the worker block. Under Ruby 4.0's strict Ractor isolation, worker blocks cannot close over outer local variables; use data: to pass configuration into the block.

Yields:

  • (Context) —

    block executes inside the worker Ractor; must call omq.sockets immediately

Raises:

  • (ArgumentError) 


386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
 
# File 'lib/omq/ractor.rb', line 386

def initialize(*sockets, serialize: true, data: nil, &block)
  raise ArgumentError, "no sockets given"  if sockets.empty?
  raise ArgumentError, "no block given"    unless block

  @sockets   = sockets
  @serialize = serialize

  # Categorize sockets
  socket_configs = sockets.map do |s|
    type_sym   = s.class.name.split("::").last.to_sym
    topic_type = TOPIC_SOCKET_TYPES.include?(type_sym)
    { readable: s.is_a?(Readable), writable: s.is_a?(Writable),
      serialize: serialize, topic_type: topic_type }
  end


  # Main Ractor creates output ports (one per writable socket)
  @output_ports = socket_configs.map { |cfg| cfg[:writable] ? ::Ractor::Port.new : nil }
  output_ports  = @output_ports

  # Setup port for the handshake (main-owned, main receives)
  setup_port = ::Ractor::Port.new

  # Build frozen context for the worker
  frozen_configs = ::Ractor.make_shareable(socket_configs)
  frozen_outputs = ::Ractor.make_shareable(output_ports)
  frozen_data    = data ? ::Ractor.make_shareable(data) : nil
  ctx = Context.new(setup_port, frozen_outputs, frozen_configs, data: frozen_data)

  # Install connection wrappers for per-connection serialization
  install_connection_wrappers(socket_configs) if serialize

  # Start the worker Ractor
  @ractor = ::Ractor.new(ctx, &block)

  # Wait for the handshake with timeout
  @input_ports = await_handshake(setup_port)
  input_ports  = @input_ports

  # Start bridges on the correct task.
  # Inside Async: spawn under current task.
  # Outside Async: dispatch to the IO thread via Reactor.run.
  @input_tasks    = []
  @output_threads = []
  @output_pipes   = []
  if Async::Task.current?
    @parent_task = Async::Task.current
    start_input_bridges(input_ports, socket_configs)
    start_output_bridges(output_ports, socket_configs)
  else
    @parent_task = Reactor.root_task
    Reactor.run do
      start_input_bridges(input_ports, socket_configs)
      start_output_bridges(output_ports, socket_configs)
    end
  end
end

Instance Method Details

#close ⇒ void

This method returns an undefined value.

Signals the worker to stop, then waits for it to finish. Sends nil through all input ports, causing proxy.receive to return nil (first time) or raise SocketClosedError.



472
473
474
475
476
 
# File 'lib/omq/ractor.rb', line 472

def close
  @input_ports.each { |p| p&.send(nil) rescue nil }
  await_ractor { @ractor.join } rescue nil
  cleanup_bridges
end

#join ⇒ void

This method returns an undefined value.

Waits for the worker Ractor to finish naturally. The worker must return from its block on its own.



449
450
451
452
453
 
# File 'lib/omq/ractor.rb', line 449

def join
  await_ractor { @ractor.join }
ensure
  cleanup_bridges
end

#value ⇒ Object

Returns the worker Ractor's return value. The worker must return from its block on its own.

Returns:

  • (Object) —

    the worker block's return value



460
461
462
463
464
 
# File 'lib/omq/ractor.rb', line 460

def value
  await_ractor { @ractor.value }
ensure
  cleanup_bridges
end