14 #ifndef STXXL_ALGO_STABLE_KSORT_HEADER
15 #define STXXL_ALGO_STABLE_KSORT_HEADER
30 #ifndef STXXL_VERBOSE_STABLE_KSORT
31 #define STXXL_VERBOSE_STABLE_KSORT STXXL_VERBOSE1
42 namespace stable_ksort_local {
44 template <
class type_,
class type_key>
47 for (type_* p = begin; p < end; p++, out++)
50 typename type_::key_type key = p->key();
51 int_type ibucket = (key - offset) >> shift;
57 template <
typename type>
69 template <
typename type>
75 template <
typename type>
82 template <
typename BIDType_,
typename AllocStrategy_>
102 mng->new_blocks(alloc_strategy_, bids->begin(), bids->end());
108 mng->new_blocks(alloc_strategy_, bids->begin(), bids->end());
114 return *(bids->begin() + i);
118 std::copy(bids->begin(), bids->end(), larger_bids->
begin());
122 return *(larger_bids->
begin() + i);
128 block_manager::get_instance()->delete_blocks(bids->begin(), bids->end());
133 template <
typename ExtIterator_>
135 bid_sequence<
typename ExtIterator_::vector_type::block_type::bid_type,
136 typename ExtIterator_::vector_type::alloc_strategy_type>* bucket_bids,
145 typedef typename ExtIterator_::vector_type::value_type value_type;
146 typedef typename value_type::key_type key_type;
147 typedef typename ExtIterator_::block_type block_type;
148 typedef buf_istream<
typename ExtIterator_::block_type,
149 typename ExtIterator_::bids_container_iterator> buf_istream_type;
153 buf_istream_type in(first.bid(), last.bid() + ((first.block_offset()) ? 1 : 0),
157 nbuckets + nwrite_buffers,
162 block_type** bucket_blocks =
new block_type*[nbuckets];
164 std::fill(bucket_sizes, bucket_sizes + nbuckets, 0);
165 std::fill(bucket_iblock, bucket_iblock + nbuckets, 0);
166 std::fill(bucket_block_offsets, bucket_block_offsets + nbuckets, 0);
168 for (i = 0; i < nbuckets; i++)
169 bucket_blocks[i] = out.get_free_block();
172 ExtIterator_ cur = first - first.block_offset();
175 for ( ; cur != first; cur++)
179 const int_type shift =
sizeof(key_type) * 8 - lognbuckets;
182 for ( ; cur != last; cur++)
184 key_type cur_key = in.current().key();
185 int_type ibucket = cur_key >> shift;
187 int_type block_offset = bucket_block_offsets[ibucket];
188 in >> (bucket_blocks[ibucket]->elem[block_offset++]);
189 if (block_offset == block_type::size)
192 int_type iblock = bucket_iblock[ibucket]++;
193 bucket_blocks[ibucket] = out.write(bucket_blocks[ibucket], bucket_bids[ibucket][iblock]);
195 bucket_block_offsets[ibucket] = block_offset;
197 for (i = 0; i < nbuckets; i++)
199 if (bucket_block_offsets[i])
201 out.write(bucket_blocks[i], bucket_bids[i][bucket_iblock[i]]);
203 bucket_sizes[i] =
int64(block_type::size) * bucket_iblock[i] +
204 bucket_block_offsets[i];
206 ", estimated size: " << ((last - first) /
int64(nbuckets)));
209 delete[] bucket_blocks;
210 delete[] bucket_block_offsets;
211 delete[] bucket_iblock;
223 template <
typename ExtIterator_>
226 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()]");
227 typedef typename ExtIterator_::vector_type::value_type value_type;
228 typedef typename value_type::key_type key_type;
229 typedef typename ExtIterator_::block_type block_type;
230 typedef typename block_type::bid_type bid_type;
231 typedef typename ExtIterator_::vector_type::alloc_strategy_type alloc_strategy;
240 config* cfg = config::get_instance();
242 assert(2 * block_type::raw_size <= M);
246 const unsigned_type min_num_read_write_buffers = (write_buffers_multiple + read_buffers_multiple) * ndisks;
247 const unsigned_type nmaxbuckets = m - min_num_read_write_buffers;
248 const unsigned int lognbuckets =
ilog2_floor(nmaxbuckets);
250 const uint64 est_bucket_size =
div_ceil((last - first) / nbuckets, block_type::size);
252 if (m < min_num_read_write_buffers + 2 || nbuckets < 2) {
253 STXXL_ERRMSG(
"stxxl::stable_ksort: Not enough memory. Blocks available: " << m <<
254 ", required for r/w buffers: " << min_num_read_write_buffers <<
255 ", required for buckets: 2, nbuckets: " << nbuckets);
256 throw bad_parameter(
"stxxl::stable_ksort(): INSUFFICIENT MEMORY provided, please increase parameter 'M'");
260 const unsigned_type nread_buffers = (m - nbuckets) * read_buffers_multiple / (read_buffers_multiple + write_buffers_multiple);
261 const unsigned_type nwrite_buffers = (m - nbuckets) * write_buffers_multiple / (read_buffers_multiple + write_buffers_multiple);
266 bucket_bids_type* bucket_bids =
new bucket_bids_type[nbuckets];
267 for (i = 0; i < nbuckets; ++i)
268 bucket_bids[i].init(est_bucket_size);
272 disk_queues::get_instance()->set_priority_op(request_queue::WRITE);
285 double io_wait_after_d = stats::get_instance()->get_io_wait_time();
290 unsigned_type max_bucket_size_bl = (m - write_buffers_multiple_bs * ndisks) / 2;
291 int64 max_bucket_size_rec =
int64(max_bucket_size_bl) * block_type::size;
292 int64 max_bucket_size_act = 0;
295 for (i = 0; i < nbuckets; i++)
297 max_bucket_size_act =
STXXL_MAX(bucket_sizes[i], max_bucket_size_act);
298 if (bucket_sizes[i] > max_bucket_size_rec)
300 STXXL_ERRMSG(
"Bucket " << i <<
" is too large: " << bucket_sizes[i] <<
301 " records, maximum: " << max_bucket_size_rec);
302 STXXL_ERRMSG(
"Recursion on buckets is not yet implemented, aborting.");
308 const int_type max_bucket_size_act_bl =
div_ceil(max_bucket_size_act, block_type::size);
310 max_bucket_size_bl <<
" to " << max_bucket_size_act_bl);
311 max_bucket_size_rec = max_bucket_size_act;
312 max_bucket_size_bl = max_bucket_size_act_bl;
313 const unsigned_type nwrite_buffers_bs = m - 2 * max_bucket_size_bl;
317 buf_ostream_type out(first.bid(), nwrite_buffers_bs);
319 disk_queues::get_instance()->set_priority_op(request_queue::READ);
321 if (first.block_offset())
324 block_type* block =
new block_type;
326 req = block->read(*first.bid());
329 for (i = 0; i < first.block_offset(); i++)
331 out << block->elem[i];
335 block_type* blocks1 =
new block_type[max_bucket_size_bl];
336 block_type* blocks2 =
new block_type[max_bucket_size_bl];
339 type_key_* refs1 =
new type_key_[max_bucket_size_rec];
340 type_key_* refs2 =
new type_key_[max_bucket_size_rec];
344 for (i = 0; i < nbucket_blocks; i++)
345 reqs1[i] = blocks1[i].read(bucket_bids[0][i]);
348 nbucket_blocks =
div_ceil(bucket_sizes[1], block_type::size);
349 for (i = 0; i < nbucket_blocks; i++)
350 reqs2[i] = blocks2[i].read(bucket_bids[1][i]);
354 const unsigned log_k1 = STXXL_MAX<unsigned>(
ilog2_ceil(max_bucket_size_rec *
sizeof(type_key_) /
STXXL_L2_SIZE), 1);
358 const unsigned int shift = (
unsigned int)(
sizeof(key_type) * 8 - lognbuckets);
359 const unsigned int shift1 = shift - log_k1;
362 " block size:" << block_type::size <<
" log_k1:" << log_k1);
366 nbucket_blocks =
div_ceil(bucket_sizes[k], block_type::size);
367 const unsigned log_k1_k = STXXL_MAX<unsigned>(
ilog2_ceil(bucket_sizes[k] *
sizeof(type_key_) /
STXXL_L2_SIZE), 1);
368 assert(log_k1_k <= log_k1);
370 std::fill(bucket1, bucket1 + k1, 0);
373 " blocks:" << nbucket_blocks <<
" log_k1:" << log_k1_k);
375 type_key_* ref_ptr = refs1;
376 key_type offset1 = offset + (key_type(1) << key_type(shift)) * key_type(k);
377 for (i = 0; i < nbucket_blocks - 1; i++)
383 const unsigned_type last_block_size = bucket_sizes[k] -
int64(nbucket_blocks - 1) * block_type::size;
388 classify_block(blocks1[i].begin(), blocks1[i].begin() + last_block_size, ref_ptr, bucket1, offset1, shift1);
391 classify(refs1, refs1 + bucket_sizes[k], refs2, bucket1, offset1, shift1);
393 type_key_* c = refs2;
394 type_key_* d = refs1;
395 for (i = 0; i < k1; i++)
397 type_key_* cEnd = refs2 + bucket1[i];
398 type_key_* dEnd = refs1 + bucket1[i];
400 const unsigned log_k2 =
ilog2_floor(bucket1[i]) - 1;
403 const unsigned shift2 = shift1 - log_k2;
406 l1sort(c, cEnd, d, bucket2, k2,
407 offset1 + (key_type(1) << key_type(shift1)) * key_type(i),
411 for (type_key_* p = d; p < dEnd; p++)
421 if (bucket2submit < nbuckets)
423 nbucket_blocks =
div_ceil(bucket_sizes[bucket2submit], block_type::size);
424 for (i = 0; i < nbucket_blocks; i++)
425 reqs1[i] = blocks1[i].read(bucket_bids[bucket2submit][i]);
428 std::swap(blocks1, blocks2);
429 std::swap(reqs1, reqs2);
439 delete[] bucket_bids;
440 delete[] bucket_sizes;
442 if (last.block_offset())
445 block_type* block =
new block_type;
449 for (i = last.block_offset(); i < block_type::size; i++)
451 out << block->elem[i];
459 STXXL_VERBOSE(
"Elapsed time : " << end - begin <<
" s. Distribution time: " <<
460 dist_end - begin <<
" s");
461 STXXL_VERBOSE(
"Time in I/O wait(ds): " << io_wait_after_d <<
" s");
470 #endif // !STXXL_ALGO_STABLE_KSORT_HEADER
#define STXXL_VERBOSE_STABLE_KSORT
unsigned long long int uint64
alloc_strategy alloc_strategy_
type_key(key_type k, type *p)
iterator end()
return mutable iterator beyond last element
simple_vector< bid_type >::iterator iterator
bid_sequence(size_type size_)
virtual void wait(bool measure_time=true)=0
Suspends calling thread until completion of the request.
AllocStrategy_ alloc_strategy
double timestamp()
Returns number of seconds since the epoch, high resolution.
const Tp & STXXL_MAX(const Tp &a, const Tp &b)
choose_int_types< my_pointer_size >::int_type int_type
bool operator>(const uint_pair &b) const
greater comparison operator
iterator begin()
return mutable iterator to first element
size_t disks_number()
Returns number of disks available to user.
simple_vector< bid_type > * bids
#define STXXL_BEGIN_NAMESPACE
void STXXL_UNUSED(const U &)
static void classify(type_key *a, type_key *aEnd, type_key *b, int_type *bucket, typename type_key::key_type offset, unsigned shift)
void init(size_type size_)
void distribute(bid_sequence< typename ExtIterator_::vector_type::block_type::bid_type, typename ExtIterator_::vector_type::alloc_strategy_type > *bucket_bids, int64 *bucket_sizes, const int_type nbuckets, const int_type lognbuckets, ExtIterator_ first, ExtIterator_ last, const int_type nread_buffers, const int_type nwrite_buffers)
simple_vector< bid_type >::size_type size_type
void l1sort(type_key *a, type_key *aEnd, type_key *b, int_type *bucket, int_type K, typename type_key::key_type offset, int shift)
unsigned int ilog2_floor(IntegerType i)
calculate the log2 floor of an integer type (by repeated bit shifts)
static void exclusive_prefix_sum(int_type *bucket, int_type K)
void new_blocks(const DiskAssignFunctor &functor, BIDIteratorClass bidbegin, BIDIteratorClass bidend, unsigned_type offset=0)
Allocates new blocks.
Simpler non-growing vector without initialization.
choose_int_types< my_pointer_size >::unsigned_type unsigned_type
unsigned int ilog2_ceil(const IntegerType &i)
calculate the log2 ceiling of an integer type (by repeated bit shifts)
Access point to disks properties. Since 1.4.0: no config files are read automatically! ...
compat::remove_const< Integral >::type div_ceil(Integral __n, Integral2 __d)
Encapsulates asynchronous buffered block writing engine.
void classify_block(type *begin, type *end, type_key *&out, int_type *bucket, typename key_extractor::key_type offset, unsigned shift, key_extractor keyobj)
void stable_ksort(ExtIterator_ first, ExtIterator_ last, unsigned_type M)
Sort records with integer keys.
#define STXXL_END_NAMESPACE