stream.h

/***************************************************************************
 * include/stxxl/bits/stream/stream.h
 *
 * Part of the STXXL. See http://stxxl.sourceforge.net
 *
 * Copyright (C) 2003-2005 Roman Dementiev <[email protected]>
 * Copyright (C) 2009, 2010 Andreas Beckmann <[email protected]>
 * Copyright (C) 2010 Johannes Singler <[email protected]>
 *
 * Distributed under the Boost Software License, Version 1.0.
 * (See accompanying file LICENSE_1_0.txt or copy at
 * http://www.boost.org/LICENSE_1_0.txt)
 **************************************************************************/

#ifndef STXXL_STREAM_HEADER
#define STXXL_STREAM_HEADER

#include <stxxl/bits/namespace.h>
#include <stxxl/bits/mng/buf_istream.h>
#include <stxxl/bits/mng/buf_ostream.h>
#include <stxxl/bits/common/tuple.h>
#include <stxxl/vector>
#include <stxxl/bits/compat_unique_ptr.h>


#ifndef STXXL_VERBOSE_MATERIALIZE
#define STXXL_VERBOSE_MATERIALIZE STXXL_VERBOSE3
#endif


__STXXL_BEGIN_NAMESPACE

namespace stream
{


 // STREAMIFY //

 template <class InputIterator_>
 class iterator2stream
 {
 InputIterator_ current_, end_;

 public:
 typedef typename std::iterator_traits<InputIterator_>::value_type value_type;

 iterator2stream(InputIterator_ begin, InputIterator_ end) :
 current_(begin), end_(end) { }

 iterator2stream(const iterator2stream & a) : current_(a.current_), end_(a.end_) { }

 const value_type & operator * () const
 {
 return *current_;
 }

 const value_type * operator -> () const
 {
 return &(*current_);
 }

 iterator2stream & operator ++ ()
 {
 assert(end_ != current_);
 ++current_;
 return *this;
 }

 bool empty() const
 {
 return (current_ == end_);
 }
 };


 template <class InputIterator_>
 iterator2stream<InputIterator_> streamify(InputIterator_ begin, InputIterator_ end)
 {
 return iterator2stream<InputIterator_>(begin, end);
 }

 template <class InputIterator_>
 struct streamify_traits
 {
 typedef iterator2stream<InputIterator_> stream_type;
 };

 template <class InputIterator_>
 class vector_iterator2stream
 {
 InputIterator_ current_, end_;
 typedef buf_istream<typename InputIterator_::block_type,
 typename InputIterator_::bids_container_iterator> buf_istream_type;

 typedef typename stxxl::compat_unique_ptr<buf_istream_type>::result buf_istream_unique_ptr_type;
 mutable buf_istream_unique_ptr_type in;

 void delete_stream()
 {
 in.reset(); // delete object
 }

 public:
 typedef vector_iterator2stream<InputIterator_> Self_;

 typedef typename std::iterator_traits<InputIterator_>::value_type value_type;

 vector_iterator2stream(InputIterator_ begin, InputIterator_ end, unsigned_type nbuffers = 0) :
 current_(begin), end_(end), in(static_cast<buf_istream_type *>(NULL))
 {
 begin.flush(); // flush container
 typename InputIterator_::bids_container_iterator end_iter = end.bid() + ((end.block_offset()) ? 1 : 0);

 if (end_iter - begin.bid() > 0)
 {
 in.reset(new buf_istream_type(begin.bid(), end_iter, nbuffers ? nbuffers :
 (2 * config::get_instance()->disks_number())));

 InputIterator_ cur = begin - begin.block_offset();

 // skip the beginning of the block
 for ( ; cur != begin; ++cur)
 ++(*in);
 }
 }

 vector_iterator2stream(const Self_ & a) :
 current_(a.current_), end_(a.end_), in(a.in.release()) { }

 const value_type & operator * () const
 {
 return **in;
 }

 const value_type * operator -> () const
 {
 return &(**in);
 }

 Self_ & operator ++ ()
 {
 assert(end_ != current_);
 ++current_;
 ++(*in);
 if (empty())
 delete_stream();

 return *this;
 }

 bool empty() const
 {
 return (current_ == end_);
 }
 virtual ~vector_iterator2stream()
 {
 delete_stream(); // not needed actually
 }
 };


 template <typename Tp_, typename AllocStr_, typename SzTp_, typename DiffTp_,
 unsigned BlkSize_, typename PgTp_, unsigned PgSz_>
 vector_iterator2stream<stxxl::vector_iterator<Tp_, AllocStr_, SzTp_, DiffTp_, BlkSize_, PgTp_, PgSz_> >
 streamify(
 stxxl::vector_iterator<Tp_, AllocStr_, SzTp_, DiffTp_, BlkSize_, PgTp_, PgSz_> begin,
 stxxl::vector_iterator<Tp_, AllocStr_, SzTp_, DiffTp_, BlkSize_, PgTp_, PgSz_> end,
 unsigned_type nbuffers = 0)
 {
 STXXL_VERBOSE1("streamify for vector_iterator range is called");
 return vector_iterator2stream<stxxl::vector_iterator<Tp_, AllocStr_, SzTp_, DiffTp_, BlkSize_, PgTp_, PgSz_> >
 (begin, end, nbuffers);
 }

 template <typename Tp_, typename AllocStr_, typename SzTp_, typename DiffTp_,
 unsigned BlkSize_, typename PgTp_, unsigned PgSz_>
 struct streamify_traits<stxxl::vector_iterator<Tp_, AllocStr_, SzTp_, DiffTp_, BlkSize_, PgTp_, PgSz_> >
 {
 typedef vector_iterator2stream<stxxl::vector_iterator<Tp_, AllocStr_, SzTp_, DiffTp_, BlkSize_, PgTp_, PgSz_> > stream_type;
 };


 template <typename Tp_, typename AllocStr_, typename SzTp_, typename DiffTp_,
 unsigned BlkSize_, typename PgTp_, unsigned PgSz_>
 vector_iterator2stream<stxxl::const_vector_iterator<Tp_, AllocStr_, SzTp_, DiffTp_, BlkSize_, PgTp_, PgSz_> >
 streamify(
 stxxl::const_vector_iterator<Tp_, AllocStr_, SzTp_, DiffTp_, BlkSize_, PgTp_, PgSz_> begin,
 stxxl::const_vector_iterator<Tp_, AllocStr_, SzTp_, DiffTp_, BlkSize_, PgTp_, PgSz_> end,
 unsigned_type nbuffers = 0)
 {
 STXXL_VERBOSE1("streamify for const_vector_iterator range is called");
 return vector_iterator2stream<stxxl::const_vector_iterator<Tp_, AllocStr_, SzTp_, DiffTp_, BlkSize_, PgTp_, PgSz_> >
 (begin, end, nbuffers);
 }

 template <typename Tp_, typename AllocStr_, typename SzTp_, typename DiffTp_,
 unsigned BlkSize_, typename PgTp_, unsigned PgSz_>
 struct streamify_traits<stxxl::const_vector_iterator<Tp_, AllocStr_, SzTp_, DiffTp_, BlkSize_, PgTp_, PgSz_> >
 {
 typedef vector_iterator2stream<stxxl::const_vector_iterator<Tp_, AllocStr_, SzTp_, DiffTp_, BlkSize_, PgTp_, PgSz_> > stream_type;
 };


 template <class InputIterator_>
 class vector_iterator2stream_sr
 {
 vector_iterator2stream<InputIterator_> * vec_it_stream;
 iterator2stream<InputIterator_> * it_stream;

 typedef typename InputIterator_::block_type block_type;

 public:
 typedef vector_iterator2stream_sr<InputIterator_> Self_;

 typedef typename std::iterator_traits<InputIterator_>::value_type value_type;

 vector_iterator2stream_sr(InputIterator_ begin, InputIterator_ end, unsigned_type nbuffers = 0)
 {
 if (end - begin < block_type::size)
 {
 STXXL_VERBOSE1("vector_iterator2stream_sr::vector_iterator2stream_sr: Choosing iterator2stream<InputIterator_>");
 it_stream = new iterator2stream<InputIterator_>(begin, end);
 vec_it_stream = NULL;
 }
 else
 {
 STXXL_VERBOSE1("vector_iterator2stream_sr::vector_iterator2stream_sr: Choosing vector_iterator2stream<InputIterator_>");
 it_stream = NULL;
 vec_it_stream = new vector_iterator2stream<InputIterator_>(begin, end, nbuffers);
 }
 }

 vector_iterator2stream_sr(const Self_ & a) : vec_it_stream(a.vec_it_stream), it_stream(a.it_stream) { }

 const value_type & operator * () const
 {
 if (it_stream)
 return **it_stream;

 return **vec_it_stream;
 }

 const value_type * operator -> () const
 {
 if (it_stream)
 return &(**it_stream);

 return &(**vec_it_stream);
 }

 Self_ & operator ++ ()
 {
 if (it_stream)
 ++(*it_stream);

 else
 ++(*vec_it_stream);


 return *this;
 }

 bool empty() const
 {
 if (it_stream)
 return it_stream->empty();

 return vec_it_stream->empty();
 }
 virtual ~vector_iterator2stream_sr()
 {
 if (it_stream)
 delete it_stream;

 else
 delete vec_it_stream;
 }
 };

 template <typename Tp_, typename AllocStr_, typename SzTp_, typename DiffTp_,
 unsigned BlkSize_, typename PgTp_, unsigned PgSz_>
 vector_iterator2stream_sr<stxxl::vector_iterator<Tp_, AllocStr_, SzTp_, DiffTp_, BlkSize_, PgTp_, PgSz_> >
 streamify_sr(
 stxxl::vector_iterator<Tp_, AllocStr_, SzTp_, DiffTp_, BlkSize_, PgTp_, PgSz_> begin,
 stxxl::vector_iterator<Tp_, AllocStr_, SzTp_, DiffTp_, BlkSize_, PgTp_, PgSz_> end,
 unsigned_type nbuffers = 0)
 {
 STXXL_VERBOSE1("streamify_sr for vector_iterator range is called");
 return vector_iterator2stream_sr<stxxl::vector_iterator<Tp_, AllocStr_, SzTp_, DiffTp_, BlkSize_, PgTp_, PgSz_> >
 (begin, end, nbuffers);
 }

 template <typename Tp_, typename AllocStr_, typename SzTp_, typename DiffTp_,
 unsigned BlkSize_, typename PgTp_, unsigned PgSz_>
 vector_iterator2stream_sr<stxxl::const_vector_iterator<Tp_, AllocStr_, SzTp_, DiffTp_, BlkSize_, PgTp_, PgSz_> >
 streamify_sr(
 stxxl::const_vector_iterator<Tp_, AllocStr_, SzTp_, DiffTp_, BlkSize_, PgTp_, PgSz_> begin,
 stxxl::const_vector_iterator<Tp_, AllocStr_, SzTp_, DiffTp_, BlkSize_, PgTp_, PgSz_> end,
 unsigned_type nbuffers = 0)
 {
 STXXL_VERBOSE1("streamify_sr for const_vector_iterator range is called");
 return vector_iterator2stream_sr<stxxl::const_vector_iterator<Tp_, AllocStr_, SzTp_, DiffTp_, BlkSize_, PgTp_, PgSz_> >
 (begin, end, nbuffers);
 }


 // MATERIALIZE //

 template <class OutputIterator_, class StreamAlgorithm_>
 OutputIterator_ materialize(StreamAlgorithm_ & in, OutputIterator_ out)
 {
 STXXL_VERBOSE_MATERIALIZE(STXXL_PRETTY_FUNCTION_NAME);
 while (!in.empty())
 {
 *out = *in;
 ++out;
 ++in;
 }
 return out;
 }


 template <class OutputIterator_, class StreamAlgorithm_>
 OutputIterator_ materialize(StreamAlgorithm_ & in, OutputIterator_ outbegin, OutputIterator_ outend)
 {
 STXXL_VERBOSE_MATERIALIZE(STXXL_PRETTY_FUNCTION_NAME);
 while ((!in.empty()) && outend != outbegin)
 {
 *outbegin = *in;
 ++outbegin;
 ++in;
 }
 return outbegin;
 }


 template <typename Tp_, typename AllocStr_, typename SzTp_, typename DiffTp_,
 unsigned BlkSize_, typename PgTp_, unsigned PgSz_, class StreamAlgorithm_>
 stxxl::vector_iterator<Tp_, AllocStr_, SzTp_, DiffTp_, BlkSize_, PgTp_, PgSz_>
 materialize(StreamAlgorithm_ & in,
 stxxl::vector_iterator<Tp_, AllocStr_, SzTp_, DiffTp_, BlkSize_, PgTp_, PgSz_> outbegin,
 stxxl::vector_iterator<Tp_, AllocStr_, SzTp_, DiffTp_, BlkSize_, PgTp_, PgSz_> outend,
 unsigned_type nbuffers = 0)
 {
 STXXL_VERBOSE_MATERIALIZE(STXXL_PRETTY_FUNCTION_NAME);
 typedef stxxl::vector_iterator<Tp_, AllocStr_, SzTp_, DiffTp_, BlkSize_, PgTp_, PgSz_> ExtIterator;
 typedef stxxl::const_vector_iterator<Tp_, AllocStr_, SzTp_, DiffTp_, BlkSize_, PgTp_, PgSz_> ConstExtIterator;
 typedef buf_ostream<typename ExtIterator::block_type, typename ExtIterator::bids_container_iterator> buf_ostream_type;


 while (outbegin.block_offset()) // go to the beginning of the block
 // of the external vector
 {
 if (in.empty() || outbegin == outend)
 return outbegin;

 *outbegin = *in;
 ++outbegin;
 ++in;
 }

 if (nbuffers == 0)
 nbuffers = 2 * config::get_instance()->disks_number();

 outbegin.flush(); // flush container

 // create buffered write stream for blocks
 buf_ostream_type outstream(outbegin.bid(), nbuffers);

 assert(outbegin.block_offset() == 0);

 // delay calling block_externally_updated() until the block is
 // completely filled (and written out) in outstream
 ConstExtIterator prev_block = outbegin;

 while (!in.empty() && outend != outbegin)
 {
 if (outbegin.block_offset() == 0) {
 if (prev_block != outbegin) {
 prev_block.block_externally_updated();
 prev_block = outbegin;
 }
 }

 *outstream = *in;
 ++outbegin;
 ++outstream;
 ++in;
 }

 ConstExtIterator const_out = outbegin;

 while (const_out.block_offset()) // filling the rest of the block
 {
 *outstream = *const_out;
 ++const_out;
 ++outstream;
 }

 if (prev_block != outbegin)
 prev_block.block_externally_updated();

 outbegin.flush();

 return outbegin;
 }


 template <typename Tp_, typename AllocStr_, typename SzTp_, typename DiffTp_,
 unsigned BlkSize_, typename PgTp_, unsigned PgSz_, class StreamAlgorithm_>
 stxxl::vector_iterator<Tp_, AllocStr_, SzTp_, DiffTp_, BlkSize_, PgTp_, PgSz_>
 materialize(StreamAlgorithm_ & in,
 stxxl::vector_iterator<Tp_, AllocStr_, SzTp_, DiffTp_, BlkSize_, PgTp_, PgSz_> out,
 unsigned_type nbuffers = 0)
 {
 STXXL_VERBOSE_MATERIALIZE(STXXL_PRETTY_FUNCTION_NAME);
 typedef stxxl::vector_iterator<Tp_, AllocStr_, SzTp_, DiffTp_, BlkSize_, PgTp_, PgSz_> ExtIterator;
 typedef stxxl::const_vector_iterator<Tp_, AllocStr_, SzTp_, DiffTp_, BlkSize_, PgTp_, PgSz_> ConstExtIterator;
 typedef buf_ostream<typename ExtIterator::block_type, typename ExtIterator::bids_container_iterator> buf_ostream_type;

 // on the I/O complexity of "materialize":
 // crossing block boundary causes O(1) I/Os
 // if you stay in a block, then materialize function accesses only the cache of the
 // vector (only one block indeed), amortized complexity should apply here

 while (out.block_offset()) // go to the beginning of the block
 // of the external vector
 {
 if (in.empty())
 return out;

 *out = *in;
 ++out;
 ++in;
 }

 if (nbuffers == 0)
 nbuffers = 2 * config::get_instance()->disks_number();


 out.flush(); // flush container

 // create buffered write stream for blocks
 buf_ostream_type outstream(out.bid(), nbuffers);

 assert(out.block_offset() == 0);

 // delay calling block_externally_updated() until the block is
 // completely filled (and written out) in outstream
 ConstExtIterator prev_block = out;

 while (!in.empty())
 {
 if (out.block_offset() == 0) {
 if (prev_block != out) {
 prev_block.block_externally_updated();
 prev_block = out;
 }
 }

 // tells the vector that the block was modified
 *outstream = *in;
 ++out;
 ++outstream;
 ++in;
 }

 ConstExtIterator const_out = out;

 while (const_out.block_offset())
 {
 *outstream = *const_out; // might cause I/Os for loading the page that
 ++const_out; // contains data beyond out
 ++outstream;
 }

 if (prev_block != out)
 prev_block.block_externally_updated();

 out.flush();

 return out;
 }


 template <class StreamAlgorithm_>
 void discard(StreamAlgorithm_ & in)
 {
 while (!in.empty())
 {
 *in;
 ++in;
 }
 }


 // GENERATE //

 template <class Generator_, typename T = typename Generator_::value_type>
 class generator2stream
 {
 public:
 typedef T value_type;

 private:
 Generator_ gen_;
 value_type current_;

 public:
 generator2stream(Generator_ g) :
 gen_(g), current_(gen_()) { }

 generator2stream(const generator2stream & a) : gen_(a.gen_), current_(a.current_) { }

 const value_type & operator * () const
 {
 return current_;
 }

 const value_type * operator -> () const
 {
 return &current_;
 }

 generator2stream & operator ++ ()
 {
 current_ = gen_();
 return *this;
 }

 bool empty() const
 {
 return false;
 }
 };

 template <class Generator_>
 generator2stream<Generator_> streamify(Generator_ gen_)
 {
 return generator2stream<Generator_>(gen_);
 }


 // TRANSFORM //

 struct Stopper { };

 template <class Operation_,
 class Input1_,
 class Input2_ = Stopper,
 class Input3_ = Stopper,
 class Input4_ = Stopper,
 class Input5_ = Stopper,
 class Input6_ = Stopper
 >
 class transform
 {
 Operation_ & op;
 Input1_ & i1;
 Input2_ & i2;
 Input3_ & i3;
 Input4_ & i4;
 Input5_ & i5;
 Input6_ & i6;

 public:
 typedef typename Operation_::value_type value_type;

 private:
 value_type current;

 public:
 transform(Operation_ & o, Input1_ & i1_, Input2_ & i2_, Input3_ & i3_, Input4_ & i4_,
 Input5_ & i5_, Input5_ & i6_) :
 op(o), i1(i1_), i2(i2_), i3(i3_), i4(i4_), i5(i5_), i6(i6_)
 {
 if (!empty())
 current = op(*i1, *i2, *i3, *i4, *i5, *i6);
 }

 const value_type & operator * () const
 {
 return current;
 }

 const value_type * operator -> () const
 {
 return &current;
 }

 transform & operator ++ ()
 {
 ++i1;
 ++i2;
 ++i3;
 ++i4;
 ++i5;
 ++i6;
 if (!empty())
 current = op(*i1, *i2, *i3, *i4, *i5, *i6);

 return *this;
 }

 bool empty() const
 {
 return i1.empty() || i2.empty() || i3.empty() ||
 i4.empty() || i5.empty() || i6.empty();
 }
 };

 // Specializations

 // TRANSFORM (1 input stream) //

 template <class Operation_,
 class Input1_
 >
 class transform<Operation_, Input1_, Stopper, Stopper, Stopper, Stopper, Stopper>
 {
 Operation_ & op;
 Input1_ & i1;

 public:
 typedef typename Operation_::value_type value_type;

 private:
 value_type current;

 public:
 transform(Operation_ & o, Input1_ & i1_) : op(o), i1(i1_)
 {
 if (!empty())
 current = op(*i1);
 }

 const value_type & operator * () const
 {
 return current;
 }

 const value_type * operator -> () const
 {
 return &current;
 }

 transform & operator ++ ()
 {
 ++i1;
 if (!empty())
 current = op(*i1);

 return *this;
 }

 bool empty() const
 {
 return i1.empty();
 }
 };


 // TRANSFORM (2 input streams) //

 template <class Operation_,
 class Input1_,
 class Input2_
 >
 class transform<Operation_, Input1_, Input2_, Stopper, Stopper, Stopper, Stopper>
 {
 Operation_ & op;
 Input1_ & i1;
 Input2_ & i2;

 public:
 typedef typename Operation_::value_type value_type;

 private:
 value_type current;

 public:
 transform(Operation_ & o, Input1_ & i1_, Input2_ & i2_) : op(o), i1(i1_), i2(i2_)
 {
 if (!empty())
 current = op(*i1, *i2);
 }

 const value_type & operator * () const
 {
 return current;
 }

 const value_type * operator -> () const
 {
 return &current;
 }

 transform & operator ++ ()
 {
 ++i1;
 ++i2;
 if (!empty())
 current = op(*i1, *i2);

 return *this;
 }

 bool empty() const
 {
 return i1.empty() || i2.empty();
 }
 };


 // TRANSFORM (3 input streams) //

 template <class Operation_,
 class Input1_,
 class Input2_,
 class Input3_
 >
 class transform<Operation_, Input1_, Input2_, Input3_, Stopper, Stopper, Stopper>
 {
 Operation_ & op;
 Input1_ & i1;
 Input2_ & i2;
 Input3_ & i3;

 public:
 typedef typename Operation_::value_type value_type;

 private:
 value_type current;

 public:
 transform(Operation_ & o, Input1_ & i1_, Input2_ & i2_, Input3_ & i3_) :
 op(o), i1(i1_), i2(i2_), i3(i3_)
 {
 if (!empty())
 current = op(*i1, *i2, *i3);
 }

 const value_type & operator * () const
 {
 return current;
 }

 const value_type * operator -> () const
 {
 return &current;
 }

 transform & operator ++ ()
 {
 ++i1;
 ++i2;
 ++i3;
 if (!empty())
 current = op(*i1, *i2, *i3);

 return *this;
 }

 bool empty() const
 {
 return i1.empty() || i2.empty() || i3.empty();
 }
 };


 // TRANSFORM (4 input streams) //

 template <class Operation_,
 class Input1_,
 class Input2_,
 class Input3_,
 class Input4_
 >
 class transform<Operation_, Input1_, Input2_, Input3_, Input4_, Stopper, Stopper>
 {
 Operation_ & op;
 Input1_ & i1;
 Input2_ & i2;
 Input3_ & i3;
 Input4_ & i4;

 public:
 typedef typename Operation_::value_type value_type;

 private:
 value_type current;

 public:
 transform(Operation_ & o, Input1_ & i1_, Input2_ & i2_, Input3_ & i3_, Input4_ & i4_) :
 op(o), i1(i1_), i2(i2_), i3(i3_), i4(i4_)
 {
 if (!empty())
 current = op(*i1, *i2, *i3, *i4);
 }

 const value_type & operator * () const
 {
 return current;
 }

 const value_type * operator -> () const
 {
 return &current;
 }

 transform & operator ++ ()
 {
 ++i1;
 ++i2;
 ++i3;
 ++i4;
 if (!empty())
 current = op(*i1, *i2, *i3, *i4);

 return *this;
 }

 bool empty() const
 {
 return i1.empty() || i2.empty() || i3.empty() || i4.empty();
 }
 };


 // TRANSFORM (5 input streams) //

 template <class Operation_,
 class Input1_,
 class Input2_,
 class Input3_,
 class Input4_,
 class Input5_
 >
 class transform<Operation_, Input1_, Input2_, Input3_, Input4_, Input5_, Stopper>
 {
 Operation_ & op;
 Input1_ & i1;
 Input2_ & i2;
 Input3_ & i3;
 Input4_ & i4;
 Input5_ & i5;

 public:
 typedef typename Operation_::value_type value_type;

 private:
 value_type current;

 public:
 transform(Operation_ & o, Input1_ & i1_, Input2_ & i2_, Input3_ & i3_, Input4_ & i4_,
 Input5_ & i5_) :
 op(o), i1(i1_), i2(i2_), i3(i3_), i4(i4_), i5(i5_)
 {
 if (!empty())
 current = op(*i1, *i2, *i3, *i4, *i5);
 }

 const value_type & operator * () const
 {
 return current;
 }

 const value_type * operator -> () const
 {
 return &current;
 }

 transform & operator ++ ()
 {
 ++i1;
 ++i2;
 ++i3;
 ++i4;
 ++i5;
 if (!empty())
 current = op(*i1, *i2, *i3, *i4, *i5);

 return *this;
 }

 bool empty() const
 {
 return i1.empty() || i2.empty() || i3.empty() || i4.empty() || i5.empty();
 }
 };


 // MAKE TUPLE //

 template <class Input1_,
 class Input2_,
 class Input3_ = Stopper,
 class Input4_ = Stopper,
 class Input5_ = Stopper,
 class Input6_ = Stopper
 >
 class make_tuple
 {
 Input1_ & i1;
 Input2_ & i2;
 Input3_ & i3;
 Input4_ & i4;
 Input5_ & i5;
 Input6_ & i6;

 public:
 typedef typename stxxl::tuple<
 typename Input1_::value_type,
 typename Input2_::value_type,
 typename Input3_::value_type,
 typename Input4_::value_type,
 typename Input5_::value_type,
 typename Input6_::value_type
 > value_type;

 private:
 value_type current;

 public:
 make_tuple(
 Input1_ & i1_,
 Input2_ & i2_,
 Input3_ & i3_,
 Input4_ & i4_,
 Input5_ & i5_,
 Input6_ & i6_
 ) :
 i1(i1_), i2(i2_), i3(i3_), i4(i4_), i5(i5_), i6(i6_),
 current(value_type(*i1, *i2, *i3, *i4, *i5, *i6)) { }

 const value_type & operator * () const
 {
 return current;
 }

 const value_type * operator -> () const
 {
 return &current;
 }

 make_tuple & operator ++ ()
 {
 ++i1;
 ++i2;
 ++i3;
 ++i4;
 ++i5;
 ++i6;

 if (!empty())
 current = value_type(*i1, *i2, *i3, *i4, *i5, *i6);

 return *this;
 }

 bool empty() const
 {
 return i1.empty() || i2.empty() || i3.empty() ||
 i4.empty() || i5.empty() || i6.empty();
 }
 };


 template <class Input1_,
 class Input2_
 >
 class make_tuple<Input1_, Input2_, Stopper, Stopper, Stopper, Stopper>
 {
 Input1_ & i1;
 Input2_ & i2;

 public:
 typedef typename stxxl::tuple<
 typename Input1_::value_type,
 typename Input2_::value_type
 > value_type;

 private:
 value_type current;

 public:
 make_tuple(
 Input1_ & i1_,
 Input2_ & i2_
 ) :
 i1(i1_), i2(i2_)
 {
 if (!empty())
 {
 current = value_type(*i1, *i2);
 }
 }

 const value_type & operator * () const
 {
 return current;
 }

 const value_type * operator -> () const
 {
 return &current;
 }

 make_tuple & operator ++ ()
 {
 ++i1;
 ++i2;

 if (!empty())
 current = value_type(*i1, *i2);

 return *this;
 }

 bool empty() const
 {
 return i1.empty() || i2.empty();
 }
 };

 template <class Input1_,
 class Input2_,
 class Input3_
 >
 class make_tuple<Input1_, Input2_, Input3_, Stopper, Stopper, Stopper>
 {
 Input1_ & i1;
 Input2_ & i2;
 Input3_ & i3;

 public:
 typedef typename stxxl::tuple<
 typename Input1_::value_type,
 typename Input2_::value_type,
 typename Input3_::value_type
 > value_type;

 private:
 value_type current;

 public:
 make_tuple(
 Input1_ & i1_,
 Input2_ & i2_,
 Input3_ & i3_
 ) :
 i1(i1_), i2(i2_), i3(i3_),
 current(value_type(*i1, *i2, *i3)) { }

 const value_type & operator * () const
 {
 return current;
 }

 const value_type * operator -> () const
 {
 return &current;
 }

 make_tuple & operator ++ ()
 {
 ++i1;
 ++i2;
 ++i3;

 if (!empty())
 current = value_type(*i1, *i2, *i3);

 return *this;
 }

 bool empty() const
 {
 return i1.empty() || i2.empty() || i3.empty();
 }
 };

 template <class Input1_,
 class Input2_,
 class Input3_,
 class Input4_
 >
 class make_tuple<Input1_, Input2_, Input3_, Input4_, Stopper, Stopper>
 {
 Input1_ & i1;
 Input2_ & i2;
 Input3_ & i3;
 Input4_ & i4;

 public:
 typedef typename stxxl::tuple<
 typename Input1_::value_type,
 typename Input2_::value_type,
 typename Input3_::value_type,
 typename Input4_::value_type
 > value_type;

 private:
 value_type current;

 public:
 make_tuple(
 Input1_ & i1_,
 Input2_ & i2_,
 Input3_ & i3_,
 Input4_ & i4_
 ) :
 i1(i1_), i2(i2_), i3(i3_), i4(i4_),
 current(value_type(*i1, *i2, *i3, *i4)) { }

 const value_type & operator * () const
 {
 return current;
 }

 const value_type * operator -> () const
 {
 return &current;
 }

 make_tuple & operator ++ ()
 {
 ++i1;
 ++i2;
 ++i3;
 ++i4;

 if (!empty())
 current = value_type(*i1, *i2, *i3, *i4);

 return *this;
 }

 bool empty() const
 {
 return i1.empty() || i2.empty() || i3.empty() ||
 i4.empty();
 }
 };

 template <
 class Input1_,
 class Input2_,
 class Input3_,
 class Input4_,
 class Input5_
 >
 class make_tuple<Input1_, Input2_, Input3_, Input4_, Input5_, Stopper>
 {
 Input1_ & i1;
 Input2_ & i2;
 Input3_ & i3;
 Input4_ & i4;
 Input5_ & i5;

 public:
 typedef typename stxxl::tuple<
 typename Input1_::value_type,
 typename Input2_::value_type,
 typename Input3_::value_type,
 typename Input4_::value_type,
 typename Input5_::value_type
 > value_type;

 private:
 value_type current;

 public:
 make_tuple(
 Input1_ & i1_,
 Input2_ & i2_,
 Input3_ & i3_,
 Input4_ & i4_,
 Input5_ & i5_
 ) :
 i1(i1_), i2(i2_), i3(i3_), i4(i4_), i5(i5_),
 current(value_type(*i1, *i2, *i3, *i4, *i5)) { }

 const value_type & operator * () const
 {
 return current;
 }

 const value_type * operator -> () const
 {
 return &current;
 }

 make_tuple & operator ++ ()
 {
 ++i1;
 ++i2;
 ++i3;
 ++i4;
 ++i5;

 if (!empty())
 current = value_type(*i1, *i2, *i3, *i4, *i5);

 return *this;
 }

 bool empty() const
 {
 return i1.empty() || i2.empty() || i3.empty() ||
 i4.empty() || i5.empty();
 }
 };


}

__STXXL_END_NAMESPACE


#include <stxxl/bits/stream/choose.h>
#include <stxxl/bits/stream/unique.h>


#endif // !STXXL_STREAM_HEADER
// vim: et:ts=4:sw=4