GRSISort
Created by P.C. Bender
Developement Team: P.C. Bender, R. Dunlop, V. Bildstein
An extension of the ROOT analysis Framework
Public Member Functions | Static Public Member Functions | Static Public Attributes | Protected Member Functions | Protected Attributes | Static Protected Attributes | Private Member Functions | Private Attributes | List of all members
TAnalysisWriteLoop Class Reference
Loops

This loop writes built events to file

Definition at line 27 of file TAnalysisWriteLoop.h.

Public Member Functions

 ~TAnalysisWriteLoop () override
 
void ClearQueue () override
 
std::string EndStatus () override
 
size_t GetItemsCurrent () override
 
size_t GetItemsPopped () override
 
size_t GetItemsPushed () override
 
size_t GetRate () override
 
std::shared_ptr< ThreadsafeQueue< std::shared_ptr< TUnpackedEvent > > > & InputQueue ()
 
bool IsPaused ()
 
bool IsRunning ()
 
void Join ()
 
std::string Name () const
 
void OnEnd () override
 
std::shared_ptr< ThreadsafeQueue< std::shared_ptr< const TFragment > > > & OutOfOrderQueue ()
 
void Pause ()
 
virtual std::string Progress ()
 
void Resume ()
 
virtual std::string Status ()
 
void Stop ()
 
void Write ()
 

Static Public Member Functions

static std::string AllThreadHeader ()
 
static std::string AllThreadProgress ()
 
static std::string AllThreadStatus ()
 
static bool AnyThreadRunning ()
 
static std::string AnyThreadStatus ()
 
static void ClearAllQueues ()
 
static void ColumnWidth (size_t val)
 
static size_t ColumnWidth ()
 
static TAnalysisWriteLoop * Get (std::string name="", std::string output_filename="")
 
static StoppableThread * Get (const std::string &name)
 
static std::vector< StoppableThread * > GetAll ()
 
static int GetNThreads ()
 
static void join_status_thread ()
 
static void PauseAll ()
 
static void Print ()
 
static void ResumeAll ()
 
static void SendStop ()
 
static void start_status_thread ()
 
static void status_out ()
 
static void status_out_loop ()
 
static void StatusWidth (size_t val)
 
static size_t StatusWidth ()
 
static void stop_status_thread ()
 
static void StopAll ()
 

Static Public Attributes

static std::thread fStatusThread
 
static bool fStatusThreadOn = false
 

Protected Member Functions

bool Iteration () override
 

Protected Attributes

std::atomic_long fInputSize {0}
 
std::atomic_size_t fItemsPopped {0}
 
std::string fName
 

Static Protected Attributes

static std::map< std::string, StoppableThread * > fThreadMap
 

Private Member Functions

 TAnalysisWriteLoop (std::string name, std::string output_filename)
 
void AddBranch (TClass *cls)
 
 ClassDefOverride (TAnalysisWriteLoop, 0)
 
bool Server ()
 

Private Attributes

std::vector< TAnalysisWriteLoopClient * > fClients
 
size_t fCurrentClient
 
std::shared_ptr< ThreadsafeQueue< std::shared_ptr< TUnpackedEvent > > > fInputQueue
 
bool fOutOfOrder
 
std::shared_ptr< ThreadsafeQueue< std::shared_ptr< const TFragment > > > fOutOfOrderQueue
 
std::string fOutputFilename
 
std::future< bool > fServerFuture
 
TServerSocket * fServerSocket
 

#include <TAnalysisWriteLoop.h>

+ Inheritance diagram for TAnalysisWriteLoop:
+ Collaboration diagram for TAnalysisWriteLoop:

Constructor & Destructor Documentation

◆ ~TAnalysisWriteLoop()

TAnalysisWriteLoop::~TAnalysisWriteLoop ( )
override

Definition at line 62 of file TAnalysisWriteLoop.cxx.

References fServerFuture, fServerSocket, and Write().

◆ TAnalysisWriteLoop()

TAnalysisWriteLoop::TAnalysisWriteLoop ( std::string  name,
std::string  output_filename 
)
private

Open a server socket looking for connections on a named service or on a specific port.

Definition at line 43 of file TAnalysisWriteLoop.cxx.

References fClients, fOutOfOrder, fOutputFilename, fServerFuture, fServerSocket, TGRSIOptions::Get(), TGRSIOptions::NumberOfClients(), TGRSIOptions::SeparateOutOfOrder(), and Server().

Referenced by Get().

Member Function Documentation

◆ AddBranch()

void TAnalysisWriteLoop::AddBranch ( TClass *  cls)
private

◆ AllThreadHeader()

std::string StoppableThread::AllThreadHeader ( )
staticinherited

Definition at line 73 of file StoppableThread.cxx.

References StoppableThread::fColumnWidth, StoppableThread::fStatusWidth, and StoppableThread::fThreadMap.

Referenced by TGRSIint::ApplyOptions().

◆ AllThreadProgress()

std::string StoppableThread::AllThreadProgress ( )
staticinherited

Definition at line 59 of file StoppableThread.cxx.

References StoppableThread::fColumnWidth, StoppableThread::fStatusWidth, and StoppableThread::fThreadMap.

Referenced by TGRSIint::LoopUntilDone().

◆ AllThreadStatus()

std::string StoppableThread::AllThreadStatus ( )
staticinherited

Definition at line 83 of file StoppableThread.cxx.

References StoppableThread::fColumnWidth, StoppableThread::fStatusWidth, and StoppableThread::fThreadMap.

Referenced by TGRSIint::LoopUntilDone().

◆ AnyThreadRunning()

bool StoppableThread::AnyThreadRunning ( )
staticinherited

Definition at line 37 of file StoppableThread.cxx.

References StoppableThread::fThreadMap.

Referenced by TChannel::GetChannel(), TGRSIint::LoopUntilDone(), and TGRSIInterruptHandler::Notify().

◆ AnyThreadStatus()

std::string StoppableThread::AnyThreadStatus ( )
staticinherited

Definition at line 47 of file StoppableThread.cxx.

References StoppableThread::fThreadMap.

◆ ClassDefOverride()

TAnalysisWriteLoop::ClassDefOverride ( TAnalysisWriteLoop  ,
0   
)
private

◆ ClearAllQueues()

void StoppableThread::ClearAllQueues ( )
staticinherited

Definition at line 166 of file StoppableThread.cxx.

References StoppableThread::fThreadMap.

Referenced by TGRSIInterruptHandler::Notify().

◆ ClearQueue()

void TAnalysisWriteLoop::ClearQueue ( )
overridevirtual

Reimplemented from StoppableThread.

Definition at line 71 of file TAnalysisWriteLoop.cxx.

References fClients, and fInputQueue.

◆ ColumnWidth() [1/2]

static void StoppableThread::ColumnWidth ( size_t  val)
inlinestaticinherited

Definition at line 75 of file StoppableThread.h.

References StoppableThread::fColumnWidth.

◆ ColumnWidth() [2/2]

static size_t StoppableThread::ColumnWidth ( )
inlinestaticinherited

Definition at line 77 of file StoppableThread.h.

References StoppableThread::fColumnWidth.

Referenced by TGRSIint::SetupPipeline().

◆ EndStatus()

std::string TAnalysisWriteLoop::EndStatus ( )
overridevirtual

Reimplemented from StoppableThread.

Definition at line 82 of file TAnalysisWriteLoop.cxx.

References StoppableThread::fInputSize, StoppableThread::fItemsPopped, and StoppableThread::Name().

◆ Get() [1/2]

TAnalysisWriteLoop * TAnalysisWriteLoop::Get ( std::string  name = "",
std::string  output_filename = "" 
)
static

Definition at line 26 of file TAnalysisWriteLoop.cxx.

References StoppableThread::Get(), and TAnalysisWriteLoop().

Referenced by TGRSIint::SetupPipeline().

◆ Get() [2/2]

StoppableThread * StoppableThread::Get ( const std::string &  name)
staticinherited

Definition at line 173 of file StoppableThread.cxx.

References StoppableThread::fThreadMap.

Referenced by ClassImp(), TTerminalLoop< T >::Get(), TFragHistLoop::Get(), TAnalysisHistLoop::Get(), Get(), TFragWriteLoop::Get(), TUnpackingLoop::Get(), TFragmentChainLoop::Get(), and TDataLoop::Get().

◆ GetAll()

std::vector< StoppableThread * > StoppableThread::GetAll ( )
staticinherited

Definition at line 315 of file StoppableThread.cxx.

References StoppableThread::fThreadMap.

◆ GetItemsCurrent()

size_t TAnalysisWriteLoop::GetItemsCurrent ( )
inlineoverridevirtual

Implements StoppableThread.

Definition at line 44 of file TAnalysisWriteLoop.h.

◆ GetItemsPopped()

size_t TAnalysisWriteLoop::GetItemsPopped ( )
inlineoverridevirtual

Implements StoppableThread.

Definition at line 43 of file TAnalysisWriteLoop.h.

◆ GetItemsPushed()

size_t TAnalysisWriteLoop::GetItemsPushed ( )
inlineoverridevirtual

Implements StoppableThread.

Definition at line 42 of file TAnalysisWriteLoop.h.

References StoppableThread::fItemsPopped.

◆ GetNThreads()

int StoppableThread::GetNThreads ( )
staticinherited

Definition at line 21 of file StoppableThread.cxx.

References StoppableThread::fThreadMap.

Referenced by StoppableThread::Print().

◆ GetRate()

size_t TAnalysisWriteLoop::GetRate ( )
inlineoverridevirtual

Implements StoppableThread.

Definition at line 45 of file TAnalysisWriteLoop.h.

◆ InputQueue()

std::shared_ptr<ThreadsafeQueue<std::shared_ptr<TUnpackedEvent> > >& TAnalysisWriteLoop::InputQueue ( )
inline

Definition at line 34 of file TAnalysisWriteLoop.h.

References fInputQueue.

Referenced by TGRSIint::SetupPipeline().

◆ IsPaused()

bool StoppableThread::IsPaused ( )
inherited

Definition at line 224 of file StoppableThread.cxx.

References StoppableThread::fPaused.

◆ IsRunning()

bool StoppableThread::IsRunning ( )
inherited

Definition at line 219 of file StoppableThread.cxx.

References StoppableThread::fRunning.

Referenced by StoppableThread::status_out().

◆ Iteration()

bool TAnalysisWriteLoop::Iteration ( )
overrideprotectedvirtual

Implements StoppableThread.

Definition at line 97 of file TAnalysisWriteLoop.cxx.

References fClients, fCurrentClient, fInputQueue, StoppableThread::fInputSize, StoppableThread::fItemsPopped, fOutOfOrder, and fOutOfOrderQueue.

◆ Join()

void StoppableThread::Join ( )
inherited

Definition at line 229 of file StoppableThread.cxx.

References StoppableThread::EndStatus(), and StoppableThread::fThread.

Referenced by StoppableThread::StopAll().

◆ join_status_thread()

void StoppableThread::join_status_thread ( )
staticinherited

Definition at line 280 of file StoppableThread.cxx.

References StoppableThread::fStatusThread, and StoppableThread::stop_status_thread().

◆ Name()

std::string StoppableThread::Name ( ) const
inlineinherited

Definition at line 58 of file StoppableThread.h.

References StoppableThread::fName.

Referenced by TAnalysisWriteLoopClient::AddBranch(), EndStatus(), TFragWriteLoop::EndStatus(), TUnpackingLoop::EndStatus(), and StoppableThread::status_out().

◆ OnEnd()

void TAnalysisWriteLoop::OnEnd ( )
overridevirtual

Reimplemented from StoppableThread.

Definition at line 90 of file TAnalysisWriteLoop.cxx.

References fClients.

◆ OutOfOrderQueue()

std::shared_ptr<ThreadsafeQueue<std::shared_ptr<const TFragment> > >& TAnalysisWriteLoop::OutOfOrderQueue ( )
inline

Definition at line 35 of file TAnalysisWriteLoop.h.

References fOutOfOrderQueue.

Referenced by TGRSIint::SetupPipeline().

◆ Pause()

void StoppableThread::Pause ( )
inherited

Definition at line 202 of file StoppableThread.cxx.

References StoppableThread::fPaused, and StoppableThread::fRunning.

◆ PauseAll()

void StoppableThread::PauseAll ( )
staticinherited

Definition at line 97 of file StoppableThread.cxx.

References StoppableThread::fThreadMap.

◆ Print()

void StoppableThread::Print ( )
staticinherited

Definition at line 254 of file StoppableThread.cxx.

References StoppableThread::fColumnWidth, StoppableThread::fStatusWidth, StoppableThread::fThreadMap, StoppableThread::GetNThreads(), and hex().

◆ Progress()

std::string StoppableThread::Progress ( )
virtualinherited

Definition at line 119 of file StoppableThread.cxx.

References StoppableThread::fColumnWidth, StoppableThread::fInputSize, and StoppableThread::fItemsPopped.

◆ Resume()

void StoppableThread::Resume ( )
inherited

Definition at line 193 of file StoppableThread.cxx.

References StoppableThread::fPaused, StoppableThread::fPausedWait, StoppableThread::fPauseMutex, and StoppableThread::fRunning.

◆ ResumeAll()

void StoppableThread::ResumeAll ( )
staticinherited

Definition at line 104 of file StoppableThread.cxx.

References StoppableThread::fThreadMap.

Referenced by TGRSIint::SetupPipeline().

◆ SendStop()

void StoppableThread::SendStop ( )
staticinherited

Definition at line 135 of file StoppableThread.cxx.

References StoppableThread::fThreadMap, and StoppableThread::Stop().

Referenced by StoppableThread::StopAll(), and TGRSIint::Terminate().

◆ Server()

bool TAnalysisWriteLoop::Server ( )
private

Definition at line 163 of file TAnalysisWriteLoop.cxx.

References fServerSocket, TParallelFileMerger::InitialMerge(), TParallelFileMerger::Merge(), pygui.grut-view::message, TParallelFileMerger::NeedFinalMerge(), NeedInitialMerge(), TParallelFileMerger::NeedMerge(), and TParallelFileMerger::RegisterClient().

Referenced by TAnalysisWriteLoop().

◆ start_status_thread()

void StoppableThread::start_status_thread ( )
staticinherited

Definition at line 265 of file StoppableThread.cxx.

References StoppableThread::fStatusThread, StoppableThread::fStatusThreadOn, and StoppableThread::status_out_loop().

Referenced by StoppableThread::StoppableThread().

◆ Status()

std::string StoppableThread::Status ( )
virtualinherited

Definition at line 111 of file StoppableThread.cxx.

References StoppableThread::fColumnWidth, StoppableThread::fInputSize, and StoppableThread::fItemsPopped.

◆ status_out()

void StoppableThread::status_out ( )
staticinherited

Definition at line 297 of file StoppableThread.cxx.

References StoppableThread::fThreadMap, StoppableThread::GetItemsCurrent(), StoppableThread::GetItemsPopped(), StoppableThread::GetItemsPushed(), StoppableThread::GetRate(), StoppableThread::IsRunning(), and StoppableThread::Name().

Referenced by StoppableThread::status_out_loop(), and StoppableThread::StopAll().

◆ status_out_loop()

void StoppableThread::status_out_loop ( )
staticinherited

Definition at line 286 of file StoppableThread.cxx.

References StoppableThread::fStatusThreadOn, and StoppableThread::status_out().

Referenced by StoppableThread::start_status_thread().

◆ StatusWidth() [1/2]

static void StoppableThread::StatusWidth ( size_t  val)
inlinestaticinherited

Definition at line 76 of file StoppableThread.h.

References StoppableThread::fStatusWidth.

◆ StatusWidth() [2/2]

static size_t StoppableThread::StatusWidth ( )
inlinestaticinherited

Definition at line 78 of file StoppableThread.h.

References StoppableThread::fStatusWidth.

Referenced by TGRSIint::SetupPipeline().

◆ Stop()

void StoppableThread::Stop ( )
inherited

Definition at line 209 of file StoppableThread.cxx.

References StoppableThread::EndStatus(), StoppableThread::fPaused, StoppableThread::fPausedWait, StoppableThread::fPauseMutex, and StoppableThread::fRunning.

Referenced by StoppableThread::SendStop().

◆ stop_status_thread()

void StoppableThread::stop_status_thread ( )
staticinherited

Definition at line 273 of file StoppableThread.cxx.

References StoppableThread::fStatusThreadOn.

Referenced by StoppableThread::join_status_thread().

◆ StopAll()

void StoppableThread::StopAll ( )
staticinherited

Definition at line 147 of file StoppableThread.cxx.

References StoppableThread::fThreadMap, StoppableThread::Join(), StoppableThread::SendStop(), and StoppableThread::status_out().

Referenced by TGRSIint::Terminate().

◆ Write()

void TAnalysisWriteLoop::Write ( )

Definition at line 132 of file TAnalysisWriteLoop.cxx.

References TGRSIOptions::AnalysisOptions(), fOutputFilename, TGRSIOptions::Get(), GValue::Get(), TSingleton< TSortingDiagnostics >::Get(), TSingleton< TPPG >::Get(), TSingleton< TRunInfo >::Get(), TChannel::GetNumberOfChannels(), GValue::Size(), TPPG::Write(), TGRSIOptions::WriteDiagnostics(), TAnalysisOptions::WriteToFile(), TRunInfo::WriteToRoot(), and TChannel::WriteToRoot().

Referenced by ~TAnalysisWriteLoop().

Member Data Documentation

◆ fClients

std::vector<TAnalysisWriteLoopClient*> TAnalysisWriteLoop::fClients
private

Definition at line 66 of file TAnalysisWriteLoop.h.

Referenced by ClearQueue(), Iteration(), OnEnd(), and TAnalysisWriteLoop().

◆ fCurrentClient

size_t TAnalysisWriteLoop::fCurrentClient
private

Definition at line 60 of file TAnalysisWriteLoop.h.

Referenced by Iteration().

◆ fInputQueue

std::shared_ptr<ThreadsafeQueue<std::shared_ptr<TUnpackedEvent> > > TAnalysisWriteLoop::fInputQueue
private

Definition at line 64 of file TAnalysisWriteLoop.h.

Referenced by ClearQueue(), InputQueue(), and Iteration().

◆ fInputSize

std::atomic_long StoppableThread::fInputSize {0}
protectedinherited

number of items in the input (queue), only updated within Iteration(), so not always fully up-to-date (signed to hold error from queue::pop)

Definition at line 95 of file StoppableThread.h.

Referenced by EndStatus(), TFragWriteLoop::EndStatus(), TAnalysisWriteLoopClient::GetInputSize(), TEventBuildingLoop::Iteration(), TDataLoop::Iteration(), TDetBuildingLoop::Iteration(), Iteration(), TUnpackingLoop::Iteration(), TFragWriteLoop::Iteration(), TFragHistLoop::Iteration(), TFragmentChainLoop::Iteration(), TAnalysisHistLoop::Iteration(), TAnalysisWriteLoopClient::Iteration(), StoppableThread::Progress(), and StoppableThread::Status().

◆ fItemsPopped

std::atomic_size_t StoppableThread::fItemsPopped {0}
protectedinherited

number of items popped from input queue

Definition at line 94 of file StoppableThread.h.

Referenced by EndStatus(), TFragWriteLoop::EndStatus(), TEventBuildingLoop::EndStatus(), TFragmentChainLoop::GetItemsPopped(), TFragWriteLoop::GetItemsPopped(), GetItemsPushed(), TFragmentChainLoop::GetItemsPushed(), TFragWriteLoop::GetItemsPushed(), TAnalysisWriteLoopClient::GetItemsPushed(), TEventBuildingLoop::Iteration(), TDataLoop::Iteration(), TDetBuildingLoop::Iteration(), Iteration(), TUnpackingLoop::Iteration(), TFragWriteLoop::Iteration(), TFragHistLoop::Iteration(), TAnalysisHistLoop::Iteration(), TAnalysisWriteLoopClient::Iteration(), TFragmentChainLoop::Iteration(), StoppableThread::Progress(), TFragmentChainLoop::Restart(), and StoppableThread::Status().

◆ fName

std::string StoppableThread::fName
protectedinherited

Definition at line 99 of file StoppableThread.h.

Referenced by StoppableThread::Name(), StoppableThread::StoppableThread(), and StoppableThread::~StoppableThread().

◆ fOutOfOrder

bool TAnalysisWriteLoop::fOutOfOrder
private

Definition at line 61 of file TAnalysisWriteLoop.h.

Referenced by Iteration(), and TAnalysisWriteLoop().

◆ fOutOfOrderQueue

std::shared_ptr<ThreadsafeQueue<std::shared_ptr<const TFragment> > > TAnalysisWriteLoop::fOutOfOrderQueue
private

Definition at line 65 of file TAnalysisWriteLoop.h.

Referenced by Iteration(), and OutOfOrderQueue().

◆ fOutputFilename

std::string TAnalysisWriteLoop::fOutputFilename
private

Definition at line 59 of file TAnalysisWriteLoop.h.

Referenced by TAnalysisWriteLoop(), and Write().

◆ fServerFuture

std::future<bool> TAnalysisWriteLoop::fServerFuture
private

Definition at line 67 of file TAnalysisWriteLoop.h.

Referenced by TAnalysisWriteLoop(), and ~TAnalysisWriteLoop().

◆ fServerSocket

TServerSocket* TAnalysisWriteLoop::fServerSocket
private

Definition at line 62 of file TAnalysisWriteLoop.h.

Referenced by Server(), TAnalysisWriteLoop(), and ~TAnalysisWriteLoop().

◆ fStatusThread

std::thread StoppableThread::fStatusThread
staticinherited

Definition at line 81 of file StoppableThread.h.

Referenced by StoppableThread::join_status_thread(), StoppableThread::start_status_thread(), and StoppableThread::~StoppableThread().

◆ fStatusThreadOn

bool StoppableThread::fStatusThreadOn = false
staticinherited

Definition at line 83 of file StoppableThread.h.

Referenced by StoppableThread::start_status_thread(), StoppableThread::status_out_loop(), StoppableThread::stop_status_thread(), StoppableThread::StoppableThread(), and StoppableThread::~StoppableThread().

◆ fThreadMap

std::map< std::string, StoppableThread * > StoppableThread::fThreadMap
staticprotectedinherited

Definition at line 91 of file StoppableThread.h.

Referenced by StoppableThread::AllThreadHeader(), StoppableThread::AllThreadProgress(), StoppableThread::AllThreadStatus(), StoppableThread::AnyThreadRunning(), StoppableThread::AnyThreadStatus(), StoppableThread::ClearAllQueues(), StoppableThread::Get(), StoppableThread::GetAll(), StoppableThread::GetNThreads(), StoppableThread::PauseAll(), StoppableThread::Print(), StoppableThread::ResumeAll(), StoppableThread::SendStop(), StoppableThread::status_out(), StoppableThread::StopAll(), StoppableThread::StoppableThread(), and StoppableThread::~StoppableThread().