class

Fiber::ExecutionContext::Parallel

Inherits Fiber::ExecutionContext / Reference / Object

Parallel execution context.

Fibers running in the same context run both concurrently and in parallel to each others, in addition to the other fibers running in other execution contexts.

The context internally keeps a number of fiber schedulers, each scheduler runs on a system thread, so multiple schedulers can run in parallel. The fibers are resumable by any scheduler in the context, and can thus move from one system thread to another at any time.

The actual parallelism is dynamic. As the need for parallelism increases, for example more fibers running longer, the more schedulers will start (and thus system threads), as the need decreases, for example not enough fibers, the schedulers will pause themselves and parallelism will decrease.

The parallelism can be as low as 1, in which case the context becomes a concurrent context (no parallelism) until resized.

For example: we can start a parallel context to run consumer fibers, while the default context produces values. Because the consumer fibers can run in parallel, we must protect accesses to the shared value variable. Running the example without Atomic#add would produce a different result every time!

require "wait_group"

consumers = Fiber::ExecutionContext::Parallel.new("consumers", 8)
channel = Channel(Int32).new(64)
wg = WaitGroup.new(32)

result = Atomic.new(0)

32.times do
  consumers.spawn do
    while value = channel.receive?
      result.add(value)
    end
  ensure
    wg.done
  end
end

1024.times { |i| channel.send(i) }
channel.close

# wait for all workers to be done
wg.wait

p result.get # => 523776

NOTE: The Parallel execution context isn't tied to a fixed set of system threads, and execution can switch to other system threads, for example when a fiber is blocked on a syscall.

Constructors

new(name : String, maximum : Int32) : self

Starts a Parallel context with a maximum parallelism. The context starts with an initial parallelism of zero. It will grow to one when a fiber is spawned, then the actual parallelism will keep increasing and decreasing as needed, but will never go past the configured maximum.

Source
new(name : String, size : Range(Nil, Int32)) : self

DEPRECATED Use Fiber::ExecutionContext::Parallel.new(String, Int32) instead.

Source
new(name : String, size : Range(Int32, Int32)) : self

DEPRECATED Use Fiber::ExecutionContext::Parallel.new(String, Int32) instead.

Source

Instance methods

capacity

The maximum number of schedulers that can be started, aka how many fibers can run in parallel or maximum parallelism of the context.

Source
inspect(io : IO) : Nil

Appends a String representation of this object which includes its class name, its object address and the values of all instance variables.

class Person
  def initialize(@name : String, @age : Int32)
  end
end

Person.new("John", 32).inspect # => #<Person:0x10fd31f20 @name="John", @age=32>
Source
name
Source
resize(maximum : Int32) : Nil

Resizes the context to the new maximum parallelism.

The new maximum can grow, in which case more schedulers are created to eventually increase the parallelism.

The new maximum can also shrink, in which case the overflow schedulers are removed and told to shutdown immediately. The actual shutdown is cooperative, so running schedulers won't stop until their current fiber tries to switch to another fiber.

Source
to_s(io : IO) : Nil

Appends a short String representation of this object which includes its class name and its object address.

class Person
  def initialize(@name : String, @age : Int32)
  end
end

Person.new("John", 32).to_s # => #<Person:0x10a199f20>
Source

Nested types