Module: Farce::Backports

Includes:
Extension.new
Defined in:
lib/farce/backports.rb

Overview

Backports for Ractor and related functionality to older Ruby versions.

Activating Backports

There are three ways of using these backports:

  • Activating them globally via: Backports.install
  • Using them within a specific scope via refinements: using Farce::Backports
  • Calling methods on Farce::Backports directly

Only the first method will also define the global constants Ractor and Ractor::Port if they are not natively available.

Examples:

Activating the backports globally

Farce::Backports.install
port = Ractor.current.default_port

Using the backports within a specific namespace via refinements

# Within a namespace
module MyNamespace
  using Farce::Backports

  def self.send_to(ractor_or_port, message)
    port = ractor_or_port.default_port if ractor_or_port.is_a?(Ractor)
    port.send(message)
  end
end

Activating the backports for a specific file

# top level of the file
using Farce::Backports

ractor  = Ractor.new { sleep 10 }
monitor = Ractor.new { puts "exit status: #{receive}" }

monitor.monitor(ractor.default_port)
monitor.join

Activating the backports for a given proc (Ruby 4.1+ only)

callback = ->(ractor) { ractor.default_port }
callback.refined(Farce::Backports).call(Ractor.current)

Calling methods on Farce::Backports directly

port = Farce::Backports.default_port(Ractor.current)

Class Method Summary collapse

Class Method Details

.alive?(ractor) ⇒ Boolean

Backport for Ractor#alive?

Examples:

using Farce::Backports
ractor = Ractor.new { sleep 10 }
ractor.alive? # => true

Parameters:

  • ractor (Ractor) —

    the ractor to check

Returns:

  • (Boolean) —

    whether the ractor is alive



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
174
175
176
177
178
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
204
205
206
207
208
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
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
 
# File 'lib/farce/backports.rb', line 136

module Backports
  # @!visibility private
  Extension = Farce::Extension.new
  private_constant :Extension
  include Extension

  if Internal.native_ractors?
    if needs_timeouts = ::Ractor.instance_method(:receive).arity.zero?
      Extension.define(::Ractor) do
        def receive(timeout: nil)
          return super() if timeout.nil?
          Internal::RactorSelector.current.ractor_receive(self, timeout:)
        end
      end

      Extension.define(::Ractor.singleton_class) do
        def select(*sources, timeout: nil, **)
          # The selector helper must reach Ruby's native wait without reentering itself.
          return super(*sources, **) if timeout.nil? && (Internal.native_ports? || sources.all?(::Ractor))
          Farce::Ractor.select(*sources, timeout:, **)
        end
      end

      def self.receive(ractor_or_port, timeout: nil)
        ractor_or_port = default_port(ractor_or_port) if ractor_or_port.is_a?(Ractor)
        return ractor_or_port.receive if timeout.nil?
        return ractor_or_port.receive(timeout:) unless Internal.native_ports?
        Internal::RactorSelector.current.ractor_receive(ractor_or_port, timeout:)
      end
    end

    if Internal.native_ports?
      unless ::Ractor.method_defined?(:alive?)
        Extension.define(::Ractor) { def alive? = Backports.alive?(self) }

        def self.alive?(ractor)
          ractor.monitor(NULL_PORT)
        ensure
          ractor.unmonitor(NULL_PORT)
        end
      end

      if needs_timeouts
        Extension.define(::Ractor::Port) do
          def receive(timeout: nil)
            return super() if timeout.nil?
            Internal::RactorSelector.current.ractor_receive(self, timeout:)
          end
        end
      end
    else
      Extension.define(::Ractor) do
        def alive?          = Backports.alive?(self)
        def default_port    = Backports.default_port(self)
        def join            = Backports.join(self)
        def monitor(port)   = Backports.monitor(self, port)
        def unmonitor(port) = Backports.unmonitor(self, port)
        def value           = Backports.value(self)
      end

      def self.alive?(ractor)
        return true if Ractor.main == ractor
        state = Internal::RactorState.for(ractor)
        return state.alive? if state
        !ractor.inspect.end_with?("terminated>")
      end

      def self.default_port(ractor) = Farce::Ractor::Port.new(ractor)

      def self.join(ractor)
        Internal::RactorState.for(ractor)&.join
        ractor
      end

      def self.monitor(ractor, port)
        return true if Ractor.main == ractor

        state = Internal::RactorState.for(ractor)
        return state.monitor(port) if state

        raise Ractor::Error, "#{ractor.inspect} cannot be monitored" unless ractor.inspect.end_with?("terminated>")

        success = true

        begin
          ractor.take
        rescue Ractor::ClosedError
          # no-op
        rescue Ractor::RemoteError
          success = false
        end

        port.send(success ? :exited : :aborted)
        false
      end

      def self.unmonitor(ractor, port)
        Internal::RactorState.for(ractor)&.unmonitor(port)
        ractor
      end

      def self.value(ractor)
        Internal::RactorState.for(ractor)&.join
        values = ::Ractor.store_if_absent("Ractor#value") { ObjectSpace::WeakKeyMap.new }
        entry  = values[ractor] ||= [Mutex.new]
        entry.first.synchronize { entry << ractor.take if entry.size == 1 }
        entry[1]
      end
    end
  end

  %i[alive? default_port join monitor unmonitor value].each do |method|
    next if respond_to?(method)
    class_eval <<~RUBY, __FILE__, __LINE__ + 1
      def self.#{method}(ractor, ...) = ractor.#{method}(...)
    RUBY
  end

  unless respond_to?(:receive)
    def self.receive(ractor_or_port, timeout: nil)
      ractor_or_port = default_port(ractor_or_port) if ractor_or_port.is_a?(Ractor)
      timeout.nil? ? ractor_or_port.receive : ractor_or_port.receive(timeout:)
    end
  end

  def self.select(...) = Farce::Ractor.select(...)

  # Installs the Farce backports for Ractor and Ractor::Port globally.
  def self.install
    Extension.install do
      if Internal.native_ractors?
        ::Ractor.const_set(:Port, Farce::Ractor::Port) unless Internal.native_ports?
      else
        Object.const_set(:Ractor, Farce::Ractor)
      end
    end
  end
end

.default_port(ractor) ⇒ Ractor::Port

Backport for Ractor.default_port

Examples:

using Farce::Backports
ractor = Ractor.new { puts receive }
port   = ractor.default_port
port.send("message")

Parameters:

  • ractor (Ractor) —

    the ractor for which to obtain the default port

Returns:



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
174
175
176
177
178
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
204
205
206
207
208
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
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
 
# File 'lib/farce/backports.rb', line 136

module Backports
  # @!visibility private
  Extension = Farce::Extension.new
  private_constant :Extension
  include Extension

  if Internal.native_ractors?
    if needs_timeouts = ::Ractor.instance_method(:receive).arity.zero?
      Extension.define(::Ractor) do
        def receive(timeout: nil)
          return super() if timeout.nil?
          Internal::RactorSelector.current.ractor_receive(self, timeout:)
        end
      end

      Extension.define(::Ractor.singleton_class) do
        def select(*sources, timeout: nil, **)
          # The selector helper must reach Ruby's native wait without reentering itself.
          return super(*sources, **) if timeout.nil? && (Internal.native_ports? || sources.all?(::Ractor))
          Farce::Ractor.select(*sources, timeout:, **)
        end
      end

      def self.receive(ractor_or_port, timeout: nil)
        ractor_or_port = default_port(ractor_or_port) if ractor_or_port.is_a?(Ractor)
        return ractor_or_port.receive if timeout.nil?
        return ractor_or_port.receive(timeout:) unless Internal.native_ports?
        Internal::RactorSelector.current.ractor_receive(ractor_or_port, timeout:)
      end
    end

    if Internal.native_ports?
      unless ::Ractor.method_defined?(:alive?)
        Extension.define(::Ractor) { def alive? = Backports.alive?(self) }

        def self.alive?(ractor)
          ractor.monitor(NULL_PORT)
        ensure
          ractor.unmonitor(NULL_PORT)
        end
      end

      if needs_timeouts
        Extension.define(::Ractor::Port) do
          def receive(timeout: nil)
            return super() if timeout.nil?
            Internal::RactorSelector.current.ractor_receive(self, timeout:)
          end
        end
      end
    else
      Extension.define(::Ractor) do
        def alive?          = Backports.alive?(self)
        def default_port    = Backports.default_port(self)
        def join            = Backports.join(self)
        def monitor(port)   = Backports.monitor(self, port)
        def unmonitor(port) = Backports.unmonitor(self, port)
        def value           = Backports.value(self)
      end

      def self.alive?(ractor)
        return true if Ractor.main == ractor
        state = Internal::RactorState.for(ractor)
        return state.alive? if state
        !ractor.inspect.end_with?("terminated>")
      end

      def self.default_port(ractor) = Farce::Ractor::Port.new(ractor)

      def self.join(ractor)
        Internal::RactorState.for(ractor)&.join
        ractor
      end

      def self.monitor(ractor, port)
        return true if Ractor.main == ractor

        state = Internal::RactorState.for(ractor)
        return state.monitor(port) if state

        raise Ractor::Error, "#{ractor.inspect} cannot be monitored" unless ractor.inspect.end_with?("terminated>")

        success = true

        begin
          ractor.take
        rescue Ractor::ClosedError
          # no-op
        rescue Ractor::RemoteError
          success = false
        end

        port.send(success ? :exited : :aborted)
        false
      end

      def self.unmonitor(ractor, port)
        Internal::RactorState.for(ractor)&.unmonitor(port)
        ractor
      end

      def self.value(ractor)
        Internal::RactorState.for(ractor)&.join
        values = ::Ractor.store_if_absent("Ractor#value") { ObjectSpace::WeakKeyMap.new }
        entry  = values[ractor] ||= [Mutex.new]
        entry.first.synchronize { entry << ractor.take if entry.size == 1 }
        entry[1]
      end
    end
  end

  %i[alive? default_port join monitor unmonitor value].each do |method|
    next if respond_to?(method)
    class_eval <<~RUBY, __FILE__, __LINE__ + 1
      def self.#{method}(ractor, ...) = ractor.#{method}(...)
    RUBY
  end

  unless respond_to?(:receive)
    def self.receive(ractor_or_port, timeout: nil)
      ractor_or_port = default_port(ractor_or_port) if ractor_or_port.is_a?(Ractor)
      timeout.nil? ? ractor_or_port.receive : ractor_or_port.receive(timeout:)
    end
  end

  def self.select(...) = Farce::Ractor.select(...)

  # Installs the Farce backports for Ractor and Ractor::Port globally.
  def self.install
    Extension.install do
      if Internal.native_ractors?
        ::Ractor.const_set(:Port, Farce::Ractor::Port) unless Internal.native_ports?
      else
        Object.const_set(:Ractor, Farce::Ractor)
      end
    end
  end
end

.install ⇒ BasicObject

Installs the Farce backports for Ractor and Ractor::Port globally.



264
265
266
267
268
269
270
271
272
 
# File 'lib/farce/backports.rb', line 264

def self.install
  Extension.install do
    if Internal.native_ractors?
      ::Ractor.const_set(:Port, Farce::Ractor::Port) unless Internal.native_ports?
    else
      Object.const_set(:Ractor, Farce::Ractor)
    end
  end
end

.join(ractor) ⇒ Ractor

Backport for Ractor#join

Examples:

using Farce::Backports
ractor = Ractor.new { sleep 10 }
ractor.join # blocks until the ractor exits

Parameters:

  • ractor (Ractor) —

    the ractor to join

Returns:

  • (Ractor) —

    the joined ractor



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
174
175
176
177
178
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
204
205
206
207
208
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
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
 
# File 'lib/farce/backports.rb', line 136

module Backports
  # @!visibility private
  Extension = Farce::Extension.new
  private_constant :Extension
  include Extension

  if Internal.native_ractors?
    if needs_timeouts = ::Ractor.instance_method(:receive).arity.zero?
      Extension.define(::Ractor) do
        def receive(timeout: nil)
          return super() if timeout.nil?
          Internal::RactorSelector.current.ractor_receive(self, timeout:)
        end
      end

      Extension.define(::Ractor.singleton_class) do
        def select(*sources, timeout: nil, **)
          # The selector helper must reach Ruby's native wait without reentering itself.
          return super(*sources, **) if timeout.nil? && (Internal.native_ports? || sources.all?(::Ractor))
          Farce::Ractor.select(*sources, timeout:, **)
        end
      end

      def self.receive(ractor_or_port, timeout: nil)
        ractor_or_port = default_port(ractor_or_port) if ractor_or_port.is_a?(Ractor)
        return ractor_or_port.receive if timeout.nil?
        return ractor_or_port.receive(timeout:) unless Internal.native_ports?
        Internal::RactorSelector.current.ractor_receive(ractor_or_port, timeout:)
      end
    end

    if Internal.native_ports?
      unless ::Ractor.method_defined?(:alive?)
        Extension.define(::Ractor) { def alive? = Backports.alive?(self) }

        def self.alive?(ractor)
          ractor.monitor(NULL_PORT)
        ensure
          ractor.unmonitor(NULL_PORT)
        end
      end

      if needs_timeouts
        Extension.define(::Ractor::Port) do
          def receive(timeout: nil)
            return super() if timeout.nil?
            Internal::RactorSelector.current.ractor_receive(self, timeout:)
          end
        end
      end
    else
      Extension.define(::Ractor) do
        def alive?          = Backports.alive?(self)
        def default_port    = Backports.default_port(self)
        def join            = Backports.join(self)
        def monitor(port)   = Backports.monitor(self, port)
        def unmonitor(port) = Backports.unmonitor(self, port)
        def value           = Backports.value(self)
      end

      def self.alive?(ractor)
        return true if Ractor.main == ractor
        state = Internal::RactorState.for(ractor)
        return state.alive? if state
        !ractor.inspect.end_with?("terminated>")
      end

      def self.default_port(ractor) = Farce::Ractor::Port.new(ractor)

      def self.join(ractor)
        Internal::RactorState.for(ractor)&.join
        ractor
      end

      def self.monitor(ractor, port)
        return true if Ractor.main == ractor

        state = Internal::RactorState.for(ractor)
        return state.monitor(port) if state

        raise Ractor::Error, "#{ractor.inspect} cannot be monitored" unless ractor.inspect.end_with?("terminated>")

        success = true

        begin
          ractor.take
        rescue Ractor::ClosedError
          # no-op
        rescue Ractor::RemoteError
          success = false
        end

        port.send(success ? :exited : :aborted)
        false
      end

      def self.unmonitor(ractor, port)
        Internal::RactorState.for(ractor)&.unmonitor(port)
        ractor
      end

      def self.value(ractor)
        Internal::RactorState.for(ractor)&.join
        values = ::Ractor.store_if_absent("Ractor#value") { ObjectSpace::WeakKeyMap.new }
        entry  = values[ractor] ||= [Mutex.new]
        entry.first.synchronize { entry << ractor.take if entry.size == 1 }
        entry[1]
      end
    end
  end

  %i[alive? default_port join monitor unmonitor value].each do |method|
    next if respond_to?(method)
    class_eval <<~RUBY, __FILE__, __LINE__ + 1
      def self.#{method}(ractor, ...) = ractor.#{method}(...)
    RUBY
  end

  unless respond_to?(:receive)
    def self.receive(ractor_or_port, timeout: nil)
      ractor_or_port = default_port(ractor_or_port) if ractor_or_port.is_a?(Ractor)
      timeout.nil? ? ractor_or_port.receive : ractor_or_port.receive(timeout:)
    end
  end

  def self.select(...) = Farce::Ractor.select(...)

  # Installs the Farce backports for Ractor and Ractor::Port globally.
  def self.install
    Extension.install do
      if Internal.native_ractors?
        ::Ractor.const_set(:Port, Farce::Ractor::Port) unless Internal.native_ports?
      else
        Object.const_set(:Ractor, Farce::Ractor)
      end
    end
  end
end

.monitor(ractor, port) ⇒ Boolean

Backport for Ractor#monitor

Examples:

using Farce::Backports
ractor = Ractor.new { raise "I'm not happy!" }
port   = Farce::Ractor::Port.new
ractor.monitor(port)
port.receive # => :aborted

Parameters:

  • ractor (Ractor) —

    the ractor to monitor

  • port (Ractor::Port) —

    the port to monitor the ractor with

Returns:

  • (Boolean) —

    whether the Ractor was still running when the monitor was set up

Raises:



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
174
175
176
177
178
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
204
205
206
207
208
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
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
 
# File 'lib/farce/backports.rb', line 136

module Backports
  # @!visibility private
  Extension = Farce::Extension.new
  private_constant :Extension
  include Extension

  if Internal.native_ractors?
    if needs_timeouts = ::Ractor.instance_method(:receive).arity.zero?
      Extension.define(::Ractor) do
        def receive(timeout: nil)
          return super() if timeout.nil?
          Internal::RactorSelector.current.ractor_receive(self, timeout:)
        end
      end

      Extension.define(::Ractor.singleton_class) do
        def select(*sources, timeout: nil, **)
          # The selector helper must reach Ruby's native wait without reentering itself.
          return super(*sources, **) if timeout.nil? && (Internal.native_ports? || sources.all?(::Ractor))
          Farce::Ractor.select(*sources, timeout:, **)
        end
      end

      def self.receive(ractor_or_port, timeout: nil)
        ractor_or_port = default_port(ractor_or_port) if ractor_or_port.is_a?(Ractor)
        return ractor_or_port.receive if timeout.nil?
        return ractor_or_port.receive(timeout:) unless Internal.native_ports?
        Internal::RactorSelector.current.ractor_receive(ractor_or_port, timeout:)
      end
    end

    if Internal.native_ports?
      unless ::Ractor.method_defined?(:alive?)
        Extension.define(::Ractor) { def alive? = Backports.alive?(self) }

        def self.alive?(ractor)
          ractor.monitor(NULL_PORT)
        ensure
          ractor.unmonitor(NULL_PORT)
        end
      end

      if needs_timeouts
        Extension.define(::Ractor::Port) do
          def receive(timeout: nil)
            return super() if timeout.nil?
            Internal::RactorSelector.current.ractor_receive(self, timeout:)
          end
        end
      end
    else
      Extension.define(::Ractor) do
        def alive?          = Backports.alive?(self)
        def default_port    = Backports.default_port(self)
        def join            = Backports.join(self)
        def monitor(port)   = Backports.monitor(self, port)
        def unmonitor(port) = Backports.unmonitor(self, port)
        def value           = Backports.value(self)
      end

      def self.alive?(ractor)
        return true if Ractor.main == ractor
        state = Internal::RactorState.for(ractor)
        return state.alive? if state
        !ractor.inspect.end_with?("terminated>")
      end

      def self.default_port(ractor) = Farce::Ractor::Port.new(ractor)

      def self.join(ractor)
        Internal::RactorState.for(ractor)&.join
        ractor
      end

      def self.monitor(ractor, port)
        return true if Ractor.main == ractor

        state = Internal::RactorState.for(ractor)
        return state.monitor(port) if state

        raise Ractor::Error, "#{ractor.inspect} cannot be monitored" unless ractor.inspect.end_with?("terminated>")

        success = true

        begin
          ractor.take
        rescue Ractor::ClosedError
          # no-op
        rescue Ractor::RemoteError
          success = false
        end

        port.send(success ? :exited : :aborted)
        false
      end

      def self.unmonitor(ractor, port)
        Internal::RactorState.for(ractor)&.unmonitor(port)
        ractor
      end

      def self.value(ractor)
        Internal::RactorState.for(ractor)&.join
        values = ::Ractor.store_if_absent("Ractor#value") { ObjectSpace::WeakKeyMap.new }
        entry  = values[ractor] ||= [Mutex.new]
        entry.first.synchronize { entry << ractor.take if entry.size == 1 }
        entry[1]
      end
    end
  end

  %i[alive? default_port join monitor unmonitor value].each do |method|
    next if respond_to?(method)
    class_eval <<~RUBY, __FILE__, __LINE__ + 1
      def self.#{method}(ractor, ...) = ractor.#{method}(...)
    RUBY
  end

  unless respond_to?(:receive)
    def self.receive(ractor_or_port, timeout: nil)
      ractor_or_port = default_port(ractor_or_port) if ractor_or_port.is_a?(Ractor)
      timeout.nil? ? ractor_or_port.receive : ractor_or_port.receive(timeout:)
    end
  end

  def self.select(...) = Farce::Ractor.select(...)

  # Installs the Farce backports for Ractor and Ractor::Port globally.
  def self.install
    Extension.install do
      if Internal.native_ractors?
        ::Ractor.const_set(:Port, Farce::Ractor::Port) unless Internal.native_ports?
      else
        Object.const_set(:Ractor, Farce::Ractor)
      end
    end
  end
end

.receive(ractor_or_port, timeout: nil) ⇒ Object

Backport for the optional timeout parameter of Ractor#receive and Ractor::Port#receive

Examples:

using Farce::Backports
port = Farce::Ractor::Port.new
port.send("message")
Farce::Backports.receive(port, timeout: 0.1) # => "message"

Parameters:

  • ractor_or_port (Ractor, Ractor::Port) —

    the ractor or port to receive from

  • timeout (Numeric, nil) (defaults to: nil) —

    the timeout for receiving

Returns:

  • (Object) —

    the received value



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
174
175
176
177
178
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
204
205
206
207
208
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
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
 
# File 'lib/farce/backports.rb', line 136

module Backports
  # @!visibility private
  Extension = Farce::Extension.new
  private_constant :Extension
  include Extension

  if Internal.native_ractors?
    if needs_timeouts = ::Ractor.instance_method(:receive).arity.zero?
      Extension.define(::Ractor) do
        def receive(timeout: nil)
          return super() if timeout.nil?
          Internal::RactorSelector.current.ractor_receive(self, timeout:)
        end
      end

      Extension.define(::Ractor.singleton_class) do
        def select(*sources, timeout: nil, **)
          # The selector helper must reach Ruby's native wait without reentering itself.
          return super(*sources, **) if timeout.nil? && (Internal.native_ports? || sources.all?(::Ractor))
          Farce::Ractor.select(*sources, timeout:, **)
        end
      end

      def self.receive(ractor_or_port, timeout: nil)
        ractor_or_port = default_port(ractor_or_port) if ractor_or_port.is_a?(Ractor)
        return ractor_or_port.receive if timeout.nil?
        return ractor_or_port.receive(timeout:) unless Internal.native_ports?
        Internal::RactorSelector.current.ractor_receive(ractor_or_port, timeout:)
      end
    end

    if Internal.native_ports?
      unless ::Ractor.method_defined?(:alive?)
        Extension.define(::Ractor) { def alive? = Backports.alive?(self) }

        def self.alive?(ractor)
          ractor.monitor(NULL_PORT)
        ensure
          ractor.unmonitor(NULL_PORT)
        end
      end

      if needs_timeouts
        Extension.define(::Ractor::Port) do
          def receive(timeout: nil)
            return super() if timeout.nil?
            Internal::RactorSelector.current.ractor_receive(self, timeout:)
          end
        end
      end
    else
      Extension.define(::Ractor) do
        def alive?          = Backports.alive?(self)
        def default_port    = Backports.default_port(self)
        def join            = Backports.join(self)
        def monitor(port)   = Backports.monitor(self, port)
        def unmonitor(port) = Backports.unmonitor(self, port)
        def value           = Backports.value(self)
      end

      def self.alive?(ractor)
        return true if Ractor.main == ractor
        state = Internal::RactorState.for(ractor)
        return state.alive? if state
        !ractor.inspect.end_with?("terminated>")
      end

      def self.default_port(ractor) = Farce::Ractor::Port.new(ractor)

      def self.join(ractor)
        Internal::RactorState.for(ractor)&.join
        ractor
      end

      def self.monitor(ractor, port)
        return true if Ractor.main == ractor

        state = Internal::RactorState.for(ractor)
        return state.monitor(port) if state

        raise Ractor::Error, "#{ractor.inspect} cannot be monitored" unless ractor.inspect.end_with?("terminated>")

        success = true

        begin
          ractor.take
        rescue Ractor::ClosedError
          # no-op
        rescue Ractor::RemoteError
          success = false
        end

        port.send(success ? :exited : :aborted)
        false
      end

      def self.unmonitor(ractor, port)
        Internal::RactorState.for(ractor)&.unmonitor(port)
        ractor
      end

      def self.value(ractor)
        Internal::RactorState.for(ractor)&.join
        values = ::Ractor.store_if_absent("Ractor#value") { ObjectSpace::WeakKeyMap.new }
        entry  = values[ractor] ||= [Mutex.new]
        entry.first.synchronize { entry << ractor.take if entry.size == 1 }
        entry[1]
      end
    end
  end

  %i[alive? default_port join monitor unmonitor value].each do |method|
    next if respond_to?(method)
    class_eval <<~RUBY, __FILE__, __LINE__ + 1
      def self.#{method}(ractor, ...) = ractor.#{method}(...)
    RUBY
  end

  unless respond_to?(:receive)
    def self.receive(ractor_or_port, timeout: nil)
      ractor_or_port = default_port(ractor_or_port) if ractor_or_port.is_a?(Ractor)
      timeout.nil? ? ractor_or_port.receive : ractor_or_port.receive(timeout:)
    end
  end

  def self.select(...) = Farce::Ractor.select(...)

  # Installs the Farce backports for Ractor and Ractor::Port globally.
  def self.install
    Extension.install do
      if Internal.native_ractors?
        ::Ractor.const_set(:Port, Farce::Ractor::Port) unless Internal.native_ports?
      else
        Object.const_set(:Ractor, Farce::Ractor)
      end
    end
  end
end

.select ⇒ BasicObject

Backport for the optional timeout parameter of Ractor.select



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
174
175
176
177
178
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
204
205
206
207
208
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
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
 
# File 'lib/farce/backports.rb', line 136

module Backports
  # @!visibility private
  Extension = Farce::Extension.new
  private_constant :Extension
  include Extension

  if Internal.native_ractors?
    if needs_timeouts = ::Ractor.instance_method(:receive).arity.zero?
      Extension.define(::Ractor) do
        def receive(timeout: nil)
          return super() if timeout.nil?
          Internal::RactorSelector.current.ractor_receive(self, timeout:)
        end
      end

      Extension.define(::Ractor.singleton_class) do
        def select(*sources, timeout: nil, **)
          # The selector helper must reach Ruby's native wait without reentering itself.
          return super(*sources, **) if timeout.nil? && (Internal.native_ports? || sources.all?(::Ractor))
          Farce::Ractor.select(*sources, timeout:, **)
        end
      end

      def self.receive(ractor_or_port, timeout: nil)
        ractor_or_port = default_port(ractor_or_port) if ractor_or_port.is_a?(Ractor)
        return ractor_or_port.receive if timeout.nil?
        return ractor_or_port.receive(timeout:) unless Internal.native_ports?
        Internal::RactorSelector.current.ractor_receive(ractor_or_port, timeout:)
      end
    end

    if Internal.native_ports?
      unless ::Ractor.method_defined?(:alive?)
        Extension.define(::Ractor) { def alive? = Backports.alive?(self) }

        def self.alive?(ractor)
          ractor.monitor(NULL_PORT)
        ensure
          ractor.unmonitor(NULL_PORT)
        end
      end

      if needs_timeouts
        Extension.define(::Ractor::Port) do
          def receive(timeout: nil)
            return super() if timeout.nil?
            Internal::RactorSelector.current.ractor_receive(self, timeout:)
          end
        end
      end
    else
      Extension.define(::Ractor) do
        def alive?          = Backports.alive?(self)
        def default_port    = Backports.default_port(self)
        def join            = Backports.join(self)
        def monitor(port)   = Backports.monitor(self, port)
        def unmonitor(port) = Backports.unmonitor(self, port)
        def value           = Backports.value(self)
      end

      def self.alive?(ractor)
        return true if Ractor.main == ractor
        state = Internal::RactorState.for(ractor)
        return state.alive? if state
        !ractor.inspect.end_with?("terminated>")
      end

      def self.default_port(ractor) = Farce::Ractor::Port.new(ractor)

      def self.join(ractor)
        Internal::RactorState.for(ractor)&.join
        ractor
      end

      def self.monitor(ractor, port)
        return true if Ractor.main == ractor

        state = Internal::RactorState.for(ractor)
        return state.monitor(port) if state

        raise Ractor::Error, "#{ractor.inspect} cannot be monitored" unless ractor.inspect.end_with?("terminated>")

        success = true

        begin
          ractor.take
        rescue Ractor::ClosedError
          # no-op
        rescue Ractor::RemoteError
          success = false
        end

        port.send(success ? :exited : :aborted)
        false
      end

      def self.unmonitor(ractor, port)
        Internal::RactorState.for(ractor)&.unmonitor(port)
        ractor
      end

      def self.value(ractor)
        Internal::RactorState.for(ractor)&.join
        values = ::Ractor.store_if_absent("Ractor#value") { ObjectSpace::WeakKeyMap.new }
        entry  = values[ractor] ||= [Mutex.new]
        entry.first.synchronize { entry << ractor.take if entry.size == 1 }
        entry[1]
      end
    end
  end

  %i[alive? default_port join monitor unmonitor value].each do |method|
    next if respond_to?(method)
    class_eval <<~RUBY, __FILE__, __LINE__ + 1
      def self.#{method}(ractor, ...) = ractor.#{method}(...)
    RUBY
  end

  unless respond_to?(:receive)
    def self.receive(ractor_or_port, timeout: nil)
      ractor_or_port = default_port(ractor_or_port) if ractor_or_port.is_a?(Ractor)
      timeout.nil? ? ractor_or_port.receive : ractor_or_port.receive(timeout:)
    end
  end

  def self.select(...) = Farce::Ractor.select(...)

  # Installs the Farce backports for Ractor and Ractor::Port globally.
  def self.install
    Extension.install do
      if Internal.native_ractors?
        ::Ractor.const_set(:Port, Farce::Ractor::Port) unless Internal.native_ports?
      else
        Object.const_set(:Ractor, Farce::Ractor)
      end
    end
  end
end

.unmonitor(ractor, port) ⇒ Ractor

Backport for Ractor#unmonitor

Examples:

using Farce::Backports
ractor = Ractor.new { sleep 10; raise "I'm not happy!" }
port   = Farce::Ractor::Port.new
ractor.monitor(port)
ractor.unmonitor(port)
port.receive # blocks forever

Parameters:

  • ractor (Ractor) —

    the ractor to unmonitor

  • port (Ractor::Port) —

    the port to unmonitor the ractor with

Returns:

  • (Ractor) —

    the unmonitored ractor



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
174
175
176
177
178
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
204
205
206
207
208
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
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
 
# File 'lib/farce/backports.rb', line 136

module Backports
  # @!visibility private
  Extension = Farce::Extension.new
  private_constant :Extension
  include Extension

  if Internal.native_ractors?
    if needs_timeouts = ::Ractor.instance_method(:receive).arity.zero?
      Extension.define(::Ractor) do
        def receive(timeout: nil)
          return super() if timeout.nil?
          Internal::RactorSelector.current.ractor_receive(self, timeout:)
        end
      end

      Extension.define(::Ractor.singleton_class) do
        def select(*sources, timeout: nil, **)
          # The selector helper must reach Ruby's native wait without reentering itself.
          return super(*sources, **) if timeout.nil? && (Internal.native_ports? || sources.all?(::Ractor))
          Farce::Ractor.select(*sources, timeout:, **)
        end
      end

      def self.receive(ractor_or_port, timeout: nil)
        ractor_or_port = default_port(ractor_or_port) if ractor_or_port.is_a?(Ractor)
        return ractor_or_port.receive if timeout.nil?
        return ractor_or_port.receive(timeout:) unless Internal.native_ports?
        Internal::RactorSelector.current.ractor_receive(ractor_or_port, timeout:)
      end
    end

    if Internal.native_ports?
      unless ::Ractor.method_defined?(:alive?)
        Extension.define(::Ractor) { def alive? = Backports.alive?(self) }

        def self.alive?(ractor)
          ractor.monitor(NULL_PORT)
        ensure
          ractor.unmonitor(NULL_PORT)
        end
      end

      if needs_timeouts
        Extension.define(::Ractor::Port) do
          def receive(timeout: nil)
            return super() if timeout.nil?
            Internal::RactorSelector.current.ractor_receive(self, timeout:)
          end
        end
      end
    else
      Extension.define(::Ractor) do
        def alive?          = Backports.alive?(self)
        def default_port    = Backports.default_port(self)
        def join            = Backports.join(self)
        def monitor(port)   = Backports.monitor(self, port)
        def unmonitor(port) = Backports.unmonitor(self, port)
        def value           = Backports.value(self)
      end

      def self.alive?(ractor)
        return true if Ractor.main == ractor
        state = Internal::RactorState.for(ractor)
        return state.alive? if state
        !ractor.inspect.end_with?("terminated>")
      end

      def self.default_port(ractor) = Farce::Ractor::Port.new(ractor)

      def self.join(ractor)
        Internal::RactorState.for(ractor)&.join
        ractor
      end

      def self.monitor(ractor, port)
        return true if Ractor.main == ractor

        state = Internal::RactorState.for(ractor)
        return state.monitor(port) if state

        raise Ractor::Error, "#{ractor.inspect} cannot be monitored" unless ractor.inspect.end_with?("terminated>")

        success = true

        begin
          ractor.take
        rescue Ractor::ClosedError
          # no-op
        rescue Ractor::RemoteError
          success = false
        end

        port.send(success ? :exited : :aborted)
        false
      end

      def self.unmonitor(ractor, port)
        Internal::RactorState.for(ractor)&.unmonitor(port)
        ractor
      end

      def self.value(ractor)
        Internal::RactorState.for(ractor)&.join
        values = ::Ractor.store_if_absent("Ractor#value") { ObjectSpace::WeakKeyMap.new }
        entry  = values[ractor] ||= [Mutex.new]
        entry.first.synchronize { entry << ractor.take if entry.size == 1 }
        entry[1]
      end
    end
  end

  %i[alive? default_port join monitor unmonitor value].each do |method|
    next if respond_to?(method)
    class_eval <<~RUBY, __FILE__, __LINE__ + 1
      def self.#{method}(ractor, ...) = ractor.#{method}(...)
    RUBY
  end

  unless respond_to?(:receive)
    def self.receive(ractor_or_port, timeout: nil)
      ractor_or_port = default_port(ractor_or_port) if ractor_or_port.is_a?(Ractor)
      timeout.nil? ? ractor_or_port.receive : ractor_or_port.receive(timeout:)
    end
  end

  def self.select(...) = Farce::Ractor.select(...)

  # Installs the Farce backports for Ractor and Ractor::Port globally.
  def self.install
    Extension.install do
      if Internal.native_ractors?
        ::Ractor.const_set(:Port, Farce::Ractor::Port) unless Internal.native_ports?
      else
        Object.const_set(:Ractor, Farce::Ractor)
      end
    end
  end
end

.value(ractor) ⇒ Object

Backport for Ractor#value

Examples:

using Farce::Backports
ractor = Ractor.new { 42 }
ractor.value # => 42

Parameters:

  • ractor (Ractor) —

    the ractor to obtain the value from

Returns:

  • (Object) —

    the value of the given ractor



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
174
175
176
177
178
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
204
205
206
207
208
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
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
 
# File 'lib/farce/backports.rb', line 136

module Backports
  # @!visibility private
  Extension = Farce::Extension.new
  private_constant :Extension
  include Extension

  if Internal.native_ractors?
    if needs_timeouts = ::Ractor.instance_method(:receive).arity.zero?
      Extension.define(::Ractor) do
        def receive(timeout: nil)
          return super() if timeout.nil?
          Internal::RactorSelector.current.ractor_receive(self, timeout:)
        end
      end

      Extension.define(::Ractor.singleton_class) do
        def select(*sources, timeout: nil, **)
          # The selector helper must reach Ruby's native wait without reentering itself.
          return super(*sources, **) if timeout.nil? && (Internal.native_ports? || sources.all?(::Ractor))
          Farce::Ractor.select(*sources, timeout:, **)
        end
      end

      def self.receive(ractor_or_port, timeout: nil)
        ractor_or_port = default_port(ractor_or_port) if ractor_or_port.is_a?(Ractor)
        return ractor_or_port.receive if timeout.nil?
        return ractor_or_port.receive(timeout:) unless Internal.native_ports?
        Internal::RactorSelector.current.ractor_receive(ractor_or_port, timeout:)
      end
    end

    if Internal.native_ports?
      unless ::Ractor.method_defined?(:alive?)
        Extension.define(::Ractor) { def alive? = Backports.alive?(self) }

        def self.alive?(ractor)
          ractor.monitor(NULL_PORT)
        ensure
          ractor.unmonitor(NULL_PORT)
        end
      end

      if needs_timeouts
        Extension.define(::Ractor::Port) do
          def receive(timeout: nil)
            return super() if timeout.nil?
            Internal::RactorSelector.current.ractor_receive(self, timeout:)
          end
        end
      end
    else
      Extension.define(::Ractor) do
        def alive?          = Backports.alive?(self)
        def default_port    = Backports.default_port(self)
        def join            = Backports.join(self)
        def monitor(port)   = Backports.monitor(self, port)
        def unmonitor(port) = Backports.unmonitor(self, port)
        def value           = Backports.value(self)
      end

      def self.alive?(ractor)
        return true if Ractor.main == ractor
        state = Internal::RactorState.for(ractor)
        return state.alive? if state
        !ractor.inspect.end_with?("terminated>")
      end

      def self.default_port(ractor) = Farce::Ractor::Port.new(ractor)

      def self.join(ractor)
        Internal::RactorState.for(ractor)&.join
        ractor
      end

      def self.monitor(ractor, port)
        return true if Ractor.main == ractor

        state = Internal::RactorState.for(ractor)
        return state.monitor(port) if state

        raise Ractor::Error, "#{ractor.inspect} cannot be monitored" unless ractor.inspect.end_with?("terminated>")

        success = true

        begin
          ractor.take
        rescue Ractor::ClosedError
          # no-op
        rescue Ractor::RemoteError
          success = false
        end

        port.send(success ? :exited : :aborted)
        false
      end

      def self.unmonitor(ractor, port)
        Internal::RactorState.for(ractor)&.unmonitor(port)
        ractor
      end

      def self.value(ractor)
        Internal::RactorState.for(ractor)&.join
        values = ::Ractor.store_if_absent("Ractor#value") { ObjectSpace::WeakKeyMap.new }
        entry  = values[ractor] ||= [Mutex.new]
        entry.first.synchronize { entry << ractor.take if entry.size == 1 }
        entry[1]
      end
    end
  end

  %i[alive? default_port join monitor unmonitor value].each do |method|
    next if respond_to?(method)
    class_eval <<~RUBY, __FILE__, __LINE__ + 1
      def self.#{method}(ractor, ...) = ractor.#{method}(...)
    RUBY
  end

  unless respond_to?(:receive)
    def self.receive(ractor_or_port, timeout: nil)
      ractor_or_port = default_port(ractor_or_port) if ractor_or_port.is_a?(Ractor)
      timeout.nil? ? ractor_or_port.receive : ractor_or_port.receive(timeout:)
    end
  end

  def self.select(...) = Farce::Ractor.select(...)

  # Installs the Farce backports for Ractor and Ractor::Port globally.
  def self.install
    Extension.install do
      if Internal.native_ractors?
        ::Ractor.const_set(:Port, Farce::Ractor::Port) unless Internal.native_ports?
      else
        Object.const_set(:Ractor, Farce::Ractor)
      end
    end
  end
end