X-Git-Url: http://git.uio.no/git/?a=blobdiff_plain;f=ANALYSIS%2FAliAnalysisAlien.cxx;h=b0579ceda09b8bc26cddb3be5c3bd029853cc4e0;hb=832977ba26abc1d64ee0f82462b914057e74edda;hp=5fd9404aecb22ea9217a5e81188d284b90453a80;hpb=2ea554963a8f8f9d349760a136577d356c3b261c;p=u%2Fmrichter%2FAliRoot.git diff --git a/ANALYSIS/AliAnalysisAlien.cxx b/ANALYSIS/AliAnalysisAlien.cxx index 5fd9404aecb..b0579ceda09 100644 --- a/ANALYSIS/AliAnalysisAlien.cxx +++ b/ANALYSIS/AliAnalysisAlien.cxx @@ -20,8 +20,11 @@ // a personalized JDL, finding and creating a dataset. //============================================================================== +#include "AliAnalysisAlien.h" + #include "Riostream.h" #include "TEnv.h" +#include "TKey.h" #include "TBits.h" #include "TError.h" #include "TROOT.h" @@ -31,6 +34,7 @@ #include "TChain.h" #include "TObjString.h" #include "TObjArray.h" +#include "TMacro.h" #include "TGrid.h" #include "TGridResult.h" #include "TGridCollection.h" @@ -39,12 +43,33 @@ #include "TGridJobStatus.h" #include "TFileMerger.h" #include "AliAnalysisManager.h" +#include "AliAnalysisTaskCfg.h" #include "AliVEventHandler.h" #include "AliAnalysisDataContainer.h" -#include "AliAnalysisAlien.h" +#include "AliMultiInputEventHandler.h" +using std::ofstream; +using std::ifstream; +using std::ios; +using std::endl; ClassImp(AliAnalysisAlien) +#if 0 +; +#endif +namespace { + Bool_t copyLocal2Alien(const char* where, const char* loc, const char* rem) + { + TString sl(Form("file:%s", loc)); + TString sr(Form("alien://%s", rem)); + Bool_t ret = TFile::Cp(sl, sr); + if (!ret) { + Warning(where, "Failed to copy %s to %s", sl.Data(), sr.Data()); + } + return ret; + } +} + //______________________________________________________________________________ AliAnalysisAlien::AliAnalysisAlien() :AliAnalysisGrid(), @@ -58,6 +83,7 @@ AliAnalysisAlien::AliAnalysisAlien() fNtestFiles(0), fNrunsPerMaster(0), fMaxMergeFiles(0), + fMaxMergeStages(0), fNsubmitted(0), fProductionMode(0), fOutputToRunNo(0), @@ -95,6 +121,7 @@ AliAnalysisAlien::AliAnalysisAlien() fJDLName(), fTerminateFiles(), fMergeExcludes(), + fRegisterExcludes(), fIncludePath(), fCloseSE(), fFriendChainName(), @@ -104,10 +131,18 @@ AliAnalysisAlien::AliAnalysisAlien() fProofCluster(), fProofDataSet(), fFileForTestMode(), - fRootVersionForProof(), fAliRootMode(), + fProofProcessOpt(), + fMergeDirName(), fInputFiles(0), - fPackages(0) + fPackages(0), + fModules(0), + fProofParam(), + fDropToShell(true), + fGridJobIDs(""), + fGridStages(""), + fFriendLibs(""), + fTreeName() { // Dummy ctor. SetDefaults(); @@ -126,6 +161,7 @@ AliAnalysisAlien::AliAnalysisAlien(const char *name) fNtestFiles(0), fNrunsPerMaster(0), fMaxMergeFiles(0), + fMaxMergeStages(0), fNsubmitted(0), fProductionMode(0), fOutputToRunNo(0), @@ -163,6 +199,7 @@ AliAnalysisAlien::AliAnalysisAlien(const char *name) fJDLName(), fTerminateFiles(), fMergeExcludes(), + fRegisterExcludes(), fIncludePath(), fCloseSE(), fFriendChainName(), @@ -172,10 +209,18 @@ AliAnalysisAlien::AliAnalysisAlien(const char *name) fProofCluster(), fProofDataSet(), fFileForTestMode(), - fRootVersionForProof(), fAliRootMode(), + fProofProcessOpt(), + fMergeDirName(), fInputFiles(0), - fPackages(0) + fPackages(0), + fModules(0), + fProofParam(), + fDropToShell(true), + fGridJobIDs(""), + fGridStages(""), + fFriendLibs(""), + fTreeName() { // Default ctor. SetDefaults(); @@ -194,6 +239,7 @@ AliAnalysisAlien::AliAnalysisAlien(const AliAnalysisAlien& other) fNtestFiles(other.fNtestFiles), fNrunsPerMaster(other.fNrunsPerMaster), fMaxMergeFiles(other.fMaxMergeFiles), + fMaxMergeStages(other.fMaxMergeStages), fNsubmitted(other.fNsubmitted), fProductionMode(other.fProductionMode), fOutputToRunNo(other.fOutputToRunNo), @@ -231,6 +277,7 @@ AliAnalysisAlien::AliAnalysisAlien(const AliAnalysisAlien& other) fJDLName(other.fJDLName), fTerminateFiles(other.fTerminateFiles), fMergeExcludes(other.fMergeExcludes), + fRegisterExcludes(other.fRegisterExcludes), fIncludePath(other.fIncludePath), fCloseSE(other.fCloseSE), fFriendChainName(other.fFriendChainName), @@ -240,10 +287,18 @@ AliAnalysisAlien::AliAnalysisAlien(const AliAnalysisAlien& other) fProofCluster(other.fProofCluster), fProofDataSet(other.fProofDataSet), fFileForTestMode(other.fFileForTestMode), - fRootVersionForProof(other.fRootVersionForProof), fAliRootMode(other.fAliRootMode), + fProofProcessOpt(other.fProofProcessOpt), + fMergeDirName(other.fMergeDirName), fInputFiles(0), - fPackages(0) + fPackages(0), + fModules(0), + fProofParam(), + fDropToShell(other.fDropToShell), + fGridJobIDs(other.fGridJobIDs), + fGridStages(other.fGridStages), + fFriendLibs(other.fFriendLibs), + fTreeName(other.fTreeName) { // Copy ctor. fGridJDL = (TGridJDL*)gROOT->ProcessLine("new TAlienJDL()"); @@ -264,16 +319,28 @@ AliAnalysisAlien::AliAnalysisAlien(const AliAnalysisAlien& other) while ((obj=next())) fPackages->Add(new TObjString(obj->GetName())); fPackages->SetOwner(); } + if (other.fModules) { + fModules = new TObjArray(); + fModules->SetOwner(); + TIter next(other.fModules); + AliAnalysisTaskCfg *mod, *crt; + while ((crt=(AliAnalysisTaskCfg*)next())) { + mod = new AliAnalysisTaskCfg(*crt); + fModules->Add(mod); + } + } } //______________________________________________________________________________ AliAnalysisAlien::~AliAnalysisAlien() { // Destructor. - if (fGridJDL) delete fGridJDL; - if (fMergingJDL) delete fMergingJDL; - if (fInputFiles) delete fInputFiles; - if (fPackages) delete fPackages; + delete fGridJDL; + delete fMergingJDL; + delete fInputFiles; + delete fPackages; + delete fModules; + fProofParam.DeleteAll(); } //______________________________________________________________________________ @@ -292,6 +359,7 @@ AliAnalysisAlien &AliAnalysisAlien::operator=(const AliAnalysisAlien& other) fNtestFiles = other.fNtestFiles; fNrunsPerMaster = other.fNrunsPerMaster; fMaxMergeFiles = other.fMaxMergeFiles; + fMaxMergeStages = other.fMaxMergeStages; fNsubmitted = other.fNsubmitted; fProductionMode = other.fProductionMode; fOutputToRunNo = other.fOutputToRunNo; @@ -329,6 +397,7 @@ AliAnalysisAlien &AliAnalysisAlien::operator=(const AliAnalysisAlien& other) fJDLName = other.fJDLName; fTerminateFiles = other.fTerminateFiles; fMergeExcludes = other.fMergeExcludes; + fRegisterExcludes = other.fRegisterExcludes; fIncludePath = other.fIncludePath; fCloseSE = other.fCloseSE; fFriendChainName = other.fFriendChainName; @@ -338,8 +407,14 @@ AliAnalysisAlien &AliAnalysisAlien::operator=(const AliAnalysisAlien& other) fProofCluster = other.fProofCluster; fProofDataSet = other.fProofDataSet; fFileForTestMode = other.fFileForTestMode; - fRootVersionForProof = other.fRootVersionForProof; fAliRootMode = other.fAliRootMode; + fProofProcessOpt = other.fProofProcessOpt; + fMergeDirName = other.fMergeDirName; + fDropToShell = other.fDropToShell; + fGridJobIDs = other.fGridJobIDs; + fGridStages = other.fGridStages; + fFriendLibs = other.fFriendLibs; + fTreeName = other.fTreeName; if (other.fInputFiles) { fInputFiles = new TObjArray(); TIter next(other.fInputFiles); @@ -354,10 +429,349 @@ AliAnalysisAlien &AliAnalysisAlien::operator=(const AliAnalysisAlien& other) while ((obj=next())) fPackages->Add(new TObjString(obj->GetName())); fPackages->SetOwner(); } + if (other.fModules) { + fModules = new TObjArray(); + fModules->SetOwner(); + TIter next(other.fModules); + AliAnalysisTaskCfg *mod, *crt; + while ((crt=(AliAnalysisTaskCfg*)next())) { + mod = new AliAnalysisTaskCfg(*crt); + fModules->Add(mod); + } + } } return *this; } +//______________________________________________________________________________ +void AliAnalysisAlien::AddAdditionalLibrary(const char *name) +{ +// Add a single additional library to be loaded. Extension must be present. + TString lib(name); + if (!lib.Contains(".")) { + Error("AddAdditionalLibrary", "Extension not defined for %s", name); + return; + } + if (fAdditionalLibs.Contains(name)) { + Warning("AddAdditionalLibrary", "Library %s already added.", name); + return; + } + if (!fAdditionalLibs.IsNull()) fAdditionalLibs += " "; + fAdditionalLibs += lib; +} + +//______________________________________________________________________________ +void AliAnalysisAlien::AddModule(AliAnalysisTaskCfg *module) +{ +// Adding a module. Checks if already existing. Becomes owned by this. + if (!module) return; + if (GetModule(module->GetName())) { + Error("AddModule", "A module having the same name %s already added", module->GetName()); + return; + } + if (!fModules) { + fModules = new TObjArray(); + fModules->SetOwner(); + } + fModules->Add(module); +} + +//______________________________________________________________________________ +void AliAnalysisAlien::AddModules(TObjArray *list) +{ +// Adding a list of modules. Checks if already existing. Becomes owned by this. + TIter next(list); + AliAnalysisTaskCfg *module; + while ((module = (AliAnalysisTaskCfg*)next())) AddModule(module); +} + +//______________________________________________________________________________ +Bool_t AliAnalysisAlien::CheckDependencies() +{ +// Check if all dependencies are satisfied. Reorder modules if needed. + Int_t nmodules = GetNmodules(); + if (!nmodules) { + Warning("CheckDependencies", "No modules added yet to check their dependencies"); + return kTRUE; + } + AliAnalysisTaskCfg *mod = 0; + AliAnalysisTaskCfg *dep = 0; + TString depname; + Int_t i, j, k; + for (i=0; iAt(i); + Int_t ndeps = mod->GetNdeps(); + Int_t istart = i; + for (j=0; jGetDependency(j); + dep = GetModule(depname); + if (!dep) { + Error("CheckDependencies","Dependency %s not added for module %s", + depname.Data(), mod->GetName()); + return kFALSE; + } + if (dep->NeedsDependency(mod->GetName())) { + Error("CheckDependencies","Modules %s and %s circularly depend on each other", + mod->GetName(), dep->GetName()); + return kFALSE; + } + Int_t idep = fModules->IndexOf(dep); + // The dependency task must come first + if (idep>i) { + // Remove at idep and move all objects below up one slot + // down to index i included. + fModules->RemoveAt(idep); + for (k=idep-1; k>=i; k--) fModules->AddAt(fModules->RemoveAt(k),k+1); + fModules->AddAt(dep, i++); + } + //Redo from istart if dependencies were inserted + if (i>istart) i=istart-1; + } + } + return kTRUE; +} + +//______________________________________________________________________________ +AliAnalysisManager *AliAnalysisAlien::CreateAnalysisManager(const char *name, const char *filename) +{ +// Create the analysis manager and optionally execute the macro in filename. + AliAnalysisManager *mgr = AliAnalysisManager::GetAnalysisManager(); + if (mgr) return mgr; + mgr = new AliAnalysisManager(name); + mgr->SetGridHandler((AliAnalysisGrid*)this); + if (strlen(filename)) { + TString line = gSystem->ExpandPathName(filename); + line.Prepend(".x "); + gROOT->ProcessLine(line.Data()); + } + return mgr; +} + +//______________________________________________________________________________ +Int_t AliAnalysisAlien::GetNmodules() const +{ +// Get number of modules. + if (!fModules) return 0; + return fModules->GetEntries(); +} + +//______________________________________________________________________________ +AliAnalysisTaskCfg *AliAnalysisAlien::GetModule(const char *name) +{ +// Get a module by name. + if (!fModules) return 0; + return (AliAnalysisTaskCfg*)fModules->FindObject(name); +} + +//______________________________________________________________________________ +Bool_t AliAnalysisAlien::LoadModule(AliAnalysisTaskCfg *mod) +{ +// Load a given module. + if (mod->IsLoaded()) return kTRUE; + AliAnalysisManager *mgr = AliAnalysisManager::GetAnalysisManager(); + if (!mgr) { + Error("LoadModule", "No analysis manager created yet. Use CreateAnalysisManager first."); + return kFALSE; + } + Int_t ndeps = mod->GetNdeps(); + TString depname; + for (Int_t j=0; jGetDependency(j); + AliAnalysisTaskCfg *dep = GetModule(depname); + if (!dep) { + Error("LoadModule","Dependency %s not existing for module %s", + depname.Data(), mod->GetName()); + return kFALSE; + } + if (!LoadModule(dep)) { + Error("LoadModule","Dependency %s for module %s could not be loaded", + depname.Data(), mod->GetName()); + return kFALSE; + } + } + // Load libraries for the module + if (!mod->CheckLoadLibraries()) { + Error("LoadModule", "Cannot load all libraries for module %s", mod->GetName()); + return kFALSE; + } + // Check if a custom file name was requested + if (strlen(mod->GetOutputFileName())) mgr->SetCommonFileName(mod->GetOutputFileName()); + + // Check if a custom terminate file name was requested + if (strlen(mod->GetTerminateFileName())) { + if (!fTerminateFiles.IsNull()) fTerminateFiles += ","; + fTerminateFiles += mod->GetTerminateFileName(); + } + + // Execute the macro + if (mod->ExecuteMacro()<0) { + Error("LoadModule", "Executing the macro %s with arguments: %s for module %s returned a negative value", + mod->GetMacroName(), mod->GetMacroArgs(), mod->GetName()); + return kFALSE; + } + // Configure dependencies + if (mod->GetConfigMacro() && mod->ExecuteConfigMacro()<0) { + Error("LoadModule", "There was an error executing the deps config macro %s for module %s", + mod->GetConfigMacro()->GetTitle(), mod->GetName()); + return kFALSE; + } + // Adjust extra libraries + Int_t nlibs = mod->GetNlibs(); + TString lib; + for (Int_t i=0; iGetLibrary(i); + lib = Form("lib%s.so", lib.Data()); + if (fAdditionalLibs.Contains(lib)) continue; + if (!fAdditionalLibs.IsNull()) fAdditionalLibs += " "; + fAdditionalLibs += lib; + } + return kTRUE; +} + +//______________________________________________________________________________ +Bool_t AliAnalysisAlien::GenerateTrain(const char *name) +{ +// Generate the full train. + fAdditionalLibs = ""; + if (!LoadModules()) return kFALSE; + AliAnalysisManager *mgr = AliAnalysisManager::GetAnalysisManager(); + if (!mgr->InitAnalysis()) return kFALSE; + mgr->RunLocalInit(); + mgr->PrintStatus(); + Int_t productionMode = fProductionMode; + SetProductionMode(); + TString macro = fAnalysisMacro; + TString executable = fExecutable; + TString validation = fValidationScript; + TString execCommand = fExecutableCommand; + SetAnalysisMacro(Form("%s.C", name)); + SetExecutable(Form("%s.sh", name)); +// SetExecutableCommand("aliroot -b -q "); + SetValidationScript(Form("%s_validation.sh", name)); + StartAnalysis(); + SetProductionMode(productionMode); + fAnalysisMacro = macro; + fExecutable = executable; + fExecutableCommand = execCommand; + fValidationScript = validation; + return kTRUE; +} + +//______________________________________________________________________________ +Bool_t AliAnalysisAlien::GenerateTest(const char *name, const char *modname) +{ +// Generate test macros for a single module or for the full train. + fAdditionalLibs = ""; + if (strlen(modname)) { + if (!CheckDependencies()) return kFALSE; + AliAnalysisTaskCfg *mod = GetModule(modname); + if (!mod) { + Error("GenerateTest", "cannot generate test for inexistent module %s", modname); + return kFALSE; + } + if (!LoadModule(mod)) return kFALSE; + if (!LoadFriendLibs()) return kFALSE; + } else if (!LoadModules()) return kFALSE; + AliAnalysisManager *mgr = AliAnalysisManager::GetAnalysisManager(); + if (!mgr->InitAnalysis()) return kFALSE; + mgr->RunLocalInit(); + mgr->PrintStatus(); + SetLocalTest(kTRUE); + Int_t productionMode = fProductionMode; + SetProductionMode(); + TString macro = fAnalysisMacro; + TString executable = fExecutable; + TString validation = fValidationScript; + TString execCommand = fExecutableCommand; + SetAnalysisMacro(Form("%s.C", name)); + SetExecutable(Form("%s.sh", name)); + fOutputFiles = GetListOfFiles("outaod"); + // Add extra files registered to the analysis manager + TString extra = GetListOfFiles("ext"); + if (!extra.IsNull()) { + extra.ReplaceAll(".root", "*.root"); + if (!fOutputFiles.IsNull()) fOutputFiles += ","; + fOutputFiles += extra; + } +// SetExecutableCommand("aliroot -b -q "); + SetValidationScript(Form("%s_validation.sh", name)); + WriteAnalysisFile(); + WriteAnalysisMacro(); + WriteExecutable(); + WriteValidationScript(); + WriteMergingMacro(); + WriteMergeExecutable(); + WriteValidationScript(kTRUE); + SetLocalTest(kFALSE); + SetProductionMode(productionMode); + fAnalysisMacro = macro; + fExecutable = executable; + fExecutableCommand = execCommand; + fValidationScript = validation; + return kTRUE; +} + +//______________________________________________________________________________ +Bool_t AliAnalysisAlien::LoadModules() +{ +// Load all modules by executing the AddTask macros. Checks first the dependencies. + fAdditionalLibs = ""; + Int_t nmodules = GetNmodules(); + if (!nmodules) { + Warning("LoadModules", "No module to be loaded"); + return kTRUE; + } + AliAnalysisManager *mgr = AliAnalysisManager::GetAnalysisManager(); + if (!mgr) { + Error("LoadModules", "No analysis manager created yet. Use CreateAnalysisManager first."); + return kFALSE; + } + if (!CheckDependencies()) return kFALSE; + nmodules = GetNmodules(); + AliAnalysisTaskCfg *mod; + for (Int_t imod=0; imodAt(imod); + if (!LoadModule(mod)) return kFALSE; + } + // Load additional friend libraries + return LoadFriendLibs(); +} + +//______________________________________________________________________________ +Bool_t AliAnalysisAlien::LoadFriendLibs() const +{ +// Load libraries required for reading friends. + if (fFriendLibs.Length()) { + TObjArray *list = 0; + TString lib; + if (fFriendLibs.Contains(",")) list = fFriendLibs.Tokenize(","); + else list = fFriendLibs.Tokenize(" "); + for (Int_t ilib=0; ilibGetEntriesFast(); ilib++) { + lib = list->At(ilib)->GetName(); + lib.ReplaceAll(".so",""); + lib.ReplaceAll(" ",""); + if (lib.BeginsWith("lib")) lib.Remove(0, 3); + lib.Prepend("lib"); + Int_t loaded = strlen(gSystem->GetLibraries(lib,"",kFALSE)); + if (!loaded) loaded = gSystem->Load(lib); + if (loaded < 0) { + Error("LoadModules", "Cannot load library for friends %s", lib.Data()); + return kFALSE; + } + } + delete list; + } + return kTRUE; +} + +//______________________________________________________________________________ +void AliAnalysisAlien::SetRunPrefix(const char *prefix) +{ +// Set the run number format. Can be a prefix or a format like "%09d" + fRunPrefix = prefix; + if (!fRunPrefix.Contains("%")) fRunPrefix += "%d"; +} + //______________________________________________________________________________ void AliAnalysisAlien::AddIncludePath(const char *path) { @@ -372,15 +786,38 @@ void AliAnalysisAlien::AddRunNumber(Int_t run) { // Add a run number to the list of runs to be processed. if (fRunNumbers.Length()) fRunNumbers += " "; - fRunNumbers += Form("%s%d", fRunPrefix.Data(), run); + fRunNumbers += Form(fRunPrefix.Data(), run); } +//______________________________________________________________________________ +void AliAnalysisAlien::AddRunList(const char* runList) +{ +// Add several runs into the list of runs; they are expected to be separated by a blank character. + TString sList = runList; + TObjArray *list = sList.Tokenize(" "); + Int_t n = list->GetEntries(); + for (Int_t i = 0; i < n; i++) { + TObjString *os = (TObjString*)list->At(i); + AddRunNumber(os->GetString().Atoi()); + } + delete list; +} + //______________________________________________________________________________ void AliAnalysisAlien::AddRunNumber(const char* run) { // Add a run number to the list of runs to be processed. - if (fRunNumbers.Length()) fRunNumbers += " "; - fRunNumbers += run; + TString runs = run; + TObjString *os; + TObjArray *arr = runs.Tokenize(" "); + TIter next(arr); + TString prefix; + prefix.Append(fRunPrefix, fRunPrefix.Index("%d")); + while ((os=(TObjString*)next())){ + if (fRunNumbers.Length()) fRunNumbers += " "; + fRunNumbers += Form("%s%s", prefix.Data(), os->GetString().Data()); + } + delete arr; } //______________________________________________________________________________ @@ -450,6 +887,11 @@ Bool_t AliAnalysisAlien::CheckFileCopy(const char *alienpath) { // Check if file copying is possible. if (fProductionMode) return kTRUE; + TString salienpath(alienpath); + if (salienpath.Contains(" ")) { + Error("CheckFileCopy", "path: <%s> contains blancs - FIX IT !",alienpath); + return kFALSE; + } if (!Connect()) { Error("CheckFileCopy", "Not connected to AliEn. File copying cannot be tested."); return kFALSE; @@ -469,7 +911,8 @@ Bool_t AliAnalysisAlien::CheckFileCopy(const char *alienpath) Error("CheckFileCopy", "Alien path %s does not seem to exist", alienpath); return kFALSE; } - TFile f("plugin_test_copy", "RECREATE"); + TString stest = "plugin_test_copy"; + TFile f(stest, "RECREATE"); // User may not have write permissions to current directory if (f.IsZombie()) { Error("CheckFileCopy", "Cannot create local test file. Do you have write access to current directory: <%s> ?", @@ -477,19 +920,19 @@ Bool_t AliAnalysisAlien::CheckFileCopy(const char *alienpath) return kFALSE; } f.Close(); - if (FileExists(Form("alien://%s/%s",alienpath, f.GetName()))) gGrid->Rm(Form("alien://%s/%s",alienpath, f.GetName())); - if (!TFile::Cp(f.GetName(), Form("alien://%s/%s",alienpath, f.GetName()))) { + if (FileExists(Form("alien://%s/%s",alienpath, stest.Data()))) gGrid->Rm(Form("alien://%s/%s",alienpath, stest.Data())); + if (!TFile::Cp(stest.Data(), Form("alien://%s/%s",alienpath, stest.Data()))) { Error("CheckFileCopy", "Cannot copy files to Alien destination: <%s> This may be temporary, or: \ \n# 1. Make sure you have write permissions there. If this is the case: \ \n# 2. Check the storage availability at: http://alimonitor.cern.ch/stats?page=SE/table \ \n# Do: export alien_CLOSE_SE=\"working_disk_SE\" \ \n# To make this permanent put in in your .bashrc (in .alienshrc is not enough) \ \n# Redo token: rm /tmp/x509up_u$UID then: alien-token-init ", alienpath); - gSystem->Unlink(f.GetName()); + gSystem->Unlink(stest.Data()); return kFALSE; } - gSystem->Unlink(f.GetName()); - gGrid->Rm(Form("%s%s",alienpath,f.GetName())); + gSystem->Unlink(stest.Data()); + gGrid->Rm(Form("%s/%s",alienpath,stest.Data())); Info("CheckFileCopy", "### ...SUCCESS ###"); return kTRUE; } @@ -504,6 +947,10 @@ Bool_t AliAnalysisAlien::CheckInputData() Error("CkeckInputData", "AliEn path to base data directory must be set.\n = Use: SetGridDataDir()"); return kFALSE; } + if (fMergeViaJDL) { + Error("CheckInputData", "Merging via jdl works only with run numbers, run range or provided xml"); + return kFALSE; + } Info("CheckInputData", "Analysis will make a single xml for base data directory %s",fGridDataDir.Data()); if (fDataPattern.Contains("tag") && TestBit(AliAnalysisGrid::kTest)) TObject::SetBit(AliAnalysisGrid::kUseTags, kTRUE); // ADDED (fix problem in determining the tag usage in test mode) @@ -574,6 +1021,7 @@ Bool_t AliAnalysisAlien::CheckInputData() // Check validity of run number(s) TObjArray *arr; TObjString *os; + TString format; Int_t nruns = 0; TString schunk, schunk2; TString path; @@ -619,12 +1067,13 @@ Bool_t AliAnalysisAlien::CheckInputData() } else { Info("CheckDataType", "Using run range [%d, %d]", fRunRange[0], fRunRange[1]); for (Int_t irun=fRunRange[0]; irun<=fRunRange[1]; irun++) { - path = Form("%s/%s%d ", fGridDataDir.Data(), fRunPrefix.Data(), irun); + format = Form("%%s/%s ", fRunPrefix.Data()); + path = Form(format.Data(), fGridDataDir.Data(), irun); if (!DirectoryExists(path)) { -// Warning("CheckInputData", "Run number %d not found in path: <%s>", irun, path.Data()); continue; } - path = Form("%s/%s%d.xml", workdir.Data(),fRunPrefix.Data(),irun); + format = Form("%%s/%s.xml", fRunPrefix.Data()); + path = Form(format.Data(), workdir.Data(),irun); TString msg = "\n##### file: "; msg += path; msg += " type: xml_collection;"; @@ -632,13 +1081,15 @@ Bool_t AliAnalysisAlien::CheckInputData() else msg += " using_tags: No"; Info("CheckDataType", "%s", msg.Data()); if (fNrunsPerMaster<2) { - AddDataFile(Form("%s%d.xml",fRunPrefix.Data(),irun)); + format = Form("%s.xml", fRunPrefix.Data()); + AddDataFile(Form(format.Data(),irun)); } else { nruns++; if (((nruns-1)%fNrunsPerMaster) == 0) { - schunk = Form("%s%d", fRunPrefix.Data(),irun); + schunk = Form(fRunPrefix.Data(),irun); } - schunk2 = Form("_%s%d.xml", fRunPrefix.Data(), irun); + format = Form("_%s.xml", fRunPrefix.Data()); + schunk2 = Form(format.Data(), irun); if ((nruns%fNrunsPerMaster)!=0 && irun != fRunRange[1]) continue; schunk += schunk2; AddDataFile(schunk); @@ -652,10 +1103,95 @@ Bool_t AliAnalysisAlien::CheckInputData() return kTRUE; } +//______________________________________________________________________________ +Int_t AliAnalysisAlien::CopyLocalDataset(const char *griddir, const char *pattern, Int_t nfiles, const char *output, const char *archivefile, const char *outputdir) +{ +// Copy data from the given grid directory according a pattern and make a local +// dataset. +// archivefile (optional) results in that the archive containing the file is copied. archivefile can contain a list of files (semicolon-separated) which are all copied + if (!Connect()) { + Error("CopyLocalDataset", "Cannot copy local dataset with no grid connection"); + return 0; + } + if (!DirectoryExists(griddir)) { + Error("CopyLocalDataset", "Data directory %s not existing.", griddir); + return 0; + } + TString command = Form("find -z -l %d %s %s", nfiles, griddir, pattern); + printf("Running command: %s\n", command.Data()); + TGridResult *res = gGrid->Command(command); + Int_t nfound = res->GetEntries(); + if (!nfound) { + Error("CopyLocalDataset", "No file found in <%s> having pattern <%s>", griddir, pattern); + return 0; + } + printf("... found %d files. Copying locally ...\n", nfound); + + // archives + TObjArray* additionalArchives = 0; + if (strlen(archivefile) > 0 && TString(archivefile).Contains(";")) { + additionalArchives = TString(archivefile).Tokenize(";"); + archivefile = additionalArchives->At(0)->GetName(); + additionalArchives->RemoveAt(0); + additionalArchives->Compress(); + } + + // Copy files locally + ofstream out; + out.open(output, ios::out); + TMap *map; + TString turl, dirname, filename, temp; + TString cdir = gSystem->WorkingDirectory(); + gSystem->MakeDirectory(outputdir); + gSystem->ChangeDirectory(outputdir); + Int_t ncopied = 0; + for (Int_t i=0; iAt(i); + turl = map->GetValue("turl")->GetName(); + filename = gSystem->BaseName(turl.Data()); + dirname = gSystem->DirName(turl.Data()); + dirname = gSystem->BaseName(dirname.Data()); + gSystem->MakeDirectory(dirname); + + TString source(turl); + TString targetFileName(filename); + + if (strlen(archivefile) > 0) { +// TODO here the archive in which the file resides should be determined +// however whereis returns only a guid, and guid2lfn does not work +// Therefore we use the one provided as argument for now + source = Form("%s/%s", gSystem->DirName(source.Data()), archivefile); + targetFileName = archivefile; + } + if (TFile::Cp(source, Form("file:./%s/%s", dirname.Data(), targetFileName.Data()))) { + Bool_t success = kTRUE; + if (additionalArchives) { + for (Int_t j=0; jGetEntriesFast(); j++) { + TString target; + target.Form("./%s/%s", dirname.Data(), additionalArchives->At(j)->GetName()); + gSystem->MakeDirectory(gSystem->DirName(target)); + success &= TFile::Cp(Form("%s/%s", gSystem->DirName(source.Data()), additionalArchives->At(j)->GetName()), Form("file:%s", target.Data())); + } + } + + if (success) { + if (strlen(archivefile) > 0) targetFileName = Form("%s#%s", targetFileName.Data(), gSystem->BaseName(turl.Data())); + out << cdir << Form("/%s/%s/%s", outputdir, dirname.Data(), targetFileName.Data()) << endl; + ncopied++; + } + } + } + gSystem->ChangeDirectory(cdir); + delete res; + delete additionalArchives; + return ncopied; +} + //______________________________________________________________________________ Bool_t AliAnalysisAlien::CreateDataset(const char *pattern) { // Create dataset for the grid data directory + run number. + const Int_t gMaxEntries = 15000; if (fProductionMode || TestBit(AliAnalysisGrid::kOffline)) return kTRUE; if (!Connect()) { Error("CreateDataset", "Cannot create dataset with no grid connection"); @@ -668,11 +1204,19 @@ Bool_t AliAnalysisAlien::CreateDataset(const char *pattern) workdir += fGridWorkingDir; // Compose the 'find' command arguments + TString format; TString command; + TString delimiter = pattern; + delimiter.Strip(); + if (delimiter.Contains(" ")) delimiter = ""; + else delimiter = " "; TString options = "-x collection "; if (TestBit(AliAnalysisGrid::kTest)) options += Form("-l %d ", fNtestFiles); + else options += Form("-l %d ", gMaxEntries); // Protection for the find command TString conditions = ""; - + Int_t nstart = 0; + Int_t ncount = 0; + Int_t stage = 0; TString file; TString path; Int_t nruns = 0; @@ -685,33 +1229,69 @@ Bool_t AliAnalysisAlien::CreateDataset(const char *pattern) if (!DirectoryExists(path)) { Error("CreateDataset", "Path to data directory %s not valid",fGridDataDir.Data()); return kFALSE; - } + } // CdWork(); if (TestBit(AliAnalysisGrid::kTest)) file = "wn.xml"; else file = Form("%s.xml", gSystem->BaseName(path)); - if (gSystem->AccessPathName(file) || TestBit(AliAnalysisGrid::kTest) || fOverwriteMode) { - command = "find "; - command += options; - command += path; - command += " "; - command += pattern; - command += conditions; - printf("command: %s\n", command.Data()); - TGridResult *res = gGrid->Command(command); - if (res) delete res; - // Write standard output to file - gROOT->ProcessLine(Form("gGrid->Stdout(); > %s", file.Data())); - Bool_t hasGrep = (gSystem->Exec("grep --version 2>/dev/null > /dev/null")==0)?kTRUE:kFALSE; - Bool_t nullFile = kFALSE; - if (!hasGrep) { - Warning("CreateDataset", "'grep' command not available on this system - cannot validate the result of the grid 'find' command"); - } else { - nullFile = (gSystem->Exec(Form("grep /event %s 2>/dev/null > /dev/null",file.Data()))==0)?kFALSE:kTRUE; - if (nullFile) { - Error("CreateDataset","Dataset %s produced by the previous find command is empty !", file.Data()); - return kFALSE; + while (1) { + ncount = 0; + stage++; + if (gSystem->AccessPathName(file) || TestBit(AliAnalysisGrid::kTest) || fOverwriteMode) { + command = "find "; + command += Form("%s -o %d ",options.Data(), nstart); + command += path; + command += delimiter; + command += pattern; + command += conditions; + printf("command: %s\n", command.Data()); + TGridResult *res = gGrid->Command(command); + if (res) delete res; + // Write standard output to file + gROOT->ProcessLine(Form("gGrid->Stdout(); > __tmp%d__%s", stage, file.Data())); + Bool_t hasGrep = (gSystem->Exec("grep --version 2>/dev/null > /dev/null")==0)?kTRUE:kFALSE; + Bool_t nullFile = kFALSE; + if (!hasGrep) { + Warning("CreateDataset", "'grep' command not available on this system - cannot validate the result of the grid 'find' command"); + } else { + nullFile = (gSystem->Exec(Form("grep -c /event __tmp%d__%s 2>/dev/null > __tmp__",stage,file.Data()))==0)?kFALSE:kTRUE; + if (nullFile) { + Error("CreateDataset","Dataset %s produced by the previous find command is empty !", file.Data()); + gSystem->Exec("rm -f __tmp*"); + return kFALSE; + } + TString line; + ifstream in; + in.open("__tmp__"); + in >> line; + in.close(); + gSystem->Exec("rm -f __tmp__"); + ncount = line.Atoi(); + } + } + if (ncount == gMaxEntries) { + Info("CreateDataset", "Dataset %s has more than 15K entries. Trying to merge...", file.Data()); + cadd = (TGridCollection*)gROOT->ProcessLine(Form("new TAlienCollection(\"__tmp%d__%s\", 1000000);",stage,file.Data())); + if (!cbase) cbase = cadd; + else { + cbase->Add(cadd); + delete cadd; } - } + nstart += ncount; + } else { + if (cbase) { + cadd = (TGridCollection*)gROOT->ProcessLine(Form("new TAlienCollection(\"__tmp%d__%s\", 1000000);",stage,file.Data())); + printf("... please wait - TAlienCollection::Add() scales badly...\n"); + cbase->Add(cadd); + delete cadd; + cbase->ExportXML(Form("file://%s", file.Data()),kFALSE,kFALSE, file, "Merged entries for a run"); + delete cbase; cbase = 0; + } else { + TFile::Cp(Form("__tmp%d__%s",stage, file.Data()), file.Data()); + } + gSystem->Exec("rm -f __tmp*"); + Info("CreateDataset", "Created dataset %s with %d files", file.Data(), nstart+ncount); + break; + } } Bool_t fileExists = FileExists(file); if (!TestBit(AliAnalysisGrid::kTest) && (!fileExists || fOverwriteMode)) { @@ -734,36 +1314,80 @@ Bool_t AliAnalysisAlien::CreateDataset(const char *pattern) TObjString *os; TIter next(arr); while ((os=(TObjString*)next())) { - path = Form("%s/%s ", fGridDataDir.Data(), os->GetString().Data()); + nstart = 0; + stage = 0; + path = Form("%s/%s/", fGridDataDir.Data(), os->GetString().Data()); if (!DirectoryExists(path)) continue; // CdWork(); if (TestBit(AliAnalysisGrid::kTest)) file = "wn.xml"; else file = Form("%s.xml", os->GetString().Data()); // If local collection file does not exist, create it via 'find' command. - if (gSystem->AccessPathName(file) || TestBit(AliAnalysisGrid::kTest) || fOverwriteMode) { - command = "find "; - command += options; - command += path; - command += pattern; - command += conditions; - TGridResult *res = gGrid->Command(command); - if (res) delete res; - // Write standard output to file - gROOT->ProcessLine(Form("gGrid->Stdout(); > %s", file.Data())); - Bool_t hasGrep = (gSystem->Exec("grep --version 2>/dev/null > /dev/null")==0)?kTRUE:kFALSE; - Bool_t nullFile = kFALSE; - if (!hasGrep) { - Warning("CreateDataset", "'grep' command not available on this system - cannot validate the result of the grid 'find' command"); - } else { - nullFile = (gSystem->Exec(Form("grep /event %s 2>/dev/null > /dev/null",file.Data()))==0)?kFALSE:kTRUE; - if (nullFile) { - Warning("CreateDataset","Dataset %s produced by: <%s> is empty !", file.Data(), command.Data()); - fRunNumbers.ReplaceAll(os->GetString().Data(), ""); - continue; + while (1) { + ncount = 0; + stage++; + if (gSystem->AccessPathName(file) || TestBit(AliAnalysisGrid::kTest) || fOverwriteMode) { + command = "find "; + command += Form("%s -o %d ",options.Data(), nstart); + command += path; + command += delimiter; + command += pattern; + command += conditions; + TGridResult *res = gGrid->Command(command); + if (res) delete res; + // Write standard output to file + gROOT->ProcessLine(Form("gGrid->Stdout(); > __tmp%d__%s", stage,file.Data())); + Bool_t hasGrep = (gSystem->Exec("grep --version 2>/dev/null > /dev/null")==0)?kTRUE:kFALSE; + Bool_t nullFile = kFALSE; + if (!hasGrep) { + Warning("CreateDataset", "'grep' command not available on this system - cannot validate the result of the grid 'find' command"); + } else { + nullFile = (gSystem->Exec(Form("grep -c /event __tmp%d__%s 2>/dev/null > __tmp__",stage,file.Data()))==0)?kFALSE:kTRUE; + if (nullFile) { + Warning("CreateDataset","Dataset %s produced by: <%s> is empty !", file.Data(), command.Data()); + gSystem->Exec("rm -f __tmp*"); + fRunNumbers.ReplaceAll(os->GetString().Data(), ""); + break; + } + TString line; + ifstream in; + in.open("__tmp__"); + in >> line; + in.close(); + gSystem->Exec("rm -f __tmp__"); + ncount = line.Atoi(); + } + nullResult = kFALSE; + } + if (ncount == gMaxEntries) { + Info("CreateDataset", "Dataset %s has more than 15K entries. Trying to merge...", file.Data()); + if (fNrunsPerMaster > 1) { + Error("CreateDataset", "File %s has more than %d entries. Please set the number of runs per master to 1 !", + file.Data(),gMaxEntries); + return kFALSE; + } + cadd = (TGridCollection*)gROOT->ProcessLine(Form("new TAlienCollection(\"__tmp%d__%s\", 1000000);",stage,file.Data())); + if (!cbase) cbase = cadd; + else { + cbase->Add(cadd); + delete cadd; } + nstart += ncount; + } else { + if (cbase && fNrunsPerMaster<2) { + cadd = (TGridCollection*)gROOT->ProcessLine(Form("new TAlienCollection(\"__tmp%d__%s\", 1000000);",stage,file.Data())); + printf("... please wait - TAlienCollection::Add() scales badly...\n"); + cbase->Add(cadd); + delete cadd; + cbase->ExportXML(Form("file://%s", file.Data()),kFALSE,kFALSE, file, "Merged entries for a run"); + delete cbase; cbase = 0; + } else { + TFile::Cp(Form("__tmp%d__%s",stage, file.Data()), file.Data()); + } + gSystem->Exec("rm -f __tmp*"); + Info("CreateDataset", "Created dataset %s with %d files", file.Data(), nstart+ncount); + break; } - nullResult = kFALSE; - } + } if (TestBit(AliAnalysisGrid::kTest)) break; // Check if there is one run per master job. if (fNrunsPerMaster<2) { @@ -821,11 +1445,15 @@ Bool_t AliAnalysisAlien::CreateDataset(const char *pattern) } else { // Process a full run range. for (Int_t irun=fRunRange[0]; irun<=fRunRange[1]; irun++) { - path = Form("%s/%s%d ", fGridDataDir.Data(), fRunPrefix.Data(), irun); + format = Form("%%s/%s/", fRunPrefix.Data()); + nstart = 0; + stage = 0; + path = Form(format.Data(), fGridDataDir.Data(), irun); if (!DirectoryExists(path)) continue; // CdWork(); + format = Form("%s.xml", fRunPrefix.Data()); if (TestBit(AliAnalysisGrid::kTest)) file = "wn.xml"; - else file = Form("%s%d.xml", fRunPrefix.Data(), irun); + else file = Form(format.Data(), irun); if (FileExists(file) && fNrunsPerMaster<2 && !TestBit(AliAnalysisGrid::kTest)) { if (fOverwriteMode) gGrid->Rm(file); else { @@ -834,28 +1462,69 @@ Bool_t AliAnalysisAlien::CreateDataset(const char *pattern) } } // If local collection file does not exist, create it via 'find' command. - if (gSystem->AccessPathName(file) || TestBit(AliAnalysisGrid::kTest) || fOverwriteMode) { - command = "find "; - command += options; - command += path; - command += pattern; - command += conditions; - TGridResult *res = gGrid->Command(command); - if (res) delete res; - // Write standard output to file - gROOT->ProcessLine(Form("gGrid->Stdout(); > %s", file.Data())); - Bool_t hasGrep = (gSystem->Exec("grep --version 2>/dev/null > /dev/null")==0)?kTRUE:kFALSE; - Bool_t nullFile = kFALSE; - if (!hasGrep) { - Warning("CreateDataset", "'grep' command not available on this system - cannot validate the result of the grid 'find' command"); - } else { - nullFile = (gSystem->Exec(Form("grep /event %s 2>/dev/null > /dev/null",file.Data()))==0)?kFALSE:kTRUE; - if (nullFile) { - Warning("CreateDataset","Dataset %s produced by: <%s> is empty !", file.Data(), command.Data()); - continue; + while (1) { + ncount = 0; + stage++; + if (gSystem->AccessPathName(file) || TestBit(AliAnalysisGrid::kTest) || fOverwriteMode) { + command = "find "; + command += Form("%s -o %d ",options.Data(), nstart); + command += path; + command += delimiter; + command += pattern; + command += conditions; + TGridResult *res = gGrid->Command(command); + if (res) delete res; + // Write standard output to file + gROOT->ProcessLine(Form("gGrid->Stdout(); > __tmp%d__%s", stage,file.Data())); + Bool_t hasGrep = (gSystem->Exec("grep --version 2>/dev/null > /dev/null")==0)?kTRUE:kFALSE; + Bool_t nullFile = kFALSE; + if (!hasGrep) { + Warning("CreateDataset", "'grep' command not available on this system - cannot validate the result of the grid 'find' command"); + } else { + nullFile = (gSystem->Exec(Form("grep -c /event __tmp%d__%s 2>/dev/null > __tmp__",stage,file.Data()))==0)?kFALSE:kTRUE; + if (nullFile) { + Warning("CreateDataset","Dataset %s produced by: <%s> is empty !", file.Data(), command.Data()); + gSystem->Exec("rm -f __tmp*"); + break; + } + TString line; + ifstream in; + in.open("__tmp__"); + in >> line; + in.close(); + gSystem->Exec("rm -f __tmp__"); + ncount = line.Atoi(); + } + nullResult = kFALSE; + } + if (ncount == gMaxEntries) { + Info("CreateDataset", "Dataset %s has more than 15K entries. Trying to merge...", file.Data()); + if (fNrunsPerMaster > 1) { + Error("CreateDataset", "File %s has more than %d entries. Please set the number of runs per master to 1 !", + file.Data(),gMaxEntries); + return kFALSE; + } + cadd = (TGridCollection*)gROOT->ProcessLine(Form("new TAlienCollection(\"__tmp%d__%s\", 1000000);",stage,file.Data())); + if (!cbase) cbase = cadd; + else { + cbase->Add(cadd); + delete cadd; } + nstart += ncount; + } else { + if (cbase && fNrunsPerMaster<2) { + cadd = (TGridCollection*)gROOT->ProcessLine(Form("new TAlienCollection(\"__tmp%d__%s\", 1000000);",stage,file.Data())); + printf("... please wait - TAlienCollection::Add() scales badly...\n"); + cbase->Add(cadd); + delete cadd; + cbase->ExportXML(Form("file://%s", file.Data()),kFALSE,kFALSE, file, "Merged entries for a run"); + delete cbase; cbase = 0; + } else { + TFile::Cp(Form("__tmp%d__%s",stage, file.Data()), file.Data()); + } + Info("CreateDataset", "Created dataset %s with %d files", file.Data(), nstart+ncount); + break; } - nullResult = kFALSE; } if (TestBit(AliAnalysisGrid::kTest)) break; // Check if there is one run per master job. @@ -883,14 +1552,15 @@ Bool_t AliAnalysisAlien::CreateDataset(const char *pattern) } printf(" Merging collection <%s> into %d runs chunk...\n",file.Data(),fNrunsPerMaster); if (((nruns-1)%fNrunsPerMaster) == 0) { - schunk = Form("%s%d", fRunPrefix.Data(), irun); + schunk = Form(fRunPrefix.Data(), irun); cbase = (TGridCollection*)gROOT->ProcessLine(Form("new TAlienCollection(\"%s\", 1000000);",file.Data())); } else { cadd = (TGridCollection*)gROOT->ProcessLine(Form("new TAlienCollection(\"%s\", 1000000);",file.Data())); cbase->Add(cadd); delete cadd; } - schunk2 = Form("%s_%s%d.xml", schunk.Data(), fRunPrefix.Data(), irun); + format = Form("%%s_%s.xml", fRunPrefix.Data()); + schunk2 = Form(format.Data(), schunk.Data(), irun); if ((nruns%fNrunsPerMaster)!=0 && irun!=fRunRange[1] && schunk2 != fInputFiles->Last()->GetName()) { continue; } @@ -931,6 +1601,7 @@ Bool_t AliAnalysisAlien::CreateJDL() { // Generate a JDL file according to current settings. The name of the file is // specified by fJDLName. + AliAnalysisManager *mgr = AliAnalysisManager::GetAnalysisManager(); Bool_t error = kFALSE; TObjArray *arr = 0; Bool_t copy = kTRUE; @@ -972,6 +1643,8 @@ Bool_t AliAnalysisAlien::CreateJDL() Error("CreateJDL", "Could not create alien output directory %s", fGridOutputDir.Data()); // error = kTRUE; } + } else { + Warning("CreateJDL", "#### Output directory %s exists! If this contains old data, jobs will fail with ERROR_SV !!! ###", fGridOutputDir.Data()); } gGrid->Cd(workdir); } @@ -992,7 +1665,11 @@ Bool_t AliAnalysisAlien::CreateJDL() if (!fArguments.IsNull()) fGridJDL->SetArguments(fArguments, "Arguments for the executable command"); if (IsOneStageMerging()) fMergingJDL->SetArguments(fGridOutputDir); - else fMergingJDL->SetArguments("$1 $2 $3"); + else { + if (fProductionMode) fMergingJDL->SetArguments("wn.xml $4"); // xml, stage + else fMergingJDL->SetArguments("wn.xml $2"); // xml, stage + } + fGridJDL->SetValue("TTL", Form("\"%d\"",fTTL)); fGridJDL->SetDescription("TTL", Form("Time after which the job is killed (%d min.)", fTTL/60)); fMergingJDL->SetValue("TTL", Form("\"%d\"",fTTL)); @@ -1005,11 +1682,17 @@ Bool_t AliAnalysisAlien::CreateJDL() if (fSplitMaxInputFileNumber > 0) { fGridJDL->SetValue("SplitMaxInputFileNumber", Form("\"%d\"", fSplitMaxInputFileNumber)); fGridJDL->SetDescription("SplitMaxInputFileNumber", "Maximum number of input files to be processed per subjob"); + } + if (!IsOneStageMerging()) { + fMergingJDL->SetValue("SplitMaxInputFileNumber", Form("\"%d\"",fMaxMergeFiles)); + fMergingJDL->SetDescription("SplitMaxInputFileNumber", "Maximum number of input files to be merged in one go"); } if (fSplitMode.Length()) { fGridJDL->SetValue("Split", Form("\"%s\"", fSplitMode.Data())); fGridJDL->SetDescription("Split", "We split per SE or file"); - } + } + fMergingJDL->SetValue("Split", "\"se\""); + fMergingJDL->SetDescription("Split", "We split per SE for merging in stages"); if (!fAliROOTVersion.IsNull()) { fGridJDL->AddToPackages("AliRoot", fAliROOTVersion,"VO_ALICE", "List of requested packages"); fMergingJDL->AddToPackages("AliRoot", fAliROOTVersion, "VO_ALICE", "List of requested packages"); @@ -1037,13 +1720,13 @@ Bool_t AliAnalysisAlien::CreateJDL() } fGridJDL->SetInputDataListFormat(fInputFormat, "Format of input data"); fGridJDL->SetInputDataList("wn.xml", "Collection name to be processed on each worker node"); + fMergingJDL->SetInputDataListFormat(fInputFormat, "Format of input data"); + fMergingJDL->SetInputDataList("wn.xml", "Collection name to be processed on each worker node"); fGridJDL->AddToInputSandbox(Form("LF:%s/%s", workdir.Data(), fAnalysisMacro.Data()), "List of input files to be uploaded to workers"); TString analysisFile = fExecutable; analysisFile.ReplaceAll(".sh", ".root"); fGridJDL->AddToInputSandbox(Form("LF:%s/%s", workdir.Data(),analysisFile.Data())); fMergingJDL->AddToInputSandbox(Form("LF:%s/%s", workdir.Data(),analysisFile.Data())); - if (IsUsingTags() && !gSystem->AccessPathName("ConfigureCuts.C")) - fGridJDL->AddToInputSandbox(Form("LF:%s/ConfigureCuts.C", workdir.Data())); if (fAdditionalLibs.Length()) { arr = fAdditionalLibs.Tokenize(" "); TIter next(arr); @@ -1062,30 +1745,38 @@ Bool_t AliAnalysisAlien::CreateJDL() fMergingJDL->AddToInputSandbox(Form("LF:%s/%s", workdir.Data(), obj->GetName())); } } + const char *comment = "List of output files and archives"; if (fOutputArchive.Length()) { - arr = fOutputArchive.Tokenize(" "); + TString outputArchive = fOutputArchive; + if (!fRegisterExcludes.IsNull()) { + arr = fRegisterExcludes.Tokenize(" "); + TIter next1(arr); + while ((os=(TObjString*)next1())) { + outputArchive.ReplaceAll(Form("%s,",os->GetString().Data()),""); + outputArchive.ReplaceAll(os->GetString(),""); + } + delete arr; + } + arr = outputArchive.Tokenize(" "); TIter next(arr); Bool_t first = kTRUE; - const char *comment = "Files to be archived"; - const char *comment1 = comment; while ((os=(TObjString*)next())) { - if (!first) comment = NULL; if (!os->GetString().Contains("@") && fCloseSE.Length()) - fGridJDL->AddToOutputArchive(Form("%s@%s",os->GetString().Data(), fCloseSE.Data()), comment); + fGridJDL->AddToSet("Output", Form("%s@%s",os->GetString().Data(), fCloseSE.Data())); else - fGridJDL->AddToOutputArchive(os->GetString(), comment); + fGridJDL->AddToSet("Output", os->GetString()); + if (first) fGridJDL->AddToSetDescription("Output", comment); first = kFALSE; } delete arr; // Output archive for the merging jdl - TString outputArchive; if (TestBit(AliAnalysisGrid::kDefaultOutputs)) { outputArchive = "log_archive.zip:std*@disk=1 "; // Add normal output files, extra files + terminate files TString files = GetListOfFiles("outextter"); - // Do not register merge excludes - if (!fMergeExcludes.IsNull()) { - arr = fMergeExcludes.Tokenize(" "); + // Do not register files in fRegisterExcludes + if (!fRegisterExcludes.IsNull()) { + arr = fRegisterExcludes.Tokenize(" "); TIter next1(arr); while ((os=(TObjString*)next1())) { files.ReplaceAll(Form("%s,",os->GetString().Data()),""); @@ -1094,22 +1785,26 @@ Bool_t AliAnalysisAlien::CreateJDL() delete arr; } files.ReplaceAll(".root", "*.root"); - outputArchive += Form("root_archive.zip:%s@disk=%d",files.Data(),fNreplicas); + + if (mgr->IsCollectThroughput()) + outputArchive += Form("root_archive.zip:%s,*.stat@disk=%d %s@disk=%d",files.Data(),fNreplicas, mgr->GetFileInfoLog(),fNreplicas); + else + outputArchive += Form("root_archive.zip:%s,*.stat@disk=%d",files.Data(),fNreplicas); } else { - outputArchive = fOutputArchive; + TString files = fOutputArchive; + files.ReplaceAll(".root", "*.root"); // nreplicas etc should be already atttached by use + outputArchive = files; } arr = outputArchive.Tokenize(" "); TIter next2(arr); - comment = comment1; first = kTRUE; while ((os=(TObjString*)next2())) { - if (!first) comment = NULL; TString currentfile = os->GetString(); - if (!IsOneStageMerging()) currentfile.ReplaceAll(".zip", "-Stage$2_$3.zip"); if (!currentfile.Contains("@") && fCloseSE.Length()) - fMergingJDL->AddToOutputArchive(Form("%s@%s",currentfile.Data(), fCloseSE.Data()), comment); + fMergingJDL->AddToSet("Output", Form("%s@%s",currentfile.Data(), fCloseSE.Data())); else - fMergingJDL->AddToOutputArchive(currentfile, comment); + fMergingJDL->AddToSet("Output", currentfile); + if (first) fMergingJDL->AddToSetDescription("Output", comment); first = kFALSE; } delete arr; @@ -1117,7 +1812,6 @@ Bool_t AliAnalysisAlien::CreateJDL() arr = fOutputFiles.Tokenize(","); TIter next(arr); Bool_t first = kTRUE; - const char *comment = "Files to be saved"; while ((os=(TObjString*)next())) { // Ignore ouputs in jdl that are also in outputarchive TString sout = os->GetString(); @@ -1125,17 +1819,21 @@ Bool_t AliAnalysisAlien::CreateJDL() sout.ReplaceAll(".root", ""); if (sout.Index("@")>0) sout.Remove(sout.Index("@")); if (fOutputArchive.Contains(sout)) continue; + // Ignore fRegisterExcludes + if (fRegisterExcludes.Contains(sout)) continue; if (!first) comment = NULL; if (!os->GetString().Contains("@") && fCloseSE.Length()) - fGridJDL->AddToOutputSandbox(Form("%s@%s",os->GetString().Data(), fCloseSE.Data()), comment); + fGridJDL->AddToSet("Output", Form("%s@%s",os->GetString().Data(), fCloseSE.Data())); else - fGridJDL->AddToOutputSandbox(os->GetString(), comment); - first = kFALSE; + fGridJDL->AddToSet("Output", os->GetString()); + if (first) fGridJDL->AddToSetDescription("Output", comment); if (fMergeExcludes.Contains(sout)) continue; if (!os->GetString().Contains("@") && fCloseSE.Length()) - fMergingJDL->AddToOutputSandbox(Form("%s@%s",os->GetString().Data(), fCloseSE.Data()), comment); + fMergingJDL->AddToSet("Output", Form("%s@%s",os->GetString().Data(), fCloseSE.Data())); else - fMergingJDL->AddToOutputSandbox(os->GetString(), comment); + fMergingJDL->AddToSet("Output", os->GetString()); + if (first) fMergingJDL->AddToSetDescription("Output", comment); + first = kFALSE; } delete arr; fGridJDL->SetPrice((UInt_t)fPrice, "AliEn price for this job"); @@ -1181,10 +1879,14 @@ Bool_t AliAnalysisAlien::CreateJDL() if (FileExists(locjdl)) gGrid->Rm(locjdl); if (FileExists(locjdl1)) gGrid->Rm(locjdl1); Info("CreateJDL", "\n##### Copying JDL file <%s> to your AliEn output directory", fJDLName.Data()); - TFile::Cp(Form("file:%s",fJDLName.Data()), Form("alien://%s", locjdl.Data())); + if (!copyLocal2Alien("CreateJDL", fJDLName, locjdl)) + Fatal("","Terminating"); +// TFile::Cp(Form("file:%s",fJDLName.Data()), Form("alien://%s", locjdl.Data())); if (fMergeViaJDL) { Info("CreateJDL", "\n##### Copying merging JDL file <%s> to your AliEn output directory", mergeJDLName.Data()); - TFile::Cp(Form("file:%s",mergeJDLName.Data()), Form("alien://%s", locjdl1.Data())); +// TFile::Cp(Form("file:%s",mergeJDLName.Data()), Form("alien://%s", locjdl1.Data())); + if (!copyLocal2Alien("CreateJDL", mergeJDLName.Data(), locjdl1)) + Fatal("","Terminating"); } } if (fAdditionalLibs.Length()) { @@ -1195,7 +1897,10 @@ Bool_t AliAnalysisAlien::CreateJDL() if (os->GetString().Contains(".so")) continue; Info("CreateJDL", "\n##### Copying dependency: <%s> to your alien workspace", os->GetString().Data()); if (FileExists(os->GetString())) gGrid->Rm(os->GetString()); - TFile::Cp(Form("file:%s",os->GetString().Data()), Form("alien://%s/%s", workdir.Data(), os->GetString().Data())); +// TFile::Cp(Form("file:%s",os->GetString().Data()), Form("alien://%s/%s", workdir.Data(), os->GetString().Data())); + if (!copyLocal2Alien("CreateJDL", os->GetString().Data(), + Form("%s/%s", workdir.Data(), os->GetString().Data()))) + Fatal("","Terminating"); } delete arr; } @@ -1205,7 +1910,10 @@ Bool_t AliAnalysisAlien::CreateJDL() while ((obj=next())) { if (FileExists(obj->GetName())) gGrid->Rm(obj->GetName()); Info("CreateJDL", "\n##### Copying dependency: <%s> to your alien workspace", obj->GetName()); - TFile::Cp(Form("file:%s",obj->GetName()), Form("alien://%s/%s", workdir.Data(), obj->GetName())); +// TFile::Cp(Form("file:%s",obj->GetName()), Form("alien://%s/%s", workdir.Data(), obj->GetName())); + if (!copyLocal2Alien("CreateJDL",obj->GetName(), + Form("%s/%s", workdir.Data(), obj->GetName()))) + Fatal("","Terminating"); } } } @@ -1223,13 +1931,24 @@ Bool_t AliAnalysisAlien::WriteJDL(Bool_t copy) TString workdir; if (!fProductionMode && !fGridWorkingDir.BeginsWith("/alice")) workdir = gGrid->GetHomeDirectory(); workdir += fGridWorkingDir; - + TString stageName = "$2"; + if (fProductionMode) stageName = "$4"; + if (!fMergeDirName.IsNull()) { + fMergingJDL->AddToInputDataCollection(Form("LF:$1/%s/Stage_%s.xml,nodownload",fMergeDirName.Data(),stageName.Data()), "Collection of files to be merged for current stage"); + fMergingJDL->SetOutputDirectory(Form("$1/%s/Stage_%s/#alien_counter_03i#",fMergeDirName.Data(),stageName.Data()), "Output directory"); + } else { + fMergingJDL->AddToInputDataCollection(Form("LF:$1/Stage_%s.xml,nodownload",stageName.Data()), "Collection of files to be merged for current stage"); + fMergingJDL->SetOutputDirectory(Form("$1/Stage_%s/#alien_counter_03i#",stageName.Data()), "Output directory"); + } if (fProductionMode) { TIter next(fInputFiles); - while ((os=next())) + while ((os=next())) { fGridJDL->AddToInputDataCollection(Form("LF:%s,nodownload", os->GetName()), "Input xml collections"); - fGridJDL->SetOutputDirectory(Form("%s/#alien_counter_04i#", fGridOutputDir.Data())); - fMergingJDL->SetOutputDirectory(fGridOutputDir); + } + if (!fOutputToRunNo) + fGridJDL->SetOutputDirectory(Form("%s/#alien_counter_04i#", fGridOutputDir.Data())); + else + fGridJDL->SetOutputDirectory(fGridOutputDir); } else { if (!fRunNumbers.Length() && !fRunRange[0]) { // One jdl with no parameters in case input data is specified by name. @@ -1250,7 +1969,6 @@ Bool_t AliAnalysisAlien::WriteJDL(Bool_t copy) else fGridJDL->SetOutputDirectory(Form("%s/$2",fGridOutputDir.Data()), "Output directory"); } else { fGridJDL->SetOutputDirectory(Form("%s/$2/#alien_counter_03i#", fGridOutputDir.Data()), "Output directory"); - fMergingJDL->SetOutputDirectory("$1", "Output directory"); } } } @@ -1258,8 +1976,17 @@ Bool_t AliAnalysisAlien::WriteJDL(Bool_t copy) // Generate the JDL as a string TString sjdl = fGridJDL->Generate(); TString sjdl1 = fMergingJDL->Generate(); - Int_t index; - sjdl.ReplaceAll("\"LF:", "\n \"LF:"); + // Final merge jdl + if (!fMergeDirName.IsNull()) { + fMergingJDL->SetOutputDirectory(Form("$1/%s",fMergeDirName.Data()), "Output directory"); + fMergingJDL->AddToInputSandbox(Form("LF:$1/%s/Stage_%s.xml",fMergeDirName.Data(),stageName.Data())); + } else { + fMergingJDL->SetOutputDirectory("$1", "Output directory"); + fMergingJDL->AddToInputSandbox(Form("LF:$1/Stage_%s.xml",stageName.Data())); + } + TString sjdl2 = fMergingJDL->Generate(); + Int_t index, index1; + sjdl.ReplaceAll("\",\"", "\",\n \""); sjdl.ReplaceAll("(member", "\n (member"); sjdl.ReplaceAll("\",\"VO_", "\",\n \"VO_"); sjdl.ReplaceAll("{", "{\n "); @@ -1267,7 +1994,7 @@ Bool_t AliAnalysisAlien::WriteJDL(Bool_t copy) sjdl.ReplaceAll("{\n \n", "{\n"); sjdl.ReplaceAll("\n\n", "\n"); sjdl.ReplaceAll("OutputDirectory", "OutputDir"); - sjdl1.ReplaceAll("\"LF:", "\n \"LF:"); + sjdl1.ReplaceAll("\",\"", "\",\n \""); sjdl1.ReplaceAll("(member", "\n (member"); sjdl1.ReplaceAll("\",\"VO_", "\",\n \"VO_"); sjdl1.ReplaceAll("{", "{\n "); @@ -1275,21 +2002,83 @@ Bool_t AliAnalysisAlien::WriteJDL(Bool_t copy) sjdl1.ReplaceAll("{\n \n", "{\n"); sjdl1.ReplaceAll("\n\n", "\n"); sjdl1.ReplaceAll("OutputDirectory", "OutputDir"); + sjdl2.ReplaceAll("\",\"", "\",\n \""); + sjdl2.ReplaceAll("(member", "\n (member"); + sjdl2.ReplaceAll("\",\"VO_", "\",\n \"VO_"); + sjdl2.ReplaceAll("{", "{\n "); + sjdl2.ReplaceAll("};", "\n};"); + sjdl2.ReplaceAll("{\n \n", "{\n"); + sjdl2.ReplaceAll("\n\n", "\n"); + sjdl2.ReplaceAll("OutputDirectory", "OutputDir"); sjdl += "JDLVariables = \n{\n \"Packages\",\n \"OutputDir\"\n};\n"; sjdl.Prepend(Form("Jobtag = {\n \"comment:%s\"\n};\n", fJobTag.Data())); index = sjdl.Index("JDLVariables"); if (index >= 0) sjdl.Insert(index, "\n# JDL variables\n"); sjdl += "Workdirectorysize = {\"5000MB\"};"; + sjdl1 += "Workdirectorysize = {\"5000MB\"};"; sjdl1 += "JDLVariables = \n{\n \"Packages\",\n \"OutputDir\"\n};\n"; index = fJobTag.Index(":"); if (index < 0) index = fJobTag.Length(); TString jobTag = fJobTag; - jobTag.Insert(index, "_Merging"); + if (fProductionMode) jobTag.Insert(index,"_Stage$4"); sjdl1.Prepend(Form("Jobtag = {\n \"comment:%s_Merging\"\n};\n", jobTag.Data())); - sjdl1.Prepend("# Generated merging jdl\n# $1 = full alien path to output directory to be merged\n# $2 = merging stage\n# $3 = merged chunk\n"); + if (fProductionMode) { + sjdl1.Prepend("# Generated merging jdl (production mode) \ + \n# $1 = full alien path to output directory to be merged \ + \n# $2 = train number \ + \n# $3 = production (like LHC10b) \ + \n# $4 = merging stage \ + \n# Stage_.xml made via: find *Stage/*root_archive.zip\n"); + sjdl2.Prepend(Form("Jobtag = {\n \"comment:%s_FinalMerging\"\n};\n", jobTag.Data())); + sjdl2.Prepend("# Generated merging jdl \ + \n# $1 = full alien path to output directory to be merged \ + \n# $2 = train number \ + \n# $3 = production (like LHC10b) \ + \n# $4 = merging stage \ + \n# Stage_.xml made via: find *Stage/*root_archive.zip\n"); + } else { + sjdl1.Prepend("# Generated merging jdl \ + \n# $1 = full alien path to output directory to be merged \ + \n# $2 = merging stage \ + \n# xml made via: find *Stage/*root_archive.zip\n"); + sjdl2.Prepend(Form("Jobtag = {\n \"comment:%s_FinalMerging\"\n};\n", jobTag.Data())); + sjdl2.Prepend("# Generated merging jdl \ + \n# $1 = full alien path to output directory to be merged \ + \n# $2 = merging stage \ + \n# xml made via: find *Stage/*root_archive.zip\n"); + } index = sjdl1.Index("JDLVariables"); if (index >= 0) sjdl1.Insert(index, "\n# JDL variables\n"); + index = sjdl2.Index("JDLVariables"); + if (index >= 0) sjdl2.Insert(index, "\n# JDL variables\n"); sjdl1 += "Workdirectorysize = {\"5000MB\"};"; + sjdl2 += "Workdirectorysize = {\"5000MB\"};"; + index = sjdl2.Index("Split ="); + if (index>=0) { + index1 = sjdl2.Index("\n", index); + sjdl2.Remove(index, index1-index+1); + } + index = sjdl2.Index("SplitMaxInputFileNumber"); + if (index>=0) { + index1 = sjdl2.Index("\n", index); + sjdl2.Remove(index, index1-index+1); + } + index = sjdl2.Index("InputDataCollection"); + if (index>=0) { + index1 = sjdl2.Index(";", index); + sjdl2.Remove(index, index1-index+1); + } + index = sjdl2.Index("InputDataListFormat"); + if (index>=0) { + index1 = sjdl2.Index("\n", index); + sjdl2.Remove(index, index1-index+1); + } + index = sjdl2.Index("InputDataList"); + if (index>=0) { + index1 = sjdl2.Index("\n", index); + sjdl2.Remove(index, index1-index+1); + } + sjdl2.ReplaceAll("wn.xml", Form("Stage_%s.xml",stageName.Data())); // Write jdl to file ofstream out; out.open(fJDLName.Data(), ios::out); @@ -1298,16 +2087,28 @@ Bool_t AliAnalysisAlien::WriteJDL(Bool_t copy) return kFALSE; } out << sjdl << endl; + out.close(); TString mergeJDLName = fExecutable; mergeJDLName.ReplaceAll(".sh", "_merge.jdl"); if (fMergeViaJDL) { ofstream out1; out1.open(mergeJDLName.Data(), ios::out); - if (out.bad()) { + if (out1.bad()) { Error("WriteJDL", "Bad file name: %s", mergeJDLName.Data()); return kFALSE; } out1 << sjdl1 << endl; + out1.close(); + ofstream out2; + TString finalJDL = mergeJDLName; + finalJDL.ReplaceAll(".jdl", "_final.jdl"); + out2.open(finalJDL.Data(), ios::out); + if (out2.bad()) { + Error("WriteJDL", "Bad file name: %s", finalJDL.Data()); + return kFALSE; + } + out2 << sjdl2 << endl; + out2.close(); } // Copy jdl to grid workspace @@ -1316,17 +2117,29 @@ Bool_t AliAnalysisAlien::WriteJDL(Bool_t copy) } else { TString locjdl = Form("%s/%s", fGridOutputDir.Data(),fJDLName.Data()); TString locjdl1 = Form("%s/%s", fGridOutputDir.Data(),mergeJDLName.Data()); + TString finalJDL = mergeJDLName; + finalJDL.ReplaceAll(".jdl", "_final.jdl"); + TString locjdl2 = Form("%s/%s", fGridOutputDir.Data(),finalJDL.Data()); if (fProductionMode) { locjdl = Form("%s/%s", workdir.Data(),fJDLName.Data()); locjdl1 = Form("%s/%s", workdir.Data(),mergeJDLName.Data()); + locjdl2 = Form("%s/%s", workdir.Data(),finalJDL.Data()); } if (FileExists(locjdl)) gGrid->Rm(locjdl); if (FileExists(locjdl1)) gGrid->Rm(locjdl1); + if (FileExists(locjdl2)) gGrid->Rm(locjdl2); Info("WriteJDL", "\n##### Copying JDL file <%s> to your AliEn output directory", fJDLName.Data()); - TFile::Cp(Form("file:%s",fJDLName.Data()), Form("alien://%s", locjdl.Data())); +// TFile::Cp(Form("file:%s",fJDLName.Data()), Form("alien://%s", locjdl.Data())); + if (!copyLocal2Alien("WriteJDL",fJDLName.Data(),locjdl.Data())) + Fatal("","Terminating"); if (fMergeViaJDL) { - Info("WriteJDL", "\n##### Copying merging JDL file <%s> to your AliEn output directory", mergeJDLName.Data()); - TFile::Cp(Form("file:%s",mergeJDLName.Data()), Form("alien://%s", locjdl1.Data())); + Info("WriteJDL", "\n##### Copying merging JDL files <%s> to your AliEn output directory", mergeJDLName.Data()); +// TFile::Cp(Form("file:%s",mergeJDLName.Data()), Form("alien://%s", locjdl1.Data())); +// TFile::Cp(Form("file:%s",finalJDL.Data()), Form("alien://%s", locjdl2.Data())); + if (!copyLocal2Alien("WriteJDL",mergeJDLName.Data(),locjdl1.Data())) + Fatal("","Terminating"); + if (!copyLocal2Alien("WriteJDL",finalJDL.Data(),locjdl2.Data())) + Fatal("","Terminating"); } } return kTRUE; @@ -1497,30 +2310,68 @@ TChain *AliAnalysisAlien::GetChainForTestMode(const char *treeName) const in.open(fFileForTestMode); Int_t count = 0; // Read the input list of files and add them to the chain - TString line; - TChain *chain = new TChain(treeName); - while (in.good()) - { + TString line; + TString streeName(treeName); + if (IsUseMCchain()) streeName = "TE"; + TChain *chain = new TChain(streeName); + TList *friends = new TList(); + TChain *cfriend = 0; + if (!fFriendChainName.IsNull()) { + TObjArray *list = fFriendChainName.Tokenize(" "); + TIter next(list); + TObjString *str; + while((str=(TObjString*)next())) { + cfriend = new TChain(streeName, str->GetName()); + friends->Add(cfriend); + chain->AddFriend(cfriend); + } + delete list; + } + TString bpath; + TIter nextfriend(friends); + while (in.good()) + { in >> line; - if (line.IsNull()) continue; + if (line.IsNull() || line.BeginsWith("#")) continue; if (count++ == fNtestFiles) break; TString esdFile(line); TFile *file = TFile::Open(esdFile); - if (file) { - if (!file->IsZombie()) chain->Add(esdFile); + if (file && !file->IsZombie()) { + chain->Add(esdFile); file->Close(); + if (!fFriendChainName.IsNull()) { + if (esdFile.Index("#") > -1) + esdFile.Remove(esdFile.Index("#")); + bpath = gSystem->DirName(esdFile); + bpath += "/"; + TString fileFriend; + nextfriend.Reset(); + while ((cfriend=(TChain*)nextfriend())) { + fileFriend = bpath; + fileFriend += cfriend->GetTitle(); + file = TFile::Open(fileFriend); + if (file && !file->IsZombie()) { + file->Close(); + cfriend->Add(fileFriend); + } else { + Fatal("GetChainForTestMode", "Cannot open friend file: %s", fileFriend.Data()); + return 0; + } + } + } } else { Error("GetChainforTestMode", "Skipping un-openable file: %s", esdFile.Data()); } - } - in.close(); - if (!chain->GetListOfFiles()->GetEntries()) { + } + in.close(); + if (!chain->GetListOfFiles()->GetEntries()) { Error("GetChainForTestMode", "No file from %s could be opened", fFileForTestMode.Data()); delete chain; + friends->Delete(); + delete friends; return NULL; - } -// chain->ls(); - return chain; + } + return chain; } //______________________________________________________________________________ @@ -1540,10 +2391,10 @@ const char *AliAnalysisAlien::GetJobStatus(Int_t jobidstart, Int_t lastid, Int_t Int_t pid; for (Int_t ijob=0; ijobAt(ijob); - pid = gROOT->ProcessLine(Form("atoi(((TAlienJobStatus*)0x%lx)->GetKey(\"queueId\"));", (ULong_t)status)); + pid = gROOT->ProcessLine(Form("atoi(((TAlienJobStatus*)%p)->GetKey(\"queueId\"));", status)); if (pidProcessLine(Form("sprintf((char*)0x%lx,((TAlienJobStatus*)0x%lx)->GetKey(\"status\"));",(ULong_t)mstatus, (ULong_t)status)); + gROOT->ProcessLine(Form("sprintf((char*)%p,((TAlienJobStatus*)%p)->GetKey(\"status\"));",mstatus, status)); } switch (status->GetStatus()) { case TGridJobStatus::kWAITING: @@ -1605,8 +2456,8 @@ void AliAnalysisAlien::Print(Option_t *) const printf("= Soft reset signal will be send to master______ CHANGE BEHAVIOR AFTER COMPLETION\n"); if (fProofReset>1) printf("= Hard reset signal will be send to master______ CHANGE BEHAVIOR AFTER COMPLETION\n"); - if (!fRootVersionForProof.IsNull()) - printf("= ROOT version requested________________________ %s\n", fRootVersionForProof.Data()); + if (!fROOTVersion.IsNull()) + printf("= ROOT version requested________________________ %s\n", fROOTVersion.Data()); else printf("= ROOT version requested________________________ default\n"); printf("= AliRoot version requested_____________________ %s\n", fAliROOTVersion.Data()); @@ -1638,7 +2489,7 @@ void AliAnalysisAlien::Print(Option_t *) const \n***** To disable, use: plugin->SetOverwriteMode(kFALSE);\n"); } printf("= Copy files to grid: __________________________ %s\n", (IsUseCopy())?"YES":"NO"); - printf("= Check if files can be copied to grid: ________ %s\n", (IsCheckCopy())?"YES":"NO"); + printf("= Check if files can be copied to grid: ________ %s\n", (IsCheckCopy())?"YES":"NO:Print"); printf("= Production mode:______________________________ %d\n", fProductionMode); printf("= Version of API requested: ____________________ %s\n", fAPIVersion.Data()); printf("= Version of ROOT requested: ___________________ %s\n", fROOTVersion.Data()); @@ -1647,13 +2498,22 @@ void AliAnalysisAlien::Print(Option_t *) const printf("= User running the plugin: _____________________ %s\n", fUser.Data()); printf("= Grid workdir relative to user $HOME: _________ %s\n", fGridWorkingDir.Data()); printf("= Grid output directory relative to workdir: ___ %s\n", fGridOutputDir.Data()); - printf("= Data base directory path requested: __________ %s\n", fGridDataDir.Data()); - printf("= Data search pattern: _________________________ %s\n", fDataPattern.Data()); + TString basedatadir = fGridDataDir; + TString pattern = fDataPattern; + pattern.Strip(); + Int_t ind = pattern.Index(" "); + if (ind>=0) { + basedatadir += "/%run%/"; + basedatadir += pattern(0, ind); + pattern = pattern(ind+1, pattern.Length()); + } + printf("= Data base directory path requested: __________ %s\n", basedatadir.Data()); + printf("= Data search pattern: _________________________ %s\n", pattern.Data()); printf("= Input data format: ___________________________ %s\n", fInputFormat.Data()); if (fRunNumbers.Length()) printf("= Run numbers to be processed: _________________ %s\n", fRunNumbers.Data()); if (fRunRange[0]) - printf("= Run range to be processed: ___________________ %s%d-%s%d\n", fRunPrefix.Data(), fRunRange[0], fRunPrefix.Data(), fRunRange[1]); + printf("= Run range to be processed: ___________________ %d-%d\n", fRunRange[0], fRunRange[1]); if (!fRunRange[0] && !fRunNumbers.Length()) { TIter next(fInputFiles); TObject *obj; @@ -1666,6 +2526,7 @@ void AliAnalysisAlien::Print(Option_t *) const printf("= List of output files to be registered: _______ %s\n", fOutputFiles.Data()); printf("= List of outputs going to be archived: ________ %s\n", fOutputArchive.Data()); printf("= List of outputs that should not be merged: ___ %s\n", fMergeExcludes.Data()); + printf("= List of outputs that should not be registered: %s\n", fRegisterExcludes.Data()); printf("= List of outputs produced during Terminate: ___ %s\n", fTerminateFiles.Data()); printf("=====================================================================\n"); printf("= Job price: ___________________________________ %d\n", fPrice); @@ -1723,11 +2584,12 @@ void AliAnalysisAlien::SetDefaults() fNreplicas = 2; fRunRange[0] = 0; fRunRange[1] = 0; + fRunPrefix = "%d"; fNrunsPerMaster = 1; fMaxMergeFiles = 100; fRunNumbers = ""; fExecutable = "analysis.sh"; - fExecutableCommand = "root -b -q"; + fExecutableCommand = "root -b -q -x"; fArguments = ""; fExecutableArgs = ""; fAnalysisMacro = "myAnalysis.C"; @@ -1757,118 +2619,127 @@ void AliAnalysisAlien::SetDefaults() } //______________________________________________________________________________ -Bool_t AliAnalysisAlien::CheckMergedFiles(const char *filename, const char *aliendir, Int_t nperchunk, Bool_t submit, const char *jdl) +void AliAnalysisAlien::SetFriendChainName(const char *name, const char *libnames) { -// Static method that checks the status of merging. This can submit merging jobs that did not produced the expected -// output. If is false (checking) returns true only when the final merged file was found. If submit is true returns -// true if the jobs were successfully submitted. - Int_t countOrig = 0; - Int_t countStage = 0; + // Set file name for the chain of friends and optionally additional libs to be loaded. + // Libs should be separated by blancs. + fFriendChainName = name; + fFriendChainName.ReplaceAll(",", " "); + fFriendChainName.Strip(); + fFriendChainName.ReplaceAll(" ", " "); + + fFriendLibs = libnames; + if (fFriendLibs.Length()) { + if(!fFriendLibs.Contains(".so")) + Fatal("SetFriendChainName()", "You should provide explicit library names (with extension)"); + fFriendLibs.ReplaceAll(",", " "); + fFriendLibs.Strip(); + fFriendLibs.ReplaceAll(" ", " "); + } +} + +//______________________________________________________________________________ +void AliAnalysisAlien::SetRootVersionForProof(const char *version) +{ +// Obsolete method. Use SetROOTVersion instead + Warning("SetRootVersionForProof", "Obsolete. Use SetROOTVersion instead"); + if (fROOTVersion.IsNull()) SetROOTVersion(version); + else Error("SetRootVersionForProof", "ROOT version already set to %s", fROOTVersion.Data()); +} + +//______________________________________________________________________________ +Bool_t AliAnalysisAlien::CheckMergedFiles(const char *filename, const char *aliendir, Int_t nperchunk, const char *jdl) +{ +// Checks current merge stage, makes xml for the next stage, counts number of files, submits next stage. + // First check if the result is already in the output directory. + if (FileExists(Form("%s/%s",aliendir,filename))) { + printf("Final merged results found. Not merging again.\n"); + return kFALSE; + } + // Now check the last stage done. Int_t stage = 0; - Int_t i; - Bool_t doneFinal = kFALSE; - TBits chunksDone; - TString saliendir(aliendir); - TString sfilename, stmp; - saliendir.ReplaceAll("//","/"); - saliendir = saliendir.Strip(TString::kTrailing, '/'); - if (!gGrid) { - ::Error("GetNregisteredFiles", "You need to be connected to AliEn."); + while (1) { + if (!FileExists(Form("%s/Stage_%d.xml",aliendir, stage+1))) break; + stage++; + } + // Next stage of merging + stage++; + TString pattern = "*root_archive.zip"; + if (stage>1) pattern = Form("Stage_%d/*root_archive.zip", stage-1); + TGridResult *res = gGrid->Command(Form("find -x Stage_%d %s %s", stage, aliendir, pattern.Data())); + if (res) delete res; + // Write standard output to file + gROOT->ProcessLine(Form("gGrid->Stdout(); > %s", Form("Stage_%d.xml",stage))); + // Count the number of files inside + ifstream ifile; + ifile.open(Form("Stage_%d.xml",stage)); + if (!ifile.good()) { + ::Error("CheckMergedFiles", "Could not redirect result of the find command to file %s", Form("Stage_%d.xml",stage)); return kFALSE; + } + TString line; + Int_t nfiles = 0; + while (!ifile.eof()) { + ifile >> line; + if (line.Contains("/event")) nfiles++; } - sfilename = filename; - sfilename.ReplaceAll(".root", "*.root"); - printf("Checking directory <%s> for merged files <%s> ...\n", aliendir, sfilename.Data()); - TString command = Form("find %s/ *%s", saliendir.Data(), sfilename.Data()); - TGridResult *res = gGrid->Command(command); - if (!res) { - ::Error("GetNregisteredFiles","Error: No result for the find command\n"); + ifile.close(); + if (!nfiles) { + ::Error("CheckMergedFiles", "Cannot start Stage_%d merging since Stage_%d did not produced yet output", stage, stage-1); return kFALSE; - } - TIter nextmap(res); - TMap *map = 0; - while ((map=(TMap*)nextmap())) { - TString turl = map->GetValue("turl")->GetName(); - if (!turl.Length()) { - // Nothing found - delete res; - return kFALSE; - } - turl.ReplaceAll("alien://", ""); - turl.ReplaceAll(saliendir, ""); - sfilename = gSystem->BaseName(turl); - turl = turl.Strip(TString::kLeading, '/'); - // Now check to what the file corresponds to: - // original output - aliendir/%03d/filename - // merged file (which stage) - aliendir/filename-Stage%02d_%04d - // final merged file - aliendir/filename - if (sfilename == turl) { - if (sfilename == filename) { - doneFinal = kTRUE; - } else { - // check stage - Int_t index = sfilename.Index("Stage"); - if (index<0) continue; - stmp = sfilename(index+5,2); - Int_t istage = atoi(stmp); - stmp = sfilename(index+8,4); - Int_t ijob = atoi(stmp); - if (istagestage) { - countStage = 0; - chunksDone.ResetAllBits(); - stage = istage; - } - countStage++; - chunksDone.SetBitNumber(ijob); - } - } else { - countOrig++; - } - if (doneFinal) { - delete res; - printf("=> Removing files from previous stages...\n"); - gGrid->Rm(Form("%s/*Stage*.root", aliendir)); - for (i=1; iRm(Form("%s/*Stage%d*.zip", aliendir, i)); - return kTRUE; - } + } else { + printf("=== Stage_%d produced %d files\n", stage-1, nfiles); + } + // Copy the file in the output directory + printf("===> Copying collection %s in the output directory %s\n", Form("Stage_%d.xml",stage), aliendir); +// TFile::Cp(Form("Stage_%d.xml",stage), Form("alien://%s/Stage_%d.xml",aliendir,stage)); + if (!copyLocal2Alien("CheckMergedFiles", Form("Stage_%d.xml",stage), + Form("%s/Stage_%d.xml",aliendir,stage))) Fatal("","Terminating"); + // Check if this is the last stage to be done. + Bool_t laststage = (nfiles=fMaxMergeStages) laststage = kTRUE; + Int_t jobId = 0; + if (laststage) { + printf("### Submiting final merging stage %d\n", stage); + TString finalJDL = jdl; + finalJDL.ReplaceAll(".jdl", "_final.jdl"); + TString query = Form("submit %s %s %d", finalJDL.Data(), aliendir, stage); + jobId = SubmitSingleJob(query); + } else { + printf("### Submiting merging stage %d\n", stage); + TString query = Form("submit %s %s %d", jdl, aliendir, stage); + jobId = SubmitSingleJob(query); } - delete res; - // Compute number of jobs that were submitted for the current stage - Int_t ntotstage = countOrig; - for (i=1; i<=stage; i++) { - if (ntotstage%nperchunk) ntotstage = (ntotstage/nperchunk)+1; - else ntotstage = (ntotstage/nperchunk); + if (!jobId) return kFALSE; + + if (!fGridJobIDs.IsNull()) fGridJobIDs.Append(" "); + fGridJobIDs.Append(Form("%d", jobId)); + if (!fGridStages.IsNull()) fGridStages.Append(" "); + fGridStages.Append(Form("%s_merge_stage%d", + laststage ? "final" : "partial", stage)); + + return kTRUE; +} + +//______________________________________________________________________________ +AliAnalysisManager *AliAnalysisAlien::LoadAnalysisManager(const char *fname) +{ +// Loat the analysis manager from a file. + TFile *file = TFile::Open(fname); + if (!file) { + ::Error("LoadAnalysisManager", "Cannot open file %s", fname); + return 0; } - // Now compare with the number of set bits in the chunksDone array - Int_t nmissing = (stage>0)?(ntotstage - countStage):0; - // Print the info - printf("*** Found %d original files\n", countOrig); - if (stage==0) printf("*** No merging completed so far.\n"); - else printf("*** Found %d out of %d files merged for stage %d\n", countStage, ntotstage, stage); - if (nmissing) printf("*** Number of merged files missing for this stage: %d -> check merging job completion\n", nmissing); - if (!submit) return doneFinal; - // Sumbit merging jobs for all missing chunks for the current stage. - TString query = Form("submit %s %s", jdl, aliendir); - Int_t ichunk = -1; - if (nmissing) { - for (i=0; iGetListOfKeys()); + AliAnalysisManager *mgr = 0; + TKey *key; + while ((key=(TKey*)nextkey())) { + if (!strcmp(key->GetClassName(), "AliAnalysisManager")) + mgr = (AliAnalysisManager*)file->Get(key->GetName()); } - // Submit next stage of merging - if (stage==0) countStage = countOrig; - Int_t nchunks = (countStage/nperchunk); - if (countStage%nperchunk) nchunks += 1; - for (i=0; iProcessLine(Form("TAlienCollection::Open(\"%s\");", collection)); + if (!coll) { + ::Error("MergeInfo", "Input XML %s collection empty.", collection); + return kFALSE; + } + // Iterate grid collection + TString outtmp; + Bool_t merged = kFALSE; + Int_t ifile = 0; + while (coll->Next()) { + TString fname = gSystem->DirName(coll->GetTURL()); + fname += "/"; + fname += output; + outtmp = Form("%d_%s", ifile, output); + if (!TFile::Cp(fname, outtmp)) { + ::Error("MergeInfo", "Could not copy %s", fname.Data()); + continue; + } + ifile++; + if (ifile<2) { + gSystem->Exec(Form("cp %s lastmerged", outtmp.Data())); + continue; + } + gSystem->Exec(Form("cat lastmerged %s > tempmerged", outtmp.Data())); + gSystem->Exec("cp tempmerged lastmerged"); + } + if (ifile) { + gSystem->Exec(Form("cp lastmerged %s", output)); + gSystem->Exec(Form("rm tempmerged lastmerged *_%s", output)); + merged = kTRUE; + } + return merged; +} + +//______________________________________________________________________________ +Bool_t AliAnalysisAlien::MergeOutput(const char *output, const char *basedir, Int_t nmaxmerge, Int_t stage) { -// Merge given output files from basedir. The file merger will merge nmaxmerge -// files in a group. Merging can be done in stages: -// stage=0 : will merge all existing files in a single stage -// stage=1 : does a find command for all files that do NOT contain the string "Stage". -// If their number is bigger that nmaxmerge, only the files from -// ichunk*nmaxmerge to ichunk*(nmaxmerge+1)-1 will get merged as output_stage_ -// stage=n : does a find command for files named Stage_*. If their number is bigger than -// nmaxmerge, merge just the chunk ichunk, otherwise write the merged output to the file -// named . +// Merge given output files from basedir. Basedir can be an alien output directory +// but also an xml file with root_archive.zip locations. The file merger will merge nmaxmerge +// files in a group (ignored for xml input). Merging can be done in stages: +// stage=0 : will merge all existing files in a single stage, supporting resume if run locally +// stage=1 : works with an xml of all root_archive.zip in the output directory +// stage>1 : works with an xml of all root_archive.zip in the Stage_ directory TString outputFile = output; TString command; TString outputChunk; TString previousChunk = ""; + TObjArray *listoffiles = new TObjArray(); +// listoffiles->SetOwner(); Int_t countChunk = 0; Int_t countZero = nmaxmerge; Bool_t merged = kTRUE; + Bool_t isGrid = kTRUE; Int_t index = outputFile.Index("@"); if (index > 0) outputFile.Remove(index); TString inputFile = outputFile; - if (stage>1) inputFile.ReplaceAll(".root", Form("-Stage%02d_*.root", stage-1)); - command = Form("find %s/ *%s", basedir, inputFile.Data()); - printf("command: %s\n", command.Data()); - TGridResult *res = gGrid->Command(command); - if (!res) { + TString sbasedir = basedir; + if (sbasedir.Contains(".xml")) { + // Merge files pointed by the xml - ignore nmaxmerge and set ichunk to 0 + nmaxmerge = 9999999; + TGridCollection *coll = (TGridCollection*)gROOT->ProcessLine(Form("TAlienCollection::Open(\"%s\");", basedir)); + if (!coll) { + ::Error("MergeOutput", "Input XML collection empty."); + return kFALSE; + } + // Iterate grid collection + while (coll->Next()) { + TString fname = gSystem->DirName(coll->GetTURL()); + fname += "/"; + fname += inputFile; + listoffiles->Add(new TNamed(fname.Data(),"")); + } + } else if (sbasedir.Contains(".txt")) { + // The file having the .txt extension is expected to contain a list of + // folders where the output files will be looked. For alien folders, + // the full folder LFN is expected (starting with alien://) + // Assume lfn's on each line + TString line; + ifstream in; + in.open(sbasedir); + if (in.fail()) { + ::Error("MergeOutput", "File %s cannot be opened. Merging stopped." ,sbasedir.Data()); + return kTRUE; + } + Int_t nfiles = 0; + while (in.good()) { + in >> line; + if (line.IsNull() || line.BeginsWith("#")) continue; + line.Strip(); + if (!line.Contains("alien:")) isGrid = kFALSE; + line += "/"; + line += outputFile; + nfiles++; + listoffiles->Add(new TNamed(line.Data(),"")); + } + in.close(); + if (!nfiles) { + ::Error("MergeOutput","Input file %s contains no files to be merged\n", sbasedir.Data()); + delete listoffiles; + return kFALSE; + } + } else { + command = Form("find %s/ *%s", basedir, inputFile.Data()); + printf("command: %s\n", command.Data()); + TGridResult *res = gGrid->Command(command); + if (!res) { + ::Error("MergeOutput","No result for the find command\n"); + delete listoffiles; + return kFALSE; + } + TIter nextmap(res); + TMap *map = 0; + while ((map=(TMap*)nextmap())) { + TObjString *objs = dynamic_cast(map->GetValue("turl")); + if (!objs || !objs->GetString().Length()) { + // Nothing found - skip this output + delete res; + delete listoffiles; + return kFALSE; + } + listoffiles->Add(new TNamed(objs->GetName(),"")); + } + delete res; + } + if (!listoffiles->GetEntries()) { ::Error("MergeOutput","No result for the find command\n"); + delete listoffiles; return kFALSE; } TFileMerger *fm = 0; - TIter nextmap(res); - TMap *map = 0; + TIter next0(listoffiles); + TObjArray *listoffilestmp = new TObjArray(); + listoffilestmp->SetOwner(); + TObject *nextfile; + TString snextfile; + // Keep only the files at upper level + Int_t countChar = 0; + while ((nextfile=next0())) { + snextfile = nextfile->GetName(); + Int_t crtCount = snextfile.CountChar('/'); + if (nextfile == listoffiles->First()) countChar = crtCount; + if (crtCount < countChar) countChar = crtCount; + } + next0.Reset(); + while ((nextfile=next0())) { + snextfile = nextfile->GetName(); + Int_t crtCount = snextfile.CountChar('/'); + if (crtCount > countChar) { + delete nextfile; + continue; + } + listoffilestmp->Add(nextfile); + } + delete listoffiles; + listoffiles = listoffilestmp; // Now contains 'good' files + listoffiles->Print(); + TIter next(listoffiles); // Check if there is a merge operation to resume. Works only for stage 0 or 1. outputChunk = outputFile; outputChunk.ReplaceAll(".root", "_*.root"); @@ -1933,31 +2935,17 @@ Bool_t AliAnalysisAlien::MergeOutput(const char *output, const char *basedir, In // Check overwrite mode and remove previous partial results if needed // Preserve old merging functionality for stage 0. if (stage==0) { - Int_t countChar = 0; if (!gSystem->Exec(Form("ls %s 2>/dev/null", outputChunk.Data()))) { while (1) { // Skip as many input files as in a chunk for (Int_t counter=0; counter(map->GetValue("turl")); - // Count the '/' characters in the path to the current file. - Int_t crtCount = objs->GetString().CountChar('/'); - if (!countChar) { - countChar = crtCount; - // Make sure we check if the same file in the parent dir exists - if (FileExists(Form("%s/../%s", basedir, output))) countChar--; - } - if (crtCount > countChar) counter--; - } - if (!map) { - ::Error("MergeOutput", "Cannot resume merging for <%s>, nentries=%d", outputFile.Data(), res->GetSize()); - delete res; - return kFALSE; + snextfile = nextfile->GetName(); } outputChunk = outputFile; outputChunk.ReplaceAll(".root", Form("_%04d.root", countChunk)); @@ -1971,112 +2959,69 @@ Bool_t AliAnalysisAlien::MergeOutput(const char *output, const char *basedir, In } countZero = nmaxmerge; - while ((map=(TMap*)nextmap())) { - TObjString *objs = dynamic_cast(map->GetValue("turl")); - if (!objs || !objs->GetString().Length()) { - // Nothing found - skip this output - delete res; - delete fm; - return kFALSE; - } - // Make sure this is a good file and not one from a subjob directory in case we merge runs - // Count the '/' characters in the path to the current file. - Int_t crtCount = objs->GetString().CountChar('/'); - if (!countChar) { - countChar = crtCount; - // Make sure we check if the same file in the parent dir exists - if (FileExists(Form("%s/../%s", basedir, output))) countChar--; - } - if (crtCount > countChar) continue; + while ((nextfile=next())) { + snextfile = nextfile->GetName(); // Loop 'find' results and get next LFN if (countZero == nmaxmerge) { // First file in chunk - create file merger and add previous chunk if any. - fm = new TFileMerger(kFALSE); + fm = new TFileMerger(isGrid); fm->SetFastMethod(kTRUE); if (previousChunk.Length()) fm->AddFile(previousChunk.Data()); outputChunk = outputFile; outputChunk.ReplaceAll(".root", Form("_%04d.root", countChunk)); } // If last file found, put merged results in the output file - if (map == res->Last()) outputChunk = outputFile; + if (nextfile == listoffiles->Last()) outputChunk = outputFile; // Add file to be merged and decrement chunk counter. - fm->AddFile(objs->GetString()); + fm->AddFile(snextfile); countZero--; - if (countZero==0 || map == res->Last()) { + if (countZero==0 || nextfile == listoffiles->Last()) { if (!fm->GetMergeList() || !fm->GetMergeList()->GetSize()) { // Nothing found - skip this output ::Warning("MergeOutput", "No <%s> files found.", inputFile.Data()); - delete res; - delete fm; - return kFALSE; + merged = kFALSE; + break; } fm->OutputFile(outputChunk); // Merge the outputs, then go to next chunk if (!fm->Merge()) { ::Error("MergeOutput", "Could not merge all <%s> files", outputFile.Data()); - delete res; - delete fm; - return kFALSE; + merged = kFALSE; + break; } else { ::Info("MergeOutputs", "\n##### Merged %d output files to <%s>", fm->GetMergeList()->GetSize(), outputChunk.Data()); gSystem->Unlink(previousChunk); } - if (map == res->Last()) { - delete res; - delete fm; - break; - } + if (nextfile == listoffiles->Last()) break; countChunk++; countZero = nmaxmerge; previousChunk = outputChunk; } } + delete listoffiles; + delete fm; return merged; } // Merging stage different than 0. // Move to the begining of the requested chunk. - outputChunk = outputFile; - if (nmaxmerge < res->GetSize()) { - if (ichunk*nmaxmerge >= res->GetSize()) { - ::Error("MergeOutput", "Cannot merge merge chunk %d grouping %d files from %d total.", ichunk, nmaxmerge, res->GetSize()); - delete res; - return kFALSE; - } - for (Int_t counter=0; counterSetFastMethod(kTRUE); - while ((map=(TMap*)nextmap())) { - // Loop 'find' results and get next LFN - TObjString *objs = dynamic_cast(map->GetValue("turl")); - if (!objs || !objs->GetString().Length()) { - // Nothing found - skip this output - delete res; - delete fm; - return kFALSE; - } - // Add file to be merged and decrement chunk counter. - fm->AddFile(objs->GetString()); - countZero--; - if (countZero==0) break; - } - delete res; + while ((nextfile=next())) fm->AddFile(nextfile->GetName()); + delete listoffiles; if (!fm->GetMergeList() || !fm->GetMergeList()->GetSize()) { // Nothing found - skip this output ::Warning("MergeOutput", "No <%s> files found.", inputFile.Data()); delete fm; return kFALSE; } - fm->OutputFile(outputChunk); + fm->OutputFile(outputFile); // Merge the outputs if (!fm->Merge()) { ::Error("MergeOutput", "Could not merge all <%s> files", outputFile.Data()); delete fm; return kFALSE; } else { - ::Info("MergeOutput", "\n##### Merged %d output files to <%s>", fm->GetMergeList()->GetSize(), outputChunk.Data()); + ::Info("MergeOutput", "\n##### Merged %d output files to <%s>", fm->GetMergeList()->GetSize(), outputFile.Data()); } delete fm; return kTRUE; @@ -2108,7 +3053,7 @@ Bool_t AliAnalysisAlien::MergeOutputs() return kFALSE; } // Get the output path - if (!fGridOutputDir.Contains("/")) fGridOutputDir = Form("/%s/%s/%s", gGrid->GetHomeDirectory(), fGridWorkingDir.Data(), fGridOutputDir.Data()); + if (!fGridOutputDir.Contains("/")) fGridOutputDir = Form("%s/%s/%s", gGrid->GetHomeDirectory(), fGridWorkingDir.Data(), fGridOutputDir.Data()); if (!DirectoryExists(fGridOutputDir)) { Error("MergeOutputs", "Grid output directory %s not found. Terminate() will NOT be executed", fGridOutputDir.Data()); return kFALSE; @@ -2123,14 +3068,16 @@ Bool_t AliAnalysisAlien::MergeOutputs() if (fFastReadOption) { Warning("MergeOutputs", "You requested FastRead option. Using xrootd flags to reduce timeouts. This may skip some files that could be accessed ! \ \n+++ NOTE: To disable this option, use: plugin->SetFastReadOption(kFALSE)"); - gEnv->SetValue("XNet.ConnectTimeout",10); - gEnv->SetValue("XNet.RequestTimeout",10); + gEnv->SetValue("XNet.ConnectTimeout",50); + gEnv->SetValue("XNet.RequestTimeout",50); gEnv->SetValue("XNet.MaxRedirectCount",2); - gEnv->SetValue("XNet.ReconnectTimeout",10); + gEnv->SetValue("XNet.ReconnectTimeout",50); gEnv->SetValue("XNet.FirstConnectMaxCnt",1); } // Make sure we change the temporary directory gSystem->Setenv("TMPDIR", gSystem->pwd()); + // Set temporary compilation directory to current one + gSystem->SetBuildDir(gSystem->pwd(), kTRUE); TObjArray *list = fOutputFiles.Tokenize(","); TIter next(list); TObjString *str; @@ -2163,17 +3110,19 @@ Bool_t AliAnalysisAlien::MergeOutputs() gSystem->Exec(Form("rm -f %s", outputChunk.Data())); } } - if (fMergeExcludes.Length() && - fMergeExcludes.Contains(outputFile.Data())) continue; + if (fMergeExcludes.Contains(outputFile.Data()) || + fRegisterExcludes.Contains(outputFile.Data())) continue; // Perform a 'find' command in the output directory, looking for registered outputs merged = MergeOutput(outputFile, fGridOutputDir, fMaxMergeFiles); if (!merged) { Error("MergeOutputs", "Terminate() will NOT be executed"); - return kFALSE; + delete list; + return kFALSE; } TFile *fileOpened = (TFile*)gROOT->GetListOfFiles()->FindObject(outputFile); if (fileOpened) fileOpened->Close(); } + delete list; return kTRUE; } @@ -2234,6 +3183,30 @@ void AliAnalysisAlien::SetPreferedSE(const char */*se*/) Warning("SetPreferedSE", "Setting a preferential SE is not allowed anymore via the plugin. Use SetNumberOfReplicas() and SetDefaultOutputs()"); } +//______________________________________________________________________________ +void AliAnalysisAlien::SetProofParameter(const char *pname, const char *value) +{ +// Set some PROOF special parameter. + TPair *pair = dynamic_cast(fProofParam.FindObject(pname)); + if (pair) { + TObject *old = pair->Key(); + TObject *val = pair->Value(); + fProofParam.Remove(old); + delete old; + delete val; + } + fProofParam.Add(new TObjString(pname), new TObjString(value)); +} + +//______________________________________________________________________________ +const char *AliAnalysisAlien::GetProofParameter(const char *pname) const +{ +// Returns a special PROOF parameter. + TPair *pair = dynamic_cast(fProofParam.FindObject(pname)); + if (!pair) return 0; + return pair->Value()->GetName(); +} + //______________________________________________________________________________ Bool_t AliAnalysisAlien::StartAnalysis(Long64_t /*nentries*/, Long64_t /*firstEntry*/) { @@ -2246,7 +3219,8 @@ Bool_t AliAnalysisAlien::StartAnalysis(Long64_t /*nentries*/, Long64_t /*firstEn } // Are we in PROOF mode ? if (mgr->IsProofMode()) { - Info("StartAnalysis", "##### Starting PROOF analysis on cluster <%s> via the plugin #####", fProofCluster.Data()); + if (testMode) Info("StartAnalysis", "##### Starting PROOF analysis with Proof Lite via the plugin #####"); + else Info("StartAnalysis", "##### Starting PROOF analysis on cluster <%s> via the plugin #####", fProofCluster.Data()); if (fProofCluster.IsNull()) { Error("StartAnalysis", "You need to specify the proof cluster name via SetProofCluster"); return kFALSE; @@ -2269,10 +3243,19 @@ Bool_t AliAnalysisAlien::StartAnalysis(Long64_t /*nentries*/, Long64_t /*firstEn Info("StartAnalysis", "Stopping the analysis. Please use SetProofReset(0) to resume."); return kFALSE; } + + if (!testMode) { + // Check if there is an old active session + Long_t nsessions = gROOT->ProcessLine(Form("TProof::Mgr(\"%s\")->QuerySessions(\"\")->GetEntries();", fProofCluster.Data())); + if (nsessions) { + Error("StartAnalysis","You have to reset your old session first\n"); + return kFALSE; + } + } // Do we need to change the ROOT version ? The success of this cannot be checked. - if (!fRootVersionForProof.IsNull() && !testMode) { - gROOT->ProcessLine(Form("TProof::Mgr(\"%s\")->SetROOTVersion(\"%s\");", - fProofCluster.Data(), fRootVersionForProof.Data())); + if (!fROOTVersion.IsNull() && !testMode) { + gROOT->ProcessLine(Form("TProof::Mgr(\"%s\")->SetROOTVersion(\"VO_ALICE@ROOT::%s\");", + fProofCluster.Data(), fROOTVersion.Data())); } // Connect to PROOF and check the status Long_t proof = 0; @@ -2297,6 +3280,13 @@ Bool_t AliAnalysisAlien::StartAnalysis(Long64_t /*nentries*/, Long64_t /*firstEn } if (fNproofWorkersPerSlave*fNproofWorkers > 0) gROOT->ProcessLine(Form("gProof->SetParallel(%d);", fNproofWorkers)); + // Set proof special parameters if any + TIter nextpp(&fProofParam); + TObject *proofparam; + while ((proofparam=nextpp())) { + TString svalue = GetProofParameter(proofparam->GetName()); + gROOT->ProcessLine(Form("gProof->SetParameter(\"%s\",%s);", proofparam->GetName(), svalue.Data())); + } // Is dataset existing ? if (!testMode) { TString dataset = fProofDataSet; @@ -2325,10 +3315,15 @@ Bool_t AliAnalysisAlien::StartAnalysis(Long64_t /*nentries*/, Long64_t /*firstEn // Check the additional libs to be loaded TString extraLibs; Bool_t parMode = kFALSE; - if (!alirootMode.IsNull()) extraLibs = "ANALYSIS:ANALYSISalice"; + if (!alirootMode.IsNull()) extraLibs = "ANALYSIS:OADB:ANALYSISalice"; // Parse the extra libs for .so if (fAdditionalLibs.Length()) { - TObjArray *list = fAdditionalLibs.Tokenize(" "); + TString additionalLibs = fAdditionalLibs; + additionalLibs.Strip(); + if (additionalLibs.Length() && fFriendLibs.Length()) + additionalLibs += " "; + additionalLibs += fFriendLibs; + TObjArray *list = additionalLibs.Tokenize(" "); TIter next(list); TObjString *str; while((str=(TObjString*)next())) { @@ -2354,12 +3349,18 @@ Bool_t AliAnalysisAlien::StartAnalysis(Long64_t /*nentries*/, Long64_t /*firstEn } if (list) delete list; } - if (!extraLibs.IsNull()) optionsList.Add(new TNamed("ALIROOT_EXTRA_LIBS",extraLibs.Data())); + if (!extraLibs.IsNull()) { + Info("StartAnalysis", "Adding extra libs: %s",extraLibs.Data()); + optionsList.Add(new TNamed("ALIROOT_EXTRA_LIBS",extraLibs.Data())); + } // Check extra includes if (!fIncludePath.IsNull()) { TString includePath = fIncludePath; includePath.ReplaceAll(" ",":"); - includePath.Strip(TString::kTrailing, ':'); + includePath.ReplaceAll("$ALICE_ROOT/",""); + includePath.ReplaceAll("${ALICE_ROOT}/",""); + includePath.ReplaceAll("-I",""); + includePath.Remove(TString::kTrailing, ':'); Info("StartAnalysis", "Adding extra includes: %s",includePath.Data()); optionsList.Add(new TNamed("ALIROOT_EXTRA_INCLUDES",includePath.Data())); } @@ -2375,18 +3376,20 @@ Bool_t AliAnalysisAlien::StartAnalysis(Long64_t /*nentries*/, Long64_t /*firstEn printf("%s %s\n", obj->GetName(), obj->GetTitle()); } if (!gROOT->ProcessLine(Form("gProof->UploadPackage(\"%s\");",alirootLite.Data())) - && !gROOT->ProcessLine(Form("gProof->EnablePackage(\"%s\", (TList*)0x%lx);",alirootLite.Data(),(ULong_t)&optionsList))) { + && !gROOT->ProcessLine(Form("gProof->EnablePackage(\"%s\", (TList*)%p);",alirootLite.Data(),&optionsList))) { Info("StartAnalysis", "AliRootProofLite enabled"); } else { Error("StartAnalysis", "There was an error trying to enable package AliRootProofLite.par"); return kFALSE; } } else { - if (gROOT->ProcessLine(Form("gProof->EnablePackage(\"VO_ALICE@AliRoot::%s\", (TList*)0x%lx);", - fAliROOTVersion.Data(), (ULong_t)&optionsList))) { - Error("StartAnalysis", "There was an error trying to enable package VO_ALICE@AliRoot::%s", fAliROOTVersion.Data()); - return kFALSE; - } + if ( ! fAliROOTVersion.IsNull() ) { + if (gROOT->ProcessLine(Form("gProof->EnablePackage(\"VO_ALICE@AliRoot::%s\", (TList*)%p, kTRUE);", + fAliROOTVersion.Data(), &optionsList))) { + Error("StartAnalysis", "There was an error trying to enable package VO_ALICE@AliRoot::%s", fAliROOTVersion.Data()); + return kFALSE; + } + } } // Enable first par files from fAdditionalLibs if (!parLibs.IsNull()) { @@ -2394,8 +3397,12 @@ Bool_t AliAnalysisAlien::StartAnalysis(Long64_t /*nentries*/, Long64_t /*firstEn TIter next(list); TObjString *package; while((package=(TObjString*)next())) { - if (gROOT->ProcessLine(Form("gProof->UploadPackage(\"%s\");", package->GetName()))) { - if (gROOT->ProcessLine(Form("gProof->EnablePackage(\"%s\",kTRUE);", package->GetName()))) { + TString spkg = package->GetName(); + spkg.ReplaceAll(".par", ""); + gSystem->Exec(TString::Format("rm -rf %s", spkg.Data())); + if (!gROOT->ProcessLine(Form("gProof->UploadPackage(\"%s\");", package->GetName()))) { + TString enablePackage = (testMode)?Form("gProof->EnablePackage(\"%s\",kFALSE);", package->GetName()):Form("gProof->EnablePackage(\"%s\",kTRUE);", package->GetName()); + if (gROOT->ProcessLine(enablePackage)) { Error("StartAnalysis", "There was an error trying to enable package %s", package->GetName()); return kFALSE; } @@ -2420,7 +3427,10 @@ Bool_t AliAnalysisAlien::StartAnalysis(Long64_t /*nentries*/, Long64_t /*firstEn while ((package=next())) { // Skip packages already enabled if (parLibs.Contains(package->GetName())) continue; - if (gROOT->ProcessLine(Form("gProof->UploadPackage(\"%s\");", package->GetName()))) { + TString spkg = package->GetName(); + spkg.ReplaceAll(".par", ""); + gSystem->Exec(TString::Format("rm -rf %s", spkg.Data())); + if (!gROOT->ProcessLine(Form("gProof->UploadPackage(\"%s\");", package->GetName()))) { if (gROOT->ProcessLine(Form("gProof->EnablePackage(\"%s\",kTRUE);", package->GetName()))) { Error("StartAnalysis", "There was an error trying to enable package %s", package->GetName()); return kFALSE; @@ -2439,7 +3449,7 @@ Bool_t AliAnalysisAlien::StartAnalysis(Long64_t /*nentries*/, Long64_t /*firstEn TObjString *str; while((str=(TObjString*)next())) { gROOT->ProcessLine(Form("gProof->Load(\"%s+g\", kTRUE);", str->GetName())); - } + } if (list) delete list; } if (testMode) { @@ -2454,7 +3464,7 @@ Bool_t AliAnalysisAlien::StartAnalysis(Long64_t /*nentries*/, Long64_t /*firstEn } TFileCollection *coll = new TFileCollection(); coll->AddFromFile(fFileForTestMode); - gROOT->ProcessLine(Form("gProof->RegisterDataSet(\"test_collection\", (TFileCollection*)0x%lx, \"OV\");", (ULong_t)coll)); + gROOT->ProcessLine(Form("gProof->RegisterDataSet(\"test_collection\", (TFileCollection*)%p, \"OV\");", coll)); gROOT->ProcessLine("gProof->ShowDataSets()"); } return kTRUE; @@ -2473,7 +3483,10 @@ Bool_t AliAnalysisAlien::StartAnalysis(Long64_t /*nentries*/, Long64_t /*firstEn } // Compose the output archive. fOutputArchive = "log_archive.zip:std*@disk=1 "; - fOutputArchive += Form("root_archive.zip:%s@disk=%d",fOutputFiles.Data(),fNreplicas); + if (mgr->IsCollectThroughput()) + fOutputArchive += Form("root_archive.zip:%s,*.stat@disk=%d %s@disk=%d",fOutputFiles.Data(),fNreplicas, mgr->GetFileInfoLog(),fNreplicas); + else + fOutputArchive += Form("root_archive.zip:%s,*.stat@disk=%d",fOutputFiles.Data(),fNreplicas); } // if (!fCloseSE.Length()) fCloseSE = gSystem->Getenv("alien_CLOSE_SE"); if (TestBit(AliAnalysisGrid::kOffline)) { @@ -2513,7 +3526,7 @@ Bool_t AliAnalysisAlien::StartAnalysis(Long64_t /*nentries*/, Long64_t /*firstEn Error("StartAnalysis", "No data to process. Please fix %s in your plugin configuration.", serror.Data()); return kFALSE; } - WriteAnalysisFile(); + WriteAnalysisFile(); WriteAnalysisMacro(); WriteExecutable(); WriteValidationScript(); @@ -2547,9 +3560,9 @@ Bool_t AliAnalysisAlien::StartAnalysis(Long64_t /*nentries*/, Long64_t /*firstEn } // Check if submitting is managed by LPM manager if (fProductionMode) { - TString prodfile = fJDLName; - prodfile.ReplaceAll(".jdl", ".prod"); - WriteProductionFile(prodfile); + //TString prodfile = fJDLName; + //prodfile.ReplaceAll(".jdl", ".prod"); + //WriteProductionFile(prodfile); Info("StartAnalysis", "Job submitting is managed by LPM. Rerun in terminate mode after jobs finished."); return kFALSE; } @@ -2557,6 +3570,8 @@ Bool_t AliAnalysisAlien::StartAnalysis(Long64_t /*nentries*/, Long64_t /*firstEn gGrid->Cd(fGridOutputDir); TGridResult *res; TString jobID = ""; + fGridJobIDs = ""; + fGridStages = ""; if (!fRunNumbers.Length() && !fRunRange[0]) { // Submit a given xml or a set of runs res = gGrid->Command(Form("submit %s", fJDLName.Data())); @@ -2574,6 +3589,12 @@ Bool_t AliAnalysisAlien::StartAnalysis(Long64_t /*nentries*/, Long64_t /*firstEn \n_______________________________________________________________________", fJDLName.Data(), cjobId); jobID = cjobId; + if (jobID.Atoi()) { + if (!fGridJobIDs.IsNull()) fGridJobIDs.Append(" "); + fGridJobIDs.Append(jobID); + if (!fGridStages.IsNull()) fGridStages.Append(" "); + fGridStages.Append("full"); + } } delete res; } else { @@ -2583,12 +3604,19 @@ Bool_t AliAnalysisAlien::StartAnalysis(Long64_t /*nentries*/, Long64_t /*firstEn } else { // Submit for a range of enumeration of runs. if (!Submit()) return kFALSE; + jobID = fGridJobIDs; } - Info("StartAnalysis", "\n#### STARTING AN ALIEN SHELL FOR YOU. EXIT WHEN YOUR JOB %s HAS FINISHED. #### \ - \n You may exit at any time and terminate the job later using the option \ - \n ##################################################################################", jobID.Data()); - gSystem->Exec("aliensh"); + if (fDropToShell) { + Info("StartAnalysis", "\n#### STARTING AN ALIEN SHELL FOR YOU. EXIT WHEN YOUR JOB %s HAS FINISHED. #### \ + \n You may exit at any time and terminate the job later using the option \ + \n ##################################################################################", jobID.Data()); + gSystem->Exec("aliensh"); + } else { + Info("StartAnalysis", "\n#### SUBMITTED JOB %s TO ALIEN QUEUE #### \ + \n Remember to terminate the job later using the option \ + \n ##################################################################################", jobID.Data()); + } return kTRUE; } @@ -2711,10 +3739,14 @@ Bool_t AliAnalysisAlien::Submit() Bool_t AliAnalysisAlien::SubmitMerging() { // Submit all merging jobs. - if (!fGridOutputDir.Contains("/")) fGridOutputDir = Form("/%s/%s/%s", gGrid->GetHomeDirectory(), fGridWorkingDir.Data(), fGridOutputDir.Data()); + if (!fGridOutputDir.Contains("/")) fGridOutputDir = Form("%s/%s/%s", gGrid->GetHomeDirectory(), fGridWorkingDir.Data(), fGridOutputDir.Data()); gGrid->Cd(fGridOutputDir); TString mergeJDLName = fExecutable; mergeJDLName.ReplaceAll(".sh", "_merge.jdl"); + if (!fInputFiles) { + Error("SubmitMerging", "You have to use explicit run numbers or run range to merge via JDL!"); + return kFALSE; + } Int_t ntosubmit = fInputFiles->GetEntries(); for (Int_t i=0; iBaseName(fInputFiles->At(i)->GetName()); @@ -2724,9 +3756,15 @@ Bool_t AliAnalysisAlien::SubmitMerging() printf("### Submitting merging job for run <%s>\n", runOutDir.Data()); runOutDir = Form("%s/%s", fGridOutputDir.Data(), runOutDir.Data()); } else { - // The output directory is the master number in 3 digits format - printf("### Submitting merging job for master <%03d>\n", i); - runOutDir = Form("%s/%03d",fGridOutputDir.Data(), i); + if (!fRunNumbers.Length() && !fRunRange[0]) { + // The output directory is the grid outdir + printf("### Submitting merging job for the full output directory %s.\n", fGridOutputDir.Data()); + runOutDir = fGridOutputDir; + } else { + // The output directory is the master number in 3 digits format + printf("### Submitting merging job for master <%03d>\n", i); + runOutDir = Form("%s/%03d",fGridOutputDir.Data(), i); + } } // Check now the number of merging stages. TObjArray *list = fOutputFiles.Tokenize(","); @@ -2737,17 +3775,29 @@ Bool_t AliAnalysisAlien::SubmitMerging() outputFile = str->GetString(); Int_t index = outputFile.Index("@"); if (index > 0) outputFile.Remove(index); - if (!fMergeExcludes.Contains(outputFile)) break; + if (!fMergeExcludes.Contains(outputFile) && + !fRegisterExcludes.Contains(outputFile)) break; } delete list; - Bool_t done = CheckMergedFiles(outputFile, runOutDir, fMaxMergeFiles, kTRUE, mergeJDLName); - if (!done) return kFALSE; + Bool_t done = CheckMergedFiles(outputFile, runOutDir, fMaxMergeFiles, mergeJDLName); + if (!done && (i==ntosubmit-1)) return kFALSE; + if (!fRunNumbers.Length() && !fRunRange[0]) break; } if (!ntosubmit) return kTRUE; - Info("StartAnalysis", "\n#### STARTING AN ALIEN SHELL FOR YOU. EXIT WHEN YOUR MERGING JOBS HAVE FINISHED. #### \ - \n You may exit at any time and terminate the job later using the option but disabling SetMergeViaJDL\ - \n ##################################################################################"); - gSystem->Exec("aliensh"); + if (fDropToShell) { + Info("StartAnalysis", "\n #### STARTING AN ALIEN SHELL FOR YOU. You can exit any time or inspect your jobs in a different shell.##########\ + \n Make sure your jobs are in a final state (you can resubmit failed ones via 'masterjob resubmit ERROR_ALL')\ + \n Rerun in 'terminate' mode to submit all merging stages, each AFTER the previous one completed. The final merged \ + \n output will be written to your alien output directory, while separate stages in . \ + \n ################################################################################################################"); + gSystem->Exec("aliensh"); + } else { + Info("StartAnalysis", "\n #### STARTED MERGING JOBS FOR YOU #### \ + \n Make sure your jobs are in a final state (you can resubmit failed ones via 'masterjob resubmit ERROR_ALL') \ + \n Rerun in 'terminate' mode to submit all merging stages, each AFTER the previous one completed. The final merged \ + \n output will be written to your alien output directory, while separate stages in . \ + \n ################################################################################################################"); + } return kTRUE; } @@ -2815,6 +3865,10 @@ Bool_t AliAnalysisAlien::SubmitNext() \n##### Your JDL %s submitted (%d to go). \nTHE JOB ID IS: %s \ \n_______________________________________________________________________", fJDLName.Data(), nmasterjobs-fNsubmitted-1, cjobId1.Data()); + if (!fGridJobIDs.IsNull()) fGridJobIDs.Append(" "); + fGridJobIDs.Append(cjobId1); + if (!fGridStages.IsNull()) fGridStages.Append(" "); + fGridStages.Append("full"); jobID += cjobId1; jobID += " "; lastmaster = cjobId1.Atoi(); @@ -2848,8 +3902,14 @@ void AliAnalysisAlien::WriteAnalysisFile() if (mgr->GetMCtruthEventHandler()) TObject::SetBit(AliAnalysisGrid::kUseMC); handler = (TObject*)mgr->GetInputEventHandler(); if (handler) { - if (handler->InheritsFrom("AliESDInputHandler")) TObject::SetBit(AliAnalysisGrid::kUseESD); - if (handler->InheritsFrom("AliAODInputHandler")) TObject::SetBit(AliAnalysisGrid::kUseAOD); + if (handler->InheritsFrom("AliMultiInputEventHandler")) { + AliMultiInputEventHandler *multiIH = (AliMultiInputEventHandler*)handler; + if (multiIH->GetFirstInputEventHandler()->InheritsFrom("AliESDInputHandler")) TObject::SetBit(AliAnalysisGrid::kUseESD); + if (multiIH->GetFirstInputEventHandler()->InheritsFrom("AliAODInputHandler")) TObject::SetBit(AliAnalysisGrid::kUseAOD); + } else { + if (handler->InheritsFrom("AliESDInputHandler")) TObject::SetBit(AliAnalysisGrid::kUseESD); + if (handler->InheritsFrom("AliAODInputHandler")) TObject::SetBit(AliAnalysisGrid::kUseAOD); + } } TDirectory *cdir = gDirectory; TFile *file = TFile::Open(analysisFile, "RECREATE"); @@ -2874,7 +3934,8 @@ void AliAnalysisAlien::WriteAnalysisFile() workdir += fGridWorkingDir; Info("WriteAnalysisFile", "\n##### Copying file <%s> containing your initialized analysis manager to your alien workspace", analysisFile.Data()); if (FileExists(analysisFile)) gGrid->Rm(analysisFile); - TFile::Cp(Form("file:%s",analysisFile.Data()), Form("alien://%s/%s", workdir.Data(),analysisFile.Data())); + if (!copyLocal2Alien("WriteAnalysisFile",analysisFile.Data(), + Form("%s/%s", workdir.Data(),analysisFile.Data()))) Fatal("","Terminating"); } } @@ -2893,16 +3954,22 @@ void AliAnalysisAlien::WriteAnalysisMacro() Bool_t hasESD = kFALSE; Bool_t hasAOD = kFALSE; Bool_t hasANALYSIS = kFALSE; + Bool_t hasOADB = kFALSE; Bool_t hasANALYSISalice = kFALSE; Bool_t hasCORRFW = kFALSE; TString func = fAnalysisMacro; TString type = "ESD"; TString comment = "// Analysis using "; - if (TObject::TestBit(AliAnalysisGrid::kUseESD)) comment += "ESD"; - if (TObject::TestBit(AliAnalysisGrid::kUseAOD)) { - type = "AOD"; - comment += "AOD"; - } + if (IsUseMCchain()) { + type = "MC"; + comment += "MC"; + } else { + if (TObject::TestBit(AliAnalysisGrid::kUseESD)) comment += "ESD"; + if (TObject::TestBit(AliAnalysisGrid::kUseAOD)) { + type = "AOD"; + comment += "AOD"; + } + } if (type!="AOD" && fFriendChainName!="") { Error("WriteAnalysisMacro", "Friend chain can be attached only to AOD"); return; @@ -2918,8 +3985,17 @@ void AliAnalysisAlien::WriteAnalysisMacro() out << " TStopwatch timer;" << endl; out << " timer.Start();" << endl << endl; // Change temp directory to current one + if (!IsLocalTest()) { + out << "// connect to AliEn and make the chain" << endl; + out << " if (!TGrid::Connect(\"alien://\")) return;" << endl; + } out << "// Set temporary merging directory to current one" << endl; out << " gSystem->Setenv(\"TMPDIR\", gSystem->pwd());" << endl << endl; + out << "// Set temporary compilation directory to current one" << endl; + out << " gSystem->SetBuildDir(gSystem->pwd(), kTRUE);" << endl << endl; + // Reset existing include path + out << "// Reset existing include path and add current directory first in the search" << endl; + out << " gSystem->SetIncludePath(\"-I.\");" << endl; if (!fExecutableCommand.Contains("aliroot")) { out << "// load base root libraries" << endl; out << " gSystem->Load(\"libTree\");" << endl; @@ -2940,9 +4016,6 @@ void AliAnalysisAlien::WriteAnalysisMacro() } if (list) delete list; } - out << "// include path" << endl; - if (fIncludePath.Length()) out << " gSystem->AddIncludePath(\"" << fIncludePath.Data() << "\");" << endl; - out << " gSystem->AddIncludePath(\"-I$ALICE_ROOT/include\");" << endl << endl; out << "// Load analysis framework libraries" << endl; TString setupPar = "AliAnalysisAlien::SetupPar"; if (!fPackages) { @@ -2952,6 +4025,7 @@ void AliAnalysisAlien::WriteAnalysisMacro() out << " gSystem->Load(\"libAOD\");" << endl; } out << " gSystem->Load(\"libANALYSIS\");" << endl; + out << " gSystem->Load(\"libOADB\");" << endl; out << " gSystem->Load(\"libANALYSISalice\");" << endl; out << " gSystem->Load(\"libCORRFW\");" << endl << endl; } else { @@ -2968,6 +4042,8 @@ void AliAnalysisAlien::WriteAnalysisMacro() pkgname == "AOD.par") hasAOD = kTRUE; if (pkgname == "ANALYSIS" || pkgname == "ANALYSIS.par") hasANALYSIS = kTRUE; + if (pkgname == "OADB" || + pkgname == "OADB.par") hasOADB = kTRUE; if (pkgname == "ANALYSISalice" || pkgname == "ANALYSISalice.par") hasANALYSISalice = kTRUE; if (pkgname == "CORRFW" || @@ -2982,6 +4058,8 @@ void AliAnalysisAlien::WriteAnalysisMacro() else out << " if (!" << setupPar << "(\"AOD\")) return;" << endl; if (!hasANALYSIS) out << " gSystem->Load(\"libANALYSIS\");" << endl; else out << " if (!" << setupPar << "(\"ANALYSIS\")) return;" << endl; + if (!hasOADB) out << " gSystem->Load(\"libOADB\");" << endl; + else out << " if (!" << setupPar << "(\"OADB\")) return;" << endl; if (!hasANALYSISalice) out << " gSystem->Load(\"libANALYSISalice\");" << endl; else out << " if (!" << setupPar << "(\"ANALYSISalice\")) return;" << endl; if (!hasCORRFW) out << " gSystem->Load(\"libCORRFW\");" << endl << endl; @@ -2998,6 +4076,8 @@ void AliAnalysisAlien::WriteAnalysisMacro() pkgname == "AOD.par" || pkgname == "ANALYSIS" || pkgname == "ANALYSIS.par" || + pkgname == "OADB" || + pkgname == "OADB.par" || pkgname == "ANALYSISalice" || pkgname == "ANALYSISalice.par" || pkgname == "CORRFW" || @@ -3005,9 +4085,29 @@ void AliAnalysisAlien::WriteAnalysisMacro() out << " if (!" << setupPar << "(\"" << obj->GetName() << "\")) return;" << endl; } } + out << "// include path" << endl; + // Get the include path from the interpreter and remove entries pointing to AliRoot + out << " TString intPath = gInterpreter->GetIncludePath();" << endl; + out << " TObjArray *listpaths = intPath.Tokenize(\" \");" << endl; + out << " TIter nextpath(listpaths);" << endl; + out << " TObjString *pname;" << endl; + out << " while ((pname=(TObjString*)nextpath())) {" << endl; + out << " TString current = pname->GetName();" << endl; + out << " if (current.Contains(\"AliRoot\") || current.Contains(\"ALICE_ROOT\")) continue;" << endl; + out << " gSystem->AddIncludePath(current);" << endl; + out << " }" << endl; + out << " if (listpaths) delete listpaths;" << endl; + if (fIncludePath.Length()) out << " gSystem->AddIncludePath(\"" << fIncludePath.Data() << "\");" << endl; + out << " gROOT->ProcessLine(\".include $ALICE_ROOT/include\");" << endl; + out << " printf(\"Include path: %s\\n\", gSystem->GetIncludePath());" << endl << endl; if (fAdditionalLibs.Length()) { out << "// Add aditional AliRoot libraries" << endl; - TObjArray *list = fAdditionalLibs.Tokenize(" "); + TString additionalLibs = fAdditionalLibs; + additionalLibs.Strip(); + if (additionalLibs.Length() && fFriendLibs.Length()) + additionalLibs += " "; + additionalLibs += fFriendLibs; + TObjArray *list = additionalLibs.Tokenize(" "); TIter next(list); TObjString *str; while((str=(TObjString*)next())) { @@ -3030,35 +4130,46 @@ void AliAnalysisAlien::WriteAnalysisMacro() if (list) delete list; } out << endl; +// out << " printf(\"Currently load libraries:\\n\");" << endl; +// out << " printf(\"%s\\n\", gSystem->GetLibraries());" << endl; if (fFastReadOption) { Warning("WriteAnalysisMacro", "!!! You requested FastRead option. Using xrootd flags to reduce timeouts in the grid jobs. This may skip some files that could be accessed !!! \ \n+++ NOTE: To disable this option, use: plugin->SetFastReadOption(kFALSE)"); out << "// fast xrootd reading enabled" << endl; out << " printf(\"!!! You requested FastRead option. Using xrootd flags to reduce timeouts. Note that this may skip some files that could be accessed !!!\");" << endl; - out << " gEnv->SetValue(\"XNet.ConnectTimeout\",10);" << endl; - out << " gEnv->SetValue(\"XNet.RequestTimeout\",10);" << endl; + out << " gEnv->SetValue(\"XNet.ConnectTimeout\",50);" << endl; + out << " gEnv->SetValue(\"XNet.RequestTimeout\",50);" << endl; out << " gEnv->SetValue(\"XNet.MaxRedirectCount\",2);" << endl; - out << " gEnv->SetValue(\"XNet.ReconnectTimeout\",10);" << endl; + out << " gEnv->SetValue(\"XNet.ReconnectTimeout\",50);" << endl; out << " gEnv->SetValue(\"XNet.FirstConnectMaxCnt\",1);" << endl << endl; - } - out << "// connect to AliEn and make the chain" << endl; - out << " if (!TGrid::Connect(\"alien://\")) return;" << endl; + } out << "// read the analysis manager from file" << endl; TString analysisFile = fExecutable; analysisFile.ReplaceAll(".sh", ".root"); - out << " TFile *file = TFile::Open(\"" << analysisFile << "\");" << endl; - out << " if (!file) return;" << endl; - out << " TIter nextkey(file->GetListOfKeys());" << endl; - out << " AliAnalysisManager *mgr = 0;" << endl; - out << " TKey *key;" << endl; - out << " while ((key=(TKey*)nextkey())) {" << endl; - out << " if (!strcmp(key->GetClassName(), \"AliAnalysisManager\"))" << endl; - out << " mgr = (AliAnalysisManager*)file->Get(key->GetName());" << endl; - out << " };" << endl; - out << " if (!mgr) {" << endl; - out << " ::Error(\"" << func.Data() << "\", \"No analysis manager found in file " << analysisFile <<"\");" << endl; - out << " return;" << endl; - out << " }" << endl << endl; + out << " AliAnalysisManager *mgr = AliAnalysisAlien::LoadAnalysisManager(\"" + << analysisFile << "\");" << endl; + out << " if (!mgr) return;" << endl; + if (IsLocalTest()) { + out << " AliAnalysisAlien *plugin = new AliAnalysisAlien();" << endl; + out << " plugin->SetRunMode(\"test\");" << endl; + if (fFileForTestMode.IsNull()) + out << " plugin->SetFileForTestMode(\"data.txt\");" << endl; + else + out << " plugin->SetFileForTestMode(\"" << fFileForTestMode << "\");" << endl; + out << " plugin->SetNtestFiles(" << fNtestFiles << ");" << endl; + if (!fFriendChainName.IsNull()) + out << " plugin->SetFriendChainName(\"" << fFriendChainName << "\",\"" << fFriendLibs << "\");" << endl; + if (IsUseMCchain()) + out << " plugin->SetUseMCchain();" << endl; + out << " mgr->SetGridHandler(plugin);" << endl; + if (AliAnalysisManager::GetAnalysisManager()) { + out << " mgr->SetDebugLevel(" << AliAnalysisManager::GetAnalysisManager()->GetDebugLevel() << ");" << endl; + out << " mgr->SetNSysInfo(" << AliAnalysisManager::GetAnalysisManager()->GetNsysInfo() << ");" << endl; + } else { + out << " mgr->SetDebugLevel(10);" << endl; + out << " mgr->SetNSysInfo(100);" << endl; + } + } out << " mgr->PrintStatus();" << endl; if (AliAnalysisManager::GetAnalysisManager()) { if (AliAnalysisManager::GetAnalysisManager()->GetDebugLevel()>3) { @@ -3070,66 +4181,33 @@ void AliAnalysisAlien::WriteAnalysisMacro() out << " AliLog::SetGlobalLogLevel(AliLog::kError);" << endl; } } - if (IsUsingTags()) { - out << " TChain *chain = CreateChainFromTags(\"wn.xml\", anatype);" << endl << endl; - } else { + if (!IsLocalTest()) { out << " TChain *chain = CreateChain(\"wn.xml\", anatype);" << endl << endl; + out << " mgr->StartAnalysis(\"localfile\", chain);" << endl; + } else { + out << " mgr->StartAnalysis(\"localfile\");" << endl; } - out << " mgr->StartAnalysis(\"localfile\", chain);" << endl; out << " timer.Stop();" << endl; out << " timer.Print();" << endl; out << "}" << endl << endl; - if (IsUsingTags()) { - out << "TChain* CreateChainFromTags(const char *xmlfile, const char *type=\"ESD\")" << endl; - out << "{" << endl; - out << "// Create a chain using tags from the xml file." << endl; - out << " TAlienCollection* coll = TAlienCollection::Open(xmlfile);" << endl; - out << " if (!coll) {" << endl; - out << " ::Error(\"CreateChainFromTags\", \"Cannot create an AliEn collection from %s\", xmlfile);" << endl; - out << " return NULL;" << endl; - out << " }" << endl; - out << " TGridResult* tagResult = coll->GetGridResult(\"\",kFALSE,kFALSE);" << endl; - out << " AliTagAnalysis *tagAna = new AliTagAnalysis(type);" << endl; - out << " tagAna->ChainGridTags(tagResult);" << endl << endl; - out << " AliRunTagCuts *runCuts = new AliRunTagCuts();" << endl; - out << " AliLHCTagCuts *lhcCuts = new AliLHCTagCuts();" << endl; - out << " AliDetectorTagCuts *detCuts = new AliDetectorTagCuts();" << endl; - out << " AliEventTagCuts *evCuts = new AliEventTagCuts();" << endl; - out << " // Check if the cuts configuration file was provided" << endl; - out << " if (!gSystem->AccessPathName(\"ConfigureCuts.C\")) {" << endl; - out << " gROOT->LoadMacro(\"ConfigureCuts.C\");" << endl; - out << " ConfigureCuts(runCuts, lhcCuts, detCuts, evCuts);" << endl; - out << " }" << endl; - if (fFriendChainName=="") { - out << " TChain *chain = tagAna->QueryTags(runCuts, lhcCuts, detCuts, evCuts);" << endl; - } else { - out << " TString tmpColl=\"tmpCollection.xml\";" << endl; - out << " tagAna->CreateXMLCollection(tmpColl.Data(),runCuts, lhcCuts, detCuts, evCuts);" << endl; - out << " TChain *chain = CreateChain(tmpColl.Data(),type);" << endl; - } - out << " if (!chain || !chain->GetNtrees()) return NULL;" << endl; - out << " chain->ls();" << endl; - out << " return chain;" << endl; - out << "}" << endl << endl; - if (gSystem->AccessPathName("ConfigureCuts.C")) { - TString msg = "\n##### You may want to provide a macro ConfigureCuts.C with a method:\n"; - msg += " void ConfigureCuts(AliRunTagCuts *runCuts,\n"; - msg += " AliLHCTagCuts *lhcCuts,\n"; - msg += " AliDetectorTagCuts *detCuts,\n"; - msg += " AliEventTagCuts *evCuts)"; - Info("WriteAnalysisMacro", "%s", msg.Data()); - } - } - if (!IsUsingTags() || fFriendChainName!="") { + if (!IsLocalTest()) { out <<"//________________________________________________________________________________" << endl; out << "TChain* CreateChain(const char *xmlfile, const char *type=\"ESD\")" << endl; out << "{" << endl; out << "// Create a chain using url's from xml file" << endl; out << " TString filename;" << endl; out << " Int_t run = 0;" << endl; - out << " TString treename = type;" << endl; - out << " treename.ToLower();" << endl; - out << " treename += \"Tree\";" << endl; + if (IsUseMCchain()) { + out << " TString treename = \"TE\";" << endl; + } else { + if (!fTreeName.IsNull()) { + out << " TString treename = \"" << fTreeName << "\";" << endl; + } else { + out << " TString treename = type;" << endl; + out << " treename.ToLower();" << endl; + out << " treename += \"Tree\";" << endl; + } + } out << " printf(\"***************************************\\n\");" << endl; out << " printf(\" Getting chain of trees %s\\n\", treename.Data());" << endl; out << " printf(\"***************************************\\n\");" << endl; @@ -3140,8 +4218,20 @@ void AliAnalysisAlien::WriteAnalysisMacro() out << " }" << endl; out << " AliAnalysisManager *mgr = AliAnalysisManager::GetAnalysisManager();" << endl; out << " TChain *chain = new TChain(treename);" << endl; - if(fFriendChainName!="") { - out << " TChain *chainFriend = new TChain(treename);" << endl; + if(!fFriendChainName.IsNull()) { + out << " TList *friends = new TList();" << endl; + out << " TIter nextfriend(friends);" << endl; + out << " TChain *cfriend = 0;" << endl; + TObjArray *list = fFriendChainName.Tokenize(" "); + TIter next(list); + TObjString *str; + while((str=(TObjString*)next())) { + out << " cfriend = new TChain(treename, \"" << str->GetName() << "\");" << endl; + out << " friends->Add(cfriend);" << endl; + out << " chain->AddFriend(cfriend);" << endl; + } + delete list; +// out << " TChain *chainFriend = new TChain(treename);" << endl; } out << " coll->Reset();" << endl; out << " while (coll->Next()) {" << endl; @@ -3155,20 +4245,31 @@ void AliAnalysisAlien::WriteAnalysisMacro() out << " }" << endl; out << " }" << endl; out << " chain->Add(filename);" << endl; - if(fFriendChainName!="") { - out << " TString fileFriend=coll->GetTURL(\"\");" << endl; - out << " fileFriend.ReplaceAll(\"AliAOD.root\",\""<Add(fileFriend.Data());" << endl; + if(!fFriendChainName.IsNull()) { + out << " TString bpath=coll->GetTURL(\"\");" << endl; + out << " if (bpath.Index(\"#\") > -1) bpath.Remove(bpath.Index(\"#\"));" << endl; + out << " bpath = gSystem->DirName(bpath);" << endl; + out << " bpath += \"/\";" << endl; + out << " TString fileFriend;" << endl; + out << " nextfriend.Reset();" << endl; + out << " while ((cfriend=(TChain*)nextfriend())) {" << endl; + out << " fileFriend = bpath;" << endl; + out << " fileFriend += cfriend->GetTitle();" << endl; + out << " TFile *file = TFile::Open(fileFriend);" << endl; + out << " if (file) {" << endl; + out << " file->Close();" << endl; + out << " cfriend->Add(fileFriend.Data());" << endl; + out << " } else {" << endl; + out << " ::Fatal(\"CreateChain\", \"Cannot open friend file: %s\", fileFriend.Data());" << endl; + out << " return 0;" << endl; + out << " }" << endl; + out << " }" << endl; } out << " }" << endl; out << " if (!chain->GetNtrees()) {" << endl; out << " ::Error(\"CreateChain\", \"No tree found from collection %s\", xmlfile);" << endl; out << " return NULL;" << endl; out << " }" << endl; - if(fFriendChainName!="") { - out << " chain->AddFriend(chainFriend);" << endl; - } out << " return chain;" << endl; out << "}" << endl << endl; } @@ -3178,7 +4279,7 @@ void AliAnalysisAlien::WriteAnalysisMacro() out << "// Compile the package and set it up." << endl; out << " TString pkgdir = package;" << endl; out << " pkgdir.ReplaceAll(\".par\",\"\");" << endl; - out << " gSystem->Exec(Form(\"tar xvzf %s.par\", pkgdir.Data()));" << endl; + out << " gSystem->Exec(TString::Format(\"tar xvzf %s.par\", pkgdir.Data()));" << endl; out << " TString cdir = gSystem->WorkingDirectory();" << endl; out << " gSystem->ChangeDirectory(pkgdir);" << endl; out << " // Check for BUILD.sh and execute" << endl; @@ -3221,13 +4322,11 @@ void AliAnalysisAlien::WriteAnalysisMacro() TString workdir = gGrid->GetHomeDirectory(); workdir += fGridWorkingDir; if (FileExists(fAnalysisMacro)) gGrid->Rm(fAnalysisMacro); - if (IsUsingTags() && !gSystem->AccessPathName("ConfigureCuts.C")) { - if (FileExists("ConfigureCuts.C")) gGrid->Rm("ConfigureCuts.C"); - Info("WriteAnalysisMacro", "\n##### Copying cuts configuration macro: to your alien workspace"); - TFile::Cp("file:ConfigureCuts.C", Form("alien://%s/ConfigureCuts.C", workdir.Data())); - } Info("WriteAnalysisMacro", "\n##### Copying analysis macro: <%s> to your alien workspace", fAnalysisMacro.Data()); - TFile::Cp(Form("file:%s",fAnalysisMacro.Data()), Form("alien://%s/%s", workdir.Data(), fAnalysisMacro.Data())); +// TFile::Cp(Form("file:%s",fAnalysisMacro.Data()), Form("alien://%s/%s", workdir.Data(), fAnalysisMacro.Data())); + if (!copyLocal2Alien("WriteAnalysisMacro",fAnalysisMacro.Data(), + Form("alien://%s/%s", workdir.Data(), + fAnalysisMacro.Data()))) Fatal("","Terminating"); } } @@ -3240,9 +4339,10 @@ void AliAnalysisAlien::WriteMergingMacro() Error("WriteMergingMacro", "No output file names defined. Are you running the right AliAnalysisAlien configuration ?"); return; } + AliAnalysisManager *mgr = AliAnalysisManager::GetAnalysisManager(); TString mergingMacro = fExecutable; mergingMacro.ReplaceAll(".sh","_merge.C"); - if (!fGridOutputDir.Contains("/")) fGridOutputDir = Form("/%s/%s/%s", gGrid->GetHomeDirectory(), fGridWorkingDir.Data(), fGridOutputDir.Data()); + if (gGrid && !fGridOutputDir.Contains("/")) fGridOutputDir = Form("%s/%s/%s", gGrid->GetHomeDirectory(), fGridWorkingDir.Data(), fGridOutputDir.Data()); if (!TestBit(AliAnalysisGrid::kSubmit)) { ofstream out; out.open(mergingMacro.Data(), ios::out); @@ -3254,16 +4354,20 @@ void AliAnalysisAlien::WriteMergingMacro() Bool_t hasESD = kFALSE; Bool_t hasAOD = kFALSE; Bool_t hasANALYSIS = kFALSE; + Bool_t hasOADB = kFALSE; Bool_t hasANALYSISalice = kFALSE; Bool_t hasCORRFW = kFALSE; TString func = mergingMacro; TString comment; func.ReplaceAll(".C", ""); - out << "void " << func.Data() << "(const char *dir, Int_t stage=0, Int_t ichunk=0)" << endl; + out << "void " << func.Data() << "(const char *dir, Int_t stage=0)" << endl; out << "{" << endl; out << "// Automatically generated merging macro executed in grid subjobs" << endl << endl; out << " TStopwatch timer;" << endl; out << " timer.Start();" << endl << endl; + // Reset existing include path + out << "// Reset existing include path and add current directory first in the search" << endl; + out << " gSystem->SetIncludePath(\"-I.\");" << endl; if (!fExecutableCommand.Contains("aliroot")) { out << "// load base root libraries" << endl; out << " gSystem->Load(\"libTree\");" << endl; @@ -3284,9 +4388,6 @@ void AliAnalysisAlien::WriteMergingMacro() } if (list) delete list; } - out << "// include path" << endl; - if (fIncludePath.Length()) out << " gSystem->AddIncludePath(\"" << fIncludePath.Data() << "\");" << endl; - out << " gSystem->AddIncludePath(\"-I$ALICE_ROOT/include\");" << endl << endl; out << "// Load analysis framework libraries" << endl; if (!fPackages) { if (!fExecutableCommand.Contains("aliroot")) { @@ -3295,6 +4396,7 @@ void AliAnalysisAlien::WriteMergingMacro() out << " gSystem->Load(\"libAOD\");" << endl; } out << " gSystem->Load(\"libANALYSIS\");" << endl; + out << " gSystem->Load(\"libOADB\");" << endl; out << " gSystem->Load(\"libANALYSISalice\");" << endl; out << " gSystem->Load(\"libCORRFW\");" << endl << endl; } else { @@ -3312,6 +4414,8 @@ void AliAnalysisAlien::WriteMergingMacro() pkgname == "AOD.par") hasAOD = kTRUE; if (pkgname == "ANALYSIS" || pkgname == "ANALYSIS.par") hasANALYSIS = kTRUE; + if (pkgname == "OADB" || + pkgname == "OADB.par") hasOADB = kTRUE; if (pkgname == "ANALYSISalice" || pkgname == "ANALYSISalice.par") hasANALYSISalice = kTRUE; if (pkgname == "CORRFW" || @@ -3324,8 +4428,11 @@ void AliAnalysisAlien::WriteMergingMacro() else out << " if (!" << setupPar << "(\"ESD\")) return;" << endl; if (!hasAOD) out << " gSystem->Load(\"libAOD\");" << endl; else out << " if (!" << setupPar << "(\"AOD\")) return;" << endl; + out << " gSystem->Load(\"libOADB\");" << endl; if (!hasANALYSIS) out << " gSystem->Load(\"libANALYSIS\");" << endl; else out << " if (!" << setupPar << "(\"ANALYSIS\")) return;" << endl; + if (!hasOADB) out << " gSystem->Load(\"libOADB\");" << endl; + else out << " if (!" << setupPar << "(\"OADB\")) return;" << endl; if (!hasANALYSISalice) out << " gSystem->Load(\"libANALYSISalice\");" << endl; else out << " if (!" << setupPar << "(\"ANALYSISalice\")) return;" << endl; if (!hasCORRFW) out << " gSystem->Load(\"libCORRFW\");" << endl << endl; @@ -3342,6 +4449,8 @@ void AliAnalysisAlien::WriteMergingMacro() pkgname == "AOD.par" || pkgname == "ANALYSIS" || pkgname == "ANALYSIS.par" || + pkgname == "OADB" || + pkgname == "OADB.par" || pkgname == "ANALYSISalice" || pkgname == "ANALYSISalice.par" || pkgname == "CORRFW" || @@ -3349,9 +4458,29 @@ void AliAnalysisAlien::WriteMergingMacro() out << " if (!" << setupPar << "(\"" << obj->GetName() << "\")) return;" << endl; } } + out << "// include path" << endl; + // Get the include path from the interpreter and remove entries pointing to AliRoot + out << " TString intPath = gInterpreter->GetIncludePath();" << endl; + out << " TObjArray *listpaths = intPath.Tokenize(\" \");" << endl; + out << " TIter nextpath(listpaths);" << endl; + out << " TObjString *pname;" << endl; + out << " while ((pname=(TObjString*)nextpath())) {" << endl; + out << " TString current = pname->GetName();" << endl; + out << " if (current.Contains(\"AliRoot\") || current.Contains(\"ALICE_ROOT\")) continue;" << endl; + out << " gSystem->AddIncludePath(current);" << endl; + out << " }" << endl; + out << " if (listpaths) delete listpaths;" << endl; + if (fIncludePath.Length()) out << " gSystem->AddIncludePath(\"" << fIncludePath.Data() << "\");" << endl; + out << " gROOT->ProcessLine(\".include $ALICE_ROOT/include\");" << endl; + out << " printf(\"Include path: %s\\n\", gSystem->GetIncludePath());" << endl << endl; if (fAdditionalLibs.Length()) { out << "// Add aditional AliRoot libraries" << endl; - TObjArray *list = fAdditionalLibs.Tokenize(" "); + TString additionalLibs = fAdditionalLibs; + additionalLibs.Strip(); + if (additionalLibs.Length() && fFriendLibs.Length()) + additionalLibs += " "; + additionalLibs += fFriendLibs; + TObjArray *list = additionalLibs.Tokenize(" "); TIter next(list); TObjString *str; while((str=(TObjString*)next())) { @@ -3377,26 +4506,40 @@ void AliAnalysisAlien::WriteMergingMacro() Warning("WriteMergingMacro", "!!! You requested FastRead option. Using xrootd flags to reduce timeouts in the grid merging jobs. Note that this may skip some files that could be accessed !!!"); out << "// fast xrootd reading enabled" << endl; out << " printf(\"!!! You requested FastRead option. Using xrootd flags to reduce timeouts. Note that this may skip some files that could be accessed !!!\");" << endl; - out << " gEnv->SetValue(\"XNet.ConnectTimeout\",10);" << endl; - out << " gEnv->SetValue(\"XNet.RequestTimeout\",10);" << endl; + out << " gEnv->SetValue(\"XNet.ConnectTimeout\",50);" << endl; + out << " gEnv->SetValue(\"XNet.RequestTimeout\",50);" << endl; out << " gEnv->SetValue(\"XNet.MaxRedirectCount\",2);" << endl; - out << " gEnv->SetValue(\"XNet.ReconnectTimeout\",10);" << endl; + out << " gEnv->SetValue(\"XNet.ReconnectTimeout\",50);" << endl; out << " gEnv->SetValue(\"XNet.FirstConnectMaxCnt\",1);" << endl << endl; } // Change temp directory to current one - out << "// Set temporary merging directory to current one" << endl; - out << " gSystem->Setenv(\"TMPDIR\", gSystem->pwd());" << endl << endl; out << "// Connect to AliEn" << endl; out << " if (!TGrid::Connect(\"alien://\")) return;" << endl; - out << " Bool_t laststage = kFALSE;" << endl; + out << "// Set temporary merging directory to current one" << endl; + out << " gSystem->Setenv(\"TMPDIR\", gSystem->pwd());" << endl << endl; + out << "// Set temporary compilation directory to current one" << endl; + out << " gSystem->SetBuildDir(gSystem->pwd(), kTRUE);" << endl << endl; out << " TString outputDir = dir;" << endl; out << " TString outputFiles = \"" << GetListOfFiles("out") << "\";" << endl; - out << " TString mergeExcludes = \"" << fMergeExcludes << "\";" << endl; + out << " TString mergeExcludes = \"" << fMergeExcludes << " " << fRegisterExcludes << "\";" << endl; out << " TObjArray *list = outputFiles.Tokenize(\",\");" << endl; out << " TIter *iter = new TIter(list);" << endl; out << " TObjString *str;" << endl; out << " TString outputFile;" << endl; out << " Bool_t merged = kTRUE;" << endl; + TString analysisFile = fExecutable; + analysisFile.ReplaceAll(".sh", ".root"); + out << " AliAnalysisManager *mgr = AliAnalysisAlien::LoadAnalysisManager(\"" + << analysisFile << "\");" << endl; + out << " if (!mgr) {" << endl; + out << " printf(\"ERROR: Analysis manager could not be extracted from file \");" << endl; + out << " return;" << endl; + out << " }" << endl; + if (IsLocalTest()) { + out << " printf(\"===================================\n\");" << endl; + out << " printf(\"Testing merging...\\n\");" << endl; + out << " printf(\"===================================\n\");" << endl; + } out << " while((str=(TObjString*)iter->Next())) {" << endl; out << " outputFile = str->GetString();" << endl; out << " if (outputFile.Contains(\"*\")) continue;" << endl; @@ -3404,44 +4547,37 @@ void AliAnalysisAlien::WriteMergingMacro() out << " if (index > 0) outputFile.Remove(index);" << endl; out << " // Skip already merged outputs" << endl; out << " if (!gSystem->AccessPathName(outputFile)) {" << endl; - out << " printf(\"Output file <%s> found. Not merging again.\",outputFile.Data());" << endl; + out << " printf(\"Output file <%s> found. Not merging again.\\n\",outputFile.Data());" << endl; out << " continue;" << endl; out << " }" << endl; out << " if (mergeExcludes.Contains(outputFile.Data())) continue;" << endl; - out << " merged = AliAnalysisAlien::MergeOutput(outputFile, outputDir, " << fMaxMergeFiles << ", stage, ichunk);" << endl; + out << " merged = AliAnalysisAlien::MergeOutput(outputFile, outputDir, " << fMaxMergeFiles << ", stage);" << endl; out << " if (!merged) {" << endl; out << " printf(\"ERROR: Cannot merge %s\\n\", outputFile.Data());" << endl; out << " return;" << endl; out << " }" << endl; - out << " // Check if this was the last stage. If yes, run terminate for the tasks." << endl; - out << " if (!gSystem->AccessPathName(outputFile)) laststage = kTRUE;" << endl; out << " }" << endl; + if (mgr && mgr->IsCollectThroughput() && !IsLocalTest()) { + out << " TString infolog = \"" << mgr->GetFileInfoLog() << "\";" << endl; + out << " AliAnalysisAlien::MergeInfo(infolog, outputDir);" << endl; + } out << " // all outputs merged, validate" << endl; out << " ofstream out;" << endl; out << " out.open(\"outputs_valid\", ios::out);" << endl; out << " out.close();" << endl; out << " // read the analysis manager from file" << endl; - TString analysisFile = fExecutable; - analysisFile.ReplaceAll(".sh", ".root"); - out << " if (!laststage) return;" << endl; - out << " TFile *file = TFile::Open(\"" << analysisFile << "\");" << endl; - out << " if (!file) return;" << endl; - out << " TIter nextkey(file->GetListOfKeys());" << endl; - out << " AliAnalysisManager *mgr = 0;" << endl; - out << " TKey *key;" << endl; - out << " while ((key=(TKey*)nextkey())) {" << endl; - out << " if (!strcmp(key->GetClassName(), \"AliAnalysisManager\"))" << endl; - out << " mgr = (AliAnalysisManager*)file->Get(key->GetName());" << endl; - out << " };" << endl; - out << " if (!mgr) {" << endl; - out << " ::Error(\"" << func.Data() << "\", \"No analysis manager found in file" << analysisFile <<"\");" << endl; - out << " return;" << endl; - out << " }" << endl << endl; + if (IsLocalTest()) { + out << " printf(\"===================================\n\");" << endl; + out << " printf(\"Testing Terminate()...\\n\");" << endl; + out << " printf(\"===================================\n\");" << endl; + } else { + out << " if (!outputDir.Contains(\"Stage\")) return;" << endl; + } out << " mgr->SetRunFromPath(mgr->GetRunFromAlienPath(dir));" << endl; out << " mgr->SetSkipTerminate(kFALSE);" << endl; out << " mgr->PrintStatus();" << endl; - if (AliAnalysisManager::GetAnalysisManager()) { - if (AliAnalysisManager::GetAnalysisManager()->GetDebugLevel()>3) { + if (mgr) { + if (mgr->GetDebugLevel()>3) { out << " gEnv->SetValue(\"XNet.Debug\", \"1\");" << endl; } else { if (TestBit(AliAnalysisGrid::kTest)) @@ -3459,7 +4595,7 @@ void AliAnalysisAlien::WriteMergingMacro() out << "// Compile the package and set it up." << endl; out << " TString pkgdir = package;" << endl; out << " pkgdir.ReplaceAll(\".par\",\"\");" << endl; - out << " gSystem->Exec(Form(\"tar xvzf %s.par\", pkgdir.Data()));" << endl; + out << " gSystem->Exec(TString::Format(\"tar xvzf %s.par\", pkgdir.Data()));" << endl; out << " TString cdir = gSystem->WorkingDirectory();" << endl; out << " gSystem->ChangeDirectory(pkgdir);" << endl; out << " // Check for BUILD.sh and execute" << endl; @@ -3502,7 +4638,9 @@ void AliAnalysisAlien::WriteMergingMacro() workdir += fGridWorkingDir; if (FileExists(mergingMacro)) gGrid->Rm(mergingMacro); Info("WriteMergingMacro", "\n##### Copying merging macro: <%s> to your alien workspace", mergingMacro.Data()); - TFile::Cp(Form("file:%s",mergingMacro.Data()), Form("alien://%s/%s", workdir.Data(), mergingMacro.Data())); +// TFile::Cp(Form("file:%s",mergingMacro.Data()), Form("alien://%s/%s", workdir.Data(), mergingMacro.Data())); + if (!copyLocal2Alien("WriteMergeMacro",mergingMacro.Data(), + Form("%s/%s", workdir.Data(), mergingMacro.Data()))) Fatal("","Terminating"); } } @@ -3514,7 +4652,7 @@ Bool_t AliAnalysisAlien::SetupPar(const char *package) // LD_LIBRARY_PATH TString pkgdir = package; pkgdir.ReplaceAll(".par",""); - gSystem->Exec(Form("tar xvzf %s.par", pkgdir.Data())); + gSystem->Exec(TString::Format("tar xzf %s.par", pkgdir.Data())); TString cdir = gSystem->WorkingDirectory(); gSystem->ChangeDirectory(pkgdir); // Check for BUILD.sh and execute @@ -3553,6 +4691,10 @@ Bool_t AliAnalysisAlien::SetupPar(const char *package) void AliAnalysisAlien::WriteExecutable() { // Generate the alien executable script. + // Patch executable with -x to catch error code + if (fExecutableCommand.Contains("root") && + fExecutableCommand.Contains("-q") && + !fExecutableCommand.Contains("-x")) fExecutableCommand += " -x"; if (!TestBit(AliAnalysisGrid::kSubmit)) { ofstream out; out.open(fExecutable.Data(), ios::out); @@ -3561,6 +4703,8 @@ void AliAnalysisAlien::WriteExecutable() return; } out << "#!/bin/bash" << endl; + // Make sure we can properly compile par files + out << "export LD_LIBRARY_PATH=.:$LD_LIBRARY_PATH" << endl; out << "echo \"=========================================\"" << endl; out << "echo \"############## PATH : ##############\"" << endl; out << "echo $PATH" << endl; @@ -3579,11 +4723,23 @@ void AliAnalysisAlien::WriteExecutable() out << "echo \"############## memory : ##############\"" << endl; out << "free -m" << endl; out << "echo \"=========================================\"" << endl << endl; - // Make sure we can properly compile par files - if (TObject::TestBit(AliAnalysisGrid::kUsePars)) out << "export LD_LIBRARY_PATH=.:$LD_LIBRARY_PATH" << endl; out << fExecutableCommand << " "; - out << fAnalysisMacro.Data() << " " << fExecutableArgs.Data() << endl << endl; - out << "echo \"======== " << fAnalysisMacro.Data() << " finished with exit code: $? ========\"" << endl; + out << fAnalysisMacro.Data() << " " << fExecutableArgs.Data() << endl; + out << "RET=$?" << endl; + out << "if [ \"$RET\" != \"0\" ];then" << endl; + out << " echo \"======== ERROR : " << fAnalysisMacro.Data() << " finished with NON zero code: $RET ========\"" << endl; + out << " if [ \"$RET\" -gt 128 ] && [ \"$RET\" -lt 160 ]; then"<GetHomeDirectory(), fExecutable.Data()); if (FileExists(executable)) gGrid->Rm(executable); Info("WriteExecutable", "\n##### Copying executable file <%s> to your AliEn bin directory", fExecutable.Data()); - TFile::Cp(Form("file:%s",fExecutable.Data()), Form("alien://%s", executable.Data())); +// TFile::Cp(Form("file:%s",fExecutable.Data()), Form("alien://%s", executable.Data())); + if (!copyLocal2Alien("WriteExecutable",fExecutable.Data(), + executable.Data())) Fatal("","Terminating"); } } @@ -3617,6 +4775,8 @@ void AliAnalysisAlien::WriteMergeExecutable() return; } out << "#!/bin/bash" << endl; + // Make sure we can properly compile par files + out << "export LD_LIBRARY_PATH=.:$LD_LIBRARY_PATH" << endl; out << "echo \"=========================================\"" << endl; out << "echo \"############## PATH : ##############\"" << endl; out << "echo $PATH" << endl; @@ -3635,15 +4795,27 @@ void AliAnalysisAlien::WriteMergeExecutable() out << "echo \"############## memory : ##############\"" << endl; out << "free -m" << endl; out << "echo \"=========================================\"" << endl << endl; - // Make sure we can properly compile par files - if (TObject::TestBit(AliAnalysisGrid::kUsePars)) out << "export LD_LIBRARY_PATH=.:$LD_LIBRARY_PATH" << endl; TString mergeMacro = fExecutable; mergeMacro.ReplaceAll(".sh", "_merge.C"); if (IsOneStageMerging()) out << "export ARG=\"" << mergeMacro << "(\\\"$1\\\")\"" << endl; else - out << "export ARG=\"" << mergeMacro << "(\\\"$1\\\",$2,$3)\"" << endl; + out << "export ARG=\"" << mergeMacro << "(\\\"$1\\\",$2)\"" << endl; out << fExecutableCommand << " " << "$ARG" << endl; + out << "RET=$?" << endl; + out << "if [ \"$RET\" != \"0\" ];then" << endl; + out << " echo \"======== ERROR : " << fAnalysisMacro.Data() << " finished with NON zero code: $RET ========\"" << endl; + out << " if [ \"$RET\" -gt 128 ] && [ \"$RET\" -lt 160 ]; then"<GetHomeDirectory(), mergeExec.Data()); if (FileExists(executable)) gGrid->Rm(executable); Info("WriteMergeExecutable", "\n##### Copying executable file <%s> to your AliEn bin directory", mergeExec.Data()); - TFile::Cp(Form("file:%s",mergeExec.Data()), Form("alien://%s", executable.Data())); +// TFile::Cp(Form("file:%s",mergeExec.Data()), Form("alien://%s", executable.Data())); + if (!copyLocal2Alien("WriteMergeExecutable", + mergeExec.Data(), executable.Data())) Fatal("","Terminating"); } } @@ -3695,7 +4869,9 @@ void AliAnalysisAlien::WriteProductionFile(const char *filename) const if (gGrid) { Info("WriteProductionFile", "\n##### Copying production file <%s> to your work directory", filename); if (FileExists(filename)) gGrid->Rm(filename); - TFile::Cp(Form("file:%s",filename), Form("alien://%s/%s", workdir.Data(),filename)); +// TFile::Cp(Form("file:%s",filename), Form("alien://%s/%s", workdir.Data(),filename)); + if (!copyLocal2Alien("WriteProductionFile", filename, + Form("%s/%s", workdir.Data(),filename))) Fatal("","Terminating"); } } @@ -3840,6 +5016,8 @@ void AliAnalysisAlien::WriteValidationScript(Bool_t merge) workdir += fGridWorkingDir; Info("WriteValidationScript", "\n##### Copying validation script <%s> to your AliEn working space", validationScript.Data()); if (FileExists(validationScript)) gGrid->Rm(validationScript); - TFile::Cp(Form("file:%s",validationScript.Data()), Form("alien://%s/%s", workdir.Data(),validationScript.Data())); +// TFile::Cp(Form("file:%s",validationScript.Data()), Form("alien://%s/%s", workdir.Data(),validationScript.Data())); + if (!copyLocal2Alien("WriteValidationScript", validationScript.Data(), + Form("%s/%s",workdir.Data(), validationScript.Data()))) Fatal("","Terminating"); } }