Class: Fractor::WrappedRactor4

Inherits:
WrappedRactor show all
Defined in:
lib/fractor/wrapped_ractor4.rb

Overview

Ruby 4.0+ specific implementation of WrappedRactor. Uses Ractor::Port for communication - main ractor creates response ports and passes them to workers when sending work.

Instance Attribute Summary collapse

Attributes inherited from WrappedRactor

#name, #ractor, #worker_class

Instance Method Summary collapse

Methods inherited from WrappedRactor

#closed?, create

Constructor Details

#initialize(name, worker_class, response_port: nil) ⇒ WrappedRactor4

Initializes the WrappedRactor with a name, worker class, and response port.

Parameters:

  • name (String) —

    Name for the ractor

  • worker_class (Class) —

    Worker class to instantiate

  • response_port (Ractor::Port, nil) (defaults to: nil) —

    The port to receive responses on (created by main ractor)



19
20
21
22
 
# File 'lib/fractor/wrapped_ractor4.rb', line 19

def initialize(name, worker_class, response_port: nil)
  super(name, worker_class)
  @response_port = response_port
end

Instance Attribute Details

#response_port ⇒ Object

Returns the value of attribute response_port.



12
13
14
 
# File 'lib/fractor/wrapped_ractor4.rb', line 12

def response_port
  @response_port
end

Instance Method Details

#close ⇒ Boolean

Closes the Ractor and its response port. In Ruby 4.0, we need to explicitly close the response port to prevent Ractor.select from hanging.

Returns:

  • (Boolean) —

    true if closed successfully



231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
 
# File 'lib/fractor/wrapped_ractor4.rb', line 231

def close
  # Close the response port first
  if @response_port
    begin
      # Send nil through the port to signal it's closing
      @response_port.send(nil) if @response_port.respond_to?(:send)
    rescue StandardError
      # Port may already be closed
    end
    @response_port = nil
  end

  # Then close the underlying ractor
  super
end

#receive_message ⇒ Hash?

Receives a message from the Ractor. In Ruby 4.0, messages come through response ports in the main loop. This method is kept for backwards compatibility with tests.

Note: In Ruby 4.0, this will block if no message is available. The test should either skip this or ensure a message was sent first.

Returns:

  • (Hash, nil) —

    The message received or nil



213
214
215
216
217
218
219
220
221
222
223
224
 
# File 'lib/fractor/wrapped_ractor4.rb', line 213

def receive_message
  # In Ruby 4.0, we receive through the response_port
  # Try to receive with a small timeout to avoid blocking indefinitely
  return nil unless @response_port

  # Use a non-blocking receive attempt
  # In Ruby 4.0, if the port is empty, this would block
  # For test compatibility, we return nil if no message is available
  @response_port.receive
rescue Ractor::ClosedError, Ractor::Error
  nil
end

#send(work) ⇒ Boolean

Sends work to the Ractor. In Ruby 4.0, sends [work, response_port] so worker can reply back. Special case: sends just :shutdown for shutdown messages.

Parameters:

  • work (Fractor::Work, Symbol) —

    The work item to process, or :shutdown

Returns:

  • (Boolean) —

    true if sent successfully, false otherwise



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

def send(work)
  if @ractor
    begin
      # In Ruby 4.0, send [work, response_port] so worker can reply
      # Special case: shutdown is sent as a symbol, not an array
      if work == :shutdown
        @ractor.send(:shutdown)
      else
        @ractor.send([work, @response_port])
      end
      true
    rescue Exception => e
      RactorLogger.warn("Error sending work to Ractor: #{e.message}",
                        ractor_name: @name)
      false
    end
  else
    RactorLogger.warn("Attempted to send work to nil Ractor",
                      ractor_name: @name)
    false
  end
end

#start ⇒ Object

Starts the underlying Ractor using the port-based pattern. In Ruby 4.0:

  • Main ractor creates response ports (one per worker)
  • Main ractor sends [work, response_port] to workers
  • Workers receive work and response_port, send results back via response_port


36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
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
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
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
169
170
171
172
173
 
# File 'lib/fractor/wrapped_ractor4.rb', line 36

def start
  RactorLogger.info("Starting Ractor #{@name} (Ruby 4.0 mode)",
                    ractor_name: @name)

  # Capture timeout value before entering ractor (Ractors can't access Fractor.config)
  # Get class-level timeout, or fall back to default of nil (no timeout)
  # Note: We avoid accessing Fractor.config from ractor creation context
  class_level_timeout = @worker_class.worker_timeout

  # In Ruby 4.0, workers don't create their own ports
  # They receive response_port from main ractor when work is sent
  @ractor = Ractor.new(@name, @worker_class,
                       class_level_timeout) do |name, worker_cls, timeout_val|
    RactorLogger.debug(
      "Ractor started with worker class #{worker_cls} and timeout #{timeout_val.inspect}", ractor_name: name
    )

    # Instantiate the specific worker inside the Ractor
    # Pass timeout as an option only if it's not nil, to avoid accessing self.class from ractor
    worker = if timeout_val.nil?
               worker_cls.new(name: name)
             else
               worker_cls.new(name: name, timeout: timeout_val)
             end

    # Main message processing loop
    loop do
      # Receive work from the main ractor (blocks until message available)
      # In Ruby 4.0, main sends [work, response_port]
      received = Ractor.receive
      RactorLogger.debug("Received #{received.inspect}", ractor_name: name)

      # Handle shutdown message
      if received == :shutdown
        RactorLogger.debug("Received shutdown message, terminating",
                           ractor_name: name)
        break
      end

      # Extract work and response_port
      # Main should send [work, response_port]
      if received.is_a?(Array) && received.size == 2
        work, response_port = received
      else
        # Legacy format for initialization or other messages
        work = received
        response_port = nil
      end

      # Handle initialize message (for backwards compatibility during startup)
      if work.is_a?(Hash) && work[:type] == :initialize
        RactorLogger.debug("Worker initialized", ractor_name: name)
        next
      end

      begin
        # Get the timeout for this specific work item
        # Priority: work.timeout > worker.timeout (nil means no timeout)
        work_timeout = if work.respond_to?(:timeout) && !work.timeout.nil?
                         work.timeout
                       else
                         worker.timeout
                       end

        # Process the work with timeout if configured
        # Note: Ruby's Timeout.timeout uses threads which don't work with Ractors.
        # We measure execution time and raise timeout error afterward if exceeded.
        result = if work_timeout
                   start_time = Time.now
                   process_result = worker.process(work)
                   elapsed = Time.now - start_time
                   if elapsed > work_timeout
                     # Raise a timeout error after the fact
                     # Note: This is a post-facto timeout check - the work has already completed
                     raise Timeout::Error,
                           "execution timed out after #{elapsed}s (limit: #{work_timeout}s)"
                   end

                   process_result
                 else
                   worker.process(work)
                 end

        RactorLogger.debug("Sending result #{result.inspect}",
                           ractor_name: name)

        # Wrap the result in a WorkResult object if not already wrapped
        work_result = if result.is_a?(Fractor::WorkResult)
                        result
                      else
                        Fractor::WorkResult.new(result: result, work: work)
                      end

        # Send the result back through the response port
        if response_port
          response_port << { type: :result, result: work_result,
                             processor: name }
        else
          RactorLogger.warn("No response port available, result lost",
                            ractor_name: name)
        end
      rescue Timeout::Error => e
        # Handle timeout errors as retriable errors
        RactorLogger.warn(
          "Timed out after #{work_timeout}s processing work #{work.inspect}", ractor_name: name
        )
        error_result = Fractor::WorkResult.new(
          error: "Worker timeout: #{e.message}",
          work: work,
          error_category: :timeout,
        )
        if response_port
          response_port << { type: :error, result: error_result,
                             processor: name }
        end
      rescue StandardError => e
        # Handle errors during processing
        RactorLogger.error("Error processing work #{work.inspect}",
                           ractor_name: name, exception: e)

        # Send an error message back through the response port
        # Ensure the original work object is included in the error result
        error_result = Fractor::WorkResult.new(error: e.message, work: work)
        if response_port
          response_port << { type: :error, result: error_result,
                             processor: name }
        end
      end
    end
  rescue Ractor::ClosedError
    RactorLogger.debug("Ractor closed", ractor_name: name)
  rescue StandardError => e
    RactorLogger.error("Unexpected error", ractor_name: name, exception: e)
  ensure
    RactorLogger.debug("Ractor shutting down", ractor_name: name)
  end
  RactorLogger.debug("Ractor instance created", ractor_name: name)
end