TPC/macros/testTPC/AliTPCjobs.cxx

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