Implements concurrent evaluation of a criterion on a set of subset candidates. More...

#include <candidate_evaluator_threaded.hpp>

Inheritance diagram for FST::Candidate_Evaluator_Threaded< RETURNTYPE, DIMTYPE, SUBSET, CRITERION >:

Classes
struct	ThreadLocal
	thread-local storage of current subset candidate, criterion clone, and tracker clone More...
struct	CandidateResult
	temporary storage of subset candidate evaluation results More...

Public Types
typedef Candidate_Evaluator< RETURNTYPE, DIMTYPE, SUBSET, CRITERION >	parent
	the abstract Candidate_Evaluator interface this class implements
typedef std::shared_ptr< CRITERION >	PCriterion
	shared pointer to the criterion this engine evaluates candidates against
typedef std::shared_ptr< SUBSET >	PSubset
	shared pointer to a candidate feature subset
typedef std::shared_ptr< Result_Tracker< RETURNTYPE, SUBSET > >	PResultTracker
	shared pointer to the result tracker fed during evaluate_stream()
typedef parent::Generator	Generator
	see Candidate_Evaluator::Generator
typedef parent::ResultCallback	ResultCallback
	see Candidate_Evaluator::ResultCallback
typedef parent::BestCallback	BestCallback
	see Candidate_Evaluator::BestCallback
typedef parent::BetterPredicate	BetterPredicate
	see Candidate_Evaluator::BetterPredicate
Public Types inherited from FST::Candidate_Evaluator< RETURNTYPE, DIMTYPE, SUBSET, CRITERION >
typedef std::shared_ptr< CRITERION >	PCriterion
	shared pointer to the criterion this engine evaluates candidates against
typedef std::shared_ptr< SUBSET >	PSubset
	shared pointer to a candidate feature subset
typedef std::shared_ptr< Result_Tracker< RETURNTYPE, SUBSET > >	PResultTracker
typedef std::function< bool(SUBSET &)>	Generator
	fills sub with the next candidate, returns false when no candidate remains (in the threaded implementation calls are serialized under a lock)
typedef std::function< void(const RETURNTYPE, const PSubset &, const DIMTYPE)>	ResultCallback
	fired once per evaluated batch candidate: (criterion value, candidate, feature tag as passed to add_candidate)
typedef std::function< void(const int, const unsigned long long, const RETURNTYPE, const PSubset &)>	BestCallback
typedef std::function< bool(const RETURNTYPE, const PSubset &, const RETURNTYPE, const PSubset &)>	BetterPredicate
	returns true if (value,subset) is better than (best value,best subset) - encodes per-search tie-breaking for evaluate_stream()

Public Member Functions
	Candidate_Evaluator_Threaded (const unsigned int max_threads=2)
virtual unsigned int	get_max_threads () const override
	1 in the sequential implementation
virtual bool	is_threaded () const override
	true for the threaded engine (any --threads value), false for the sequential one – distinguishes the engine kind, not whether get_max_threads()>1
virtual void	initialize (const PCriterion &crit, const DIMTYPE max_candidates, ResultCallback on_result=ResultCallback()) override
	Opens a new batch to be evaluated against `crit`, which must already be initialize()d. `max_candidates` must be greater or equal than the number of candidates that will be added - the result storage is sized here once, so that workers never race a vector reallocation; it must be nonzero. See the criterion-clone warning in the class documentation.
virtual bool	add_candidate (const PSubset &sub, const DIMTYPE feature=0) override
	adds candidate subset to the evaluation pool; returns false once the result storage reserved by initialize() is full, i.e. when more candidates are added than max_candidates announced
virtual bool	collect () override
	completes the batch: waits for all evaluations and fires pending per-result callbacks in add-order
virtual bool	getBestResult (RETURNTYPE &result, PSubset &sub, DIMTYPE &feature) override
	completes the batch, then returns (without copying sub contents) the candidate that yielded the highest criterion value; ties go to the candidate added first
bool	getFirstResult (RETURNTYPE &result, PSubset &sub, DIMTYPE &feature)
	completes the batch and starts iteration over ALL of its results in ADD-ORDER (not by value); returns a pointer, does not copy sub contents
bool	getNextResult (RETURNTYPE &result, PSubset &sub, DIMTYPE &feature)
	advances the getFirstResult() iteration by one result; false once the batch is exhausted
bool	results_to_tracker (const PResultTracker tracker)
	completes the batch and feeds every candidate's (value,subset) into `tracker` in add-order; false if the batch was empty or failed
virtual bool	evaluate_stream (Generator generator, const PCriterion &crit, const PResultTracker tracker, BetterPredicate better, BestCallback on_best, RETURNTYPE &result, const PSubset bestsub) override
	streaming mode: the master thread calls `generator` until exhaustion, workers evaluate concurrently, the master replays the results in generation order (see the class notes)
virtual void	invalidate_clones () override
	drops the cached per-worker criterion clones (see Candidate_Evaluator::invalidate_clones()); the next initialize()/evaluate_stream() re-clones from the criterion it is handed
virtual std::ostream &	print (std::ostream &os) const override
	writes a human-readable identification of the engine; base default writes nothing

Protected Types
enum	StreamSlotState { SLOT_EMPTY =0 , SLOT_PENDING , SLOT_RUNNING , SLOT_DONE , SLOT_FAILED }

Protected Member Functions
void	pooled_worker (unsigned int tidx)
	the persistent worker loop: park, claim, evaluate, count, repeat
void	finish_batch ()
	stops and joins workers, then fires pending per-result callbacks (once per batch)
void	record_worker_exception ()
	captures the currently-handled exception (first one wins) for rethrow on the master by finish_batch()/evaluate_stream(); called from a worker's catch(...) block, never lets an exception escape the worker
void	stream_worker (unsigned int tidx)
	claims pending slots, evaluates with the worker's criterion clone

Protected Attributes
const unsigned int	_max_threads
	size of the persistent worker pool, as given to the constructor
PCriterion	template_crit
	to detect change of employed criterion inbetween search() calls
bool	initialized
	true after the first initialize() call (guards clone (re)creation logic)
bool	working
	true while a batch is open (between initialize()/add_candidate() and finish_batch())
std::vector< ThreadLocal >	tlocal
	per-worker criterion clone + status, one entry per worker index; sized once in the constructor, never resized (no reallocation races)
std::vector< std::thread >	_pworkers
DIMTYPE	_processed_candidates
	candidates with finished processing, or being currently processed
DIMTYPE	_available_candidates
	to sychronize workers with adding new candidates to cres[]
DIMTYPE	_completed_candidates
	candidates whose result is stored - batch done when it reaches _available_candidates
bool	_pool_started
	the persistent pool exists (lazy spawn at first initialize())
bool	_pshutdown
	destructor flag: workers drain pending work and exit
std::mutex	mutex_candidate_getter
	guards the candidate counters (workers claim/complete, master adds/waits)
std::condition_variable	condition_available
	wakes parked workers: new candidate available (or shutdown)
std::condition_variable	condition_batch_done
	wakes the master: a candidate completed (finish_batch waits on the counter)
std::vector< CandidateResult >	cres
	one entry per batch candidate, indexed by add-order (the position add_candidate() assigned it)
DIMTYPE	_reading_result
	iterating index for get*Result readers
ResultCallback	_on_result
	optional per-candidate callback, fired in add-order when the batch completes
bool	_callbacks_fired
	ensures _on_result fires exactly once per batch
std::atomic< bool >	_batch_failed
	batch-mode twin of the sequential engine's _failed: a worker saw evaluate()==false
std::exception_ptr	_worker_exception
	first exception a criterion threw on a worker; rethrown on the master after join
std::mutex	mutex_worker_exception
	guards _worker_exception (shared by batch and streaming workers; the modes never overlap)
std::vector< PSubset >	_sslot_sub
	ring: master-generated candidates
std::vector< RETURNTYPE >	_sslot_val
	ring: evaluation results
std::vector< unsigned char >	_sslot_state
	ring: StreamSlotState per slot
std::list< std::size_t >	_spending
	pending slot indices in generation order
std::mutex	mutex_stream
	guards the ring bookkeeping
std::condition_variable	_scv_work
	wakes workers on new pending slots / shutdown
std::condition_variable	_scv_done
	wakes the master when a slot completes
bool	_sstop
	shutdown flag for stream workers

Detailed Description

template<class RETURNTYPE, typename DIMTYPE, class SUBSET, class CRITERION>
class FST::Candidate_Evaluator_Threaded< RETURNTYPE, DIMTYPE, SUBSET, CRITERION >

Implements concurrent evaluation of a criterion on a set of subset candidates.

Concurrently evaluates subset candidates using clones of the supplied criterion and at most max_threads worker threads. It implements the abstract Candidate_Evaluator strategy interface, so a search can take either this or Candidate_Evaluator_Sequential through the same seam and behave identically apart from speed.

Two modes are offered, as in the interface:

BATCH mode - initialize(), a sequence of add_candidate() calls, then collect() or getBestResult(); getBestResult() returns the candidate that yielded the highest criterion value, getFirstResult()/getNextResult() iterate over ALL results of the batch in add-order. The optional per-result callback fires for all candidates in add-order when the batch completes.
STREAMING mode - evaluate_stream() consumes candidates from a generator until it is exhausted and maintains the best-so-far itself.

Warning: The engine CLONES the criterion (once per worker) the first time it sees a given criterion object, and reuses those clones for every later batch or stream in which the same criterion object is passed again; a clone is created anew only when the criterion pointer handed to initialize()/evaluate_stream() differs from the one seen before. The clones' internal state is NOT re-synchronized with the original in between. If the original's evaluation context changes behind its back - the typical case being a wrapper criterion whose data accessor has moved to another data split - the clones keep evaluating in the old context and the results are silently wrong. When that can happen, call invalidate_clones() before the next batch or stream (or re-create the engine, or the search step that owns it) so that a fresh set of clones is made from the criterion in its new context.

Note: Batch workers form a PERSISTENT pool: spawned once at the first initialize(), parked on a condition variable between batches, joined only in the destructor. Batch completion is detected by a completion counter, not by joining, so a batch costs two mutexed counter sweeps instead of max_threads thread spawns plus joins. This matters for users that submit many tiny batches (Branch & Bound submits one per node expansion), which the spawn-per-batch alternative made slower threaded than sequential. Scheduling, result placement (cres[] indexed by add-order) and callback order do not depend on it, so the determinism guarantees below hold either way.; evaluate_stream(): the MASTER thread generates all candidates into a bounded ring (so any randomness in the generator draws from the master's fst_rand stream, exactly as sequential execution would), worker threads evaluate them concurrently with criterion clones, and the master replays completed slots strictly in generation order - tracker adds, best-updates (identical tie-breaking) and on_best callbacks all fire on the master. Results, tracker contents and callback sequence are therefore the same as with Candidate_Evaluator_Sequential at the same seed, for any thread count. Note that letting workers pull from the generator instead would make them draw from worker-thread_local RNG streams, which no seed can make reproducible.; Failure semantics match the sequential engine: a criterion evaluate() returning false marks the batch/stream failed (collect()/getBestResult()/ evaluate_stream() then return false, per-result callbacks do not fire), and an exception thrown by a criterion on a worker is captured and rethrown on the MASTER thread after the workers join - never left to escape the worker, which would std::terminate the process with no FST diagnostic.

Examples: demo52t.cpp.

Member Enumeration Documentation

◆ StreamSlotState

template<class RETURNTYPE, typename DIMTYPE, class SUBSET, class CRITERION>

enum FST::Candidate_Evaluator_Threaded::StreamSlotState

protected

shared state of the ordered streaming mode (evaluate_stream): a bounded ring of master-generated candidate slots, evaluated by workers, replayed in order by the master. A slot travels SLOT_EMPTY (free for the master to generate into) -> SLOT_PENDING (queued in _spending, not claimed yet) -> SLOT_RUNNING (a worker is evaluating it) -> SLOT_DONE or SLOT_FAILED (the criterion returned false or threw), and back to SLOT_EMPTY once the master has replayed it. All transitions happen under mutex_stream.

Constructor & Destructor Documentation

◆ Candidate_Evaluator_Threaded()

template<class RETURNTYPE, typename DIMTYPE, class SUBSET, class CRITERION>

FST::Candidate_Evaluator_Threaded< RETURNTYPE, DIMTYPE, SUBSET, CRITERION >::Candidate_Evaluator_Threaded ( const unsigned int max_threads = 2 )

inlineexplicit

Parameters

max_threads size of the persistent worker pool (lazily spawned at the first initialize() call, parked between batches, joined only in the destructor)

Member Function Documentation

◆ add_candidate()

template<class RETURNTYPE, typename DIMTYPE, class SUBSET, class CRITERION>

bool FST::Candidate_Evaluator_Threaded< RETURNTYPE, DIMTYPE, SUBSET, CRITERION >::add_candidate	(	const PSubset &	sub,
		const DIMTYPE	feature = 0 )

overridevirtual

adds candidate subset to the evaluation pool; returns false once the result storage reserved by initialize() is full, i.e. when more candidates are added than max_candidates announced

Implements FST::Candidate_Evaluator< RETURNTYPE, DIMTYPE, SUBSET, CRITERION >.

◆ collect()

template<class RETURNTYPE, typename DIMTYPE, class SUBSET, class CRITERION>

bool FST::Candidate_Evaluator_Threaded< RETURNTYPE, DIMTYPE, SUBSET, CRITERION >::collect ( )

overridevirtual

completes the batch: waits for all evaluations and fires pending per-result callbacks in add-order

Implements FST::Candidate_Evaluator< RETURNTYPE, DIMTYPE, SUBSET, CRITERION >.

◆ evaluate_stream()

template<class RETURNTYPE, typename DIMTYPE, class SUBSET, class CRITERION>

bool FST::Candidate_Evaluator_Threaded< RETURNTYPE, DIMTYPE, SUBSET, CRITERION >::evaluate_stream	(	Generator	generator,
		const PCriterion &	crit,
		const PResultTracker	tracker,
		BetterPredicate	better,
		BestCallback	on_best,
		RETURNTYPE &	result,
		const PSubset	bestsub )

overridevirtual

streaming mode: the master thread calls generator until exhaustion, workers evaluate concurrently, the master replays the results in generation order (see the class notes)

Implements FST::Candidate_Evaluator< RETURNTYPE, DIMTYPE, SUBSET, CRITERION >.

◆ get_max_threads()

template<class RETURNTYPE, typename DIMTYPE, class SUBSET, class CRITERION>

virtual unsigned int FST::Candidate_Evaluator_Threaded< RETURNTYPE, DIMTYPE, SUBSET, CRITERION >::get_max_threads ( ) const

inlineoverridevirtual

1 in the sequential implementation

Implements FST::Candidate_Evaluator< RETURNTYPE, DIMTYPE, SUBSET, CRITERION >.

◆ getBestResult()

template<class RETURNTYPE, typename DIMTYPE, class SUBSET, class CRITERION>

bool FST::Candidate_Evaluator_Threaded< RETURNTYPE, DIMTYPE, SUBSET, CRITERION >::getBestResult	(	RETURNTYPE &	result,
		PSubset &	sub,
		DIMTYPE &	feature )

overridevirtual

completes the batch, then returns (without copying sub contents) the candidate that yielded the highest criterion value; ties go to the candidate added first

Implements FST::Candidate_Evaluator< RETURNTYPE, DIMTYPE, SUBSET, CRITERION >.

◆ initialize()

template<class RETURNTYPE, typename DIMTYPE, class SUBSET, class CRITERION>

void FST::Candidate_Evaluator_Threaded< RETURNTYPE, DIMTYPE, SUBSET, CRITERION >::initialize	(	const PCriterion &	crit,
		const DIMTYPE	max_candidates,
		ResultCallback	on_result = ResultCallback() )

overridevirtual

Opens a new batch to be evaluated against crit, which must already be initialize()d. max_candidates must be greater or equal than the number of candidates that will be added - the result storage is sized here once, so that workers never race a vector reallocation; it must be nonzero. See the criterion-clone warning in the class documentation.

Implements FST::Candidate_Evaluator< RETURNTYPE, DIMTYPE, SUBSET, CRITERION >.

◆ invalidate_clones()

template<class RETURNTYPE, typename DIMTYPE, class SUBSET, class CRITERION>

void FST::Candidate_Evaluator_Threaded< RETURNTYPE, DIMTYPE, SUBSET, CRITERION >::invalidate_clones ( )

overridevirtual

drops the cached per-worker criterion clones (see Candidate_Evaluator::invalidate_clones()); the next initialize()/evaluate_stream() re-clones from the criterion it is handed

Reimplemented from FST::Candidate_Evaluator< RETURNTYPE, DIMTYPE, SUBSET, CRITERION >.

◆ is_threaded()

template<class RETURNTYPE, typename DIMTYPE, class SUBSET, class CRITERION>

virtual bool FST::Candidate_Evaluator_Threaded< RETURNTYPE, DIMTYPE, SUBSET, CRITERION >::is_threaded ( ) const

inlineoverridevirtual

true for the threaded engine (any --threads value), false for the sequential one – distinguishes the engine kind, not whether get_max_threads()>1

Implements FST::Candidate_Evaluator< RETURNTYPE, DIMTYPE, SUBSET, CRITERION >.

◆ print()

template<class RETURNTYPE, typename DIMTYPE, class SUBSET, class CRITERION>

virtual std::ostream & FST::Candidate_Evaluator_Threaded< RETURNTYPE, DIMTYPE, SUBSET, CRITERION >::print ( std::ostream & os ) const

inlineoverridevirtual

writes a human-readable identification of the engine; base default writes nothing

Reimplemented from FST::Candidate_Evaluator< RETURNTYPE, DIMTYPE, SUBSET, CRITERION >.

Member Data Documentation

◆ _pworkers

template<class RETURNTYPE, typename DIMTYPE, class SUBSET, class CRITERION>

std::vector<std::thread> FST::Candidate_Evaluator_Threaded< RETURNTYPE, DIMTYPE, SUBSET, CRITERION >::_pworkers

protected

persistent batch-worker pool: workers spawn ONCE at first initialize() and PARK on condition_available between batches instead of being joined and respawned per batch - the per-batch overhead that makes many-small-batch users (Branch & Bound submits one batch per node expansion) slower threaded than sequential. Batch completion is signalled by the _completed_candidates counter (condition_batch_done), never by joining; workers join only in the destructor. Result placement (cres[] by add-order index) and callback order do not depend on the pooling.

The documentation for this class was generated from the following file:

/Users/psms/dev/devtest1/_src_search/candidate_evaluator_threaded.hpp

Classes

Public Types

Public Member Functions

Protected Types

Protected Member Functions

Protected Attributes

Detailed Description

Member Enumeration Documentation

◆ StreamSlotState

Constructor & Destructor Documentation

◆ Candidate_Evaluator_Threaded()

Member Function Documentation

◆ add_candidate()

◆ collect()

◆ evaluate_stream()

◆ get_max_threads()

◆ getBestResult()

◆ initialize()

◆ invalidate_clones()

◆ is_threaded()

◆ print()

Member Data Documentation

◆ _pworkers