Flow is a lightweight C++ inter-thread flow library, built on top of the boost::thread library. It allows dispatch of parts of a complex computation process to one or more additional threads, in a transparent and idiomatic way. It has been written in early 2008 by me (Victor Nicollet), and it is available under the terms of the new BSD license.

The flow library implements thread-safe FIFO queues (flows), and provides facilities for manipulating them that are compatible with the Standard C++ Library.

Downloads

Using this library requires Boost.Threads and Boost.Optional to be installed and available to the program. To use the elements defined in the flow library, one only has to include the header file flow.hpp: since all code is templated, there is no need for separate compiling and linking.

• flow.hpp : include this file to use any entity in the library.
• prime.cpp : a naive prime number test using several cores.
• sort.cpp : a constant-time sort using an auxiliary thread.

Concepts

A flow is a FIFO queue where enqueueing and dequeueing are atomic. The type flows::flow<T> represents a flow of elements of type T.

Flows are manipulated through readers and writers, which are objects used to read from and write to the flow. Reading will remove the oldest element from the flow, while writing will append a new element to the flow, in a FIFO fashion. Attempting to read from an empty flow will block until another thread writes to the flow. Readers are represented by class flows::reader<T>, and writers are represented by class flows::writer<T>.

When all writers of a flow disappear, the flow is closed: all the readers are notified that no more data will be appended to the flow, triggering an end-of-flow state once all the remaining elements have been removed from the flow.

When all readers of a flow disappear, the flow is silent: all the writers are notified that nobody will ever read the data written to the flow, providing opportunities for optimization.

When all readers and writers of a flow disappear, that flow is automatically garbage-collected and is destroyed by the thread that destroyed the last reader or writer (along with all the elements it still contained).

Documentation

class flows::reader<T>

This class represents a reader, which may be associated to a flow (valid reader), or not be associated to any flow (invalid reader). The class itself can be summarized as:

template <typename T>
class reader
{
public:
  typedef T                  value_type;
  typedef boost::optional<T> option_type;
  typedef /* unspecified */  iterator; 

  iterator    begin();
  iterator    end();
              reader();
              reader(const reader&);
  reader&     operator =(const reader&);
  void        close();
  option_type get();
};

The default constructor creates an invalid reader. A valid reader can only be obtained from flows::flow<T>. Assigning to a reader dissociates that reader from its flow (if any), which may lead to that flow being silenced or even destroyed if this was the last reader for that flow. Note that flows::reader<T>::close() is equivalent to assigning a default-constructed reader: the reader becomes invalid and detached from its associated flow (if any).

The flows::reader<T>::get() function behaves as follows:

• If the reader is invalid, returns no value.
• If the reader is valid, but the associated flow is closed and empty, returns no value.
• If the reader is valid, but the associated flow is empty, blocks until a value becomes available or the flow is closed, and tries again.
• If the reader is valid and the associated flow is not empty, atomically removes the oldest value and returns it.

As a consequence, if two or more readers are associated to the same flow, every element in the flow will be read by at most one reader (and all elements will be either read by a reader or destroyed along with the flow).

flows::reader<T>::begin() and flows::reader<T>::end() are the first and past-the-end iterators of the read sequence of elements. They are input iterators (in the sense defined by the Standard C++Library) which represent the sequence of elements read by a particular reader:

• Creating an iterator using begin() or incrementing an iterator reads the next value from the flow by calling the flows::reader<T>::get() function on the associated reader. This function call may block until the flow is written to. If it returns a value, then the iterator is bound to that value. Otherwise, the iterator becomes equal to the past-the-end iterator.
• Dereferencing an iterator returns the value bound to the iterator. This is guaranteed to occur in constant time.

Iterators are invalidated on every assignment to the reader.

An analogy for read iterators is the std::istream_iterator, but the data is read from a flow using flows::reader<T>::get(), instead of being read from a stream using operator>>().

class flows::writer<T>

This class represents a writer, which may be associated to a flow (valid writer), or not be associated to any flow (invalid writer). The class can be summarized as:

template <typename T>
class writer
{
  typedef T                 value_type;
  typedef /* unspecified */ iterator; 

  iterator begin();
           writer();
           writer(const writer&);
  writer&  operator =(const writer&);
  void     close();
  bool     silent();
  void     put(const T&);
};

The default constructor creates an invalid writer. The only way to create a valid writer is to use flows::flow<T>. Assigning to a writer dissociates that writer from its flow (if any), and then binds it to the flow of the writer that was assigned to it (if any). This may lead to the flow becoming closed or even garbage-collected. flows::writer<T>::close() is equivalent to (but more idiomatic than) assigning a default-constructed writer.

The flows::writer<T>::put() function atomically writes a copy of its argument to the flow associated to the writer. If there is no associated flow, the behavior of this function is undefined.

The flows::writer<T>::silent() function returns whether the associated flow is silent or not. If there is no associated flow, the behavior of this function is undefined. This is provided as optimization hints to avoid computing values only to place them in a silent flow. It can be used to notify search threads that the one of them has found what was searched. See the above example on prime numbers for an illustration.

The flows::writer<T>::begin() function returns the output iterator for this writer. This iterator behaves as an std::back_insert_iterator, but calls flows::writer<T>::put() on its associated writer instead of calling insert(). This iterator is invalidated by assignment to the writer.

class flows::flow<T>

This class represents an active flow, and allows creating fresh readers and writers associated to that flow. For that reason, for as long as this instance exists, the underlying flow can be neither closed nor silenced (because a new reader or writer could be obtained from it at any time). Therefore, this instance should be used to prepare a computation, and destroyed before the computation starts.

When the instance is destroyed, the flow itself remains in memory for as long as readers or writers are associated to it. Once all readers and writers are gone, the flow is destroyed immediately, destroying any unread data left.

Note that the library provides several functions that dispatch computations to other threads without requiring the explicit creation of a flow from user code. Ideally, flows::flow<T> should only be used in situations where these functions do not meet the needs of the program, and even then it should only be used with care.

The class can be summarized as:

template <typename T> 
class flow
{
  flow(); 

  const writer<T>& input() const;
  const reader<T>& output() const;
};

As expected, flows::flow<T>::input() returns a writer associated to the flow, and flows::flow<T>::output() returns a reader associated to the flow. Unlike the naming approach of the C++ Standard Library, where an input stream is a reader and an output stream is a writer, the input of a flow is the writer (where the flow gets its input) and the output of a flow is the reader (where the flow sends its input).

Generators

A generator is a separate thread that computes values and outputs them to a flow. The original thread keeps a reader associated to that flow, and may read values from the flow as they become available.

Two functions allow the creation of generators:

template <typename T, typename F>
reader<T> generate(F f); 

template <typename T, typename F, typename I>
reader<T> generate(F f, I b, I e);

The first function creates a new flow and a new thread. Then, in that new thread, it calls f(w) where f is the argument to the generate function, and w is a writer associated to that flow. Simultaneously, it returns a reader associated to the flow in the original thread. The created thread ends when the argument function returns.

The second function creates one new flow, and creates a thread for each iterator i between b and e. Each thread calls f(w,*i), and the function simultaneously returns a reader for the flow. Each created thread ends when its function call returns.

That is, the first function creates a generator, and the second function creates one generator for each iterator in the passed sequence.

Processing

A processor is a thread which reads from a flow, does some computations, and writes to another flow. The most generic way of doing this (by specifying the exact function that reads and writes) is with the flows::process<T>() function:

template<typename T, typename U, typename F>
reader<T> process(const reader<U> &source, F f);

This function creates a new flow and a new thread which runs f(source,dest), where dest is a writer associated to the new flow. It immediately returns a reader associated to the new flow.

A special case of processing is mapping an function to a flow: the processor reads a value, applies a function to it, and writes the result to another flow. Mapping is done using the flows::transform<T>() function:

template <typename T, typename U, typename F>
reader<T> transform(const reader<U> &source, F f, unsigned n = 1);

This function creates n new threads and a single flow, and returns a reader associated to that flow. Every thread reads a value x from the source reader, and writes f(x) to the new flow. Note that if n = 1, the order is preserved, but not if n > 1. The new flow is closed when all the data from the original flow has been processed.

Accumulating

An accumulator is a thread which reads from a flow and applies a binary operation to collapse all the values into a single one, using an initial value for the first application. Once all the data is processed, it writes the final value to an output flow.

Accumulation is performed using the function flows::accumulate<T>() function:

template <typename T, typename U, typename F>
reader<T> accumulate(const reader<U> &source, T initial, F f);

This function creates a flow and thread. The thread performs the operation f(initial,*it) for it between source.begin() and source.end(), then writes initial to the flow. The function returns the reader associated to the flow immediately.

Further work

The implementation could be improved. First, it could be adapted to work with lock-free structures. Second, it could also be adapted to use worker threads instead of spawning new threads for every operation.

The interface could also use some improvement. For instance, it’s still too easy to create deadlocks by forgetting to close a writer in the wrong place. And the code does not interact with the standard library as well as it should.

Implicit Iterator Intervals

The Standard C++ Library is built around the concept of iterators. On the one hand, output iterators can be incremented and assigned values, and on the other hand input iterators traverse a range between a start iterator (usually referred to as the begin iterator) and a past-the-end iterator (the end iterator). All containers follow this convention, either through member functions begin() and end() or through pointer arithmetics based on base pointer and size. Also, several other range concepts (for instance, input from an istream)  also follow this convention.

On the other hand, all standard library algorithms work on iterator ranges, which are passed implicitly by providing the two iterators independently.

The problem with implicit iterator intervals is that they don’t play nice with functions that transform intervals: for instance, a filter function would turn an interval into the elements of that interval which satisfy a certain predicate, or a sub-interval function would restrict iteration to a fixed contiguous sub-interval. In the current situation, since it’s not possible to dispatch the result of a single function call to two arguments (the expected begin and end iterators), the result of filter or sub-interval functions has to be stored as a variable, which implies that the type has to be explicitly spelled out (and it’s often too complex to write down).

Explicit Iterator Intervals

Instead of having algorithms operate on two argument iterators, I propose to combine the two iterators into a single object, called an iterator interval. The type of this iterator interval is unspecified: expression templates are used to avoid unnecessary recourse to runtime polymorphism. The only constraint is that an iterator interval must specify at least the following expressions

T::iterator is an input iterator. Then, given const T t :
t.begin() is the first iterator in the interval.
t.end() is the past-the-end iterator in the interval.

Note that this rules out the standard library containers as intervals: these will return a const_iterator instead of a normal iterator. This is because a container is not an interval, it merely has an interval that represents all its content. The mutability of a container determines the mutability of its members, whereas the mutability of an interval does not affect the mutability of the elements of the represented sequence.

For starters, a few helper functions that create intervals from most things which can have intervals:

namespace interval
{
  namespace detail
  {
    template<typename It>
    class interval_type
    {
      It begin_iterator, end_iterator;
    public:
      interval_type(It begin, It end)
        : begin_iterator(begin), end_iterator(end) {} 

      typedef It iterator; 

      iterator begin() const { return this -> begin_iterator; }
      iterator end() const { return this -> end_iterator; }
    };
  } 

  template<typename T, std::size_t N>
  detail::interval_type<T> of(T (&array)[N])
  {
    return detail::interval_type<T*>(array, array + N);
  } 

  template<typename It>
  detail::interval_type<It> of(It begin, It end)
  {
    return detail::interval_type<It>(begin,end);
  } 

  template<typename T>
  detail::interval_type< std::istream_iterator<T> > of(std::istream &in)
  {
    return detail::interval_type< std::istream_iterator<T> >
      (std::istream_iterator<T>(in),
       std::istream_iterator<T>());
  } 

  template<typename T>
  detail::interval_type< typename T::const_iterator > of(const T &t)
  {
    return detail::interval_type< typename T::const_iterator >
      (t.begin(), t.end());
  } 

  template<typename T>
  detail::interval_type< typename T::iterator > of(T &t)
  {
    return detail::interval_type< typename T::iterator >
      (t.begin(), t.end());
  }
}

So, to create an interval from a container, you can write interval::of(my_vector), to create one from an array you can write interval::of(my_array), and to create one from an input stream you can write interval::of<Type>(my_stream). Once you have obtained an interval, you can start using it.

Algorithms

For the purposes of this article, only two elementary algorithms have been rewritten. One could consider rewriting others as well, using the same principle. The key of rewriting algorithms is to have them take a single iterator interval instead of two iterators.

namespace interval
{
  template<typename I, typename It>
  void copy(const I &i, const It &o)
  {
    std::copy(i.begin(), i.end(), o);
  } 

  template<typename I, typename F>
  void for_each(const I &i, const F &f)
  {
    std::for_each(i.begin(), i.end(), f);
  }
}

So, now, it’s possible to write code such as:

interval::copy(interval::of<int>(std::cin), std::back_inserter(my_vector));

Filtering

The real benefit of using intervals is that now intervals can be transformed. An example transformation is to eliminate the elements in an interval which do not satisfy a certain predicate. This involves defining a new iterator type which wraps around the old one, but which will advance internally so that it always references a value which satisfies the predicate. The obvious limitation is that all iterators become forward iterators (the random access property, if present, cannot be kept).

namespace interval
{
  namespace detail
  {
    template<typename It, typename T>
    class filtered_interval_iterator
    {
      It internal;
      const T *owner; 

      void advance()
      {
        while (this -> internal != this -> owner -> end_iterator
               && ! this -> owner -> filter(* this -> internal))
          ++ this -> internal;
      } 

    public: 

      typedef typename std::iterator_traits<It>::value_type value_type;
      typedef typename std::iterator_traits<It>::reference reference;
      typedef typename std::iterator_traits<It>::pointer pointer;
      typedef typename std::iterator_traits<It>::difference_type difference_type;
      typedef std::input_iterator_tag iterator_category; 

      filtered_interval_iterator(It internal, const T &owner)
        : internal(internal), owner(&owner)
      { this -> advance(); } 

      filtered_interval_iterator &operator++()
      {
        ++ this -> internal;
        this -> advance();
        return *this;
      } 

      filtered_interval_iterator operator++(int)
      {
        filtered_interval_iterator temp(*this);
        ++ *this;
        return temp;
      } 

      reference operator*() const { return *this->internal; } 

      bool operator == (const filtered_interval_iterator &o)
      {
        return this -> internal == o.internal;
      } 

      bool operator != (const filtered_interval_iterator &o)
      {
        return this -> internal != o.internal;
      }
    }; 

    template<typename It, typename F>
    class filtered_interval_type
    {
      F filter;
      It begin_iterator, end_iterator; 

      friend class filtered_interval_iterator<It,filtered_interval_type>; 

    public: 

      filtered_interval_type(It begin, It end, F filter)
        : filter(filter), begin_iterator(begin), end_iterator(end) {} 

      typedef filtered_interval_iterator<It,filtered_interval_type> iterator; 

      iterator begin() const
      {
        return iterator(this -> begin_iterator, *this);
      } 

      iterator end() const
      {
        return iterator(this -> end_iterator, *this);
      }
    };
  } 

  template<typename I, typename F>
  detail::filtered_interval_type<typename I::iterator,F> filter(const I &interval, F flt)
  {
    return detail::filtered_interval_type<typename I::iterator,F>
      (interval.begin(), interval.end(), flt);
  }
}

Filtering takes an interval and a predicate, and filters the interval with the predicate, resulting in a new interval. So, to print to standard output all even numbers read from standard input:

bool even(int x) { return x%2==0; } 

interval::copy( interval::filter( interval::of<int>(std::cin), even ),
                std::ostream_iterator<int>(std::cout, "\n"));

Sub-intervals

Another functionality that can be easily implemented is the ability to select only a subset of an interval. The implementation increments the begin-iterator until it reaches the start distance, then forces any iterator which has moved the total length to equal the past-the-end iterator to mark the end of the interval.

namespace interval
{
  namespace detail
  {
    template<typename It>
    class sub_interval_iterator
    {
      It internal;
      typename std::iterator_traits<It>::difference_type left; 

    public: 

      typedef typename std::iterator_traits<It>::value_type value_type;
      typedef typename std::iterator_traits<It>::reference reference;
      typedef typename std::iterator_traits<It>::pointer pointer;
      typedef typename std::iterator_traits<It>::difference_type difference_type;
      typedef std::input_iterator_tag iterator_category; 

      sub_interval_iterator(It internal, difference_type left)
        : internal(internal), left(left)
      { } 

      sub_interval_iterator &operator++()
      {
        -- this -> left;
        ++ this -> internal;
        return *this;
      } 

      sub_interval_iterator operator++(int)
      {
        sub_interval_iterator temp(*this);
        ++ *this;
        return temp;
      } 

      reference operator*() const { return *this->internal; } 

      bool operator == (const sub_interval_iterator &o)
      {
        return this -> left == 0 && o.left == 0
          || this -> internal == o.internal;
      } 

      bool operator != (const sub_interval_iterator &o)
      {
        return this -> internal != o.internal
          && this -> left != o.left;;
      }
    }; 

    template<typename It>
    class sub_interval_type
    {
    public:
      typedef typename std::iterator_traits<It>::difference_type distance;
    private:
      distance length;
      It begin_iterator, end_iterator;
    public:
      sub_interval_type(It begin, It end, distance start, distance length)
        : length(length), begin_iterator(begin), end_iterator(end)
      {
        while (start > 0 && begin != end) { --start; ++begin; }
        begin_iterator = begin;
      } 

      typedef sub_interval_iterator<It> iterator; 

      iterator begin() const
      {
        return iterator(this -> begin_iterator, this -> length);
      } 

      iterator end() const
      {
        return iterator(this -> end_iterator, 0);
      }
    };
  } 

  template<typename I>
  detail::sub_interval_type<typename I::iterator>
  sub(const I& i, typename I::iterator::difference_type start, typename I::iterator::difference_type length)
  {
    return detail::sub_interval_type<typename I::iterator>(i.begin(), i.end(), start, length);
  }
}

With this function, one can specify the start and length of the sub-interval, just like with a typical sub-string function.

Assignment

How to assign an interval to a container? Containers have constructors and assignment functions that work on two iterators, not a single interval. However, it’s possible to work around this, in part, by using a cast function:

namespace interval
{
  namespace detail
  {
    template<typename I, typename C>
    struct assign_type
    {
      I i;
      assign_type(const I &i) : i(i) {}
      operator C() const { return C(i.begin(), i.end()); }
    };
  } 

  template<typename C, typename I>
  detail::assign_type<I,C> assign(const I &i)
  {
    return detail::assign_type<I,C>(i);
  }
}

The return value of the assignment function above can now be assigned to the appropriate container.

The final example reads even integers from standard input, stores them in a vector, then outputs all the contents of the vector between index 2 and index 12:

bool even(int i) { return i % 2 == 0; } 

int main()
{
  std::vector<int> values = interval::assign< std::vector<int> >
   ( interval::filter ( interval::of<int>(std::cin), even )); 

  interval::copy
   ( interval::sub( interval::of(values), 2, 10 ),
     std::ostream_iterator<int>(std::cout, "\n") );
}

Possible extensions

It is currently not possible to retrieve the original iterator, but this could be done easily (just extend the interval type to handle this) and would be useful (for instance when using std::find across an interval).  Also, iterator types are not conserved: filtering conserves bidirectional iterators and sub-interval extraction conserves random access iterators (plus, on random access iterators, sub-interval extraction is much easier than the above version, which has to handle forward iterators).