FrameDispatcher.cpp

Go to the documentation of this file.

/***************************************************************************
 *   Copyright (C) 2011 by Olivier Stezowski                               *
 *   stezow(AT)ipnl.in2p3.fr                                               *
 *                                                                         *
 *   This program is free software; you can redistribute it and/or modify  *
 *   it under the terms of the GNU General Public License as published by  *
 *   the Free Software Foundation; either version 2 of the License, or     *
 *   (at your option) any later version.                                   *
 *                                                                         *
 *   This program is distributed in the hope that it will be useful,       *
 *   but WITHOUT ANY WARRANTY; without even the implied warranty of        *
 *   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the         *
 *   GNU General Public License for more details.                          *
 *                                                                         *
 *   You should have received a copy of the GNU General Public License     *
 *   along with this program; if not, write to the                         *
 *   Free Software Foundation, Inc.,                                       *
 *   59 Temple Place - Suite 330, Boston, MA  02111-1307, USA.             *
 ***************************************************************************/

#ifndef Gw_FrameDispatcher
#include "FrameDispatcher.h"
#endif

// root includes
#include "TSystem.h"
#include "TROOT.h"
#include "TClass.h"

// ADF include

using namespace ADF;
using namespace Gw;

Short_t FrameDispatcher::gDoTimer = 0;

ClassImp(FrameDispatcher);

FrameDispatcher::FrameDispatcher() : 
        Watcher("FD","An unamed Frame Dispatcher",0x0,0x0),
        NarvalConsumer(),
        fListOfTriggers(),
        fTimeControl(),
        fWhichBin(0u),
        fTriggerRate(0x0),
        fBufferSize(0x0),
        fRealTime (new TList()),
        fCPUTime  (new TList()),
        fCurrentValues()
{
        // to force reading a local ADF.conf if it exists
        UInt_t error_code;
        NarvalInterface::process_config("./",&error_code);
        
        // set stack so that triggers are deleted by fListOfTriggers
        fListOfTriggers.SetOwner(true);
        
        // histograms that shows statistics
        fTriggerRate  = MakeTH1<TH1D>("RateWatchers","Gives the number of time at least one Watcher is called", 10,0,10,"Performances");
        fBufferSize   = MakeTH1<TH1F>("BufferSize","Size of the buffer [MB] treated", kMaxCycle,0,kMaxCycle,"Performances") ;

        std::pair<Double_t, Double_t> p; fCurrentValues.push_back(p);
        TH1F* h;
        h = MakeTH1<TH1F>(Form("%s_RealTime",GetName()),"Real time [ms] to execute the Dispacher", kMaxCycle,0,kMaxCycle,"Performances");
        fRealTime->Add(h);
        h =  MakeTH1<TH1F>(Form("%s_CPUTime",GetName()),"CPU time [ms] to execute the Dispacher", kMaxCycle,0,kMaxCycle,"Performances");        
        fCPUTime->Add(h);

        // objects are owned by the pool 
        fRealTime->SetOwner(false); fCPUTime->SetOwner(false);
}

FrameDispatcher::FrameDispatcher(const char *name, const char *title, TDirectory *mother_watcher_dir, TDirectory *mother_tag_watcher_dir) : 
        Watcher(name,title,mother_watcher_dir,mother_tag_watcher_dir),
        NarvalConsumer(),
        fListOfTriggers(),
        fTimeControl(),
        fWhichBin(0u),
        fTriggerRate(0x0),
        fBufferSize(0x0),       
        fRealTime (new TList()),
        fCPUTime  (new TList()),
        fCurrentValues()
{
        // to force reading a local ADF.conf if it exists
        UInt_t error_code;
        NarvalInterface::process_config("./",&error_code);
        
        // set stack so that triggers are deleted by fListOfTriggers
        fListOfTriggers.SetOwner(true);
        
        // histograms that shows statistics
        fTriggerRate  = MakeTH1<TH1D>("RateWatchers","Gives the number of time at least one Watcher is called", 10,0,10,"Performances");
        fBufferSize   = MakeTH1<TH1F>("BufferSize","Size of the buffer [MB] treated", kMaxCycle,0,kMaxCycle,"Performances") ;

        std::pair<Double_t, Double_t> p; fCurrentValues.push_back(p);
        TH1F* h;
        h = MakeTH1<TH1F>(Form("%s_RealTime",GetName()),"Real time [ms] to execute the Dispacher", kMaxCycle,0,kMaxCycle,"Performances");
        fRealTime->Add(h);
        h =  MakeTH1<TH1F>(Form("%s_CPUTime",GetName()),"CPU time [ms] to execute the Dispacher", kMaxCycle,0,kMaxCycle,"Performances");        
        fCPUTime->Add(h);
        
        // objects are owned by the pool 
        fRealTime->SetOwner(false); fCPUTime->SetOwner(false);
}

FrameDispatcher::~FrameDispatcher()
{
//      printf("Delete FrameDispatcher::FrameDispatcher \n");
        // print statistics on the log system
        Log.ClearMessage(); 
        
        Log.SetProcessMethod("~FrameDispatcher()");
        GetFrameIO().Print( Log() );
        Log << dolog;   
        
        if ( fTrigger )
                { delete fTrigger; fTrigger = 0x0; }
        if ( fRealTime )
                { delete fRealTime; fRealTime = 0x0; }
        if ( fCPUTime )
                { delete fCPUTime; fCPUTime = 0x0; }
//      printf("Delete FrameDispatcher::FrameDispatcher \n");   
}

Bool_t FrameDispatcher::SetTrigger(ADF::DFTrigger *trigger)
{
        if ( trigger && !trigger->IsZombie() ) {
                fTrigger = trigger;
                GetFrameIO().Register(fTrigger);
                GetFrameIO().SetStatus(BaseFrameIO::kIdle);
                
                return true;
        }
        return false;
}

void FrameDispatcher::DoCanvas(TCanvas *c, Option_t *o)
{
        TString option = o;
        
        if ( option == "" ) {
                c->Divide(2,2,0.001,0.001);
                c->cd(1); 
                fTriggerRate->Draw();   
                c->cd(2); 
                fBufferSize->Draw();
                c->cd(3); 
                fRealTime->At(0)->Draw();
                for (Int_t i = 1u; i < fRealTime->GetSize(); i++) {
                        if ( fRealTime->At(i) )
                                fRealTime->At(i)->Draw("same");
                }
                c->cd(4); 
                fCPUTime->At(0)->Draw();
                for (Int_t i = 1u; i < fCPUTime->GetSize(); i++) {
                        fCPUTime->At(i)->Draw("same");
                }
        }
}

void FrameDispatcher::SetDirectory(TDirectory *mother_dir_of_watcher, TDirectory *mother_dir_for_tag, Bool_t load_objects)
{       
        Watcher::SetDirectory(mother_dir_of_watcher,mother_dir_for_tag,load_objects);

        /*
        if ( MakeDir(fTopFolder,mother_dir_of_watcher,true) ) {
                
                Watcher::SetDirectory(mother_dir_of_watcher,mother_dir_for_tag,load_objects);
        }
         */
        
        
        return;
        
        
        // now change the directories of the watchers
        /*
        TIter loopontask(GetListOfTasks());
        Watcher *watcher;
        while ( ((watcher=(Watcher*)loopontask())) ) {
                TDirectory *w_dir = 0x0, *w_tag_dir= 0x0;
                if ( fDirectory ) {
                        TDirectory *tmp = fDirectory->GetDirectory(GetName());
                        if ( tmp == 0x0 ) {
                                tmp = fDirectory->mkdir(GetName(),GetTitle());
                        }
                        if ( tmp ) {
                                w_dir = tmp->mkdir(watcher->GetName(),watcher->GetTitle());     
                        }
                }
                if ( fTagDirectory ) {
                        TDirectory *tmp = fTagDirectory->GetDirectory(GetName());
                        if ( tmp == 0x0 ) {
                                tmp = fTagDirectory->mkdir(GetName(),GetTitle());
                        }
                        if ( tmp ) {
                                w_dir = tmp->mkdir(watcher->GetName(),watcher->GetTitle());     
                        }
                }       
                watcher->SetDirectory(w_dir, w_tag_dir, load_objects);
        }
        return;
         */
        /*
        TDirectory *c_dir = fDirectory, *c_tag_dir = fTagDirectory;
        
        // check and creates sub-directories with the name of the watcher 
        Watcher::SetDirectory(mother_dir_of_watcher,mother_dir_for_tag,load_objects);
        if ( c_dir == fDirectory && c_tag_dir == fTagDirectory)
                return;
        
        // now change the directories of the watchers
        TIter loopontask(GetListOfTasks());
        Watcher *watcher;
        while ( ((watcher=(Watcher*)loopontask())) ) {
                watcher->SetDirectory(fDirectory, fTagDirectory, load_objects);
        }
        
        // for clarity, objects and tagged objects owned by this are stored in a sub-directory FD
        // so creates it if required and move object

        if ( fDirectory != 0x0 ) { 
//      TDirectory *tmp = fDirectory->GetDirectory("FD");
//              if ( tmp == 0x0 ) {
//                      tmp = fDirectory->mkdir("FD","Histograms to control Watcher's executions"); */
//                      if ( tmp )
//                              fDirectory = tmp;
//                      else 
//                              fDirectory = 0x0; 
//              } 
/*              else fDirectory = tmp;  */

//              DirToDir(c_dir, fDirectory, GetPoolOfObject(), load_objects);
//              DirToDir(c_dir, tmp, GetPoolOfObject(), load_objects);

//      }
//      if ( fTagDirectory != 0x0 ) { 
                /*
                TDirectory *tmp = fTagDirectory->GetDirectory("FD");
                if ( tmp == 0x0 ) {
                        tmp = fTagDirectory->mkdir("FD","Histograms to control Watcher's executions"); */
/*                      if ( tmp )
                                fTagDirectory = tmp;
                        else 
                                fTagDirectory = 0x0; 
                }*/
/*              else fTagDirectory = tmp;       */      
//              DirToDir(c_dir, tmp, GetReference(), load_objects);
//              DirToDir(c_tag_dir, fTagDirectory, GetReference(), load_objects);
//      }
        // now change the directory of the objects in the pool
//      DirToDir(c_tag_dir, fTagDirectory, GetReference(), load_objects);


}

void FrameDispatcher::Add(TTask *task)
{
        if ( task == 0x0 || fTrigger == 0x0 )
                return;
        if ( fTrigger->IsZombie() )
                return;
        
        Watcher *watcher = 
                dynamic_cast<Watcher *>(task);
        if ( watcher ) 
                Add(watcher,fTrigger);
}

void FrameDispatcher::Add(Watcher *watch, ADF::DFTrigger *trig)
{
        if ( watch == 0x0 || trig == 0x0 )
                return;
        if ( trig->IsZombie() )
                return;
        
        // Add a new watcher, associates the trigger, add to the list of watchers, set the directory and allocate histograms
        TFolder *topfolder = 
                (TFolder *)gROOT->GetRootFolder()->FindObject("//root/GwWatchers");
        TFolder *w_folder = (TFolder *)gROOT->GetRootFolder()->FindObjectAny( Form("//root/GwWatchers/%s", watch->GetName() ) );
        if ( w_folder ) { // shoud aways be true ... 
                topfolder->Remove(w_folder);
                fTopFolder->Add(w_folder);
        }
        
        if ( fDirectory ) {
                Bool_t ok_dir = false;
                if ( fTagDirectory ) {
                        ok_dir = MakeDir(fTopFolder, fDirectory, true);
                        if ( fTagDirectory != fDirectory ) {
                                ok_dir = ok_dir && MakeDir(fTopFolder, fTagDirectory, true);
                        }               
                }
                if (ok_dir) {
                        watch->SetDirectory(fDirectory->GetDirectory(fTopFolder->GetName()), fTagDirectory->GetDirectory(fTopFolder->GetName()), true);
                }
        }

        watch->SetTrigger(trig);        
        TTask::Add(watch); 
        
        /*      
        // Add a new watcher, associates the trigger, add to the list of watchers, set the directory and allocate histograms

        if ( fDirectory ) {
                if ( fTagDirectory ) {
                        watch->SetDirectory(fDirectory, fTagDirectory);
                }
                else {
                        watch->SetDirectory(fDirectory, 0x0);
                }
        }
        else {
                if ( fTagDirectory ) {
                        watch->SetDirectory(0x0, fTagDirectory);                        
                }
                else {
                        watch->SetDirectory(0x0, 0x0);                  
                }               
        }
         */

        // now histograms for that watcher
        TH1F* h;
        h =  MakeTH1<TH1F>(Form("%s_RealTime_%lu",watch->GetName(),fCurrentValues.size()),"Real time [ms] to execute the task", kMaxCycle,0,kMaxCycle,"Performances");
        fRealTime->Add(h);
        h = MakeTH1<TH1F>(Form("%s_CPUTime_%lu",watch->GetName(),fCurrentValues.size()),"CPU time [ms] to execute the task", kMaxCycle,0,kMaxCycle,"Performances"); 
        fCPUTime->Add (h);
                
        std::pair<Double_t, Double_t> p; fCurrentValues.push_back(p);
        
        // no need to add it to the list of to be deleted triggers
        if ( trig == fTrigger ) 
                return;
                
        // try and find if this trigger is already known.
        // if not, add it and Register it in FrameIO
        Bool_t to_add = true;

        for ( UInt_t i = 0u; i < fListOfTriggers.GetSize(); i++ ) {
                if ( fListOfTriggers.At(i) == trig )
                        { to_add = false; break; }      
        }
        if ( to_add ) {
                // 
                GetFrameIO().Register(trig);
                GetFrameIO().SetStatus(BaseFrameIO::kIdle);
                        
                fListOfTriggers.Add(trig);      
        }
}

UInt_t FrameDispatcher::ProcessBlock (FrameBlock &block) 
{       
        TStopwatch loop_time; 
        
        if ( block.GetSize() == 0 ) {
                Log << warning << " Empty block, nothing to dispach " << nline;
                return 0u;
        }
        
        if ( gDoTimer ) {
                // init the value for that loop
                loop_time.Start(true);
                
                for (UInt_t i = 0u; i < fCurrentValues.size(); i++ ) {
                        fCurrentValues[i].first  = 0.0;
                        fCurrentValues[i].second = 0.0; 
                }
        }
        // attach the input/output buffer to the FrameIO system
        GetFrameIO().Attach(&block,0x0);
        
        if ( gDebug ) 
                printf("FrameDispatcher::ProcessBlock \n");

        // start the processing
        UInt_t error_code = 0u; UInt_t size_input_i, size_input_j;
        while ( GetFrameIO().Notify(false) ) { // the trigger has just fired 
                
                fTriggerRate->Fill(0);
        
                if ( gDebug ) 
                        Log << info << "Dispacher called " << GetName() << nline;
                
                // if this task is no more active, break the processing
                if ( ! IsActive() )
                        break;
                        
                // call Read for all InputFrames so that it it done just once
                if ( fTrigger->IsFired() ) {
                        size_input_i = fTrigger->GetNbInputFrame();
                        for (UInt_t i = 0u; i < size_input_i; i++) {
                                if ( fTrigger->IsIndividualFired(i) )
                                        fTrigger->GetInputFrame(i)->Read();
                        }
                }
                size_input_i = fListOfTriggers.GetSize();
                for (UInt_t i = 0; i < size_input_i ; i++ ) { 
                        if ( fListOfTriggers.At(i)->IsFired() ) {
                                size_input_j = fListOfTriggers.At(i)->GetNbInputFrame(); // +1  to Test try catch speed !
                                for (UInt_t j = 0u; j < size_input_j ; j++) 
                                        if ( fListOfTriggers.At(i)->IsIndividualFired(j) )
                                                fListOfTriggers.At(i)->GetInputFrame(j)->Read();
                        }
                }

                // All subtasks ready to be treated
                CleanTasks();
                
                TTask *task; TObjLink *lnk = GetListOfTasks()->FirstLink();
                UInt_t slot = 1u;
                //
                while (lnk) {
                        task = (TTask *)lnk->GetObject();
                        if ( !task->IsActive() ) {
                                lnk = lnk->Next();
                                continue;       
                        }
                        Watcher *watcher = 
                                dynamic_cast<Watcher *>(task);
                        if ( watcher ) {
                                // the trigger has fired... so tell the task should not be executed
                                if ( watcher->GetTrigger()->IsFired() ) {
                                        
                                        if ( gDoTimer > 0 ) {
                                                fTimeControl.Start(kTRUE);
                                        }
                                        watcher->Exec("");
                                        if ( gDoTimer > 0 ) {
                                                fTimeControl.Stop();    
                                                
                                                fCurrentValues[slot].first  += fTimeControl.RealTime();
                                                fCurrentValues[slot].second += fTimeControl.CpuTime();                                          
                                        }
                                        fTriggerRate->Fill(slot);
                                        
                                        watcher->Pass();
                                }
                                else { 
                                        watcher->Pass(); 
                                }
                                slot++;
                        }                       
                        lnk = lnk->Next();
                }               
                /*
                TIter loopontask(GetListOfTasks());
                TTask *task;
                UInt_t slot = 1u;
                while ( ((task=(TTask*)loopontask())) ) {
                        if ( !task->IsActive() )
                                continue;       
                        Watcher *watcher = 
                                dynamic_cast<Watcher *>(task);
                        if ( watcher ) {
                                // the trigger has fired... so tell the task should not be executed
                                if ( watcher->GetTrigger()->IsFired() ) {
                                        
                                        if ( gDoTimer > 0 ) {
                                                fTimeControl.Start(kTRUE);
                                        }
                                        watcher->Exec("");
                                        if ( gDoTimer > 0 ) {
                                                fTimeControl.Stop();    
                                        
                                                fCurrentValues[slot].first  += fTimeControl.RealTime();
                                                fCurrentValues[slot].second += fTimeControl.CpuTime();
                                        
                                                fTriggerRate->Fill(slot);
                                        }

                                        watcher->Pass();
                                }
                                else { watcher->Pass(); }
                                slot++;
                        }                       
                }
                 */
        }       
        
        GetFrameIO().Detach(&block,0x0);
        
        // to avoid being stack in an infinite loop
        if ( !gROOT->IsBatch() ) {
                gSystem->DispatchOneEvent(true);
                gSystem->ProcessEvents();
        }
        
        // now fill the different histograms
        fBufferSize->Fill(fWhichBin, block.GetSize()/Float_t(aMByte));

        
        if ( gDoTimer > 0 ) {
                TH1 *hr, *hc;
                for (UInt_t i = 0u; i < fCurrentValues.size(); i++ ) {
                        
                        if ( i == 0u ) 
                                continue;
                        
                        hr = (TH1 *)(fRealTime->At(i));
                        hc = (TH1 *)(fCPUTime->At(i));
                                
                        hr->Fill(fWhichBin, 1000*fCurrentValues[i].first  ); 
                        hc->Fill(fWhichBin, 1000*fCurrentValues[i].second  ); 
                }
                loop_time.Stop();

                hr = (TH1 *)(fRealTime->At(0));
                hc = (TH1 *)(fCPUTime->At(0));
                
                hr->Fill(fWhichBin, 1000*loop_time.RealTime() ); 
                hc->Fill(fWhichBin, 1000*loop_time.CpuTime() ); 
        }
        
        if ( fWhichBin == kMaxCycle )
                fWhichBin = 0u;
        else
                fWhichBin++;

        return error_code;
}

void FrameDispatcher::Zero(Option_t *opt1, Option_t *opt2)
{       
        // call Zero for all watchers 
        TIter next(GetListOfTasks());
        TTask *task; 
        //
        while ( (task = (TTask *)next()) ) {
                Watcher *w = dynamic_cast<Watcher *> (task);
                if ( w ) {              
                        w->Zero(opt1,opt2);
                }
        }       
        Watcher::Zero(opt1,opt2);
}

// void FrameDispatcher::ExecuteTasks(Option_t*/* option*/)
// {
//    // Execute all the subtasks of a task.
// 
//    TIter next(fTasks);
//    TTask *task;
//    while((task=(TTask*)next())) {
//       if (fgBreakPoint) return;
//       if (!task->IsActive()) continue;
//       if (task->fHasExecuted) {
//          task->ExecuteTasks(option);
//          continue;
//       }
//       if (task->fBreakin == 1) {
//          printf("Break at entry of task: %s\n",task->GetName());
//          fgBreakPoint = this;
//          task->fBreakin++;
//          return;
//       }
// 
//       if (gDebug > 1) {
//          TROOT::IndentLevel();
//          cout<<"Execute task:"<<task->GetName()<<" : "<<task->GetTitle()<<endl;
//          TROOT::IncreaseDirLevel();
//       }
//       task->Exec(option);
//       task->fHasExecuted = kTRUE;
//       task->ExecuteTasks(option);
//       if (gDebug > 1) TROOT::DecreaseDirLevel();
//       if (task->fBreakout == 1) {
//          printf("Break at exit of task: %s\n",task->GetName());
//          fgBreakPoint = this;
//          task->fBreakout++;
//          return;
//       }
//    }
// }