rayon Documentation

Technical Documentation: rayon-rs/rayon

ℹ️ Provenance: Multi-Tier Ingestion · CodeWiki Reference

Overview (`README.md`) Copy Section

Rayon

Rayon is a data-parallelism library for Rust. It is extremely
lightweight and makes it easy to convert a sequential computation into
a parallel one. It also guarantees data-race freedom. (You may also
enjoy this blog post about Rayon, which gives more background
and details about how it works, or this video, from the Rust
Belt Rust conference.) Rayon is
available on crates.io, and
API documentation is available on docs.rs.

Parallel iterators and more Copy Chapter

Rayon makes it drop-dead simple to convert sequential iterators into
parallel ones: usually, you just change your foo.iter() call into
foo.par_iter(), and Rayon does the rest:

use rayon::prelude::*;
fn sum_of_squares(input: &[i32]) -> i32 {
    input.par_iter() // <-- just change that!
         .map(|&i| i * i)
         .sum()
}

Parallel iterators take care of deciding how to divide your data
into tasks; it will dynamically adapt for maximum performance. If you
need more flexibility than that, Rayon also offers the join and
scope functions, which let you create parallel tasks on your own.
For even more control, you can create custom thread pools rather than
using Rayon's default, global thread pool.

No data races Copy Chapter

You may have heard that parallel execution can produce all kinds of
crazy bugs. Well, rest easy. Rayon's APIs all guarantee data-race
freedom, which generally rules out most parallel bugs (though not
all). In other words, if your code compiles, it typically does the
same thing it did before.

For the most part, parallel iterators in particular are guaranteed to
produce the same results as their sequential counterparts. One caveat:
If your iterator has side effects (for example, sending methods to
other threads through a Rust channel or writing to disk), those side
effects may occur in a different order. Note also that, in some cases,
parallel iterators offer alternative versions of the sequential
iterator methods that can have higher performance.

Using Rayon Copy Chapter

Rayon is available on crates.io. The
recommended way to use it is to add a line into your Cargo.toml such
as:

[dependencies]
rayon = "1.12"

To use the parallel iterator APIs, a number of traits have to be in
scope. The easiest way to bring those things into scope is to use the
Rayon prelude. In
each module where you would like to use the parallel iterator APIs,
just add:

use rayon::prelude::*;

Rayon currently requires rustc 1.85.0 or greater.

Usage with WebAssembly Copy Section

By default, when building to WebAssembly, Rayon will treat it as any
other platform without multithreading support and will fall back to
sequential iteration. This allows existing code to compile and run
successfully with no changes necessary, but it will run slower as it
will only use a single CPU core.

You can build Rayon-based projects with proper multithreading support
for the Web, but you'll need an adapter and some project configuration
to account for differences between WebAssembly threads and threads on
the other platforms.

Check out the
wasm-bindgen-rayon
docs for more details.

Contribution Copy Chapter

Rayon is an open source project! If you'd like to contribute to Rayon,
check out
the list of "help wanted" issues.
These are all (or should be) issues that are suitable for getting
started, and they generally include a detailed set of instructions for
what to do. Please ask questions if anything is unclear! Also, check
out the
Guide to Development
page on the wiki. Note that all code submitted in PRs to Rayon is
assumed to
be licensed under Rayon's dual MIT/Apache 2.0 licensing.

Quick demo Copy Chapter

To see Rayon in action, check out the rayon-demo directory, which
includes a number of demos of code using Rayon. For example, run this
command to get a visualization of an N-body simulation. To see the
effect of using Rayon, press s to run sequentially and p to run in
parallel.

> cd rayon-demo
> cargo run --release -- nbody visualize

For more information on demos, try:

> cd rayon-demo
> cargo run --release -- --help

Other questions? Copy Chapter

See the Rayon FAQ.

License Copy Chapter

Rayon is distributed under the terms of both the MIT license and the
Apache License (Version 2.0). See LICENSE-APACHE and
LICENSE-MIT for details. Opening a pull request is
assumed to signal agreement with these licensing terms.

Overview (`rayon-core/README.md`) Copy Section

Rayon-core represents the "core, stable" APIs of Rayon: join, scope, and so forth, as well as the ability to create custom thread pools with ThreadPool.

Maybe worth mentioning: users are not necessarily intended to directly access rayon-core; all its APIs are mirrored in the rayon crate. To that end, the examples in the docs use rayon::join and so forth rather than rayon_core::join.

rayon-core aims to never, or almost never, have a breaking change to its API, because each revision of rayon-core also houses the global thread pool (and hence if you have two simultaneous versions of rayon-core, you have two thread pools).

Please see Rayon Docs for details about using Rayon.

Rayon-core currently requires rustc 1.85.0 or greater.

Overview (`rayon-demo/data/README.md`) Copy Section

Data for various benchmarks. All in public domain or otherwise freely
licensed.

Overview (`rayon-demo/examples/README.md`) Copy Section

We use this directory for interactive tests that can't be run in an
automatic fashion. For examples of how to use Rayon, or benchmarks,
see rayon-demo.

Overview (`rayon-core/src/sleep/README.md`) Copy Section

Introduction: the sleep module

The code in this module governs when worker threads should go to
sleep. The system used in this code was introduced in Rayon RFC #5.
There is also a video walkthrough available. Both of those may be
valuable resources to understanding the code, though naturally they
will also grow stale over time. The comments in this file are
extracted from the RFC and meant to be kept up to date.

The `Sleep` struct

The Sleep struct is embedded into each registry. It performs several functions:

• It tracks when workers are awake or asleep.
• It decides how long a worker should look for work before it goes to sleep,
  via a callback that is invoked periodically from the worker's search loop.
• It is notified when latches are set, jobs are published, or other
  events occur, and it will go and wake the appropriate threads if
  they are sleeping.

Thread states

There are three main thread states:

• An active thread is one that is actively executing a job.
• An idle thread is one that is searching for work to do. It will be
  trying to steal work or pop work from the global injector queue.
• A sleeping thread is one that is blocked on a condition variable,
  waiting to be awoken.

We sometimes refer to the final two states collectively as inactive.
Threads begin as idle but transition to idle and finally sleeping when
they're unable to find work to do.

Sleepy threads Copy Chapter

There is one other special state worth mentioning. During the idle state,
threads can get sleepy. A sleepy thread is still idle, in that it is still
searching for work, but it is about to go to sleep after it does one more
search (or some other number, potentially). When a thread enters the sleepy
state, it signals (via the jobs event counter, described below) that it is
about to go to sleep. If new work is published, this will lead to the counter
being adjusted. When the thread actually goes to sleep, it will (hopefully, but
not guaranteed) see that the counter has changed and elect not to sleep, but
instead to search again. See the section on the jobs event counter for more
details.

The counters

One of the key structs in the sleep module is AtomicCounters, found in
counters.rs. It packs three counters into one atomically managed value:

• Two thread counters, which track the number of threads in a particular state.
• The jobs event counter, which is used to signal when new work is available.
  It (sort of) tracks the number of jobs posted, but not quite, and it can rollover.

Thread counters Copy Chapter

There are two thread counters, one that tracks inactive threads and one that
tracks sleeping threads. From this, one can deduce the number of threads
that are idle by subtracting sleeping threads from inactive threads. We track
the counters in this way because it permits simpler atomic operations. One can
increment the number of sleeping threads (and thus decrease the number of idle
threads) simply by doing one atomic increment, for example. Similarly, one can
decrease the number of sleeping threads (and increase the number of idle
threads) through one atomic decrement.

These counters are adjusted as follows:

• When a thread enters the idle state: increment the inactive thread counter.
• When a thread enters the sleeping state: increment the sleeping thread counter.
• When a thread awakens a sleeping thread: decrement the sleeping thread counter.
  • Subtle point: the thread that awakens the sleeping thread decrements the
    counter, not the thread that is sleeping. This is because there is a delay
    between signaling a thread to wake and the thread actually waking:
    decrementing the counter when awakening the thread means that other threads
    that may be posting work will see the up-to-date value that much faster.
• When a thread finds work, exiting the idle state: decrement the inactive
  thread counter.

Jobs event counter Copy Chapter

The final counter is the jobs event counter. The role of this counter is to
help sleepy threads detect when new work is posted in a lightweight fashion. In
its simplest form, we would simply have a counter that gets incremented each
time a new job is posted. This way, when a thread gets sleepy, it could read the
counter, and then compare to see if the value has changed before it actually
goes to sleep. But this turns out to be too expensive in practice, so we use a
somewhat more complex scheme.

The idea is that the counter toggles between two states, depending on whether
its value is even or odd (or, equivalently, on the value of its low bit):

• Even -- If the low bit is zero, then it means that there has been no new work
  since the last thread got sleepy.
• Odd -- If the low bit is one, then it means that new work was posted since
  the last thread got sleepy.

New work is posted Copy Section

When new work is posted, we check the value of the counter: if it is even,
then we increment it by one, so that it becomes odd.

Worker thread gets sleepy Copy Section

When a worker thread gets sleepy, it will read the value of the counter. If the
counter is odd, it will increment the counter so that it is even. Either way, it
remembers the final value of the counter. The final value will be used later,
when the thread is going to sleep. If at that time the counter has not changed,
then we can assume no new jobs have been posted (though note the remote
possibility of rollover, discussed in detail below).

Protocol for a worker thread to post work

The full protocol for a thread to post work is as follows

• If the work is posted into the injection queue, then execute a seq-cst fence (see below).
• Load the counters, incrementing the JEC if it is even so that it is odd.
• Check if there are idle threads available to handle this new job. If not,
  and there are sleeping threads, then wake one or more threads.

Protocol for a worker thread to fall asleep

The full protocol for a thread to fall asleep is as follows:

• After completing all its jobs, the worker goes idle and begins to
  search for work. As it searches, it counts "rounds". In each round,
  it searches all other work threads' queues, plus the 'injector queue' for
  work injected from the outside. If work is found in this search, the thread
  becomes active again and hence restarts this protocol from the top.
• After a certain number of rounds, the thread "gets sleepy" and executes get_sleepy
  above, remembering the final_value of the JEC. It does one more search for work.
• If no work is found, the thread atomically:
  • Checks the JEC to see that it has not changed from final_value.
    • If it has, then the thread goes back to searching for work. We reset to
      just before we got sleepy, so that we will do one more search
      before attempting to sleep again (rather than searching for many rounds).
  • Increments the number of sleeping threads by 1.
• The thread then executes a seq-cst fence operation (see below).
• The thread then does one final check for injected jobs (see below). If any
  are available, it returns to the 'pre-sleepy' state as if the JEC had changed.
• The thread waits to be signaled. Once signaled, it returns to the idle state.

The jobs event counter and deadlock

As described in the section on the JEC, the main concern around going to sleep
is avoiding a race condition wherein:

• Thread A looks for work, finds none.
• Thread B posts work but sees no sleeping threads.
• Thread A goes to sleep.

The JEC protocol largely prevents this, but due to rollover, this prevention is
not complete. It is possible -- if unlikely -- that enough activity occurs for
Thread A to observe the same JEC value that it saw when getting sleepy. If the
new work being published came from inside the thread pool, then this race
condition isn't too harmful. It means that we have fewer workers processing the
work then we should, but we won't deadlock. This seems like an acceptable risk
given that this is unlikely in practice.

However, if the work was posted as an external job, that is a problem. In that
case, it's possible that all of our workers could go to sleep, and the external
job would never get processed. To prevent that, the sleeping protocol includes
one final check to see if the injector queue is empty before fully falling
asleep. Note that this final check occurs after the number of sleeping
threads has been incremented. We are not concerned therefore with races against
injections that occur after that increment, only before.

Unfortunately, there is one rather subtle point concerning this final check:
we wish to avoid the possibility that:

• work is pushed into the injection queue by an outside thread X,
• the sleepy thread S sees the JEC but it has rolled over and is equal
• the sleepy thread S reads the injection queue but does not see the work posted by X.

This is possible because the C++ memory model typically offers guarantees of the
form "if you see the access A, then you must see those other accesses" -- but it
doesn't guarantee that you will see the access A (i.e., if you think of
processors with independent caches, you may be operating on very out of date
cache state).

Using seq-cst fences to prevent deadlock Copy Chapter

To overcome this problem, we have inserted two sequentially consistent fence
operations into the protocols above:

• One fence occurs after work is posted into the injection queue, but before the
  counters are read (including the number of sleeping threads).
  • Note that no fence is needed for work posted to internal queues, since it is ok
    to overlook work in that case.
• One fence occurs after the number of sleeping threads is incremented, but
  before the injection queue is read.

Proof sketch Copy Section

What follows is a "proof sketch" that the protocol is deadlock free. We model
two relevant bits of memory, the job injector queue J and the atomic counters C.

Consider the actions of the injecting thread:

• PushJob: Job is injected, which can be modeled as an atomic write to J with release semantics.
• PushFence: A sequentially consistent fence is executed.
• ReadSleepers: The counters C are read (they may also be incremented, but we just consider the read that comes first).

Meanwhile, the sleepy thread does the following:

• IncSleepers: The number of sleeping threads is incremented, which is atomic exchange to C.
• SleepFence: A sequentially consistent fence is executed.
• ReadJob: We look to see if the queue is empty, which is a read of J with acquire semantics.

Either PushFence or SleepFence must come first:

• If PushFence comes first, then PushJob must be visible to ReadJob.
• If SleepFence comes first, then IncSleepers is visible to ReadSleepers.

Overview (`rayon-demo/data/tsp/README.md`) Copy Section

Inputs for the Traveling Salesman Problem solver. These are in TSPLIB
format.

Sources:

• dj10.tsp: derived from dj38.tsp
• dj15.tsp: derived from dj38.tsp
• dj38.tsp: https://www.math.uwaterloo.ca/tsp/world/dj38.tsp

Overview (`src/iter/plumbing/README.md`) Copy Section

Parallel Iterators

These are some notes on the design of the parallel iterator traits.
This file does not describe how to use parallel iterators.

The challenge Copy Chapter

Parallel iterators are more complicated than sequential iterators.
The reason is that they have to be able to split themselves up and
operate in parallel across the two halves.

The current design for parallel iterators has two distinct modes in
which they can be used; as we will see, not all iterators support both
modes (which is why there are two):

• Pull mode (the Producer and UnindexedProducer traits): in this mode,
  the iterator is asked to produce the next item using a call to next. This
  is basically like a normal iterator, but with a twist: you can split the
  iterator in half to produce disjoint items in separate threads.
  • in the Producer trait, splitting is done with split_at, which accepts
    an index where the split should be performed. Only indexed iterators can
    work in this mode, as they know exactly how much data they will produce,
    and how to locate the requested index.
  • in the UnindexedProducer trait, splitting is done with split, which
    simply requests that the producer divide itself approximately in half.
    This is useful when the exact length and/or layout is unknown, as with
    String characters, or when the length might exceed usize, as with
    Range on 32-bit platforms.
    • In theory, any Producer could act unindexed, but we don't currently
      use that possibility. When you know the exact length, a split can
      simply be implemented as split_at(length/2).
• Push mode (the Consumer and UnindexedConsumer traits): in
  this mode, the iterator instead is given each item in turn, which
  is then processed. This is the opposite of a normal iterator. It's
  more like a for_each call: each time a new item is produced, the
  consume method is called with that item. (The traits themselves are
  a bit more complex, as they support state that can be threaded
  through and ultimately reduced.) Like producers, there are two
  variants of consumers which differ in how the split is performed:
  • in the Consumer trait, splitting is done with split_at, which
    accepts an index where the split should be performed. All
    iterators can work in this mode. The resulting halves thus have an
    idea about how much data they expect to consume.
  • in the UnindexedConsumer trait, splitting is done with
    split_off_left. There is no index: the resulting halves must be
    prepared to process any amount of data, and they don't know where that
    data falls in the overall stream.
    • Not all consumers can operate in this mode. It works for
      for_each and reduce, for example, but it does not work for
      collect_into_vec, since in that case the position of each item is
      important for knowing where it ends up in the target collection.

How iterator execution proceeds Copy Chapter

We'll walk through this example iterator chain to start. This chain
demonstrates more-or-less the full complexity of what can happen.

vec1.par_iter()
    .zip(vec2.par_iter())
    .flat_map(some_function)
    .for_each(some_other_function)

To handle an iterator chain, we start by creating consumers. This
works from the end. So in this case, the call to for_each is the
final step, so it will create a ForEachConsumer that, given an item,
just calls some_other_function with that item. (ForEachConsumer is
a very simple consumer because it doesn't need to thread any state
between items at all.)

Now, the for_each call will pass this consumer to the base iterator,
which is the flat_map. It will do this by calling the drive_unindexed
method on the ParallelIterator trait. drive_unindexed basically
says "produce items for this iterator and feed them to this consumer";
it only works for unindexed consumers.

(As an aside, it is interesting that only some consumers can work in
unindexed mode, but all producers can drive an unindexed consumer.
In contrast, only some producers can drive an indexed consumer, but
all consumers can be supplied indexes. Isn't variance neat.)

As it happens, FlatMap only works with unindexed consumers anyway.
This is because flat-map basically has no idea how many items it will
produce. If you ask flat-map to produce the 22nd item, it can't do it,
at least not without some intermediate state. It doesn't know whether
processing the first item will create 1 item, 3 items, or 100;
therefore, to produce an arbitrary item, it would basically just have
to start at the beginning and execute sequentially, which is not what
we want. But for unindexed consumers, this doesn't matter, since they
don't need to know how much data they will get.

Therefore, FlatMap can wrap the ForEachConsumer with a
FlatMapConsumer that feeds to it. This FlatMapConsumer will be
given one item. It will then invoke some_function to get a parallel
iterator out. It will then ask this new parallel iterator to drive the
ForEachConsumer. The drive_unindexed method on flat_map can then
pass the FlatMapConsumer up the chain to the previous item, which is
zip. At this point, something interesting happens.

Switching from push to pull mode Copy Chapter

If you think about zip, it can't really be implemented as a
consumer, at least not without an intermediate thread and some
channels or something (or maybe coroutines). The problem is that it
has to walk two iterators in lockstep. Basically, it can't call two
drive methods simultaneously, it can only call one at a time. So at
this point, the zip iterator needs to switch from push mode into
pull mode.

You'll note that Zip is only usable if its inputs implement
IndexedParallelIterator, meaning that they can produce data starting
at random points in the stream. This need to switch to push mode is
exactly why. If we want to split a zip iterator at position 22, we
need to be able to start zipping items from index 22 right away,
without having to start from index 0.

Anyway, so at this point, the drive_unindexed method for Zip stops
creating consumers. Instead, it creates a producer, a ZipProducer,
to be exact, and calls the bridge function in the internals
module. Creating a ZipProducer will in turn create producers for
the two iterators being zipped. This is possible because they both
implement IndexedParallelIterator.

The bridge function will then connect the consumer, which is
handling the flat_map and for_each, with the producer, which is
handling the zip and its predecessors. It will split down until the
chunks seem reasonably small, then pull items from the producer and
feed them to the consumer.

The base case Copy Chapter

The other time that bridge gets used is when we bottom out in an
indexed producer, such as a slice or range. There is also a
bridge_unindexed equivalent for - you guessed it - unindexed producers,
such as string characters.

base_iter.with_producer(|base_producer| {
    // here, `base_producer` is the producer for `base_iter`
});

struct MapProducer<'f, P, F: 'f> {
    base: P,
    map_op: &'f F,
}

impl<I, F> IndexedParallelIterator for Map<I, F>
    where I: IndexedParallelIterator,
          F: MapOp<I::Item>,
{
    fn with_producer<CB>(self, callback: CB) -> CB::Output {
        let map_op = &self.map_op;
        self.base_iter.with_producer(|base_producer| {
            // Here `producer` is the producer for `self.base_iter`.
            // Wrap that to make a `MapProducer`
            let map_producer = MapProducer {
                base: base_producer,
                map_op: map_op
            };

            // invoke the callback with the wrapped version
            callback(map_producer)
        });
    }
});

pub trait IndexedParallelIterator: ParallelIterator {
    type Producer;
    fn with_producer<CB, R>(self, callback: CB) -> R
        where CB: FnOnce(Self::Producer) -> R;
    ...
}

impl<I, F> IndexedParallelIterator for Map<I, F>
    where I: IndexedParallelIterator,
          F: MapOp<I::Item>,
{
    type MapProducer = MapProducer<'f, P::Producer, F>;
    //                             ^^ wait, what is this `'f`?

    fn with_producer<CB, R>(self, callback: CB) -> R
        where CB: FnOnce(Self::Producer) -> R
    {
        let map_op = &self.map_op;
        //  ^^^^^^ `'f` is (conceptually) the lifetime of this reference,
        //         so it will be different for each call to `with_producer`!
    }
}

If we had associated type constructors, we could solve this
problem that way. But there is another solution. We can use a
dedicated callback trait like ProducerCallback, instead of FnOnce:

pub trait ProducerCallback<T> {
    type Output;
    fn callback<P>(self, producer: P) -> Self::Output
        where P: Producer<Item=T>;
}

Using this trait, the signature of with_producer() looks like this:

fn with_producer<CB: ProducerCallback<Self::Item>>(self, callback: CB) -> CB::Output;

Notice that this signature never has to name the producer type --
there is no associated type Producer anymore. This is because the
callback() method is generically over all producers P.

The problem is that now the || sugar doesn't work anymore. So we
have to manually create the callback struct, which is a mite tedious.
So our MapProducer code looks like this:

impl<I, F> IndexedParallelIterator for Map<I, F>
    where I: IndexedParallelIterator,
          F: MapOp<I::Item>,
{
    fn with_producer<CB>(self, callback: CB) -> CB::Output
        where CB: ProducerCallback<Self::Item>
    {
        return self.base.with_producer(Callback { callback: callback, map_op: self.map_op });
        //                             ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
        //                             Manual version of the closure sugar: create an instance
        //                             of a struct that implements `ProducerCallback`.

        // The struct declaration. Each field is something that need to capture from the
        // creating scope.
        struct Callback<CB, F> {
            callback: CB,
            map_op: F,
        }

        // Implement the `ProducerCallback` trait. This is pure boilerplate.
        impl<T, F, CB> ProducerCallback<T> for Callback<CB, F>
            where F: MapOp<T>,
                  CB: ProducerCallback<F::Output>
        {
            type Output = CB::Output;

            fn callback<P>(self, base: P) -> CB::Output
                where P: Producer<Item=T>
            {
                // The body of the closure is here:
                let producer = MapProducer { base: base,
                                             map_op: &self.map_op };
                self.callback.callback(producer)
            }
        }
    }
}

OK, a bit tedious, but it works!

--- METRICS ---

• Files Extracted: 7
• Estimated Token Budget: ~7888 tokens
• Canonical Reference: https://codewiki.google/github.com/rayon-rs/rayon