[rayon] How do I convert a `ParallelIterator` into a regular `Iterator` without collecting it in a `Vec` first?

help
1

TL; DR: How do I convert a rayon::ParallelIterator into a regular Iterator? The best I found so far was to first collect it into a Vec, then create an iterator from that Vec with into_iter(), but this feels stupid.

I have a function that generates data and a second one that consume them (and write the data in a file).

pub fn write_data<It: Iterator<Item = (usize, f64)>>(data: It, filename: &str) {
    /* write data in `filename` */
}   

fn main() {
   let data = vec![1, 2, 3];
   let data = vec.into_iter().map(/* ... some transformations */);
   write_data(data, "out.txt");
}

This works perfectly fine. For performance reason, I wanted to use rayon, so I replaced into_iter() by into_par_iter().

fn main() {
   let data = vec![1, 2, 3];
   let data = vec
       .into_par_iter() // this line
       .map(/* ... some transformations */);
   write_data(data, "out.txt");
}

The compilers complains that "rayon::iter::map::Map<...> is not an iterator". The best I found so far was to first collect it into a Vec, then iterate over that Vec, but I'm sure there is a better way.

fn main() {
   let data = vec![1, 2, 3];
   let data = vec
       .into_par_iter()
       .map(/* ... some transformations */);

   // this feels stupid
   let data: Vec<_> = data.collect();
   let data = data.into_iter();

   write_data(data, "out.txt");
}

I am pretty sure that there is a way to convert a ParallelIterator into a regular Iterator, I just cannot find it in the doc:

1 Like
2

You could use for_each instead of map to send results via a channel, and use the channel's receiver as the non-parallel iterator.

1 Like
3

Turning a par_iter into a regular Iterator doesn't work in any trivial way. Think about what par_iter does in an abstract fashion: it takes some collection, spins up a bunch of threads, processes chunks of that collection across those threads, and then re-assembles the output into a new collection if necessary. How would you be able to iterate over that linearly? What would an iterator's next() method return if the collections's chunks are being processed and completed in some arbitrary order?

Even if you don't care about the order, you'll have to attach something like the aforementioned output channel in order to deal with it.

4

I'm not saying it's easy, especially if we want to minimize the amount of synchronization, and as often there is a trade-off between throughput and latency.

One way to do it is to split the work in small chunks. Each worker would consume one of those chunks, process it, return the result in a buffer. Then they would start again with another chunk until all the work have been done. One process (the coordinator) would be responsible to split the work in chunk (by reading the input iterator), then collecting and ordering the output of the workers. The ordered output could then be consumed as a regular iterator. When fully consumed, an output buffer can be either desallocated or re-used.

At long as each chunk of work takes more or less the same time, you will have more or less the same amount of output buffer than your number of worker. This means that the amount of memory required would be relatively low compared to collecting everything into a Vec. The worst case is when the first chunk of work take more time than all the other chunks combined. In that case, all the other output buffers would need to be stored, which means that it would take exactly as much memory as collecting everything in the Vec (plus some bookkeeping). To mitigate the memory consumption, and at the cost of throughput, it is possible to stop computing new chunks if there is already too much output buffer ready. Using this strategy allow to consume an infinite amount of work with a limited memory.

Each output buffer can be thread local (I don't know if the same terminology is used in Rust than in C++), which means that no synchronization is needed when the workers write data into them. However, when the work is finished, they need to signal it to the coordinator and request new work to do. This requires synchronization. The bigger the amount of work in each chunk, the better the throughput will be (as long as all worker have work to do obviously), but at the cost of increased latency. Finding the right balance is anything but trivial, and the requirements can change from project to project.

5

This topic was automatically closed 90 days after the last reply. We invite you to open a new topic if you have further questions or comments.