random_shuffle.h

/***************************************************************************
 * include/stxxl/bits/algo/random_shuffle.h
 *
 * Part of the STXXL. See http://stxxl.sourceforge.net
 *
 * Copyright (C) 2007 Manuel Krings
 * Copyright (C) 2007 Markus Westphal
 * Copyright (C) 2009, 2010 Andreas Beckmann <[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_RANDOM_SHUFFLE_HEADER
#define STXXL_RANDOM_SHUFFLE_HEADER

// TODO: improve main memory consumption in recursion
// (free stacks buffers)
// TODO: shuffle small input in internal memory


#include <stxxl/bits/stream/stream.h>
#include <stxxl/scan>
#include <stxxl/stack>


__STXXL_BEGIN_NAMESPACE




template <typename ExtIterator_,
 typename RandomNumberGenerator_,
 unsigned BlockSize_,
 unsigned PageSize_,
 typename AllocStrategy_>
void random_shuffle(ExtIterator_ first,
 ExtIterator_ last,
 RandomNumberGenerator_ & rand,
 unsigned_type M,
 AllocStrategy_ AS = STXXL_DEFAULT_ALLOC_STRATEGY())
{
 STXXL_UNUSED(AS); // FIXME: Why is this not being used?
 typedef typename ExtIterator_::value_type value_type;
 typedef typename stxxl::STACK_GENERATOR<value_type, stxxl::external,
 stxxl::grow_shrink2, PageSize_,
 BlockSize_, void, 0, AllocStrategy_>::result stack_type;
 typedef typename stack_type::block_type block_type;

 STXXL_VERBOSE1("random_shuffle: Plain Version");
 STXXL_STATIC_ASSERT(int(BlockSize_) < 0 && "This implementation was never tested. Please report to the stxxl developers if you have an ExtIterator_ that works with this implementation.");

 stxxl::int64 n = last - first; // the number of input elements

 // make sure we have at least 6 blocks + 1 page
 if (M < 6 * BlockSize_ + PageSize_ * BlockSize_) {
 STXXL_ERRMSG("random_shuffle: insufficient memory, " << M << " bytes supplied,");
 M = 6 * BlockSize_ + PageSize_ * BlockSize_;
 STXXL_ERRMSG("random_shuffle: increasing to " << M << " bytes (6 blocks + 1 page)");
 }

 int_type k = M / (3 * BlockSize_); // number of buckets


 stxxl::int64 i, j, size = 0;

 value_type * temp_array;
 typedef typename stxxl::VECTOR_GENERATOR<value_type,
 PageSize_, 4, BlockSize_, AllocStrategy_>::result temp_vector_type;
 temp_vector_type * temp_vector;

 STXXL_VERBOSE1("random_shuffle: " << M / BlockSize_ - k << " write buffers for " << k << " buckets");
 stxxl::read_write_pool<block_type> pool(0, M / BlockSize_ - k); // no read buffers and M/B-k write buffers

 stack_type ** buckets;

 // create and put buckets into container
 buckets = new stack_type *[k];
 for (j = 0; j < k; j++)
 buckets[j] = new stack_type(pool, 0);


 typedef typename stream::streamify_traits<ExtIterator_>::stream_type input_stream;
 input_stream in = stxxl::stream::streamify(first, last);

 // distribute input into random buckets
 int_type random_bucket = 0;
 for (i = 0; i < n; ++i) {
 random_bucket = rand(k);
 buckets[random_bucket]->push(*in); // reading the current input element
 ++in; // go to the next input element
 }

 // resize buffers
 pool.resize_write(0);
 pool.resize_prefetch(PageSize_);

 unsigned_type space_left = M - k * BlockSize_ -
 PageSize_ * BlockSize_; // remaining int space
 ExtIterator_ Writer = first;
 ExtIterator_ it = first;

 for (i = 0; i < k; i++) {
 STXXL_VERBOSE1("random_shuffle: bucket no " << i << " contains " << buckets[i]->size() << " elements");
 }

 // shuffle each bucket
 for (i = 0; i < k; i++) {
 buckets[i]->set_prefetch_aggr(PageSize_);
 size = buckets[i]->size();

 // does the bucket fit into memory?
 if (size * sizeof(value_type) < space_left) {
 STXXL_VERBOSE1("random_shuffle: no recursion");

 // copy bucket into temp. array
 temp_array = new value_type[size];
 for (j = 0; j < size; j++) {
 temp_array[j] = buckets[i]->top();
 buckets[i]->pop();
 }

 // shuffle
 potentially_parallel::
 random_shuffle(temp_array, temp_array + size, rand);

 // write back
 for (j = 0; j < size; j++) {
 *Writer = temp_array[j];
 ++Writer;
 }

 // free memory
 delete[] temp_array;
 }
 else {
 STXXL_VERBOSE1("random_shuffle: recursion");

 // copy bucket into temp. stxxl::vector
 temp_vector = new temp_vector_type(size);

 for (j = 0; j < size; j++) {
 (*temp_vector)[j] = buckets[i]->top();
 buckets[i]->pop();
 }

 pool.resize_prefetch(0);
 space_left += PageSize_ * BlockSize_;
 STXXL_VERBOSE1("random_shuffle: Space left: " << space_left);

 // recursive shuffle
 stxxl::random_shuffle(temp_vector->begin(),
 temp_vector->end(), rand, space_left);

 pool.resize_prefetch(PageSize_);

 // write back
 for (j = 0; j < size; j++) {
 *Writer = (*temp_vector)[j];
 ++Writer;
 }

 // free memory
 delete temp_vector;
 }

 // free bucket
 delete buckets[i];
 space_left += BlockSize_;
 }

 delete[] buckets;
}

template <typename Tp_, typename AllocStrategy_, typename SzTp_, typename DiffTp_,
 unsigned BlockSize_, typename PgTp_, unsigned PageSize_, typename RandomNumberGenerator_>
void random_shuffle(stxxl::vector_iterator<Tp_, AllocStrategy_, SzTp_, DiffTp_, BlockSize_, PgTp_, PageSize_> first,
 stxxl::vector_iterator<Tp_, AllocStrategy_, SzTp_, DiffTp_, BlockSize_, PgTp_, PageSize_> last,
 RandomNumberGenerator_ & rand,
 unsigned_type M)
{
 typedef stxxl::vector_iterator<Tp_, AllocStrategy_, SzTp_, DiffTp_, BlockSize_, PgTp_, PageSize_> ExtIterator_;
 typedef typename ExtIterator_::value_type value_type;
 typedef typename stxxl::STACK_GENERATOR<value_type, stxxl::external,
 stxxl::grow_shrink2, PageSize_, BlockSize_>::result stack_type;
 typedef typename stack_type::block_type block_type;

 STXXL_VERBOSE1("random_shuffle: Vector Version");

 // make sure we have at least 6 blocks + 1 page
 if (M < 6 * BlockSize_ + PageSize_ * BlockSize_) {
 STXXL_ERRMSG("random_shuffle: insufficient memory, " << M << " bytes supplied,");
 M = 6 * BlockSize_ + PageSize_ * BlockSize_;
 STXXL_ERRMSG("random_shuffle: increasing to " << M << " bytes (6 blocks + 1 page)");
 }

 stxxl::int64 n = last - first; // the number of input elements
 int_type k = M / (3 * BlockSize_); // number of buckets

 stxxl::int64 i, j, size = 0;

 value_type * temp_array;
 typedef typename stxxl::VECTOR_GENERATOR<value_type,
 PageSize_, 4, BlockSize_, AllocStrategy_>::result temp_vector_type;
 temp_vector_type * temp_vector;

 stxxl::read_write_pool<block_type> pool(0, M / BlockSize_ - k); // no read buffers and M/B-k write buffers

 stack_type ** buckets;

 // create and put buckets into container
 buckets = new stack_type *[k];
 for (j = 0; j < k; j++)
 buckets[j] = new stack_type(pool, 0);


 typedef buf_istream<block_type, typename ExtIterator_::bids_container_iterator> buf_istream_type;
 typedef buf_ostream<block_type, typename ExtIterator_::bids_container_iterator> buf_ostream_type;

 first.flush(); // flush container

 // create prefetching stream,
 buf_istream_type in(first.bid(), last.bid() + ((last.block_offset()) ? 1 : 0), 2);
 // create buffered write stream for blocks
 buf_ostream_type out(first.bid(), 2);

 ExtIterator_ _cur = first - first.block_offset();

 // leave part of the block before _begin untouched (e.g. copy)
 for ( ; _cur != first; ++_cur)
 {
 typename ExtIterator_::value_type tmp;
 in >> tmp;
 out << tmp;
 }


 // distribute input into random buckets
 int_type random_bucket = 0;
 for (i = 0; i < n; ++i, ++_cur) {
 random_bucket = rand(k);
 typename ExtIterator_::value_type tmp;
 in >> tmp;
 buckets[random_bucket]->push(tmp); // reading the current input element
 }

 // resize buffers
 pool.resize_write(0);
 pool.resize_prefetch(PageSize_);

 unsigned_type space_left = M - k * BlockSize_ -
 PageSize_ * BlockSize_; // remaining int space

 for (i = 0; i < k; i++) {
 STXXL_VERBOSE1("random_shuffle: bucket no " << i << " contains " << buckets[i]->size() << " elements");
 }

 // shuffle each bucket
 for (i = 0; i < k; i++) {
 buckets[i]->set_prefetch_aggr(PageSize_);
 size = buckets[i]->size();

 // does the bucket fit into memory?
 if (size * sizeof(value_type) < space_left) {
 STXXL_VERBOSE1("random_shuffle: no recursion");

 // copy bucket into temp. array
 temp_array = new value_type[size];
 for (j = 0; j < size; j++) {
 temp_array[j] = buckets[i]->top();
 buckets[i]->pop();
 }

 // shuffle
 potentially_parallel::
 random_shuffle(temp_array, temp_array + size, rand);

 // write back
 for (j = 0; j < size; j++) {
 typename ExtIterator_::value_type tmp;
 tmp = temp_array[j];
 out << tmp;
 }

 // free memory
 delete[] temp_array;
 } else {
 STXXL_VERBOSE1("random_shuffle: recursion");
 // copy bucket into temp. stxxl::vector
 temp_vector = new temp_vector_type(size);

 for (j = 0; j < size; j++) {
 (*temp_vector)[j] = buckets[i]->top();
 buckets[i]->pop();
 }

 pool.resize_prefetch(0);
 space_left += PageSize_ * BlockSize_;

 STXXL_VERBOSE1("random_shuffle: Space left: " << space_left);

 // recursive shuffle
 stxxl::random_shuffle(temp_vector->begin(),
 temp_vector->end(), rand, space_left);

 pool.resize_prefetch(PageSize_);

 // write back
 for (j = 0; j < size; j++) {
 typename ExtIterator_::value_type tmp;
 tmp = (*temp_vector)[j];
 out << tmp;
 }

 // free memory
 delete temp_vector;
 }

 // free bucket
 delete buckets[i];
 space_left += BlockSize_;
 }

 delete[] buckets;

 // leave part of the block after _end untouched
 if (last.block_offset())
 {
 ExtIterator_ _last_block_end = last + (block_type::size - last.block_offset());
 for ( ; _cur != _last_block_end; ++_cur)
 {
 typename ExtIterator_::value_type tmp;
 in >> tmp;
 out << tmp;
 }
 }
}

template <typename Tp_, typename AllocStrategy_, typename SzTp_, typename DiffTp_,
 unsigned BlockSize_, typename PgTp_, unsigned PageSize_>
inline
void random_shuffle(stxxl::vector_iterator<Tp_, AllocStrategy_, SzTp_, DiffTp_, BlockSize_, PgTp_, PageSize_> first,
 stxxl::vector_iterator<Tp_, AllocStrategy_, SzTp_, DiffTp_, BlockSize_, PgTp_, PageSize_> last,
 unsigned_type M)
{
 stxxl::random_number<> rand;
 stxxl::random_shuffle(first, last, rand, M);
}


__STXXL_END_NAMESPACE

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