`Folder` API doesn't allow for internal iteration optimisation of `filter`-like adapters

[huonw]

In #631 it was found that the filter and filter_map parallel iterator adaptors can't have their Folder's changed to maximally benefit from any optimisations that their base folder has for consuming a whole iterator via consume_iter. These adaptors may consume an arbitrary number of elements before yielding anything, and so need to regularly manually check whether their base is full, or else they may get stuck unnecessarily consuming a whole split.

The Problem: filter shouldn't use naive serial adaptors

For instance, consider this potential version of filter's Folder that naively just uses (serial) iterator adaptors:

struct FilterFolder<'p, C, P: 'p> {
    base: C,
    filter_op: &'p P,
}

impl<'p, C, P, T> Folder<T> for FilterFolder<'p, C, P>
where
    C: Folder<T>,
    P: Fn(&T) -> bool + 'p,
{
    type Result = C::Result;

    fn consume(self, item: T) -> Self {
        let filter_op = self.filter_op;
        if filter_op(&item) {
            let base = self.base.consume(item);
            FilterFolder {
                base: base,
                filter_op: filter_op,
            }
        } else {
            self
        }
    }

    fn consume_iter<I>(mut self, iter: I) -> Self
    where
        I: IntoIterator<Item = T>
    {
        self.base = self.base.consume_iter(iter.into_iter().filter(self.filter_op));
        self
    }


    fn complete(self) -> Self::Result {
        self.base.complete()
    }

    fn full(&self) -> bool {
        self.base.full()
    }
}

This benefits from both any optimisations self.base has for handling whole iterators, and from internal-iteration optimisations that iter (and iter.filter(_)) has, and can be significantly faster.

However, while consume_itering, self.base.full() will only get checked for each element that the filter adaptor actually yields (that is, self.base will handle short-circuiting when it is itself full). If filter doesn't yield anything for an extended period, it may be seconds/minutes/hours between checks, which can incorrectly miss when other parallel threads have finished a task. Consider:

#[test]
fn filter_octillion() {
    let x = octillion().filter(|&x| x > OCTILLION / 2).find_any(|_| true);
    assert!(x.is_some());
}

The first split is huge and filter won't yield anything, but later splits will very quickly find an element. Using the consume_iter above, the first split won't realise that find_any has finished, and will continue to process the whole split rather than short circuiting.

Failed Fix: take_while

One way to fix this would be to limit item using self.base.full(), so that it stops attempting to process items once another thread has finished:

self.base = self.base.consume_iter(iter.into_iter().take_while(|_| !self.base.full()).filter(self.filter_op));

But this doesn't and can't work: consume_iter consumes self.base. I don't think there's any variation along these lines that can work.

Proposal: Split full into an independent object

A possible thing that would address this is allowing breaking a "is full" object out of a folder, that can be handled independently from the original object, something like:

/// A type that can check whether a Folder is full.
trait FullChecker {
    fn full(&self) -> bool;
}

trait Folder<F> {
    // no change:
    type Result;
    fn consume(...);
    fn consume_iter(...);
    fn complete(...);

    // new:
    type Full: FullChecker;
    fn full_checker(&self) -> Self::Full;
    fn full(&self) -> bool { self.full_checker().full() }
}

There could be convenience types covering the common cases (i.e. most things other than find_first/find_last):

struct NeverFull;
impl FullChecker for NeverFull { fn full(&self) -> bool { false } }

struct FlagFull<'a>(&'a AtomicBool);
impl<'a> FullChecker for FlagFull<'a> {
    fn full(&self) -> bool { self.0.load(Ordering::Relaxed) }
}

For instance:

• FilterFolder::consume_iter then becomes:

  let base_full = self.base.full_checker();
  self.base = self.base.consume_iter(iter.into_iter().take_while(|_| !base_full.full()).filter(self.filter_op));

• MapFolder (and FilterFolder) becomes something like:

  impl<...> Folder<T> for MapFolder<...> {
      // ...
  
      type Full = C::Full;
      fn full_checker(&self) -> C::Full { self.base.full_checker() }
  }

• FindFolder becomes something like

  impl<'p, ...> Folder<T> for FindFolder<'p, ...> {
      // ...
  
      type Full = FlagFull<'p>;
      fn full_checker(&self) -> FlagFull<'p> { FlagFull(self.found) }
  }

• ReduceFolder becomes something like:

  impl<...> Folder<T> for ReduceFolder<...> {
      // ...
  
      type Full = NeverFull;
      fn full_checker(&self) -> NeverFull { NeverFull }
  }

[cuviper]

Thanks for the nice write-up!