• Main Page
  • Related Pages
  • Modules
  • Namespaces
  • Classes
  • Files
  • Examples
  • File List

stable_ksort.h

00001 /***************************************************************************
00002  *  include/stxxl/bits/algo/stable_ksort.h
00003  *
00004  *  Part of the STXXL. See http://stxxl.sourceforge.net
00005  *
00006  *  Copyright (C) 2003 Roman Dementiev <dementiev@mpi-sb.mpg.de>
00007  *
00008  *  Distributed under the Boost Software License, Version 1.0.
00009  *  (See accompanying file LICENSE_1_0.txt or copy at
00010  *  http://www.boost.org/LICENSE_1_0.txt)
00011  **************************************************************************/
00012 
00013 #ifndef STXXL_STABLE_KSORT_HEADER
00014 #define STXXL_STABLE_KSORT_HEADER
00015 
00016 // it is a first try: distribution sort without sampling
00017 // I rework the stable_ksort when I would have a time
00018 
00019 
00020 #include <stxxl/bits/mng/mng.h>
00021 #include <stxxl/bits/mng/buf_istream.h>
00022 #include <stxxl/bits/mng/buf_ostream.h>
00023 #include <stxxl/bits/common/simple_vector.h>
00024 #include <stxxl/bits/algo/intksort.h>
00025 #include <stxxl/bits/algo/sort_base.h>
00026 #include <stxxl/bits/common/utils.h>
00027 
00028 
00029 #ifndef STXXL_VERBOSE_STABLE_KSORT
00030 #define STXXL_VERBOSE_STABLE_KSORT STXXL_VERBOSE1
00031 #endif
00032 
00033 
00034 __STXXL_BEGIN_NAMESPACE
00035 
00038 
00041 namespace stable_ksort_local
00042 {
00043     template <class type_, class type_key>
00044     void classify_block(type_ * begin, type_ * end, type_key * & out, int_type * bucket, unsigned_type offset, unsigned shift)
00045     {
00046         for (type_ * p = begin; p < end; p++, out++) // count & create references
00047         {
00048             out->ptr = p;
00049             typename type_::key_type key = p->key();
00050             int_type ibucket = (key - offset) >> shift;
00051             out->key = key;
00052             bucket[ibucket]++;
00053         }
00054     }
00055 
00056     template <typename type>
00057     struct type_key
00058     {
00059         typedef typename type::key_type key_type;
00060         key_type key;
00061         type * ptr;
00062 
00063         type_key() { }
00064         type_key(key_type k, type * p) : key(k), ptr(p)
00065         { }
00066     };
00067 
00068     template <typename type>
00069     bool operator < (const type_key<type> & a, const type_key<type> & b)
00070     {
00071         return a.key < b.key;
00072     }
00073 
00074     template <typename type>
00075     bool operator > (const type_key<type> & a, const type_key<type> & b)
00076     {
00077         return a.key > b.key;
00078     }
00079 
00080 
00081     template <typename BIDType_, typename AllocStrategy_>
00082     class bid_sequence
00083     {
00084     public:
00085         typedef BIDType_ bid_type;
00086         typedef bid_type & reference;
00087         typedef AllocStrategy_ alloc_strategy;
00088         typedef typename simple_vector<bid_type>::size_type size_type;
00089         typedef typename simple_vector<bid_type>::iterator iterator;
00090 
00091     protected:
00092         simple_vector<bid_type> * bids;
00093         alloc_strategy alloc_strategy_;
00094 
00095     public:
00096         bid_sequence() { }
00097         bid_sequence(size_type size_)
00098         {
00099             bids = new simple_vector<bid_type>(size_);
00100             block_manager * mng = block_manager::get_instance();
00101             mng->new_blocks(alloc_strategy_, bids->begin(), bids->end());
00102         }
00103         void init(size_type size_)
00104         {
00105             bids = new simple_vector<bid_type>(size_);
00106             block_manager * mng = block_manager::get_instance();
00107             mng->new_blocks(alloc_strategy_, bids->begin(), bids->end());
00108         }
00109         reference operator [] (size_type i)
00110         {
00111             size_type size_ = size();             // cache size in a register
00112             if (i < size_)
00113                 return *(bids->begin() + i);
00114 
00115             block_manager * mng = block_manager::get_instance();
00116             simple_vector<bid_type> * larger_bids = new simple_vector<bid_type>((i + 1) * 2);
00117             std::copy(bids->begin(), bids->end(), larger_bids->begin());
00118             mng->new_blocks(alloc_strategy_, larger_bids->begin() + size_, larger_bids->end());
00119             delete bids;
00120             bids = larger_bids;
00121             return *(larger_bids->begin() + i);
00122         }
00123         size_type size() { return bids->size(); }
00124         iterator begin() { return bids->begin(); }
00125         ~bid_sequence()
00126         {
00127             block_manager::get_instance()->delete_blocks(bids->begin(), bids->end());
00128             delete bids;
00129         }
00130     };
00131 
00132     template <typename ExtIterator_>
00133     void distribute(
00134         bid_sequence<typename ExtIterator_::vector_type::block_type::bid_type,
00135                      typename ExtIterator_::vector_type::alloc_strategy_type> * bucket_bids,
00136         int64 * bucket_sizes,
00137         const int_type nbuckets,
00138         const int_type lognbuckets,
00139         ExtIterator_ first,
00140         ExtIterator_ last,
00141         const int_type nread_buffers,
00142         const int_type nwrite_buffers)
00143     {
00144         typedef typename ExtIterator_::vector_type::value_type value_type;
00145         typedef typename value_type::key_type key_type;
00146         typedef typename ExtIterator_::block_type block_type;
00147         typedef typename block_type::bid_type bid_type;
00148         typedef buf_istream<typename ExtIterator_::block_type,
00149                             typename ExtIterator_::bids_container_iterator> buf_istream_type;
00150 
00151         int_type i = 0;
00152 
00153         buf_istream_type in(first.bid(), last.bid() + ((first.block_offset()) ? 1 : 0),
00154                             nread_buffers);
00155 
00156         buffered_writer<block_type> out(
00157             nbuckets + nwrite_buffers,
00158             nwrite_buffers);
00159 
00160         unsigned_type * bucket_block_offsets = new unsigned_type[nbuckets];
00161         unsigned_type * bucket_iblock = new unsigned_type[nbuckets];
00162         block_type ** bucket_blocks = new block_type *[nbuckets];
00163 
00164         std::fill(bucket_sizes, bucket_sizes + nbuckets, 0);
00165         std::fill(bucket_iblock, bucket_iblock + nbuckets, 0);
00166         std::fill(bucket_block_offsets, bucket_block_offsets + nbuckets, 0);
00167 
00168         for (i = 0; i < nbuckets; i++)
00169             bucket_blocks[i] = out.get_free_block();
00170 
00171 
00172         ExtIterator_ cur = first - first.block_offset();
00173 
00174         // skip part of the block before first untouched
00175         for ( ; cur != first; cur++)
00176             ++in;
00177 
00178 
00179         const int_type shift = sizeof(key_type) * 8 - lognbuckets;
00180         // search in the the range [_begin,_end)
00181         STXXL_VERBOSE_STABLE_KSORT("Shift by: " << shift << " bits, lognbuckets: " << lognbuckets);
00182         for ( ; cur != last; cur++)
00183         {
00184             key_type cur_key = in.current().key();
00185             int_type ibucket = cur_key >> shift;
00186 
00187             int_type block_offset = bucket_block_offsets[ibucket];
00188             in >> (bucket_blocks[ibucket]->elem[block_offset++]);
00189             if (block_offset == block_type::size)
00190             {
00191                 block_offset = 0;
00192                 int_type iblock = bucket_iblock[ibucket]++;
00193                 bucket_blocks[ibucket] = out.write(bucket_blocks[ibucket], bucket_bids[ibucket][iblock]);
00194             }
00195             bucket_block_offsets[ibucket] = block_offset;
00196         }
00197         for (i = 0; i < nbuckets; i++)
00198         {
00199             if (bucket_block_offsets[i])
00200             {
00201                 out.write(bucket_blocks[i], bucket_bids[i][bucket_iblock[i]]);
00202             }
00203             bucket_sizes[i] = int64(block_type::size) * bucket_iblock[i] +
00204                               bucket_block_offsets[i];
00205             STXXL_VERBOSE_STABLE_KSORT("Bucket " << i << " has size " << bucket_sizes[i] <<
00206                                        ", estimated size: " << ((last - first) / int64(nbuckets)));
00207         }
00208 
00209         delete[] bucket_blocks;
00210         delete[] bucket_block_offsets;
00211         delete[] bucket_iblock;
00212     }
00213 }
00214 
00222 template <typename ExtIterator_>
00223 void stable_ksort(ExtIterator_ first, ExtIterator_ last, unsigned_type M)
00224 {
00225     STXXL_MSG("Warning: stable_ksort is not yet fully implemented, it assumes that the keys are uniformly distributed between [0,(std::numeric_limits<key_type>::max)()]");
00226     typedef typename ExtIterator_::vector_type::value_type value_type;
00227     typedef typename value_type::key_type key_type;
00228     typedef typename ExtIterator_::block_type block_type;
00229     typedef typename block_type::bid_type bid_type;
00230     typedef typename ExtIterator_::vector_type::alloc_strategy_type alloc_strategy;
00231     typedef stable_ksort_local::bid_sequence<bid_type, alloc_strategy> bucket_bids_type;
00232     typedef stable_ksort_local::type_key<value_type> type_key_;
00233 
00234     first.flush();     // flush container
00235 
00236     double begin = timestamp();
00237 
00238     unsigned_type i = 0;
00239     config * cfg = config::get_instance();
00240     const unsigned_type m = M / block_type::raw_size;
00241     assert(2 * block_type::raw_size <= M);
00242     const unsigned_type write_buffers_multiple = 2;
00243     const unsigned_type read_buffers_multiple = 2;
00244     const unsigned_type ndisks = cfg->disks_number();
00245     const unsigned_type min_num_read_write_buffers = (write_buffers_multiple + read_buffers_multiple) * ndisks;
00246     const unsigned_type nmaxbuckets = m - min_num_read_write_buffers;
00247     const unsigned_type lognbuckets = log2_floor(nmaxbuckets);
00248     const unsigned_type nbuckets = 1 << lognbuckets;
00249     const unsigned_type est_bucket_size = div_ceil((last - first) / nbuckets, block_type::size);      //in blocks
00250 
00251     if (m < min_num_read_write_buffers + 2 || nbuckets < 2) {
00252         STXXL_ERRMSG("stxxl::stable_ksort: Not enough memory. Blocks available: " << m <<
00253                      ", required for r/w buffers: " << min_num_read_write_buffers <<
00254                      ", required for buckets: 2, nbuckets: " << nbuckets);
00255         throw bad_parameter("stxxl::stable_ksort(): INSUFFICIENT MEMORY provided, please increase parameter 'M'");
00256     }
00257     STXXL_VERBOSE_STABLE_KSORT("Elements to sort: " << (last - first));
00258     STXXL_VERBOSE_STABLE_KSORT("Number of buckets has to be reduced from " << nmaxbuckets << " to " << nbuckets);
00259     const unsigned_type nread_buffers = (m - nbuckets) * read_buffers_multiple / (read_buffers_multiple + write_buffers_multiple);
00260     const unsigned_type nwrite_buffers = (m - nbuckets) * write_buffers_multiple / (read_buffers_multiple + write_buffers_multiple);
00261 
00262     STXXL_VERBOSE_STABLE_KSORT("Read buffers in distribution phase: " << nread_buffers);
00263     STXXL_VERBOSE_STABLE_KSORT("Write buffers in distribution phase: " << nwrite_buffers);
00264 
00265     bucket_bids_type * bucket_bids = new bucket_bids_type[nbuckets];
00266     for (i = 0; i < nbuckets; ++i)
00267         bucket_bids[i].init(est_bucket_size);
00268 
00269     int64 * bucket_sizes = new int64[nbuckets];
00270 
00271     disk_queues::get_instance()->set_priority_op(disk_queue::WRITE);
00272 
00273     stable_ksort_local::distribute(
00274         bucket_bids,
00275         bucket_sizes,
00276         nbuckets,
00277         lognbuckets,
00278         first,
00279         last,
00280         nread_buffers,
00281         nwrite_buffers);
00282 
00283     double dist_end = timestamp(), end;
00284     double io_wait_after_d = stats::get_instance()->get_io_wait_time();
00285 
00286     {
00287         // sort buckets
00288         unsigned_type write_buffers_multiple_bs = 2;
00289         unsigned_type max_bucket_size_bl = (m - write_buffers_multiple_bs * ndisks) / 2; // in number of blocks
00290         int64 max_bucket_size_rec = int64(max_bucket_size_bl) * block_type::size;        // in number of records
00291         int64 max_bucket_size_act = 0;                                                   // actual max bucket size
00292         // establish output stream
00293 
00294         for (i = 0; i < nbuckets; i++)
00295         {
00296             max_bucket_size_act = STXXL_MAX(bucket_sizes[i], max_bucket_size_act);
00297             if (bucket_sizes[i] > max_bucket_size_rec)
00298             {
00299                 STXXL_ERRMSG("Bucket " << i << " is too large: " << bucket_sizes[i] <<
00300                              " records, maximum: " << max_bucket_size_rec);
00301                 STXXL_ERRMSG("Recursion on buckets is not yet implemented, aborting.");
00302                 abort();
00303             }
00304         }
00305         // here we can increase write_buffers_multiple_b knowing max(bucket_sizes[i])
00306         // ... and decrease max_bucket_size_bl
00307         const int_type max_bucket_size_act_bl = div_ceil(max_bucket_size_act, block_type::size);
00308         STXXL_VERBOSE_STABLE_KSORT("Reducing required number of required blocks per bucket from " <<
00309                                    max_bucket_size_bl << " to " << max_bucket_size_act_bl);
00310         max_bucket_size_rec = max_bucket_size_act;
00311         max_bucket_size_bl = max_bucket_size_act_bl;
00312         const unsigned_type nwrite_buffers_bs = m - 2 * max_bucket_size_bl;
00313         STXXL_VERBOSE_STABLE_KSORT("Write buffers in bucket sorting phase: " << nwrite_buffers_bs);
00314 
00315         typedef buf_ostream<block_type, typename ExtIterator_::bids_container_iterator> buf_ostream_type;
00316         buf_ostream_type out(first.bid(), nwrite_buffers_bs);
00317 
00318         disk_queues::get_instance()->set_priority_op(disk_queue::READ);
00319 
00320         if (first.block_offset())
00321         {
00322             // has to skip part of the first block
00323             block_type * block = new block_type;
00324             request_ptr req;
00325             req = block->read(*first.bid());
00326             req->wait();
00327 
00328             for (i = 0; i < first.block_offset(); i++)
00329             {
00330                 out << block->elem[i];
00331             }
00332             delete block;
00333         }
00334         block_type * blocks1 = new block_type[max_bucket_size_bl];
00335         block_type * blocks2 = new block_type[max_bucket_size_bl];
00336         request_ptr * reqs1 = new request_ptr[max_bucket_size_bl];
00337         request_ptr * reqs2 = new request_ptr[max_bucket_size_bl];
00338         type_key_ * refs1 = new type_key_[max_bucket_size_rec];
00339         type_key_ * refs2 = new type_key_[max_bucket_size_rec];
00340 
00341         // submit reading first 2 buckets (Peter's scheme)
00342         unsigned_type nbucket_blocks = div_ceil(bucket_sizes[0], block_type::size);
00343         for (i = 0; i < nbucket_blocks; i++)
00344             reqs1[i] = blocks1[i].read(bucket_bids[0][i]);
00345 
00346 
00347         nbucket_blocks = div_ceil(bucket_sizes[1], block_type::size);
00348         for (i = 0; i < nbucket_blocks; i++)
00349             reqs2[i] = blocks2[i].read(bucket_bids[1][i]);
00350 
00351 
00352         key_type offset = 0;
00353         const unsigned log_k1 = STXXL_MAX<unsigned>(log2_ceil(max_bucket_size_rec * sizeof(type_key_) / STXXL_L2_SIZE), 1);
00354         unsigned_type k1 = 1 << log_k1;
00355         int_type * bucket1 = new int_type[k1];
00356 
00357         const unsigned shift = sizeof(key_type) * 8 - lognbuckets;
00358         const unsigned shift1 = shift - log_k1;
00359 
00360         STXXL_VERBOSE_STABLE_KSORT("Classifying " << nbuckets << " buckets, max size:" << max_bucket_size_rec <<
00361                                    " block size:" << block_type::size << " log_k1:" << log_k1);
00362 
00363         for (unsigned_type k = 0; k < nbuckets; k++)
00364         {
00365             nbucket_blocks = div_ceil(bucket_sizes[k], block_type::size);
00366             const unsigned log_k1_k = STXXL_MAX<unsigned>(log2_ceil(bucket_sizes[k] * sizeof(type_key_) / STXXL_L2_SIZE), 1);
00367             assert(log_k1_k <= log_k1);
00368             k1 = 1 << log_k1_k;
00369             std::fill(bucket1, bucket1 + k1, 0);
00370 
00371             STXXL_VERBOSE_STABLE_KSORT("Classifying bucket " << k << " size:" << bucket_sizes[k] <<
00372                                        " blocks:" << nbucket_blocks << " log_k1:" << log_k1_k);
00373             // classify first nbucket_blocks-1 blocks, they are full
00374             type_key_ * ref_ptr = refs1;
00375             key_type offset1 = offset + (key_type(1) << key_type(shift)) * key_type(k);
00376             for (i = 0; i < nbucket_blocks - 1; i++)
00377             {
00378                 reqs1[i]->wait();
00379                 stable_ksort_local::classify_block(blocks1[i].begin(), blocks1[i].end(), ref_ptr, bucket1, offset1, shift1 /*,k1*/);
00380             }
00381             // last block might be non-full
00382             const unsigned_type last_block_size = bucket_sizes[k] - int64(nbucket_blocks - 1) * block_type::size;
00383             reqs1[i]->wait();
00384 
00385             //STXXL_MSG("block_type::size: "<<block_type::size<<" last_block_size:"<<last_block_size);
00386 
00387             classify_block(blocks1[i].begin(), blocks1[i].begin() + last_block_size, ref_ptr, bucket1, offset1, shift1);
00388 
00389             exclusive_prefix_sum(bucket1, k1);
00390             classify(refs1, refs1 + bucket_sizes[k], refs2, bucket1, offset1, shift1);
00391 
00392             type_key_ * c = refs2;
00393             type_key_ * d = refs1;
00394             for (i = 0; i < k1; i++)
00395             {
00396                 type_key_ * cEnd = refs2 + bucket1[i];
00397                 type_key_ * dEnd = refs1 + bucket1[i];
00398 
00399                 const unsigned log_k2 = log2_floor(bucket1[i]) - 1;        // adaptive bucket size
00400                 const unsigned_type k2 = 1 << log_k2;
00401                 int_type * bucket2 = new int_type[k2];
00402                 const unsigned shift2 = shift1 - log_k2;
00403 
00404                 // STXXL_MSG("Sorting bucket "<<k<<":"<<i);
00405                 l1sort(c, cEnd, d, bucket2, k2,
00406                        offset1 + (key_type(1) << key_type(shift1)) * key_type(i),
00407                        shift2);
00408 
00409                 // write out all
00410                 for (type_key_ * p = d; p < dEnd; p++)
00411                     out << (*(p->ptr));
00412 
00413 
00414                 delete[] bucket2;
00415                 c = cEnd;
00416                 d = dEnd;
00417             }
00418             // submit next read
00419             const unsigned_type bucket2submit = k + 2;
00420             if (bucket2submit < nbuckets)
00421             {
00422                 nbucket_blocks = div_ceil(bucket_sizes[bucket2submit], block_type::size);
00423                 for (i = 0; i < nbucket_blocks; i++)
00424                     reqs1[i] = blocks1[i].read(bucket_bids[bucket2submit][i]);
00425             }
00426 
00427             std::swap(blocks1, blocks2);
00428             std::swap(reqs1, reqs2);
00429         }
00430 
00431         delete[] bucket1;
00432         delete[] refs1;
00433         delete[] refs2;
00434         delete[] blocks1;
00435         delete[] blocks2;
00436         delete[] reqs1;
00437         delete[] reqs2;
00438         delete[] bucket_bids;
00439         delete[] bucket_sizes;
00440 
00441         if (last.block_offset())
00442         {
00443             // has to skip part of the first block
00444             block_type * block = new block_type;
00445             request_ptr req = block->read(*last.bid());
00446             req->wait();
00447 
00448             for (i = last.block_offset(); i < block_type::size; i++)
00449             {
00450                 out << block->elem[i];
00451             }
00452             delete block;
00453         }
00454 
00455         end = timestamp();
00456     }
00457 
00458     STXXL_VERBOSE("Elapsed time        : " << end - begin << " s. Distribution time: " <<
00459                   dist_end - begin << " s");
00460     STXXL_VERBOSE("Time in I/O wait(ds): " << io_wait_after_d << " s");
00461     STXXL_VERBOSE(*stats::get_instance());
00462     STXXL_UNUSED(begin + dist_end + end + io_wait_after_d);
00463 }
00464 
00466 
00467 __STXXL_END_NAMESPACE
00468 
00469 #endif // !STXXL_STABLE_KSORT_HEADER

Generated on Sun Oct 17 2010 06:13:44 for Stxxl by  doxygen 1.7.1