MQAStar.h

Go to the documentation of this file.

/*--------------------------------------------------------------------------*/
/*----------------------------- File MQAStar.h -----------------------------*/
/*--------------------------------------------------------------------------*/
/*--------------------------------------------------------------------------*/
/*------------------------------ DEFINITIONS -------------------------------*/
/*--------------------------------------------------------------------------*/
 
#ifndef __PBeFS_MQAStar
 #define __PBeFS_MQAStar
                      /* self-identification: #endif at the end of the file */
 
/*--------------------------------------------------------------------------*/
/*-------------------------------- INCLUDES --------------------------------*/
/*--------------------------------------------------------------------------*/
 
#include <atomic>
 
#include <cstdint>
 
#include <memory>
 
#include <mutex>
 
#include <optional>
 
#include <ranges>
 
#include <type_traits>
 
#include <vector>
 
#include "PBeFS/astar/ConcurrentClosed.h"
 
#include "PBeFS/astar/HDAStar.h"
 
#include "ADS/concurrent/MultiQueue.h"
 
/*--------------------------------------------------------------------------*/
/*------------------------------- NAMESPACE --------------------------------*/
/*--------------------------------------------------------------------------*/

namespace PBeFS_di_unipi_it
{
 
/*--------------------------------------------------------------------------*/
/*----------------------------- CLASS MQAStar ------------------------------*/
/*--------------------------------------------------------------------------*/

 
template< Domain D , ADS::WorkerExecutor Executor = ADS::ThreadPool >
class MQAStar {
 
/*------------------------ PUBLIC PART OF THE CLASS ------------------------*/
 
 public:
 
/*------------------------------ PUBLIC TYPES ------------------------------*/
 
 using State = state_t< D >;                       
 using Cost = cost_t< D >;                         
 using NodeT = Node< State , Cost >;               
 using SolutionT = Solution< State , Cost >;       
 
/*------------------------------ CONSTRUCTOR -------------------------------*/

 
 explicit MQAStar( const D & domain , const ParallelOptions & options = {} )
  : f_domain( domain ) , f_options( options ) {
  f_options.normalize();
  }
 
/*------------------------- OTHER INITIALIZATIONS --------------------------*/
 
 void set_executor( Executor & ex ) noexcept { f_ex = &ex; }
 
/*-------------------------------- SOLVING ---------------------------------*/
 
 [[nodiscard]] SolutionT solve( void ) const
 {
  const unsigned p = f_options.threads;
  Stopwatch clock;
  Shared sh( f_domain , p , f_options.seed );
  std::vector< std::unique_ptr< Worker > > w( p );
  for( unsigned i = 0 ; i < p ; ++i )
   w[ i ] = std::make_unique< Worker >( *this , sh );
 
  {  // the root, inserted by the first worker before the start
   const State s0 = f_domain.initial_state();
   NodeT * root = w[ 0 ]->update( s0 , Cost( 0 ) ,
                                  Cost( f_domain.heuristic( s0 ) ) ,
                                  nullptr , no_op );
   auto hd = sh.open.handle();
   hd.push( key( root->g , root->h ) , OpenEntry< NodeT >{ root , root->g } );
   }
 
  run_workers( f_ex , p , [ & ]( unsigned i ) { w[ i ]->run(); } );
 
  SolutionT sol;
  SearchStats & st = sol.stats;
  for( auto & x : w ) {
   st.expanded += x->f_st.expanded;
   st.reexpanded += x->f_st.reexpanded;
   st.generated += x->f_st.generated;
   st.duplicates += x->f_st.duplicates;
   st.stored += x->f_local.stored();
   st.expanded_by.push_back( x->f_st.expanded );
   st.stored_by.push_back( x->f_local.stored() );
   }
  if( sh.limit_hit.load() )
   sol.status = Status::node_limit;
  else
   if( sh.goal ) {
    sol.status = Status::solved;
    sol.cost = sh.goal->g;
    sol.path = trace_path( sh.goal );
    }
  st.seconds = clock.seconds();
  return( sol );
  }
 
/*----------------------- PRIVATE PART OF THE CLASS ------------------------*/
 
 private:
 
 using Lock = typename Executor::lock_type;
 
 using Closed = ConcurrentClosed< D , Lock >;      

 static constexpr bool kPacked = std::is_integral_v< Cost >;
 using Key = std::conditional_t< kPacked , std::uint64_t , Cost >;
 
 static Key key( Cost g , Cost h ) {
  if constexpr( kPacked )
   return( packed_fg( g , h ) );
  else
   return( g + h );
  }
 
 static Cost f_of( Key k ) {
  if constexpr( kPacked )
   return( Cost( k >> 32 ) );
  else
   return( k );
  }

 using Mq = ADS::MultiQueue< Key , OpenEntry< NodeT > >;
 
/*- - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -*/
 // the data shared by the workers
 
 struct Shared {
  Shared( const D & d , unsigned p , std::uint64_t seed )
   : open( 2 * std::size_t( p ) , 1 , seed ) , closed( d ) {}
 
  Mq open;                                      // the OPEN list
  // the entries in OPEN or being expanded
  alignas( cache_line ) std::atomic< std::size_t > pending{ 1 };
  Closed closed;                                // the CLOSED list
  alignas( cache_line ) std::atomic< Cost > incumbent{
                                                   infinite_cost< Cost >() };
  std::atomic< bool > done{ false };
  std::atomic< bool > limit_hit{ false };
  std::mutex goal_mutex;                        // protects goal
  NodeT * goal = nullptr;                       // the best goal found
  };
 
/*- - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -*/
 // a worker: its pool of nodes and its counters
 
 struct alignas( cache_line ) Worker {
 
  Worker( const MQAStar & a , Shared & s ) : f_a( a ) , f_sh( s ) {}
 
  // the main loop
  void run( void ) {
   ADS::Backoff backoff;
   auto hd = f_sh.open.handle();
   std::vector< OpenEntry< NodeT > > fresh;
   while( ! f_sh.done.load( std::memory_order_acquire ) ) {
    const auto e = hd.tryPop();
    if( ! e ) {
     if( f_sh.pending.load( std::memory_order_acquire ) == 0 ) {
      f_sh.done.store( true , std::memory_order_release );
      break;
      }
     backoff.pause();
     continue;
     }
    backoff.reset();
    fresh.clear();
    if( f_of( e->key ) < f_sh.incumbent.load() )
     expand( e->item , fresh );
    // the entries of F counted and the one popped dropped in one step,
    // before the pushes, so that m never misses an entry [see line 13]
    if( fresh.size() != 1 )
     f_sh.pending.fetch_add( fresh.size() - 1 , std::memory_order_acq_rel );
    auto entries = fresh | std::views::transform(
                            []( const OpenEntry< NodeT > & x ) {
                             return( typename Mq::Entry{
                                      key( x.g , x.node->h ) , x } ); } );
    hd.pushAll( entries.begin() , entries.end() );
    }
   }
 
  // expands the node of e unless the entry is stale; the nodes to put in
  // OPEN, with the g they had under their lock, are appended to fresh
  void expand( const OpenEntry< NodeT > & e ,
               std::vector< OpenEntry< NodeT > > & fresh ) {
   NodeT * n = e.node;
   State s;
   Cost g , h;
   std::uint8_t op;
   {
    std::lock_guard< Lock > guard( f_sh.closed.lock( n ) );
    if( e.stale() )
     return;
    n->closed = true;
    if( n->expanded )
     ++f_st.reexpanded;
    n->expanded = true;
    s = n->state;
    g = n->g;
    h = n->h;
    op = n->op;
    }
   ++f_st.expanded;
 
   if( f_a.f_domain.is_goal( s ) ) {
    std::lock_guard< std::mutex > guard( f_sh.goal_mutex );
    if( g < f_sh.incumbent.load() ) {
     f_sh.incumbent.store( g );
     f_sh.goal = n;
     }
    return;
    }
 
   // the successors first, but the parent [see OperatorDomain], with the
   // places of CLOSED that their updates read prefetched, so that the
   // misses of all of them overlap
   f_succ.clear();
   for_each_child( f_a.f_domain , s , h , op , [ & ]( const State & t ,
                                                      Cost c , Cost ht ,
                                                      std::uint8_t ot ) {
    ++f_st.generated;
    const Cost gt = g + c;
    if( gt + ht >= f_sh.incumbent.load( std::memory_order_relaxed ) )
     return;
    const std::uint64_t hash = f_a.f_domain.hash( t );
    f_sh.closed.prefetch( hash );
    f_succ.push_back( { t , gt , ht , hash , ot } );
    } );
   for( const auto & v : f_succ )
    if( NodeT * x = f_sh.closed.update( v.state , v.hash , v.g , v.h , n ,
                                        v.op , f_local , f_st.duplicates ) )
     fresh.push_back( OpenEntry< NodeT >{ x , v.g } );
 
   if( f_a.f_options.node_limit &&
       ( f_local.stored() >= f_a.f_options.node_limit ) ) {
    f_sh.limit_hit.store( true );
    f_sh.done.store( true , std::memory_order_release );
    }
   }
 
  // records the path to t of cost g through parent and op; returns the
  // node of t if it has to go into OPEN, nullptr if it is a duplicate
  NodeT * update( const State & t , Cost g , Cost h , NodeT * parent ,
                  std::uint8_t op ) {
   return( f_sh.closed.update( t , f_a.f_domain.hash( t ) , g , h ,
                               parent , op , f_local , f_st.duplicates ) );
   }
 
  const MQAStar & f_a;                          // the search
  Shared & f_sh;                                // the shared data
  typename Closed::Local f_local;               // the nodes it allocated
  std::vector< typename Closed::Successor > f_succ;  // of the expansion
  SearchStats f_st;                             // its counters
  };
 
/*- - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -*/
 
 const D & f_domain;                                   
 ParallelOptions f_options;                            
 Executor * f_ex = nullptr;                            
 
 };  // end( class( MQAStar ) )
 
/*--------------------------------------------------------------------------*/
 
}  // end( namespace PBeFS_di_unipi_it )
 
/*--------------------------------------------------------------------------*/
 
#endif  /* __PBeFS_MQAStar */
 
/*--------------------------------------------------------------------------*/
/*--------------------------- End File MQAStar.h ---------------------------*/
/*--------------------------------------------------------------------------*/