Thread pool for event-by-event reconstruction. More...

#include <JMultiThreadedReconstruction.hh>

Classes
struct	output_type
	Output data type. More...

struct	queue_type_t
	Auxiliary data structure to maintain time order of events for writing. More...

Public Types
typedef JFit_t::input_type	input_type

typedef JLANG::JObjectOutput< JEvt >	writer_type

Public Member Functions
	JMultiThreadedReconstruction (const JFit_t &fit, writer_type &writer, const size_t ns, const size_t backlog=std::numeric_limits< size_t >::max())
	Constructor.

	~JMultiThreadedReconstruction ()
	Destructor.

void	enqueue (input_type &data)
	Add data in queue.

void	flush ()
	Flush output.

Private Types
typedef std::priority_queue< output_type, std::vector< output_type > >	queue_type
	Type definition of output queue.

Private Attributes
JRECONSTRUCTION::JMultiThreadedReconstruction::queue_type_t	output

writer_type &	writer

std::vector< std::thread >	workers

std::queue< input_type >	input

std::mutex	in

std::mutex	out

std::condition_variable	cv

std::condition_variable	cw

bool	stop

size_t	backlog

Detailed Description

template<class JFit_t>
class JRECONSTRUCTION::JMultiThreadedReconstruction< JFit_t >

Thread pool for event-by-event reconstruction.

Definition at line 32 of file JMultiThreadedReconstruction.hh.

Member Typedef Documentation

◆ input_type

template<class JFit_t >

JFit_t::input_type JRECONSTRUCTION::JMultiThreadedReconstruction< JFit_t >::input_type

Definition at line 35 of file JMultiThreadedReconstruction.hh.

◆ writer_type

template<class JFit_t >

JLANG::JObjectOutput<JEvt> JRECONSTRUCTION::JMultiThreadedReconstruction< JFit_t >::writer_type

Definition at line 36 of file JMultiThreadedReconstruction.hh.

◆ queue_type

template<class JFit_t >

std::priority_queue<output_type, std::vector<output_type> > JRECONSTRUCTION::JMultiThreadedReconstruction< JFit_t >::queue_type

private

Type definition of output queue.

Definition at line 223 of file JMultiThreadedReconstruction.hh.

Constructor & Destructor Documentation

◆ JMultiThreadedReconstruction()

template<class JFit_t >

JRECONSTRUCTION::JMultiThreadedReconstruction< JFit_t >::JMultiThreadedReconstruction	(	const JFit_t &	fit,
		writer_type &	writer,
		const size_t	ns,
		const size_t	backlog = std::numeric_limits<size_t>::max() )

inline

Constructor.

Parameters

fit	fit
writer	writer
ns	number of threads
backlog	backlog

Definition at line 47 of file JMultiThreadedReconstruction.hh.

                                                                                         :
      writer(writer),
      stop(false),
      backlog(backlog)
    {
      using namespace std;
      using namespace JPP;
 
      for (size_t id = 0; id < ns; ++id) {
        output[id] = 0;
      }
 
      for (size_t id = 0; id < ns; ++id) {
 
        thread worker([this, fit, id]() {
 
          input_type data;
 
          for (JFit_t f1(fit); ; ) {
 
            {
              unique_lock<mutex> lock(in);
 
              cv.wait(lock, [this]() { return stop || !input.empty(); });
 
              if (stop && input.empty()) {
                return;
              }
 
              swap(data, input.front());
 
              input.pop();
            }
 
            cw.notify_one();
 
            output_type evt(id, data.getDAQEventHeader(), f1(data));
 
            {
              unique_lock<mutex> lock(out);
 
              output.push(evt);
            }
          }
        });
        
        workers.emplace_back(std::move(worker));
      }
    }

◆ ~JMultiThreadedReconstruction()

template<class JFit_t >

JRECONSTRUCTION::JMultiThreadedReconstruction< JFit_t >::~JMultiThreadedReconstruction ( )

inline

Destructor.

Definition at line 104 of file JMultiThreadedReconstruction.hh.

    {
      using namespace std;
 
      {
        unique_lock<mutex> lock(in);
 
        stop = true;
      }
 
      cv.notify_all();
 
      for (auto& worker : workers) {
        worker.join();
      }
 
      // write remaining output
 
      while (!output.empty()) {
 
        writer.put(output.top());
 
        output.pop();
      }
    }

Member Function Documentation

◆ enqueue()

template<class JFit_t >

void JRECONSTRUCTION::JMultiThreadedReconstruction< JFit_t >::enqueue ( input_type & data )

inline

Add data in queue.

Parameters

data data

Definition at line 136 of file JMultiThreadedReconstruction.hh.

    {
      using namespace std;
 
      {
        unique_lock<mutex> lock(in);
 
        cw.wait(lock, [this]() { return stop || input.size() < backlog; });
 
        if (stop) {
          throw runtime_error("The thread pool has been stopped.");
        }
 
        input.emplace(std::move(data));
      }
 
      cv.notify_one();
 
      flush();
    }

◆ flush()

template<class JFit_t >

void JRECONSTRUCTION::JMultiThreadedReconstruction< JFit_t >::flush ( )

inline

Flush output.

Definition at line 161 of file JMultiThreadedReconstruction.hh.

    {
      unique_lock<mutex> lock(out);
 
      while (output.is_ready()) {
 
        writer.put(output.top());
 
        output.pop();
      }
    }