TPC/macros/testTPC/AliTPCjobs.cxx

   1 #include <fstream>
   2 #include <TFile.h>
   3 #include <TSystem.h>
   4
   5 /*
   6 .L $ALICE_ROOT/TPC/macros/testTPC/AliTPCjobs.cxx+
   7 AliTPCJobs jobs;
   8 jobs.fJobFile="job.list"
   9 jobs.ProcessAllJobs();
  10 */
  11 class AliTPCJobs : public TNamed{
  12 public:
  13   AliTPCJobs();
  14   void ProcessAllJobs();
  15   Bool_t GetNextJob();
  16   void ProcessJob(TString jobID, TString inputData, TString outputDir, TString   action);
  17
  18   void    SetLock(TString jobID);
  19   void    SetDone(TString jobID);
  20   Bool_t  IsLocked(TString jobID);
  21
  22   TString  fJobFile;
  23   TString  fWorkDir;
  24   ClassDef(AliTPCJobs,0)
  25 };
  26
  27  ClassImp(AliTPCJobs)
  28
  29 AliTPCJobs::AliTPCJobs(){
  30   //
  31   //
  32   //
  33   gSystem->Load("libXrdClient.so");
  34   gSystem->Load("libNetx.so");
  35 }
  36
  37 void AliTPCJobs::ProcessAllJobs(){
  38   //
  39   //
  40   //
  41   Int_t counter=0;
  42   while (GetNextJob()){
  43     //
  44     printf("PROCESSING JOB\t%d\n",counter);
  45     counter++;
  46     if (!GetNextJob()) break;
  47   }
  48 }
  49
  50
  51 Bool_t AliTPCJobs::GetNextJob(){
  52   ifstream in;
  53   in.open(fJobFile);
  54   TString id;
  55   TString inputData;
  56   TString outputDir;
  57   TString action;
  58   Bool_t hasJob=kFALSE;
  59   while(in.good()) {
  60     in>>id;
  61     in>>inputData;
  62     in>>outputDir;
  63     in>>action;
  64     if (!IsLocked(id)){
  65       hasJob=kTRUE;
  66       break;
  67     }
  68   }
  69   printf("Process %s\n",id.Data());
  70   if (hasJob) ProcessJob(id,inputData,outputDir, action);
  71   return hasJob;
  72 }
  73
  74
  75 void    AliTPCJobs::SetLock(TString jobID){
  76   printf("touch out/%s.lock\n",jobID.Data());
  77   gSystem->Exec(Form("touch out/%s.lock",jobID.Data()));
  78 }
  79
  80 void    AliTPCJobs::SetDone(TString jobID){
  81   printf("touch out/%s.done\n",jobID.Data());
  82   gSystem->Exec(Form("touch out/%s.done", jobID.Data()));
  83 }
  84
  85
  86 Bool_t    AliTPCJobs::IsLocked(TString jobID){
  87   TString path = "out/";
  88   path+=jobID;
  89   path+=".lock";
  90   Long_t pid; Long_t psize; Long_t pflags; Long_t pmodtime;
  91   Int_t status = gSystem->GetPathInfo(path,&pid,&psize,&pflags,&pmodtime);
  92   return (status==0);
  93 }
  94
  95
  96 void AliTPCJobs::ProcessJob(TString jobID, TString inputData, TString outputDir, TString   action){
  97   //
  98   //
  99   // 1. Create lock file
 100   // 2. Get Input data
 101   // 3. Process data
 102   // 4. Create Done file
 103   SetLock(jobID);
 104   if (action.Contains("COPY")){
 105     char command[10000];
 106       sprintf(command,"xrdcp -d 1 -DIFirstConnectMaxCnt 2 -DIConnectTimeout 2 -DIRequestTimeout 2 -DIMaxRedirectcount 2 -DIRedirCntTimeout 2 %s\t%s\n",inputData.Data(), outputDir.Data());
 107     printf("Exec\t%s\n", command);
 108     gSystem->Exec(command);
 109     //TFile::Cp(inputData.Data(), outputDir.Data());
 110   }else{
 111     char command[10000];
 112     sprintf(command,"$ALICE_ROOT/TPC/macros/testTPC/action.sh %s %s %s %s", jobID.Data(), inputData.Data(), outputDir.Data(), action.Data());
 113     printf("%s\n\n",command);
 114     gSystem->Exec(command);
 115     printf("\n\n");
 116
 117   }
 118
 119   SetDone(jobID);
 120 }