STEER/TFndMonitorMan.cxx

// @(#)fROOT/STEER:$Name:  $:$Id: TFndMonitorMan.cxx,v 1.19 2007/09/24 07:32:41 Diego_Faso Exp $
// Author: Diego Faso <mailto:faso@to.infn.it>, 2006/06/01

//                         TFndMonitorMan                          //
//                                                                 //
// Implementation of all methods needed in order to control the    //
//  online monitoring process                                      //
//                                                                 //
// NOTE: all GUI methods are collected in the TFndMonitorGUI class //
//                                                                 //

#include "TROOT.h"
#include "TApplication.h"

#include "TFndRun.h"
#include "TThread.h"
#include "TFndMonitorMan.h"

ClassImp(TFndMonitorMan)

//_______________________________
TFndMonitorMan::TFndMonitorMan():
  fStartProdThread(),fCheckProdThread(),fProducerCommand()
{
  gROOT->Info("TFndMonitorMan::TFndMonitorMan","called");
  if(!fndrun) new TFndRun("MonitorGUI","Finuda");
}

//_______________________________
TFndMonitorMan::~TFndMonitorMan(){

  StopCheckRunningThread();

  if(fStartProdThread){
    if(fStartProdThread->GetState() == TThread::kRunningState){
      Printf("killing \"Start new producer thread\"");
      fStartProdThread->Kill();
      TThread::Delete(fStartProdThread);
      delete fStartProdThread;
    }
    else Printf("Start new producer thread found, but not running"); 
  }
  else Printf("Start new producer thread not found");
}

//______________________
void TFndMonitorMan::AutoRestartOnlineProd(Int_t mode){
  // mode value:
  //            0: clear and restart
  //            1: restart and recovery

  KillCurrentProducer();
  if(mode==0)  RemoveMapFiles();
  // ---
  if(fStartProdThread){
    if(fStartProdThread->GetState() == TThread::kRunningState){
      Printf("killing \"Start new producer thread\"");
      fStartProdThread->Kill();
      TThread::Delete(fStartProdThread);
      delete fStartProdThread;
    }
  }
  // ---
  fStartProdThread = new TThread("Monitor_producer",RunProdThread,(void *)this);
  
  gROOT->Info("SendProducerCommand","\"%s\"",fProducerCommand.Data());
  Printf("Sending command: \"%s\"",fProducerCommand.Data());
  fStartProdThread->Run();
  Printf("...done");
  
}

//______________________
void TFndMonitorMan::StopCheckRunningThread(){

  if(fCheckProdThread){
    if(fCheckProdThread->GetState() == TThread::kRunningState){
      Printf("killing \"Check producer thread\"");
      fCheckProdThread->Kill();
      TThread::Delete(fCheckProdThread);
      delete fCheckProdThread;
      fCheckProdThread = 0;
    }
    else Printf("Start new producer thread found, but not running"); 
  }
  else Printf("Start new producer thread not found");
  
}

//______________________
void* TFndMonitorMan::CheckRunningProdThread(void *arg){
  // the producer must always run in online mode.
  // This thread will check the geb2hdt process:
  //  - If online producer is terminated abnormally,
  //     it will restart automatically.
  //  - There is also limit to the maximum duration:
  //     too long processes will be restarted

  TThread::SetCancelOff();
  TThread::SetCancelDeferred();

  TDatime start_time;
  start_time.Set();
  UInt_t stim = start_time.Convert();

  UInt_t MaxDurationSec = 3600; // 1 hours
  //  UInt_t MaxDurationSec = 7200; // 2 hours
  //  UInt_t MaxDurationSec = 10800; // 3 hours
  //  UInt_t MaxDurationSec = 55; // test line

  while(1){
    TThread::CancelPoint();

    TDatime cur_time;
    cur_time.Set();
    UInt_t eltim = cur_time.Convert() - stim;
    if(eltim < 5){
      usleep(700000);
      continue;
    }    

    // --- check proces status
    enum pr_st{
      pr_st_unknown = 0,
      pr_st_running = 1,
      pr_st_stopped = 2,
      pr_st_broken = 3
    };
    pr_st pr_st_val = pr_st_unknown;

    Int_t gsys_res = gSystem->Exec("pgrep geb2hdt > /dev/null");

    Bool_t is_stopfile_present = kTRUE;
    if(gSystem->AccessPathName(".froot_onlstopped") ) is_stopfile_present = kFALSE;
    
    TString p_status = "unknown status";
    if(gsys_res == 0){
      pr_st_val = pr_st_running;
      p_status = "running";
    }
    else  if(gsys_res != 0 && is_stopfile_present){
      pr_st_val = pr_st_stopped;
      p_status = "stopped";
    }
    else if(gsys_res != 0 && !is_stopfile_present){
      pr_st_val = pr_st_broken;
      p_status = "broken";
    }
    
    // ---
    TString msg = "";
    msg.Form("*** CHECKING ONLINE PRODUCER: %s",p_status.Data());
    TString msg1 = "";
    msg1.Form(" (elapsed time: %u s)",eltim);
    if( pr_st_val != pr_st_stopped ) msg+=msg1;
    else{ // reset time evaluation if stopped
      start_time.Set();
      stim = start_time.Convert();            
    }
    
    if(pr_st_val != pr_st_running) Printf(msg);
    if(eltim > MaxDurationSec) Printf("*** Online Process too long: restarting it.");
    if(pr_st_val == pr_st_broken || eltim > MaxDurationSec){
      Printf("\n\n\n Restarting online producer (automatic check is running)");
      gSystem->Exec("date");
      ((TFndMonitorMan *)arg) -> AutoRestartOnlineProd(1);
      start_time.Set();
      stim = start_time.Convert();
    }
    //    
    usleep(10000000); // check timing
  }
}

//______________________
void* TFndMonitorMan::RunProdThread(void *arg){
  // used only to start the new producer within
  // a separate thread

  Printf("executing command...");
  gSystem->Exec(((TFndMonitorMan *)arg)->GetProducerCommand());
  return 0;
}


//______________________
void TFndMonitorMan::SendProducerCommand(TString appl,TString db_host,TString data_path,TString run_type,Int_t run_num,Int_t n_events){

  TString run_name = BuildRunName(run_type,run_num);

  fndrun->SetRunType(run_type);
  if(run_type.CompareTo("ONLM")) fndrun->SetRunNumber(run_num);
  else{
    fndrun->SetRunNumber(0);
    run_name = run_type;
  }
  
  //  (depracated macro) froot 'geb2hdt.C++("localhost","$RDT","FINU02181",100) ' > FINU02181.dat
  fProducerCommand.Form("%s %s %s %s %d %d > %s.dat" ,appl.Data(),
                        db_host.Data(),
                        data_path.Data(),
                        run_type.Data(),
                        run_num,
                        n_events,
                        run_name.Data());

  fStartProdThread = new TThread("Monitor_producer",RunProdThread,(void *)this);

  gROOT->Info("SendProducerCommand","\"%s\"",fProducerCommand.Data());
  Printf("Sending command: \"%s\"",fProducerCommand.Data());
  fStartProdThread->Run();
  Printf("...done");

  // --- check thread for producer running (online mode only!)
  if(run_type.CompareTo("ONLM") == 0){
    if(!fCheckProdThread) fCheckProdThread = new TThread("Monitor_controller",CheckRunningProdThread,(void *)this);
    if(fCheckProdThread->GetState() != TThread::kRunningState){
      Printf("Starting control for running producer");
      fCheckProdThread->Run();
      Printf("...done");
    }
  }
}

//______________________________________
// void TFndMonitorMan::SendProducerCommand(TString appl,TString db_host,TString data_path,TString run_name,Int_t n_events){
//   // the command line within the froot shell is:
//   // .x geb2hdt.C++("fnddiskfe.lnf.infn.it","$RDT","FINU02181",100)
//   //
//   // from bash shell the command is:
//   //  (depracated macro) froot 'geb2hdt.C++("localhost","$RDT","FINU02181",100) ' > FINU02181.dat
//   //  (depracated macro) geb2hdt localhost $RDT FINU02181 100 > FINU02181.dat
  
//   //#define GEB2HDT_MACRO // uncomment this line to use the (deprecated) geb2hdt.C macro
//   //  RemoveMapFiles();

//   // #ifdef GEB2HDT_MACRO
//   //   fProducerCommand.Form("froot \'%s/%s.C++(\"%s\",\"%s\",\"%s\",%d) \' > %s.dat",FROOT::ExpandPathName("$FROOTSYS/mcr_called").Data(),appl.Data(),db_host.Data(),data_path.Data(),run_name.Data(),n_events,run_name.Data());
//   // #else
//   fProducerCommand.Form("%s %s %s %s %d > %s.dat" ,appl.Data(),db_host.Data(),data_path.Data(),run_name.Data(),n_events,run_name.Data());
//   // #endif
  
//   fStartProdThread = new TThread("Monitor_producer",RunProdThread,(void *)this);

//   gROOT->Info("SendProducerCommand","\"%s\"",fProducerCommand.Data());
//   Printf("Sending command: \"%s\"",fProducerCommand.Data());
//   fStartProdThread->Run();
//   Printf("...done");

// }

//_________________________________________
void TFndMonitorMan::KillCurrentProducer(){
  // NOTE: not only current process, but all
  // active processes with the same name are killed

  Printf("Killing current process...");

  TString pid_kill = TString("kill -9 ");

  FILE *f = fopen(ExpandPathName(".froot_pid"),"r");
  Char_t file[200][200];
  if(!f){
    gROOT->Warning("TFndMonitorMan::KillCurrentProducer()","PID file not found");
    return;
  }

  TString cur_line = TString();
  int l=0;
  const Char_t *prod_pid = 0;
  while (fgets(file[l],200,f)) { 
    Char_t *line=file[l++];
    if(strlen(line)<2) continue;
    line[strlen(line)-1]='\0';
    cur_line = line;
    Int_t SpacePos = (Int_t) cur_line.Index(" ");
    if(!strncmp(cur_line.Data(),"prod:",SpacePos)){ 
      prod_pid = ((TSubString)(cur_line.SubString("",SpacePos+1))).Data(); // read content after space
    }
  }
  pid_kill += prod_pid;
  Printf(" ******* Killing current producer: sending command: \"%s\" *******",(Char_t *)(pid_kill.Data()));
  gSystem->Exec(pid_kill.Data());
  Printf("                        ******* current producer killed *******");
  fclose(f);
  //  delete f; // do not delete f

  //  gSystem->Exec("pkill \'geb2hdt fnddiskfe.lnf.infn.it $PWD ONLM\'");

}

//______________________________________________________________________________
void TFndMonitorMan::KillOtherProducers(){
  // all active geb2hdt in background are killed
  // they must have been executed in thgis way:
  // froot -q -b geb2hdt.C

  gSystem->Exec("pkill froot,geb2");

}

//______________________________________________________________________________
void TFndMonitorMan::KillYourself(){
  // To be completed

  //  HandleMenu(M_CLOSE_ALL); // close consumers (displays)
  cout << "Killing application" << endl;
  gApplication->Terminate(0);
}

//______________________________________________________________________________
void TFndMonitorMan::RemoveMapFiles(){
  // used shared memories (*.map) must be removed,
  // otherwise next time you run the online consumer
  // monitor you could get old histograms, since new shared
  // memories are not created until the socket connection is established

  //  gSystem->ProcessEvents();
  cout << "... removing used shared memories..." << endl;
  gSystem->Exec("rm -f $FND_SHR/*.map");
  gSystem->Exec("rm ./*.map");
  cout << "...done." << endl;
}

---