Module: Concurrent

Run a block that reads and writes TVars as a single atomic transaction. With respect to the value of TVar objects, the transaction is atomic, in that it either happens or it does not, consistent, in that the TVar objects involved will never enter an illegal state, and isolated, in that transactions never interfere with each other. You may recognise these properties from database transactions.

There are some very important and unusual semantics that you must be aware of:

• Most importantly, the block that you pass to atomically may be executed more than once. In most cases your code should be free of side-effects, except for via TVar.

• If an exception escapes an atomically block it will abort the transaction.

• It is undefined behaviour to use callcc or Fiber with atomically.

• If you create a new thread within an atomically, it will not be part of the transaction. Creating a thread counts as a side-effect.

Transactions within transactions are flattened to a single transaction.

Number of processors cores available for process scheduling. This method takes in account the CPU quota if the process is inside a cgroup with a dedicated CPU quota (typically Docker). Otherwise it returns the same value as #processor_count but as a Float.

For performance reasons the calculated value will be memoized on the first call.

The maximum number of processors cores available for process scheduling. Returns nil if there is no enforced limit, or a Float if the process is inside a cgroup with a dedicated CPU quota (typically Docker).

Note that nothing prevents setting a CPU quota higher than the actual number of cores on the system.

For performance reasons the calculated value will be memoized on the first call.

The CPU shares requested by the process. For performance reasons the calculated value will be memoized on the first call.

Create a simple logger with provided level and output.

Create a stdlib logger with provided level and output. If you use this deprecated method you might need to add logger to your Gemfile to avoid warnings from Ruby 3.3.5+.

Dataflow allows you to create a task that will be scheduled when all of its data dependencies are available. Data dependencies are Future values. The dataflow task itself is also a Future value, so you can build up a graph of these tasks, each of which is run when all the data and other tasks it depends on are available or completed.

Our syntax is somewhat related to that of Akka's flow and Habanero Java's DataDrivenFuture. However unlike Akka we don't schedule a task at all until it is ready to run, and unlike Habanero Java we pass the data values into the task instead of dereferencing them again in the task.

The theory of dataflow goes back to the 70s. In the terminology of the literature, our implementation is coarse-grained, in that each task can be many instructions, and dynamic in that you can create more tasks within other tasks.

Example

A dataflow task is created with the dataflow method, passing in a block.

task = Concurrent::dataflow { 14 } 

This produces a simple Future value. The task will run immediately, as it has no dependencies. We can also specify Future values that must be available before a task will run. When we do this we get the value of those futures passed to our block.

a = Concurrent::dataflow { 1 } b = Concurrent::dataflow { 2 } c = Concurrent::dataflow(a, b) { |av, bv| av + bv } 

Using the dataflow method you can build up a directed acyclic graph (DAG) of tasks that depend on each other, and have the tasks run as soon as their dependencies are ready and there is CPU capacity to schedule them. This can help you create a program that uses more of the CPU resources available to you.

Derivation

This section describes how we could derive dataflow from other primitives in this library.

Consider a naive fibonacci calculator.

def fib(n) if n < 2 n else fib(n - 1) + fib(n - 2) end end puts fib(14) #=> 377 

We could modify this to use futures.

def fib(n) if n < 2 Concurrent::Future.new { n } else n1 = fib(n - 1).execute n2 = fib(n - 2).execute Concurrent::Future.new { n1.value + n2.value } end end f = fib(14) #=> #f.execute #=> # sleep(0.5) puts f.value #=> 377 

One of the drawbacks of this approach is that all the futures start, and then most of them immediately block on their dependencies. We know that there's no point executing those futures until their dependencies are ready, so let's not execute each future until all their dependencies are ready.

To do this we'll create an object that counts the number of times it observes a future finishing before it does something - and for us that something will be to execute the next future.

class CountingObserver def initialize(count, &block) @count = count @block = block end def update(time, value, reason) @count -= 1 if @count <= 0 @block.call() end end end def fib(n) if n < 2 Concurrent::Future.new { n }.execute else n1 = fib(n - 1) n2 = fib(n - 2) result = Concurrent::Future.new { n1.value + n2.value } barrier = CountingObserver.new(2) { result.execute } n1.add_observer barrier n2.add_observer barrier n1.execute n2.execute result end end 

We can wrap this up in a dataflow utility.

f = fib(14) #=> #sleep(0.5) puts f.value #=> 377  def dataflow(*inputs, &block) result = Concurrent::Future.new(&block) if inputs.empty? result.execute else barrier = CountingObserver.new(inputs.size) { result.execute } inputs.each do |input| input.add_observer barrier end end result end def fib(n) if n < 2 dataflow { n } else n1 = fib(n - 1) n2 = fib(n - 2) dataflow(n1, n2) { n1.value + n2.value } end end f = fib(14) #=> #sleep(0.5) puts f.value #=> 377 

Since we know that the futures the dataflow computation depends on are already going to be available when the future is executed, we might as well pass the values into the block so we don't have to reference the futures inside the block. This allows us to write the dataflow block as straight non-concurrent code without reference to futures.

def dataflow(*inputs, &block) result = Concurrent::Future.new do values = inputs.map { |input| input.value } block.call(*values) end if inputs.empty? result.execute else barrier = CountingObserver.new(inputs.size) { result.execute } inputs.each do |input| input.add_observer barrier end end result end def fib(n) if n < 2 Concurrent::dataflow { n } else n1 = fib(n - 1) n2 = fib(n - 2) Concurrent::dataflow(n1, n2) { |v1, v2| v1 + v2 } end end f = fib(14) #=> #sleep(0.5) puts f.value #=> 377 

Disables AtExit handlers including pool auto-termination handlers. When disabled it will be the application programmer's responsibility to ensure that the handlers are shutdown properly prior to application exit by calling AtExit.run method.

General access point to global executors.

Global thread pool optimized for short, fast operations.

Global thread pool optimized for long, blocking (IO) tasks.

Global thread pool user for global timers.

Leave a transaction without committing or aborting - see Concurrent::atomically.

Returns the current time as tracked by the application monotonic clock.

Number of physical processor cores on the current system. For performance reasons the calculated value will be memoized on the first call.

On Windows the Win32 API will be queried for the NumberOfCores from Win32_Processor. This will return the total number "of cores for the current instance of the processor." On Unix-like operating systems either the hwprefs or sysctl utility will be called in a subshell and the returned value will be used. In the rare case where none of these methods work or an exception is raised the function will simply return 1.

Number of processors seen by the OS and used for process scheduling. For performance reasons the calculated value will be memoized on the first call.

When running under JRuby the Java runtime call java.lang.Runtime.getRuntime.availableProcessors will be used. According to the Java documentation this "value may change during a particular invocation of the virtual machine... [applications] should therefore occasionally poll this property." We still memoize this value once under JRuby.

Otherwise Ruby's Etc.nprocessors will be used.

Use logger created by #create_simple_logger to log concurrent-ruby messages.

Use logger created by #create_stdlib_logger to log concurrent-ruby messages.

Waits for another thread to arrive at this exchange point (unless the current thread is interrupted), and then transfers the given object to it, receiving its object in return. The timeout value indicates the approximate number of seconds the method should block while waiting for the exchange. When the timeout value is nil the method will block indefinitely.

On timeout a TimeoutError exception will be raised.

Create a new thread pool.

Waits for another thread to arrive at this exchange point (unless the current thread is interrupted), and then transfers the given object to it, receiving its object in return. The timeout value indicates the approximate number of seconds the method should block while waiting for the exchange. When the timeout value is nil the method will block indefinitely.

The return value will be a Maybe set to Just on success or Nothing on timeout.