//----------------------------------------------------------------------------
|
/// @file merge_blocks.hpp
|
/// @brief contains the class merge_blocks, which is part of the
|
/// block_indirect_sort algorithm
|
///
|
/// @author Copyright (c) 2016 Francisco Jose Tapia (fjtapia@gmail.com )\n
|
/// Distributed under the Boost Software License, Version 1.0.\n
|
/// ( See accompanying file LICENSE_1_0.txt or copy at
|
/// http://www.boost.org/LICENSE_1_0.txt )
|
/// @version 0.1
|
///
|
/// @remarks
|
//-----------------------------------------------------------------------------
|
#ifndef __BOOST_SORT_PARALLEL_DETAIL_MERGE_BLOCKS_HPP
|
#define __BOOST_SORT_PARALLEL_DETAIL_MERGE_BLOCKS_HPP
|
|
#include <atomic>
|
#include <boost/sort/block_indirect_sort/blk_detail/backbone.hpp>
|
#include <boost/sort/common/range.hpp>
|
#include <future>
|
#include <iostream>
|
#include <iterator>
|
|
namespace boost
|
{
|
namespace sort
|
{
|
namespace blk_detail
|
{
|
//----------------------------------------------------------------------------
|
// USING SENTENCES
|
//----------------------------------------------------------------------------
|
namespace bsc = boost::sort::common;
|
namespace bscu = bsc::util;
|
using bsc::range;
|
using bsc::is_mergeable;
|
using bsc::merge_uncontiguous;
|
//
|
///---------------------------------------------------------------------------
|
/// @struct merge_blocks
|
/// @brief This class merge the blocks. The blocks to merge are defined by two
|
/// ranges of positions in the index of the backbone
|
//----------------------------------------------------------------------------
|
template<uint32_t Block_size, uint32_t Group_size, class Iter_t, class Compare>
|
struct merge_blocks
|
{
|
//-----------------------------------------------------------------------
|
// D E F I N I T I O N S
|
//-----------------------------------------------------------------------
|
typedef typename std::iterator_traits<Iter_t>::value_type value_t;
|
typedef std::atomic<uint32_t> atomic_t;
|
typedef range<size_t> range_pos;
|
typedef range<Iter_t> range_it;
|
typedef range<value_t *> range_buf;
|
typedef std::function<void(void)> function_t;
|
typedef backbone<Block_size, Iter_t, Compare> backbone_t;
|
typedef compare_block_pos<Block_size, Iter_t, Compare> compare_block_pos_t;
|
|
//------------------------------------------------------------------------
|
// V A R I A B L E S
|
//------------------------------------------------------------------------
|
// Object with the elements to sort and all internal data structures of the
|
// algorithm
|
backbone_t &bk;
|
//
|
//------------------------------------------------------------------------
|
// F U N C T I O N S
|
//------------------------------------------------------------------------
|
merge_blocks(backbone_t &bkb, size_t pos_index1, size_t pos_index2,
|
size_t pos_index3);
|
|
void tail_process(std::vector<block_pos> &vblkpos1,
|
std::vector<block_pos> &vblkpos2);
|
|
void cut_range(range_pos rng);
|
|
void merge_range_pos(range_pos rng);
|
|
void extract_ranges(range_pos range_input);
|
//
|
//------------------------------------------------------------------------
|
// function : function_merge_range_pos
|
/// @brief create a function_t with a call to merge_range_pos, and insert
|
/// in the stack of the backbone
|
//
|
/// @param rng_input : range of positions of blocks in the index to merge
|
/// @param son_counter : atomic variable which is decremented when finish
|
/// the function. This variable is used for to know
|
/// when are finished all the function_t created
|
/// inside an object
|
/// @param error : global indicator of error.
|
///
|
//------------------------------------------------------------------------
|
void function_merge_range_pos(const range_pos &rng_input, atomic_t &counter,
|
bool &error)
|
{
|
bscu::atomic_add(counter, 1);
|
function_t f1 = [this, rng_input, &counter, &error]( ) -> void
|
{
|
if (not error)
|
{
|
try
|
{
|
this->merge_range_pos (rng_input);
|
}
|
catch (std::bad_alloc &ba)
|
{
|
error = true;
|
};
|
}
|
bscu::atomic_sub (counter, 1);
|
};
|
bk.works.emplace_back(f1);
|
}
|
;
|
//
|
//------------------------------------------------------------------------
|
// function : function_cut_range
|
/// @brief create a function_t with a call to cut_range, and inser in
|
/// the stack of the backbone
|
//
|
/// @param rng_input : range of positions in the index to cut
|
/// @param counter : atomic variable which is decremented when finish
|
/// the function. This variable is used for to know
|
/// when are finished all the function_t created
|
/// inside an object
|
/// @param error : global indicator of error.
|
//------------------------------------------------------------------------
|
void function_cut_range(const range_pos &rng_input, atomic_t &counter,
|
bool &error)
|
{
|
bscu::atomic_add(counter, 1);
|
function_t f1 = [this, rng_input, &counter, &error]( ) -> void
|
{
|
if (not error)
|
{
|
try
|
{
|
this->cut_range (rng_input);
|
}
|
catch (std::bad_alloc &)
|
{
|
error = true;
|
};
|
}
|
bscu::atomic_sub (counter, 1);
|
};
|
bk.works.emplace_back(f1);
|
}
|
|
|
//----------------------------------------------------------------------------
|
};
|
// end struct merge_blocks
|
//----------------------------------------------------------------------------
|
//
|
//############################################################################
|
// ##
|
// ##
|
// N O N I N L I N E F U N C T I O N S ##
|
// ##
|
// ##
|
//############################################################################
|
//
|
//-------------------------------------------------------------------------
|
// function : merge_blocks
|
/// @brief make the indirect merge of the two range_pos defined by their index
|
/// position [pos_index1, pos_index2 ) and [ pos_index2, pos_index3 )
|
//
|
/// @param bkb : backbone with all the data to sort , and the internal data
|
/// structures of the algorithm
|
/// @param pos_index1 : first position of the first range in the index
|
/// @param pos_index2 : last position of the first range and first position
|
/// of the second range in the index
|
/// @param pos_index3 : last position of the second range in the index
|
//-------------------------------------------------------------------------
|
template<uint32_t Block_size, uint32_t Group_size, class Iter_t, class Compare>
|
merge_blocks<Block_size, Group_size, Iter_t, Compare>
|
::merge_blocks( backbone_t &bkb, size_t pos_index1, size_t pos_index2,
|
size_t pos_index3) : bk(bkb)
|
{
|
size_t nblock1 = pos_index2 - pos_index1;
|
size_t nblock2 = pos_index3 - pos_index2;
|
if (nblock1 == 0 or nblock2 == 0) return;
|
|
//-----------------------------------------------------------------------
|
// Merging of the two intervals
|
//-----------------------------------------------------------------------
|
std::vector<block_pos> vpos1, vpos2;
|
vpos1.reserve(nblock1 + 1);
|
vpos2.reserve(nblock2 + 1);
|
|
for (size_t i = pos_index1; i < pos_index2; ++i)
|
{
|
vpos1.emplace_back(bk.index[i].pos(), true);
|
};
|
|
for (size_t i = pos_index2; i < pos_index3; ++i)
|
{
|
vpos2.emplace_back(bk.index[i].pos(), false);
|
};
|
//-------------------------------------------------------------------
|
// tail process
|
//-------------------------------------------------------------------
|
if (vpos2.back().pos() == (bk.nblock - 1)
|
and bk.range_tail.first != bk.range_tail.last)
|
{
|
tail_process(vpos1, vpos2);
|
nblock1 = vpos1.size();
|
nblock2 = vpos2.size();
|
};
|
|
compare_block_pos_t cmp_blk(bk.global_range.first, bk.cmp);
|
if (bk.error) return;
|
bscu::merge(vpos1.begin(), vpos1.end(), vpos2.begin(), vpos2.end(),
|
bk.index.begin() + pos_index1, cmp_blk);
|
if (bk.error) return;
|
// Extracting the ranges for to merge the elements
|
extract_ranges(range_pos(pos_index1, pos_index1 + nblock1 + nblock2));
|
}
|
|
|
//
|
//-------------------------------------------------------------------------
|
// function : tail_process
|
/// @brief make the process when the second vector of block_pos to merge is
|
/// the last, and have an incomplete block ( tail)
|
//
|
/// @param vblkpos1 : first vector of block_pos elements to merge
|
/// @param vblkpos2 : second vector of block_pos elements to merge
|
//-------------------------------------------------------------------------
|
template<uint32_t Block_size, uint32_t Group_size, class Iter_t, class Compare>
|
void merge_blocks<Block_size, Group_size, Iter_t, Compare>
|
::tail_process( std::vector<block_pos> &vblkpos1,
|
std::vector<block_pos> &vblkpos2 )
|
{
|
if (vblkpos1.size() == 0 or vblkpos2.size() == 0) return;
|
|
vblkpos2.pop_back();
|
|
size_t posback1 = vblkpos1.back().pos();
|
range_it range_back1 = bk.get_range(posback1);
|
|
if (bsc::is_mergeable(range_back1, bk.range_tail, bk.cmp))
|
{
|
bsc::merge_uncontiguous(range_back1, bk.range_tail, bk.get_range_buf(),
|
bk.cmp);
|
if (vblkpos1.size() > 1)
|
{
|
size_t pos_aux = vblkpos1[vblkpos1.size() - 2].pos();
|
range_it range_aux = bk.get_range(pos_aux);
|
|
if (bsc::is_mergeable(range_aux, range_back1, bk.cmp))
|
{
|
vblkpos2.emplace_back(posback1, false);
|
vblkpos1.pop_back();
|
};
|
};
|
};
|
}
|
|
//
|
//-------------------------------------------------------------------------
|
// function : cut_range
|
/// @brief when the rng_input is greather than Group_size, this function divide
|
/// it in several parts creating function_t elements, which are inserted
|
/// in the concurrent stack of the backbone
|
//
|
/// @param rng_input : range to divide
|
//-------------------------------------------------------------------------
|
template<uint32_t Block_size, uint32_t Group_size, class Iter_t, class Compare>
|
void merge_blocks<Block_size, Group_size, Iter_t, Compare>
|
::cut_range(range_pos rng_input)
|
{
|
if (rng_input.size() < Group_size)
|
{
|
merge_range_pos(rng_input);
|
return;
|
};
|
|
atomic_t counter(0);
|
size_t npart = (rng_input.size() + Group_size - 1) / Group_size;
|
size_t size_part = rng_input.size() / npart;
|
|
size_t pos_ini = rng_input.first;
|
size_t pos_last = rng_input.last;
|
|
while (pos_ini < pos_last)
|
{
|
size_t pos = pos_ini + size_part;
|
while (pos < pos_last
|
and bk.index[pos - 1].side() == bk.index[pos].side())
|
{
|
++pos;
|
};
|
if (pos < pos_last)
|
{
|
merge_uncontiguous(bk.get_range(bk.index[pos - 1].pos()),
|
bk.get_range(bk.index[pos].pos()),
|
bk.get_range_buf(), bk.cmp);
|
}
|
else pos = pos_last;
|
if ((pos - pos_ini) > 1)
|
{
|
range_pos rng_aux(pos_ini, pos);
|
function_merge_range_pos(rng_aux, counter, bk.error);
|
};
|
pos_ini = pos;
|
};
|
bk.exec(counter); // wait until finish all the ranges
|
}
|
|
|
//
|
//-------------------------------------------------------------------------
|
// function : merge_range_pos
|
/// @brief make the indirect merge of the blocks inside the rng_input
|
//
|
/// @param rng_input : range of positions of the blocks to merge
|
//-------------------------------------------------------------------------
|
template<uint32_t Block_size, uint32_t Group_size, class Iter_t, class Compare>
|
void merge_blocks<Block_size, Group_size, Iter_t, Compare>
|
::merge_range_pos(range_pos rng_input)
|
{
|
if (rng_input.size() < 2) return;
|
range_buf rbuf = bk.get_range_buf();
|
|
range_it rng_prev = bk.get_range(bk.index[rng_input.first].pos());
|
move_forward(rbuf, rng_prev);
|
range_it rng_posx(rng_prev);
|
|
for (size_t posx = rng_input.first + 1; posx != rng_input.last; ++posx)
|
{
|
rng_posx = bk.get_range(bk.index[posx].pos());
|
bsc::merge_flow(rng_prev, rbuf, rng_posx, bk.cmp);
|
rng_prev = rng_posx;
|
|
};
|
move_forward(rng_posx, rbuf);
|
}
|
//
|
//-------------------------------------------------------------------------
|
// function : extract_ranges
|
/// @brief from a big range of positions of blocks in the index. Examine which
|
/// are mergeable, and generate a couple of ranges for to be merged.
|
/// With the ranges obtained generate function_t elements and are
|
/// inserted in the concurrent stack.
|
/// When the range obtained is smaller than Group_size, generate a
|
/// function_t calling to merge_range_pos, when is greater, generate a
|
/// function_t calling to cut_range
|
//
|
/// @param rpos range_input : range of the position in the index, where must
|
/// extract the ranges to merge
|
//-------------------------------------------------------------------------
|
template<uint32_t Block_size, uint32_t Group_size, class Iter_t, class Compare>
|
void merge_blocks<Block_size, Group_size, Iter_t, Compare>
|
::extract_ranges(range_pos range_input)
|
{
|
if (range_input.size() < 2) return;
|
atomic_t counter(0);
|
|
// The names with x are positions of the index
|
size_t posx_ini = range_input.first;
|
block_pos bp_posx_ini = bk.index[posx_ini];
|
|
range_it rng_max = bk.get_range(bp_posx_ini.pos());
|
bool side_max = bp_posx_ini.side();
|
|
block_pos bp_posx;
|
range_it rng_posx = rng_max;
|
bool side_posx = side_max;
|
|
for (size_t posx = posx_ini + 1; posx <= range_input.last; ++posx)
|
{
|
bool final = (posx == range_input.last);
|
bool mergeable = false;
|
|
if (not final)
|
{
|
bp_posx = bk.index[posx];
|
rng_posx = bk.get_range(bp_posx.pos());
|
side_posx = bp_posx.side();
|
mergeable = (side_max != side_posx
|
and is_mergeable(rng_max, rng_posx, bk.cmp));
|
};
|
if (bk.error) return;
|
if (final or not mergeable)
|
{
|
range_pos rp_final(posx_ini, posx);
|
if (rp_final.size() > 1)
|
{
|
if (rp_final.size() > Group_size)
|
{
|
function_cut_range(rp_final, counter, bk.error);
|
}
|
else
|
{
|
function_merge_range_pos(rp_final, counter, bk.error);
|
};
|
};
|
posx_ini = posx;
|
if (not final)
|
{
|
rng_max = rng_posx;
|
side_max = side_posx;
|
};
|
}
|
else
|
{
|
if (bk.cmp(*(rng_max.back()), *(rng_posx.back())))
|
{
|
rng_max = rng_posx;
|
side_max = side_posx;
|
};
|
};
|
};
|
bk.exec(counter);
|
}
|
//
|
//****************************************************************************
|
}; // End namespace blk_detail
|
}; // End namespace sort
|
}; // End namespace boost
|
//****************************************************************************
|
//
|
#endif
|
