class
Fiber::ExecutionContext::Parallel
Overview
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 schedulerswill start (and thus system threads), as the need decreases, for example not enough fibers, the schedulerswill pause themselves and parallelismwill decrease.
The parallelism can be as low as 1, inwhich 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 sharedvalue variable. Running
the examplewithoutAtomic#addwould 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 TheParallel execution context isn't tied to a fixed set of system
threads, and execution can switch to other system threads, for examplewhen
a fiber is blocked on a syscall.
Included Modules
Direct Known Subclasses
Defined in:
fiber/execution_context/parallel.crfiber/execution_context/parallel/scheduler.cr
Constructors
-
.new(name : String, maximum : Int32) : self
Starts a
Parallelcontextwith amaximum parallelism. -
.new(name : String, size : Range(Nil, Int32)) : self
DEPRECATED Use Fiber::ExecutionContext::Parallel.new(String, Int32) instead.
-
.new(name : String, size : Range(Int32, Int32)) : self
DEPRECATED Use Fiber::ExecutionContext::Parallel.new(String, Int32) instead.
Instance Method Summary
-
#capacity : Int32
The maximum number of schedulers that can be started, aka how many fibers can run in parallel or maximum parallelism of the context.
-
#inspect(io : IO) : Nil
Appends a String representation of this objectwhich includes its class name, its object address and the values of all instance variables.
- #name : String
-
#resize(maximum : Int32) : Nil
Resizes the context to the newmaximum parallelism.
-
#scale_up(to size : Int32 = capacity) : Nil
Manually scales the context up tosize running schedulers now,withoutwaiting for auto-scaling.
-
#to_s(io : IO) : Nil
Appends a short String representation of this objectwhich includes its class name and its object address.
Instance methods inherited from module Fiber::ExecutionContext
spawn(*, name : String | Nil = nil, &block : -> ) : Fiber
spawn
Constructor methods inherited from module Fiber::ExecutionContext
current : ExecutionContext
current
Class methods inherited from module Fiber::ExecutionContext
current? : ExecutionContext | Nil
current?,
default : ExecutionContext::Parallel
default,
default_workers_count : Int32
default_workers_count,
each(&) : Nil
each,
thread_keepalive : Time::Span
thread_keepalive,
thread_keepalive=(thread_keepalive : Time::Span)
thread_keepalive=
Instance methods inherited from class Reference
==(other : self)==(other : JSON::Any)
==(other : YAML::Any)
==(other) ==, dup dup, hash(hasher) hash, initialize initialize, inspect(io : IO) : Nil inspect, object_id : UInt64 object_id, pretty_print(pp) : Nil pretty_print, same?(other : Reference) : Bool
same?(other : Nil) same?, to_s(io : IO) : Nil to_s
Constructor methods inherited from class Reference
new
new,
unsafe_construct(address : Pointer, *args, **opts) : self
unsafe_construct
Class methods inherited from class Reference
pre_initialize(address : Pointer)
pre_initialize
Instance methods inherited from class Object
! : Bool
!,
!=(other)
!=,
!~(other)
!~,
==(other)
==,
===(other : JSON::Any)===(other : YAML::Any)
===(other) ===, =~(other) =~, as(type : Class) as, as?(type : Class) as?, class class, dup dup, hash(hasher)
hash hash, in?(collection : Object) : Bool
in?(*values : Object) : Bool in?, inspect(io : IO) : Nil
inspect : String inspect, is_a?(type : Class) : Bool is_a?, itself itself, nil? : Bool nil?, not_nil!(message)
not_nil! not_nil!, pretty_inspect(width = 79, newline = "\n", indent = 0) : String pretty_inspect, pretty_print(pp : PrettyPrint) : Nil pretty_print, responds_to?(name : Symbol) : Bool responds_to?, tap(&) tap, to_json(io : IO) : Nil
to_json : String to_json, to_pretty_json(indent : String = " ") : String
to_pretty_json(io : IO, indent : String = " ") : Nil to_pretty_json, to_s(io : IO) : Nil
to_s : String to_s, to_yaml(io : IO) : Nil
to_yaml : String to_yaml, try(&) try, unsafe_as(type : T.class) forall T unsafe_as
Class methods inherited from class Object
from_json(string_or_io : String | IO, root : String)from_json(string_or_io : String | IO) from_json, from_yaml(string_or_io : String | IO) from_yaml
Macros inherited from class Object
class_getter(*names, &block)
class_getter,
class_getter!(*names)
class_getter!,
class_getter?(*names, &block)
class_getter?,
class_property(*names, &block)
class_property,
class_property!(*names)
class_property!,
class_property?(*names, &block)
class_property?,
class_setter(*names)
class_setter,
def_clone
def_clone,
def_equals(*fields)
def_equals,
def_equals_and_hash(*fields)
def_equals_and_hash,
def_hash(*fields)
def_hash,
delegate(*methods, to object)
delegate,
forward_missing_to(delegate)
forward_missing_to,
getter(*names, &block)
getter,
getter!(*names)
getter!,
getter?(*names, &block)
getter?,
property(*names, &block)
property,
property!(*names)
property!,
property?(*names, &block)
property?,
setter(*names)
setter
Constructor Detail
Starts aParallel contextwith amaximum parallelism. The context
startswith an initial parallelism of zero. Itwill grow to onewhen a
fiber is spawned, then the actual parallelismwill keep increasing and
decreasing as needed, butwill never go past the configuredmaximum.
DEPRECATED Use Fiber::ExecutionContext::Parallel.new(String, Int32) instead.
DEPRECATED Use Fiber::ExecutionContext::Parallel.new(String, Int32) instead.
Instance Method Detail
The maximum number of schedulers that can be started, aka how many fibers can run in parallel or maximum parallelism of the context.
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>
Resizes the context to the newmaximum parallelism.
The newmaximum can grow, inwhich case more schedulers are created to
eventually increase the parallelism through auto-scaling or manual scaling
(see#scale_up).
The newmaximum can also shrink, inwhich case the overflow schedulers are removed and told to shutdown immediately. The actual shutdown is cooperative, so running schedulerswon't stop until their current fiber tries to switch to another fiber.
Manually scales the context up tosize running schedulers now,without waiting for auto-scaling.
Wakeswaiting schedulers and starts system threads until the parallelism reachessize orcapacity,whichever is smaller (no overscaling). Only scales up: does nothing ifsize is lower or equal to the current number of running schedulers.
Schedulerswill immediately return towait if there isn't enough fibers enqueued in the context.
See#resize to increase, or shrink, the maximum parallelism of the context.