Program Listing for File FastFlowExecutor.h

↰ Return to documentation for file (ADS/parallel/FastFlowExecutor.h)

/*--------------------------------------------------------------------------*/
/*------------------------ File FastFlowExecutor.h -------------------------*/
/*--------------------------------------------------------------------------*/
/*--------------------------------------------------------------------------*/
/*----------------------------- DEFINITIONS --------------------------------*/
/*--------------------------------------------------------------------------*/

#ifndef __ADS_FastFlowExecutor
 #define __ADS_FastFlowExecutor
                      /* self-identification: #endif constant */

#if defined( ADS_WITH_FASTFLOW )

/*--------------------------------------------------------------------------*/
/*------------------------------ INCLUDES ----------------------------------*/
/*--------------------------------------------------------------------------*/

#if ! defined( ADS_FASTFLOW_MAPPING ) && ! defined( NO_DEFAULT_MAPPING )
 #define NO_DEFAULT_MAPPING
#endif

#include "ADS/Contract.h"
#include "ADS/parallel/Executor.h"

#include <ff/ff.hpp>
#include <ff/parallel_for.hpp>

#include <algorithm>
#include <cstddef>
#include <exception>
#include <mutex>
#include <thread>

/*--------------------------------------------------------------------------*/
/*--------------------------- NAMESPACE ------------------------------------*/
/*--------------------------------------------------------------------------*/

namespace ADS {

/*--------------------------------------------------------------------------*/
/*------------------------- CLASS FastFlowExecutor -------------------------*/
/*--------------------------------------------------------------------------*/

class FastFlowExecutor {

/*--------------------------------------------------------------------------*/
 public:

 explicit FastFlowExecutor( std::size_t p = 0 , bool spin = true )
  : f_p( p ? p : std::max< std::size_t >(
                  std::thread::hardware_concurrency() , 1 ) ) ,
    f_for( long( f_p ) , spin , spin ) {
  ADS_PRE( f_p >= 1 );
  f_for.disableScheduler( true );
  }

 [[nodiscard]] std::size_t concurrency( void ) const { return( f_p ); }

 template< class F >
 void bulk( std::size_t n , F && f ) {
  if( ( n <= 1 ) || ( f_p == 1 ) || tl_inside ) {
   for( std::size_t i = 0 ; i < n ; ++i )
    f( i );
   return;
   }
  std::lock_guard< std::mutex > g( f_lock );
  detail::FirstError error;
  f_for.parallel_for( 0 , long( n ) , 1 , 1 , [ & ]( const long i ) {
   if( error.failed() )
    return;
   tl_inside = true;
   try {
    f( std::size_t( i ) );
    }
   catch( ... ) {
    error.keep();
    }
   tl_inside = false;
   } , long( std::min( f_p , n ) ) );
  error.rethrow();
  }

/*--------------------------------------------------------------------------*/
 private:

 static inline thread_local bool tl_inside = false;

 std::size_t f_p;
 ff::ParallelFor f_for;
 std::mutex f_lock;
 };  // end( class FastFlowExecutor )

/*--------------------------------------------------------------------------*/

}  // end( namespace ADS )

#endif  /* ADS_WITH_FASTFLOW */

/*--------------------------------------------------------------------------*/

#endif  /* FastFlowExecutor.h included */

/*--------------------------------------------------------------------------*/
/*---------------------- End File FastFlowExecutor.h -----------------------*/
/*--------------------------------------------------------------------------*/