[boost] [futures] future producer/consumer streams and prior art

From: Braddock Gaskill (braddock_at_[hidden])
Date: 2007-03-25 18:50:46

I don't really know where the 'future' concept came from (maybe someone does?),
but I've been looking at some old parallel programming papers, and was
surprised to find the concept of a "single-assignment variable" in a 1996 paper
by Ian Foster entitled "Compositional Parallel Programming Languages", which in
turn credited the concept to a 1982 "Data flow languages" paper by W. Ackerman.
Single assignment variables are essentially the futures concept.

Most interesting was a description of a very neat producer/consumer idiom using
these "single-assignment variables" in an academic parallel language.  I
have implemented and tested this idiom in C++ using futures and
shared_ptrs below (uses my futures implementation, which is pretty generic).

It is essentially a "future stream" concept, where a linked list of futures is
constructed and then eaten by the producer and consumer respectively.

Is this new, or old hat?  Either way, I'll be adding "future_stream" templates
to my implementation!

Braddock Gaskill
Dockside Vision Inc

// define my "future stream" int types...make pretty templates someday
struct stream_item;
typedef shared_ptr<stream_item> stream_item_p;
typedef future<stream_item_p> future_stream;
typedef promise<stream_item_p> promised_stream;
struct stream_item {
  stream_item(int v, promised_stream p) : value(v), next(p) {}
  int value;
  future_stream next;
};

// produces 24 int values, 0 - 23
void producer(promised_stream cur) {
  promised_stream next;
  for (int i=0; i<24; ++i) {
    next = promised_stream();
    cur.set(stream_item_p(new stream_item(i, next)));
    cur = next;
  }
  cur.set(stream_item_p()); //set a null next to terminate
}

// consumes values and sums them until the stream ends
int consumer(future_stream stream) {
  int accum=0;
  while (stream.get()) {
    accum += stream.get()->value;
    stream = stream.get()->next;
  }
  return accum;
}

void TestCase13() { //producer/consumer list of futures
  promised_stream stream;
  // launch producer
  boost::thread t(boost::bind(producer, stream));
  int accum = consumer(stream);
  BOOST_CHECK_EQUAL(accum, 276);
  t.join();
}