class Fiber::ExecutionContext::Concurrent

Overview

Concurrent-only execution context.

Fibers running in the same context can only run concurrently and never in parallel to each others. However, they still run in parallel to fibers running in other execution contexts.

Fibers in this context can use simpler and faster synchronization primitives between themselves (for example no atomics or thread safety required), but data shared with other contexts needs to be protected (e.g. Mutex), and communication with fibers in other contexts requires safe primitives, for example Channel.

A blocking fiber blocks the entire context, and thus all the other fibers in the context.

For example: we can start a concurrent context to run consumer fibers, while the default context produces values. Because the consumer fibers will never run in parallel and don't yield between reading result then writing it, we are not required to synchronize accesses to the value:

require "wait_group"

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

result = 0

32.times do
  consumers.spawn do
    while value = channel.receive?
      # safe, but only for this example:
      result = result + 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 # => 523776

In practice, we still recommended to always protect shared accesses to a variable, for example using Atomic#add to increment result or a Mutex for more complex operations.

Defined in:

fiber/execution_context/concurrent.cr

Constructors

Instance Method Summary

Instance methods inherited from class Fiber::ExecutionContext::Parallel

capacity : Int32 capacity, inspect(io : IO) : Nil inspect, name : String name, resize(maximum : Int32) : Nil resize, size : Int32 size, to_s(io : IO) : Nil to_s

Constructor methods inherited from class Fiber::ExecutionContext::Parallel

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

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, default : ExecutionContext default

Class methods inherited from module Fiber::ExecutionContext

current? : ExecutionContext | Nil current?, default_workers_count : Int32 default_workers_count, each(&) : Nil each

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

def self.default : self #

[View source]
def self.new(name : String) : self #

[View source]

Instance Method Detail

def resize(maximum : Int32) : Nil #
Description copied from class Fiber::ExecutionContext::Parallel

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.


[View source]