Program Listing for File DistributedSort.h
↰ Return to documentation for file (ADS/distributed/DistributedSort.h)
/*--------------------------------------------------------------------------*/
/*------------------------- File DistributedSort.h -------------------------*/
/*--------------------------------------------------------------------------*/
/*--------------------------------------------------------------------------*/
/*----------------------------- DEFINITIONS --------------------------------*/
/*--------------------------------------------------------------------------*/
#ifndef __ADS_DistributedSort
#define __ADS_DistributedSort
/* self-identification: #endif constant */
#if defined( ADS_WITH_MPI )
/*--------------------------------------------------------------------------*/
/*------------------------------ INCLUDES ----------------------------------*/
/*--------------------------------------------------------------------------*/
#include "ADS/Contract.h"
#include "ADS/parallel/Executor.h"
#include "ADS/parallel/ParallelMerge.h"
#include "ADS/parallel/ParallelSort.h"
#include "ADS/priority/MultiwayMerge.h"
#include "ADS/sort/Sort.h"
// the C interface of MPI only: the bindings of C++, removed from the
// standard, do not compile cleanly
#ifndef OMPI_SKIP_MPICXX
#define OMPI_SKIP_MPICXX 1
#endif
#ifndef MPICH_SKIP_MPICXX
#define MPICH_SKIP_MPICXX 1
#endif
#include <mpi.h>
#include <algorithm>
#include <climits>
#include <cstddef>
#include <functional>
#include <stdexcept>
#include <type_traits>
#include <utility>
#include <vector>
/*--------------------------------------------------------------------------*/
/*--------------------------- NAMESPACE ------------------------------------*/
/*--------------------------------------------------------------------------*/
namespace ADS::distributed {
template< Executor E , class T , class C = std::less<> >
void sort( E & ex , MPI_Comm comm , std::vector< T > & A , C less = C() );
template< class T , class C = std::less<> >
void sort( MPI_Comm comm , std::vector< T > & A , C less = C() )
{
Sequential s;
sort( s , comm , A , less );
}
/*--------------------------------------------------------------------------*/
namespace detail {
inline void check( int code , const char * what )
{
if( code != MPI_SUCCESS )
throw( std::runtime_error( std::string( "ADS::distributed::sort: " ) +
what ) );
}
} // end( namespace detail )
template< Executor E , class T , class C >
void sort( E & ex , MPI_Comm comm , std::vector< T > & A , C less )
{
static_assert( std::is_trivially_copyable_v< T > ,
"the items travel as bytes" );
using detail::check;
int P = 1 , me = 0;
check( MPI_Comm_size( comm , &P ) , "MPI_Comm_size" );
check( MPI_Comm_rank( comm , &me ) , "MPI_Comm_rank" );
ADS::sort( ex , A , less );
if( P == 1 )
return;
const std::size_t n = A.size() , p = std::size_t( P );
// P - 1 regular samples of each process, and a flag for an empty one
std::vector< T > mine( p - 1 );
std::vector< int > has( p ) , counts( p ) , got( p );
const int full = n > 0;
for( std::size_t j = 0 ; j + 1 < p && n ; ++j )
mine[ j ] = A[ ( j + 1 ) * n / p ];
check( MPI_Allgather( &full , 1 , MPI_INT , has.data() , 1 , MPI_INT ,
comm ) , "MPI_Allgather" );
std::vector< T > all( p * ( p - 1 ) );
const int bytes = int( ( p - 1 ) * sizeof( T ) );
check( MPI_Allgather( mine.data() , bytes , MPI_BYTE , all.data() , bytes ,
MPI_BYTE , comm ) , "MPI_Allgather" );
std::vector< T > S; // the samples of the full ones
for( std::size_t q = 0 ; q < p ; ++q )
if( has[ q ] )
S.insert( S.end() , all.begin() + std::ptrdiff_t( q * ( p - 1 ) ) ,
all.begin() + std::ptrdiff_t( ( q + 1 ) * ( p - 1 ) ) );
ADS::sort( S , less );
// the cuts: piece j of this process are its items in ( s_{j-1} , s_j ],
// a run of splitters equal to x splitting the items equal to x evenly
std::vector< std::size_t > cut( p + 1 , 0 );
cut[ p ] = n;
for( std::size_t j = 1 ; j < p ; ++j ) {
if( S.empty() ) {
cut[ j ] = n;
continue;
}
const T & s = S[ std::min( S.size() - 1 , j * S.size() / p ) ];
const auto lo = std::size_t( std::lower_bound( A.begin() , A.end() , s ,
less ) - A.begin() ) ,
hi = std::size_t( std::upper_bound( A.begin() , A.end() , s ,
less ) - A.begin() );
// the splitters equal to s: [ first , last ) of the indices j
std::size_t first = j , last = j + 1;
while( ( first > 1 ) &&
! less( S[ std::min( S.size() - 1 , ( first - 1 ) * S.size() /
p ) ] , s ) )
--first;
while( ( last < p ) &&
! less( s , S[ std::min( S.size() - 1 , last * S.size() / p ) ] ) )
++last;
// the equal items [ lo , hi ) shared among the cuts first .. last - 1
cut[ j ] = lo + ( hi - lo ) * ( j - first + 1 ) / ( last - first + 1 );
}
for( std::size_t j = 1 ; j <= p ; ++j )
cut[ j ] = std::max( cut[ j ] , cut[ j - 1 ] );
for( std::size_t j = 0 ; j < p ; ++j ) {
const std::size_t b = ( cut[ j + 1 ] - cut[ j ] ) * sizeof( T );
if( b > std::size_t( INT_MAX ) )
throw( std::length_error( "ADS::distributed::sort: a piece of more "
"than 2^31 bytes" ) );
counts[ j ] = int( b );
}
check( MPI_Alltoall( counts.data() , 1 , MPI_INT , got.data() , 1 ,
MPI_INT , comm ) , "MPI_Alltoall" );
std::vector< int > sd( p , 0 ) , rd( p , 0 );
std::size_t total = 0;
for( std::size_t j = 0 ; j < p ; ++j ) {
sd[ j ] = int( cut[ j ] * sizeof( T ) );
rd[ j ] = int( total );
total += std::size_t( got[ j ] );
if( total > std::size_t( INT_MAX ) )
throw( std::length_error( "ADS::distributed::sort: more than 2^31 "
"bytes to receive" ) );
}
std::vector< T > R( total / sizeof( T ) );
check( MPI_Alltoallv( A.data() , counts.data() , sd.data() , MPI_BYTE ,
R.data() , got.data() , rd.data() , MPI_BYTE ,
comm ) , "MPI_Alltoallv" );
// the P sorted pieces received, merged
std::vector< std::pair< const T * , const T * > > runs;
for( std::size_t j = 0 ; j < p ; ++j )
runs.emplace_back( R.data() + rd[ j ] / int( sizeof( T ) ) ,
R.data() + ( rd[ j ] + got[ j ] ) / int( sizeof( T ) ) );
A.resize( R.size() );
ADS::multiwayMerge( ex , runs , A.begin() , less );
ADS_AUDIT( std::is_sorted( A.begin() , A.end() , less ) );
}
/*--------------------------------------------------------------------------*/
} // end( namespace ADS::distributed )
#endif /* ADS_WITH_MPI */
/*--------------------------------------------------------------------------*/
#endif /* DistributedSort.h included */
/*--------------------------------------------------------------------------*/
/*---------------------- End File DistributedSort.h ------------------------*/
/*--------------------------------------------------------------------------*/