X-Git-Url: http://git.uio.no/git/?a=blobdiff_plain;f=ANALYSIS%2FAliAnalysisAlien.cxx;h=47517705da7155173dadf2ce493a6adeb27e082f;hb=94d06b79a195dec517c374e4978148b5aaf4cfe1;hp=1a566b7c6ec87709834042fc49244f892de17fd4;hpb=7ea40ac8051aa16b7b6ce380a61739ab90e32d8d;p=u%2Fmrichter%2FAliRoot.git diff --git a/ANALYSIS/AliAnalysisAlien.cxx b/ANALYSIS/AliAnalysisAlien.cxx index 1a566b7c6ec..47517705da7 100644 --- a/ANALYSIS/AliAnalysisAlien.cxx +++ b/ANALYSIS/AliAnalysisAlien.cxx @@ -24,6 +24,7 @@ #include "Riostream.h" #include "TEnv.h" +#include "TKey.h" #include "TBits.h" #include "TError.h" #include "TROOT.h" @@ -33,6 +34,7 @@ #include "TChain.h" #include "TObjString.h" #include "TObjArray.h" +#include "TMacro.h" #include "TGrid.h" #include "TGridResult.h" #include "TGridCollection.h" @@ -41,12 +43,29 @@ #include "TGridJobStatus.h" #include "TFileMerger.h" #include "AliAnalysisManager.h" +#include "AliAnalysisTaskCfg.h" #include "AliVEventHandler.h" #include "AliAnalysisDataContainer.h" #include "AliMultiInputEventHandler.h" 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(), @@ -98,6 +117,7 @@ AliAnalysisAlien::AliAnalysisAlien() fJDLName(), fTerminateFiles(), fMergeExcludes(), + fRegisterExcludes(), fIncludePath(), fCloseSE(), fFriendChainName(), @@ -111,7 +131,9 @@ AliAnalysisAlien::AliAnalysisAlien() fAliRootMode(), fMergeDirName(), fInputFiles(0), - fPackages(0) + fPackages(0), + fModules(0), + fProofParam() { // Dummy ctor. SetDefaults(); @@ -168,6 +190,7 @@ AliAnalysisAlien::AliAnalysisAlien(const char *name) fJDLName(), fTerminateFiles(), fMergeExcludes(), + fRegisterExcludes(), fIncludePath(), fCloseSE(), fFriendChainName(), @@ -181,7 +204,9 @@ AliAnalysisAlien::AliAnalysisAlien(const char *name) fAliRootMode(), fMergeDirName(), fInputFiles(0), - fPackages(0) + fPackages(0), + fModules(0), + fProofParam() { // Default ctor. SetDefaults(); @@ -238,6 +263,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), @@ -251,7 +277,9 @@ AliAnalysisAlien::AliAnalysisAlien(const AliAnalysisAlien& other) fAliRootMode(other.fAliRootMode), fMergeDirName(other.fMergeDirName), fInputFiles(0), - fPackages(0) + fPackages(0), + fModules(0), + fProofParam() { // Copy ctor. fGridJDL = (TGridJDL*)gROOT->ProcessLine("new TAlienJDL()"); @@ -272,16 +300,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(); } //______________________________________________________________________________ @@ -338,6 +378,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; @@ -364,10 +405,301 @@ 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; + } 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)); +// SetExecutableCommand("aliroot -b -q "); + SetValidationScript(Form("%s_validation.sh", name)); + WriteAnalysisFile(); + WriteAnalysisMacro(); + WriteExecutable(); + WriteValidationScript(); + 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; + } + return kTRUE; +} + //______________________________________________________________________________ void AliAnalysisAlien::SetRunPrefix(const char *prefix) { @@ -491,6 +823,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; @@ -510,7 +847,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> ?", @@ -518,19 +856,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; } @@ -701,10 +1039,94 @@ Bool_t AliAnalysisAlien::CheckInputData() return kTRUE; } +//______________________________________________________________________________ +Bool_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 kFALSE; + } + if (!DirectoryExists(griddir)) { + Error("CopyLocalDataset", "Data directory %s not existing.", griddir); + return kFALSE; + } + 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 kFALSE; + } + 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); + 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; + } + } + } + gSystem->ChangeDirectory(cdir); + delete res; + if (additionalArchives) + delete additionalArchives; + return kTRUE; +} + //______________________________________________________________________________ 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"); @@ -721,8 +1143,11 @@ Bool_t AliAnalysisAlien::CreateDataset(const char *pattern) TString command; 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; @@ -739,29 +1164,65 @@ Bool_t AliAnalysisAlien::CreateDataset(const char *pattern) // 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 += " "; + 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)) { @@ -784,36 +1245,79 @@ Bool_t AliAnalysisAlien::CreateDataset(const char *pattern) TObjString *os; TIter next(arr); while ((os=(TObjString*)next())) { + 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 += 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) { @@ -872,6 +1376,8 @@ Bool_t AliAnalysisAlien::CreateDataset(const char *pattern) // Process a full run range. for (Int_t irun=fRunRange[0]; irun<=fRunRange[1]; irun++) { format = Form("%%s/%s ", fRunPrefix.Data()); + nstart = 0; + stage = 0; path = Form(format.Data(), fGridDataDir.Data(), irun); if (!DirectoryExists(path)) continue; // CdWork(); @@ -886,28 +1392,68 @@ 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 += 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. @@ -1109,8 +1655,6 @@ Bool_t AliAnalysisAlien::CreateJDL() 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); @@ -1130,7 +1674,17 @@ Bool_t AliAnalysisAlien::CreateJDL() } } 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"; @@ -1145,14 +1699,13 @@ Bool_t AliAnalysisAlien::CreateJDL() } 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()),""); @@ -1193,6 +1746,8 @@ 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); @@ -1249,10 +1804,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()) { @@ -1263,7 +1822,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; } @@ -1273,7 +1835,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"); } } } @@ -1305,7 +1870,10 @@ Bool_t AliAnalysisAlien::WriteJDL(Bool_t copy) while ((os=next())) { fGridJDL->AddToInputDataCollection(Form("LF:%s,nodownload", os->GetName()), "Input xml collections"); } - fGridJDL->SetOutputDirectory(Form("%s/#alien_counter_04i#", fGridOutputDir.Data())); + 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. @@ -1486,11 +2054,17 @@ Bool_t AliAnalysisAlien::WriteJDL(Bool_t copy) 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 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())); +// 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; @@ -1661,30 +2235,48 @@ 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; + TChain *chain = new TChain(treeName); + TChain *chainFriend = 0; + if (!fFriendChainName.IsNull()) chainFriend = new TChain(treeName); + while (in.good()) + { in >> line; if (line.IsNull()) 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("#")); + esdFile = gSystem->DirName(esdFile); + esdFile += "/" + fFriendChainName; + file = TFile::Open(esdFile); + if (file && !file->IsZombie()) { + file->Close(); + chainFriend->Add(esdFile); + } else { + Fatal("GetChainForTestMode", "Cannot open friend file: %s", esdFile.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; + delete chainFriend; return NULL; - } + } // chain->ls(); - return chain; + if (!fFriendChainName.IsNull()) chain->AddFriend(chainFriend); + return chain; } //______________________________________________________________________________ @@ -1830,6 +2422,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); @@ -1966,7 +2559,9 @@ Bool_t AliAnalysisAlien::CheckMergedFiles(const char *filename, const char *alie } // 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)); +// 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; @@ -1986,6 +2581,27 @@ Bool_t AliAnalysisAlien::CheckMergedFiles(const char *filename, const char *alie 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; + } + TIter nextkey(file->GetListOfKeys()); + AliAnalysisManager *mgr = 0; + TKey *key; + while ((key=(TKey*)nextkey())) { + if (!strcmp(key->GetClassName(), "AliAnalysisManager")) + mgr = (AliAnalysisManager*)file->Get(key->GetName()); + } + if (!mgr) + ::Error("LoadAnalysisManager", "No analysis manager found in file %s", fname); + return mgr; +} + //______________________________________________________________________________ Int_t AliAnalysisAlien::SubmitSingleJob(const char *query) { @@ -2137,7 +2753,7 @@ Bool_t AliAnalysisAlien::MergeOutput(const char *output, const char *basedir, In // 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(kTRUE); fm->SetFastMethod(kTRUE); if (previousChunk.Length()) fm->AddFile(previousChunk.Data()); outputChunk = outputFile; @@ -2177,7 +2793,7 @@ Bool_t AliAnalysisAlien::MergeOutput(const char *output, const char *basedir, In } // Merging stage different than 0. // Move to the begining of the requested chunk. - fm = new TFileMerger(kFALSE); + fm = new TFileMerger(kTRUE); fm->SetFastMethod(kTRUE); while ((nextfile=next())) fm->AddFile(nextfile->GetName()); delete listoffiles; @@ -2283,8 +2899,8 @@ 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) { @@ -2354,6 +2970,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*/) { @@ -2366,7 +3006,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; @@ -2389,6 +3030,15 @@ 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\");", @@ -2415,13 +3065,15 @@ Bool_t AliAnalysisAlien::StartAnalysis(Long64_t /*nentries*/, Long64_t /*firstEn Error("StartAnalysis", "Could not connect to PROOF cluster <%s>", fProofCluster.Data()); return kFALSE; } - 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; - } 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; @@ -2450,7 +3102,7 @@ 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(" "); @@ -2653,7 +3305,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(); @@ -2687,9 +3339,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; } @@ -2887,7 +3539,8 @@ 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, mergeJDLName); @@ -3033,7 +3686,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"); } } @@ -3083,6 +3737,10 @@ 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; @@ -3231,25 +3889,32 @@ void AliAnalysisAlien::WriteAnalysisMacro() out << " gEnv->SetValue(\"XNet.MaxRedirectCount\",2);" << 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 << "\");" << 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) { @@ -3261,57 +3926,16 @@ 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; @@ -3352,9 +3976,18 @@ void AliAnalysisAlien::WriteAnalysisMacro() out << " chain->Add(filename);" << endl; if(fFriendChainName!="") { out << " TString fileFriend=coll->GetTURL(\"\");" << endl; - out << " fileFriend.ReplaceAll(\"AliAOD.root\",\""<Add(fileFriend.Data());" << endl; + out << " if (fileFriend.Index(\"#\") > -1) fileFriend.Remove(fileFriend.Index(\"#\"));" << endl; + out << " fileFriend = gSystem->DirName(fileFriend);" << endl; + out << " fileFriend += \"/\";" << endl; + out << " fileFriend += \"" << fFriendChainName << "\";"; + out << " TFile *file = TFile::Open(fileFriend);" << endl; + out << " if (file) {" << endl; + out << " file->Close();" << endl; + out << " chainFriend->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 << " if (!chain->GetNtrees()) {" << endl; @@ -3416,13 +4049,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"); } } @@ -3437,7 +4068,7 @@ void AliAnalysisAlien::WriteMergingMacro() } 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); @@ -3603,15 +4234,15 @@ void AliAnalysisAlien::WriteMergingMacro() out << " gEnv->SetValue(\"XNet.FirstConnectMaxCnt\",1);" << endl << endl; } // Change temp directory to current one + out << "// Connect to AliEn" << 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; - out << "// Connect to AliEn" << endl; - out << " if (!TGrid::Connect(\"alien://\")) return;" << 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; @@ -3642,19 +4273,9 @@ void AliAnalysisAlien::WriteMergingMacro() TString analysisFile = fExecutable; analysisFile.ReplaceAll(".sh", ".root"); out << " if (!outputDir.Contains(\"Stage\")) 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; + out << " AliAnalysisManager *mgr = AliAnalysisAlien::LoadAnalysisManager(\"" + << analysisFile << "\");" << endl; + out << " if (!mgr) return;" << endl; out << " mgr->SetRunFromPath(mgr->GetRunFromAlienPath(dir));" << endl; out << " mgr->SetSkipTerminate(kFALSE);" << endl; out << " mgr->PrintStatus();" << endl; @@ -3720,7 +4341,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"); } } @@ -3800,8 +4423,22 @@ void AliAnalysisAlien::WriteExecutable() out << "free -m" << endl; out << "echo \"=========================================\"" << endl << 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"); } } @@ -3862,6 +4501,20 @@ void AliAnalysisAlien::WriteMergeExecutable() else 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"); } } @@ -3913,7 +4568,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"); } } @@ -4058,6 +4715,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"); } }