HDAStarMPI.h

Go to the documentation of this file.

/*--------------------------------------------------------------------------*/
/*--------------------------- File HDAStarMPI.h ----------------------------*/
/*--------------------------------------------------------------------------*/
/*--------------------------------------------------------------------------*/
/*------------------------------ DEFINITIONS -------------------------------*/
/*--------------------------------------------------------------------------*/
 
#ifndef __PBeFS_HDAStarMPI
 #define __PBeFS_HDAStarMPI
                      /* self-identification: #endif at the end of the file */
 
/*--------------------------------------------------------------------------*/
/*-------------------------------- INCLUDES --------------------------------*/
/*--------------------------------------------------------------------------*/
 
#include <algorithm>
 
#include <stdexcept>
 
#include <type_traits>
 
#include <variant>
 
#include <vector>
 
#include <mpi.h>
 
#include "PBeFS/astar/HDAStar.h"
 
#include "PBeFS/core/SearchUtils.h"
 
#include "ADS/distributed/MPIMailbox.h"
 
/*--------------------------------------------------------------------------*/
/*------------------------------- NAMESPACE --------------------------------*/
/*--------------------------------------------------------------------------*/

namespace PBeFS_di_unipi_it
{
 
/*--------------------------------------------------------------------------*/
/*---------------------------- CLASS HDAStarMPI ----------------------------*/
/*--------------------------------------------------------------------------*/

 
template< Domain D , class Dist = StateHashDistribution< D > >
class HDAStarMPI {
 
 static_assert( std::is_trivially_copyable_v< state_t< D > > ,
                "HDAStarMPI: the states have to be trivially copyable" );
 
/*------------------------ PUBLIC PART OF THE CLASS ------------------------*/
 
 public:
 
/*------------------------------ PUBLIC TYPES ------------------------------*/
 
 using State = state_t< D >;                       
 using Cost = cost_t< D >;                         
 using SolutionT = Solution< State , Cost >;       
 
/*------------------------------ CONSTRUCTOR -------------------------------*/

 
 HDAStarMPI( const D & domain , const Dist & dist ,
             const ParallelOptions & options = {} ,
             MPI_Comm comm = MPI_COMM_WORLD )
  : f_domain( domain ) , f_dist( dist ) , f_options( options ) ,
    f_comm( comm ) {
  f_options.normalize();
  }
 
/*-------------------------------- SOLVING ---------------------------------*/
 
 [[nodiscard]] SolutionT solve( void ) const
 {
  Stopwatch clock;
  Process w( *this );
  w.run();
  SolutionT sol;
  w.gather( sol );
  sol.stats.seconds = clock.seconds();
  return( sol );
  }
 
/*----------------------- PRIVATE PART OF THE CLASS ------------------------*/
 
 private:
 
 // a node: the parent is a state, owned by another process maybe
 struct MNode {
  State state;
  Cost g;
  Cost h;
  State parent;
  bool has_parent;
  bool closed;
  bool expanded;
  [[no_unique_address]] OperatorOf< D > o;
  std::uint32_t key;                                // the key of state
  };
 
 // a node travelling to its owner
 struct Msg {
  State s;
  Cost g;
  Cost h;
  State parent;
  bool has_parent;
  [[no_unique_address]] OperatorOf< D > o;
  std::uint32_t key;                                // the key of s
  };
 
 // a cost of a goal, sent to every process
 struct Incumbent {
  Cost c;
  };
 
 // the messages: the nodes and the incumbents
 using Mail = std::variant< Msg , Incumbent >;
 
 // the OPEN list: buckets for integer costs, the heap otherwise
 using Open_t = std::conditional_t< std::is_integral_v< Cost > ,
                                    BucketOpenList< MNode > ,
                                    HeapOpenList< MNode > >;
 
/*- - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -*/
 // the search on one process
 
 struct Process {
 
  explicit Process( const HDAStarMPI & a )
   : f_a( a ) , f_comm( a.f_comm ) ,
     f_mail( a.f_comm , a.f_options.batch ) , f_dist( a.f_dist ) {
   MPI_Comm_rank( f_comm , & f_rank );
   int np;
   MPI_Comm_size( f_comm , & np );
   f_p = unsigned( np );
   }
 
  // the main loop
  void run( void ) {
   const State s0 = f_a.f_domain.initial_state();
   if( f_dist.owner( s0 , f_p ) == unsigned( f_rank ) )
    insert( Msg{ s0 , Cost( 0 ) , Cost( f_a.f_domain.heuristic( s0 ) ) , s0 ,
                 false , no_op , distribution_key< D >( f_dist , s0 ) } ,
            f_a.f_domain.hash( s0 ) );
 
   unsigned since_flush = 0;
   for( ;; ) {
    receive();
    if( MNode * n = next() ) {
     expand( n );
     if( ++since_flush >= f_a.f_options.flush_period ) {
      since_flush = 0;
      f_mail.flush();
      }
     if( f_a.f_options.node_limit && ! f_mail.stopped() &&
         ( f_pool.size() >= f_a.f_options.node_limit ) ) {
      f_limit_hit = true;
      f_mail.stop();
      }
     continue;
     }
    if( f_mail.quiescent() )      // nothing to expand nor to send
     break;
    }
   }
 
  // the best node of OPEN with f < incumbent, not removed; nullptr if none
  MNode * next_peek( void ) {
   if( f_mail.stopped() )
    return( nullptr );
   while( ! f_open.empty() ) {
    const OpenEntry< MNode > e = f_open.top();
    if( e.node->closed || ( e.node->g != e.g ) ) {
     f_open.pop();
     continue;
     }
    if( e.node->g + e.node->h >= f_incumbent )
     return( nullptr );
    return( e.node );
    }
   return( nullptr );
   }
 
  // the best node of OPEN with f < incumbent, removed; nullptr if none
  MNode * next( void ) {
   MNode * n = next_peek();
   if( n )
    f_open.pop();
   return( n );
   }
 
  // expands n, or records it as the incumbent if it is a goal
  void expand( MNode * n ) {
   n->closed = true;
   ++f_st.expanded;
   if( n->expanded )
    ++f_st.reexpanded;
   n->expanded = true;
 
   if( f_a.f_domain.is_goal( n->state ) ) {
    if( n->g < f_incumbent ) {
     f_incumbent = n->g;
     f_goal = n;
     for( unsigned q = 0 ; q < f_p ; ++q )
      if( q != unsigned( f_rank ) )
       f_mail.send( q , Mail( Incumbent{ n->g } ) );
     f_mail.flush();
     }
    return;
    }
 
   const auto owner = successor_owners< D >( f_dist , n->state , n->key ,
                                             f_p );
   for_each_child( f_a.f_domain , n->state , n->h , n->o.op ,
                   [ & ]( const State & t , Cost c , Cost ht ,
                          std::uint8_t ot ) {
    ++f_st.generated;
    const Cost g = n->g + c;
    if( g + ht >= f_incumbent )
     return;
    const auto [ dst , key ] = owner( t );
    const Msg m{ t , g , ht , n->state , true , ot , key };
    if( dst == unsigned( f_rank ) ) {
     insert( m , f_a.f_domain.hash( t ) );
     return;
     }
    f_mail.send( dst , Mail( m ) );
    ++f_st.sent;
    } );
   }
 
  // adds the node of m, whose state has hash h, to OPEN, unless it is a
  // duplicate
  void insert( const Msg & m , std::uint64_t h ) {
   if( f_mail.stopped() || ( m.g + m.h >= f_incumbent ) )
    return;
   auto [ x , fresh ] = f_table.find_or_insert( m.s , h , [ & ] {
    MNode * y = f_pool.allocate();
    *y = MNode{ m.s , m.g , m.h , m.parent , m.has_parent , false , false ,
                m.o , m.key };
    return( y );
    } );
   if( fresh )
    f_open.push( x );
   else
    if( m.g < x->g ) {
     x->g = m.g;
     x->parent = m.parent;
     x->has_parent = m.has_parent;
     x->o = m.o;
     x->closed = false;
     f_open.push( x );
     }
    else
     ++f_st.duplicates;
   }
 
  // receives and processes all the messages that have arrived, the nodes
  // of a batch hashed and prefetched first
  void receive( void ) {
   v_msg.clear();
   f_mail.poll( [ & ]( const Mail & x ) {
    if( const Msg * m = std::get_if< Msg >( & x ) )
     v_msg.push_back( *m );
    else
     f_incumbent = std::min( f_incumbent , std::get< Incumbent >( x ).c );
    } );
   const std::size_t k = v_msg.size();
   v_hash.resize( k );
   for( std::size_t i = 0 ; i < k ; ++i ) {
    v_hash[ i ] = f_a.f_domain.hash( v_msg[ i ].s );
    f_table.prefetch( v_hash[ i ] );
    }
   for( std::size_t i = 0 ; i < k ; ++i )
    insert( v_msg[ i ] , v_hash[ i ] );
   }
 
  // fills the Solution on every process: counters, status, path
  void gather( SolutionT & sol ) {
   SearchStats & st = sol.stats;
   unsigned long long mine[ 7 ] = { f_st.expanded , f_st.reexpanded ,
                                    f_st.generated , f_st.duplicates ,
                                    f_st.sent , f_mail.batches() ,
                                    f_pool.size() };
   std::vector< unsigned long long > all( 7 * std::size_t( f_p ) );
   MPI_Allgather( mine , 7 , MPI_UNSIGNED_LONG_LONG , all.data() , 7 ,
                  MPI_UNSIGNED_LONG_LONG , f_comm );
   for( unsigned q = 0 ; q < f_p ; ++q ) {
    const auto * x = all.data() + 7 * std::size_t( q );
    st.expanded += x[ 0 ];
    st.reexpanded += x[ 1 ];
    st.generated += x[ 2 ];
    st.duplicates += x[ 3 ];
    st.sent += x[ 4 ];
    st.messages += x[ 5 ];
    st.stored += x[ 6 ];
    st.expanded_by.push_back( x[ 0 ] );
    st.stored_by.push_back( x[ 6 ] );
    }
 
   int limit = f_limit_hit ? 1 : 0 , any_limit = 0;
   MPI_Allreduce( & limit , & any_limit , 1 , MPI_INT , MPI_MAX , f_comm );
   if( any_limit ) {
    sol.status = Status::node_limit;
    return;
    }
 
   struct { double c; int r; } best{ f_goal ? double( f_goal->g )
                                            : infinite_cost< double >() ,
                                     f_rank } , win{};
   MPI_Allreduce( & best , & win , 1 , MPI_DOUBLE_INT , MPI_MINLOC ,
                  f_comm );
   if( win.c == infinite_cost< double >() )
    return;                                    // no goal: unsolvable
 
   // the path, from the goal back to the root, one owner at a time
   struct Step {
    State s;
    Cost g;
    State parent;
    bool has_parent;
    } step{};
   if( f_rank == win.r )
    step = Step{ f_goal->state , f_goal->g , f_goal->parent ,
                 f_goal->has_parent };
   MPI_Bcast( & step , int( sizeof( Step ) ) , MPI_BYTE , win.r , f_comm );
   sol.status = Status::solved;
   sol.cost = step.g;
   std::vector< State > path{ step.s };
   while( step.has_parent ) {
    const State s = step.parent;
    const int owner = int( f_dist.owner( s , f_p ) );
    if( f_rank == owner ) {
     const MNode * x = f_table.find( s , f_a.f_domain.hash( s ) );
     if( ! x )
      throw( std::logic_error( "HDAStarMPI::solve: lost node" ) );
     step = Step{ x->state , x->g , x->parent , x->has_parent };
     }
    MPI_Bcast( & step , int( sizeof( Step ) ) , MPI_BYTE , owner , f_comm );
    path.push_back( step.s );
    }
   sol.path.assign( path.rbegin() , path.rend() );
   }
 
  const HDAStarMPI & f_a;                       // the search
  MPI_Comm f_comm;                              // the communicator
  ADS::MPIMailbox< Mail > f_mail;               // its messages
  int f_rank;                                   // this process
  unsigned f_p;                                 // the number of processes
  Dist f_dist;                                  // the distribution
  NodePool< MNode > f_pool;                     // its nodes
  NodeTable< MNode > f_table;                   // its CLOSED list
  Open_t f_open;                                // its OPEN list
  std::vector< Msg > v_msg;                     // the nodes received
  std::vector< std::uint64_t > v_hash;          // their hashes
  Cost f_incumbent = infinite_cost< Cost >();   // the best cost known
  MNode * f_goal = nullptr;                     // its best goal
  bool f_limit_hit = false;                     // it hit the node limit
  SearchStats f_st;                             // its counters
  };
 
/*- - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -*/
 
 const D & f_domain;                                   
 const Dist & f_dist;                    
 ParallelOptions f_options;                            
 MPI_Comm f_comm;                                      
 
 };  // end( class( HDAStarMPI ) )
 
/*--------------------------------------------------------------------------*/
 
}  // end( namespace PBeFS_di_unipi_it )
 
/*--------------------------------------------------------------------------*/
 
#endif  /* __PBeFS_HDAStarMPI */
 
/*--------------------------------------------------------------------------*/
/*------------------------- End File HDAStarMPI.h --------------------------*/
/*--------------------------------------------------------------------------*/