PWGLF/FORWARD/trains/ProofRailway.C

   1 /**
   2  * @file   ProofRailway.C
   3  * @author Christian Holm Christensen <cholm@master.hehi.nbi.dk>
   4  * @date   Tue Oct 16 18:58:37 2012
   5  *
   6  * @brief
   7  *
   8  *
   9  * @ingroup pwglf_forward_trains_helper
  10  *
  11  */
  12 #ifndef PROOFHELPER_C
  13 #define PROOFHELPER_C
  14 #include "Railway.C"
  15 #ifndef __CINT__
  16 # include "OutputUtilities.C"
  17 # include "ParUtilities.C"
  18 # include "ChainBuilder.C"
  19 # include <TUrl.h>
  20 # include <TString.h>
  21 # include <TProof.h>
  22 # include <TProofLog.h>
  23 # include <TProofDebug.h>
  24 # include <AliAnalysisManager.h>
  25 # include <TEnv.h>
  26 # include <TChain.h>
  27 // For SendFile
  28 # include <TSystem.h>
  29 # include <TSlave.h>
  30 # include <TSocket.h>
  31 # include <cerrno>
  32 #else
  33 class TUrl;
  34 class TChain;
  35 #endif
  36
  37 // ===================================================================
  38 /**
  39  * Handle analysis on a Proof farm.
  40  *
  41  * This helper is triggered by URIs of the form
  42  *
  43  * @code
  44  * proof://[<user>@]<host>[:<port>]/<dsname>[?<options>][#<treename>]
  45  * @endcode
  46  * where
  47  * <dl>
  48  *   <dt>&lt;user&gt;</dt>
  49  *   <dd>Optional user name</dd>
  50  *   <dt>&lt;host&gt;</dt>
  51  *   <dd>PROOF cluster master host</dd>
  52  *   <dt>&lt;port&gt;</dt>
  53  *   <dd>Optional PROOF cluster port on master host</dd>
  54  *   <dt>&lt;dsname&gt;</dt>
  55  *   <dd>Data set name</dd>
  56  *   <dt>&lt;treename&gt;</dt>
  57  *   <dd>Optional tree name in data set, often <tt>esdTree</tt> or
  58  *   <tt>aodTree</tt></dd>
  59  *   <dt>&lt;options&gt;</dt>
  60  *   <dd>List of options separated by an &amp;
  61  *     <dl>
  62  *       <dt><tt>workers=N[x]</tt></dt>
  63  *       <dd>Set the number of workers to use.  If <tt>x</tt> is appended,
  64  *         then it's maximum number of workers per slave</dd>
  65  *       <dt><tt>dsname</tt>[=&lt;output dataset&gt;]</dt>
  66  *       <dd>Register tree output (e.g., AOD) as a new data set on the
  67  *         PROOF cluster. If &lt;output dataset&gt; is not specified, take
  68  *         the name of the train.</dd>
  69  *       <dt><tt>par[=all]</tt></dt>
  70  *       <dd>Use PAR files.  If the value <tt>all</tt> is given, then also
  71  *         PAR files of STEERBase, ESD, AOD, ANALYSIS, OADB, ANALYSISalice
  72  *         are used. </dd>
  73  *       <dt><tt>mode=[default,rec,sim,train,custom]</tt></dt>
  74  *       <dd>Set the AliROOT mode.  If not specified <tt>default</tt>
  75  *         is assumed.  See also CreateAliROOTPar</dd>
  76  *       <dt><tt>storage=&lt;url&gt;</tt></dt>
  77  *       <dd>Specify a non-default storage location for special output
  78  *         (e.g., AOD trees).  &lt;url&gt; should be a valid XRootd
  79  *         server URI accessible to the slaves - e.g.,
  80  *         <tt>root://lxplus.cern.ch:10930//tmp</tt>.</dd>
  81  *     </dl>
  82  *   </dd>
  83  * </dl>
  84  *
  85  * @ingroup pwglf_forward_trains_helper
  86  */
  87 struct ProofRailway : public Railway
  88 {
  89   /**
  90    * Constructor
  91    *
  92    * @param url  Url
  93    * @param verbose Verbosity level
  94    */
  95   ProofRailway(const TUrl& url, Int_t verbose)
  96     : Railway(url, verbose),
  97       fExtraLibs(""),
  98       fExtraPars(""),
  99       fExtraSrcs(""),
 100       fUsePars(false),
 101       fBasePars(false),
 102       fTestBuild(true),
 103       fAuxFiles()
 104   {
 105     fOptions.Add("workers",  "N[x]", "Number of workers to use", 0);
 106     fOptions.Add("dsname",   "NAME", "Make output dataset", "");
 107     fOptions.Add("par",      "tasks|all", "Use par files",           "tasks");
 108     fOptions.Add("mode",     "default|rec|sim", "AliROOT mode",      "default");
 109     fOptions.Add("storage",  "URL", "Location for external storage", "");
 110     fOptions.Add("wrapper",  "CMD", "Wrapper command", "");
 111     fOptions.Add("clear",    "PKGS", "Clear packages ','-separated", "");
 112     fOptions.Add("reset",    "soft|hard", "Reset cluster", "hard");
 113     fOptions.Add("feedback", "Enable feedback mechanism");
 114     fOptions.Add("env",      "SCRIPT", "Script to set-up environment","-none-");
 115     fOptions.Add("offset",   "EVENTS", "Skip this number of events", 0);
 116     fOptions.Add("testpar",  "Test build PARs");
 117     if (!fUrl.GetUser() || fUrl.GetUser()[0] == '\0')
 118       fUrl.SetUser(gSystem->GetUserInfo()->fUser);
 119     fAuxFiles.SetOwner();
 120   }
 121   ProofRailway(const ProofRailway& o)
 122     : Railway(o),
 123       fExtraLibs(""),
 124       fExtraPars(""),
 125       fExtraSrcs(""),
 126       fUsePars(false),
 127       fBasePars(false),
 128       fTestBuild(true),
 129       fAuxFiles()
 130   {}
 131   ProofRailway& operator=(const ProofRailway& o)
 132   {
 133     if (&o == this) return *this;
 134     Railway::operator=(o);
 135     fExtraLibs = o.fExtraLibs;
 136     fExtraPars = o.fExtraPars;
 137     fExtraSrcs = o.fExtraSrcs;
 138     fUsePars   = o.fUsePars;
 139     fBasePars  = o.fBasePars;
 140     fTestBuild = o.fTestBuild;
 141     // fAuxFiles;
 142     return *this;
 143   }
 144   /**
 145    * Destructor
 146    */
 147   virtual ~ProofRailway() {}
 148   /**
 149    * Load a library/PAR/script
 150    *
 151    * @param name   Name
 152    * @param slaves If true, also load on slaves
 153    *
 154    * @return true on success
 155    */
 156   virtual Bool_t LoadLibrary(const TString& name,
 157                              Bool_t slaves=true)
 158   {
 159     Bool_t isBase = false;
 160     if (!fBasePars) {
 161       if (name.EqualTo("STEERBase")      ||
 162           name.EqualTo("ESD")            ||
 163           name.EqualTo("AOD")            ||
 164           name.EqualTo("ANALYSIS")       ||
 165           name.EqualTo("OADB")           ||
 166           name.EqualTo("ANALYSISalice"))
 167         isBase = true;
 168     }
 169     if (!fUsePars || isBase) {
 170       Int_t ret = gSystem->Load(MakeLibraryName(name));
 171       if (ret < 0) return false;
 172       if (slaves) fExtraLibs.Append(Form(":%s", name.Data()));
 173     }
 174     else {
 175       if (!ParUtilities::Find(name)) {
 176         Error("ProofRailway::LoadLibrary", "Failed to find PAR file %s",
 177               name.Data());
 178         return false;
 179       }
 180       if (fTestBuild && !ParUtilities::Build(name)) {
 181         Error("ProofRailway::LoadLibrary", "Failed to build PAR file %s",
 182               name.Data());
 183         return false;
 184       }
 185       if (gProof->UploadPackage(name.Data(), TProof::kRemoveOld) < 0) {
 186         Error("ProofRailway::LoadLibrary", "Failed to upload PAR file %s",
 187               name.Data());
 188         return false;
 189       }
 190       fExtraPars.Append(Form(":%s", name.Data()));
 191     }
 192     return true;
 193   }
 194   /**
 195    * Load a source file, and compile it
 196    *
 197    * @param name Name of the source file
 198    * @param copy If true, copy not link
 199    *
 200    * @return true on success
 201    */
 202   virtual Bool_t LoadSource(const TString& name, bool copy=false)
 203   {
 204     if (!Railway::LoadSource(name, copy)) return false;
 205     fExtraSrcs.Append(Form(":%s", gSystem->BaseName(name.Data())));
 206     return true;
 207   }
 208   /**
 209    * Set-up to load the AliROOT libraries
 210    *
 211    * @return true on success
 212    */
 213   virtual Bool_t LoadAliROOT()
 214   {
 215     if (!gSystem->Getenv("ALICE_ROOT")) {
 216       Error("ProofRailway::LoadAliROOT", "Local AliROOT not available");
 217       return false;
 218     }
 219
 220     Bool_t tmp = fUsePars;
 221     fUsePars   = fBasePars;
 222     if (!LoadLibrary("STEERBase"))     return false;
 223     if (!LoadLibrary("ESD"))           return false;
 224     if (!LoadLibrary("AOD"))           return false;
 225     if (!LoadLibrary("ANALYSIS"))      return false;
 226     if (!LoadLibrary("OADB"))          return false;
 227     if (!LoadLibrary("ANALYSISalice")) return false;
 228     fUsePars = tmp;
 229
 230     return CreateAliROOTPar();
 231   }
 232   /**
 233    * Get the name of the AliROOT par file to use
 234    *
 235    * @return String
 236    */
 237   virtual const char* AliROOTParName() const
 238   {
 239     return "ALIROOT";
 240   }
 241   /**
 242    * Create an AliROOT par file from the executing AliROOT.  This PAR
 243    * file basically uses the environment of the client - that is, we
 244    * assume that the used AliROOT is accessible on the slaves - e.g.,
 245    * via an NFS export.
 246    *
 247    * Note, the SETUP.C script take one argument - a TList of TNamed
 248    * parameters.  Parameters processed are
 249    *
 250    * - ALIROOT_MODE=[default,aliroot,rec,sim,train]
 251    *   - default: Load base analysis libraries
 252    *   - aliroot: Load $ALICE_ROOT/macros/loadlibs.C
 253    *   - rec:     Load $ALICE_ROOT/macros/loadlibsrec.C
 254    *   - sim:     Load $ALICE_ROOT/macros/loadlibssim.C
 255    * - ALIROOT_EXTRA_LIBS Colon separated list of additional (Ali)ROOT
 256    *   libraries to load on the slaves.
 257    *
 258    * The generated PAR file is uploaded but not enabled until we have
 259    * populated fExtraLibs.  The enabling takes place at the end of the
 260    * set-up.
 261    *
 262    * @return true on success, false otherwise.     */
 263   virtual Bool_t CreateAliROOTPar()
 264   {
 265     if (fBasePars) return true;
 266
 267     TString parName(AliROOTParName());
 268     TString parFile(Form("%s.par", parName.Data()));
 269
 270     // --- Check if we have the drirectory already -------------------
 271     if (gSystem->AccessPathName(parName.Data()) == 0) {
 272       // Let's remove it to get a clean slate
 273       if (gSystem->Exec(Form("rm -rf %s", parName.Data())) != 0) {
 274         Error("ProofRailway", "Failed to remove %s", parName.Data());
 275         return false;
 276       }
 277     }
 278     // --- Check if the PAR file is there, and remove it if so -------
 279     if (gSystem->AccessPathName(parFile.Data()) == 0) {
 280       if (gSystem->Unlink(parFile.Data()) != 0) {
 281         Error("ProofRailway::CreateAliROOTPar", "Failed to remove %s",
 282               parFile.Data());
 283         return false;
 284       }
 285     }
 286
 287
 288     // Set-up directories
 289     if (gSystem->MakeDirectory(parName) < 0) {
 290       Error("ProofRailway::CreateAliROOTPar", "Could not make directory '%s'",
 291             parName.Data());
 292       return false;
 293     }
 294
 295     if (gSystem->MakeDirectory(Form("%s/PROOF-INF", parName.Data()))) {
 296       Error("ProofRailway::CreateAliROOTPar",
 297             "Could not make directory %s/PROOF-INF",
 298             parName.Data());
 299       return false;
 300     }
 301
 302     std::ofstream b(Form("%s/PROOF-INF/BUILD.sh",parName.Data()));
 303     if (!b) {
 304       Error("ProofRailway::CreateAliROOTPar",
 305             "Failed to make BUILD.sh shell script");
 306       return false;
 307     }
 308     b << "#!/bin/sh\n\n"
 309       << "# echo Nothing to do\n"
 310       << "exit 0\n"
 311       << std::endl;
 312     b.close();
 313     gSystem->Exec(Form("chmod a+x %s/PROOF-INF/BUILD.sh",parName.Data()));
 314
 315     TString envScript = fOptions.Get("env");
 316     if (envScript.EqualTo("-none-", TString::kIgnoreCase))
 317       envScript = "";
 318     if (!envScript.IsNull()) {
 319       // If an environment script was specified, copy that to the par
 320       if (gSystem->AccessPathName(envScript.Data()) == 0) {
 321         // Copy script
 322         if (gSystem->Exec(Form("cp %s %s/PROOF-INF/", envScript.Data(),
 323                                parName.Data())) != 0) {
 324           Error("ProofRailway", "Failed to copy %s", envScript.Data());
 325           return false;
 326         }
 327       }
 328       else {
 329         Warning("CreateALIROOTPar", "Couldn't read %s", envScript.Data());
 330         envScript = "";
 331       }
 332     }
 333     std::ofstream s(Form("%s/PROOF-INF/SETUP.C", parName.Data()));
 334     if (!s) {
 335       Error("ProofRailway::CreateAliROOTPar",
 336             "Failed to make SETUP.C ROOT script");
 337       return false;
 338     }
 339     s << "void SETUP(TList* opts) {\n";
 340     if (envScript.IsNull()) {
 341       s << "  gSystem->Setenv(\"ALICE\",\""
 342       << gSystem->Getenv("ALICE") << "\");\n"
 343       << "  gSystem->Setenv(\"ALICE_ROOT\",\""
 344       << gSystem->Getenv("ALICE_ROOT") << "\");\n"
 345       << "  gSystem->Setenv(\"ALICE_TARGET\",\""
 346         << gSystem->Getenv("ALICE_TARGET") << "\");\n";
 347     if (gSystem->Getenv("OADB_PATH"))
 348       s << "  gSystem->Setenv(\"OADB_PATH\",\""
 349         << gSystem->Getenv("OADB_PATH") << "\");\n";
 350     }
 351     else {
 352       s << "  gROOT->Macro(\"PROOF-INF/" << gSystem->BaseName(envScript.Data())
 353         << "\");\n";
 354     }
 355     s   << "  gSystem->AddDynamicPath("
 356         << "\"$(ALICE_ROOT)/lib/tgt_$(ALICE_TARGET)\");\n";
 357     s << "  \n"
 358       << "  // Info(\"SETUP\",\"Loading ROOT libraries\");\n"
 359       << "  gSystem->Load(\"libTree\");\n"
 360       << "  gSystem->Load(\"libGeom\");\n"
 361       << "  gSystem->Load(\"libVMC\");\n"
 362       << "  gSystem->Load(\"libPhysics\");\n"
 363       << "  gSystem->Load(\"libMinuit\");\n"
 364       << "  \n";
 365     s << "  // Info(\"SETUP\",\"Parameter list:\");\n"
 366       << "  if (!opts) return;\n"
 367       << "  //opts->ls();\n"
 368       << "  \n";
 369     s << "  TObject* par = opts->FindObject(\"ALIROOT_MODE\");\n"
 370       << "  if (par) {\n"
 371       << "    // Info(\"SETUP\",\"ALIROOT mode: %s\", par->GetTitle());\n"
 372       << "    TString mode(par->GetTitle());\n"
 373       << "    if (mode.EqualTo(\"default\",TString::kIgnoreCase)) {\n"
 374       << "      gSystem->Load(\"libSTEERBase\");\n"
 375       << "      gSystem->Load(\"libESD\");\n"
 376       << "      gSystem->Load(\"libAOD\");\n"
 377       << "      gSystem->Load(\"libANALYSIS\");\n"
 378       << "      gSystem->Load(\"libOADB\");\n"
 379       << "      gSystem->Load(\"libANALYSISalice\");\n"
 380       << "    }\n"
 381       << "    else if (mode.EqualTo(\"aliroot\",TString::kIgnoreCase)) \n"
 382       << "      gROOT->Macro(\"$ALICE_ROOT/macros/loadlibs.C\");\n"
 383       << "    else if (mode.EqualTo(\"rec\",TString::kIgnoreCase)) \n"
 384       << "      gROOT->Macro(\"$ALICE_ROOT/macros/loadlibsrec.C\");\n"
 385       << "    else if (mode.EqualTo(\"sim\",TString::kIgnoreCase)) \n"
 386       << "      gROOT->Macro(\"$ALICE_ROOT/macros/loadlibssim.C\");\n"
 387       << "    else if (mode.EqualTo(\"train\",TString::kIgnoreCase)) \n"
 388       << "      gROOT->Macro(\"$ALICE_ROOT/macros/loadlibstrain.C\");\n"
 389       << "    else if (mode.EqualTo(\"custom\",TString::kIgnoreCase)) \n"
 390       << "      gROOT->Macro(\"$ALICE_ROOT/macros/loadlibstrain.C\");\n"
 391       << "  }\n"
 392       << "  \n";
 393     s << "  par = opts->FindObject(\"ALIROOT_EXTRA_LIBS\");\n"
 394       << "  if (par) {\n"
 395       << "    Info(\"SETUP\",\"Libaries to load: %s\\n\",par->GetTitle());\n"
 396       << "    TString tit(par->GetTitle());\n"
 397       << "    TObjArray* tokens = tit.Tokenize(\":\");\n"
 398       << "    TObject*   lib    = 0;\n"
 399       << "    TIter      next(tokens);\n"
 400       << "    while ((lib = next())) {\n"
 401       << "      TString libName(lib->GetName());\n"
 402       << "      if (!libName.BeginsWith(\"lib\")) libName.Prepend(\"lib\");\n"
 403       << "      // Info(\"SETUP\",\"Loading %s ...\",libName.Data());\n"
 404       << "      gSystem->Load(libName.Data());\n"
 405       << "    }\n"
 406       << "  }\n"
 407       << "}\n"
 408       << std::endl;
 409     s.close();
 410
 411     Int_t ret = gSystem->Exec(Form("tar -czf %s %s",
 412                                    parFile.Data(), parName.Data()));
 413     if (ret != 0) {
 414       Error("ProofRailway::CreateAliROOTPar", "Failed to pack up PAR file %s",
 415             parFile.Data());
 416       return false;
 417     }
 418
 419     ret = gProof->UploadPackage(parFile.Data(),TProof::kRemoveOld);
 420     if (ret != 0) {
 421       Error("ProofRailway::CreateAliROOTPar",
 422             "Failed to upload the AliROOT PAR file");
 423       return false;
 424     }
 425     // Note, the PAR isn't enabled until much later when we've
 426     // collected all the needed libraries in fExtraLibs
 427     return true;
 428   }
 429   /**
 430    * Get the mode identifier
 431    *
 432    * @return Always kProof
 433    */
 434   virtual UShort_t Mode() const { return kProof; }
 435   /**
 436    * Get the mode string used for AliAnalysisManager::StartAnalysis
 437    */
 438   virtual const char* ModeString() const { return "proof"; }
 439   /**
 440    * Set-up done before task set-ups
 441    *
 442    * @return true on success
 443    */
 444   virtual Bool_t PreSetup()
 445   {
 446     // --- Set prefered GSI method ---------------------------------
 447     gEnv->SetValue("XSec.GSI.DelegProxy", "2");
 448
 449     // --- Add ALICE_ROOT directory to search path for packages ----
 450     // Info("ProofRailway::PreSetup", "Set location of packages");
 451     gEnv->SetValue("Proof.GlobalPackageDirs",
 452                    Form("%s:%s",
 453                         gEnv->GetValue("Proof.GlobalPackageDirs", "."),
 454                         gSystem->Getenv("ALICE_ROOT")));
 455
 456     // --- Forming the URI we use to connect with --------------------
 457     TUrl connect(fUrl);
 458     connect.SetAnchor("");
 459     connect.SetFile("");
 460     connect.SetOptions("");
 461
 462     // --- Check if we need to reset first ---------------------------
 463     if (fOptions.Has("reset")) {
 464       TString reset = fOptions.Get("reset");
 465       Bool_t  hard  = (reset.IsNull() ||
 466                        reset.EqualTo("hard", TString::kIgnoreCase));
 467       Info("ProofRailway::PreSetup", "Doing a %s reset of %s",
 468            hard ? "hard" : "soft", connect.GetUrl());
 469       TProof::Reset(connect.GetUrl(), hard);
 470       Int_t secs = 3;
 471       Info("ProofRailway::PreSetup",
 472            "Waiting for %d second%s for things to settle", secs,
 473            secs > 1 ? "s" : "");
 474       gSystem->Sleep(1000*secs);
 475     }
 476
 477     // --- Check if we're using a wrapper ----------------------------
 478     if (fOptions.Has("wrapper")) {
 479       TString wrapper = fOptions.Get("wrapper");
 480       if (wrapper.IsNull())
 481         // In case of no argument, use GDB
 482         // Just run and backtrace
 483         wrapper = "/usr/bin/gdb --batch -ex run -ex bt --args";
 484       Info("ProofRailway::PreSetup", "Using wrapper command: %s",
 485            wrapper.Data());
 486       TProof::AddEnvVar("PROOF_WRAPPERCMD", wrapper);
 487     }
 488
 489     // --- PAR parameters --------------------------------------------
 490     fUsePars   = fOptions.Has("par");
 491     fBasePars  = (fUsePars &&
 492                   fOptions.Get("par").EqualTo("all",TString::kIgnoreCase));
 493     fTestBuild = fOptions.Has("testpar");
 494
 495     // --- Connect to the cluster ------------------------------------
 496     TString opts;
 497     if (fOptions.Has("workers"))
 498       opts.Append(Form("workers=%s", fOptions.Get("workers").Data()));
 499
 500     Info("ProofRailway::PreSetup", "Connecting to %s with %soptions %s",
 501          connect.GetUrl(),
 502          opts.IsNull() ? "no " : "",
 503          opts.Data());
 504     TString proto(connect.GetProtocol());
 505     if (proto.BeginsWith("lite") && fOptions.Has("workers"))
 506       TProof::Open(opts);
 507     else
 508       TProof::Open(connect.GetUrl(), opts);
 509     // TProof::Open(connect.GetHost(), opts);
 510     if (!gProof) {
 511       Error("ProofRailway::PreSetup", "Failed to open Proof connection %s",
 512             connect.GetUrl());
 513       return false;
 514     }
 515
 516     // --- Check if we need to clear packages ------------------------
 517     if (fOptions.Has("clear")) {
 518       TString pkgs = fOptions.Get("clear");
 519       if (pkgs.IsNull() || pkgs.EqualTo("all", TString::kIgnoreCase)) {
 520         // No value given, clear all
 521         if (gProof->ClearPackages() != 0)
 522           Warning("ProofRailway::PreSetup", "Failed to lear all packages");
 523       }
 524       else {
 525         // Tokenize on ',' and clear each package
 526         TObjArray* pars = pkgs.Tokenize(",");
 527         TObject*   pkg  = 0;
 528         TIter      next(pars);
 529         while ((pkg = next())) {
 530           if (gProof->ClearPackage(pkg->GetName()) != 0)
 531             Warning("ProofRailway::PreSetup", "Failed to clear package %s",
 532                     pkg->GetName());
 533         }
 534         pars->Delete();
 535       }
 536     }
 537     return true;
 538   }
 539   /**
 540    * Set-up done after the task set-ups
 541    *
 542    * @return true on success
 543    */
 544   virtual Bool_t PostSetup()
 545   {
 546     AliAnalysisManager* mgr = AliAnalysisManager::GetAnalysisManager();
 547     if (!mgr) {
 548       Error("ProofRailway::PostSetup", "No analysis manager defined");
 549       return false;
 550     }
 551
 552     // --- Check for output ------------------------------------------
 553     if (fOptions.Has("dsname"))
 554       OutputUtilities::RegisterDataset(fOptions.Get("dsname"));
 555     if (fOptions.Has("storage"))
 556       OutputUtilities::RegisterStorage(fOptions.Get("storage"));
 557
 558     // --- Check for feedback mechanism ------------------------------
 559     if (!fOptions.Has("feedback"))
 560       gProof->ClearFeedback();
 561
 562     // --- If we are not using PARs for Base, enable special PAR -----
 563     if (!fBasePars) {
 564       TString tmp(fExtraLibs.Strip(TString::kBoth,':'));
 565       TList* params = new TList;
 566       params->SetOwner(true);
 567       params->Add(new TNamed("ALIROOT_EXTRA_LIBS", tmp.Data()));
 568       if (fOptions.Has("mode"))
 569         params->Add(new TNamed("ALIROOT_MODE", fOptions.Get("mode").Data()));
 570       else
 571         params->Add(new TNamed("ALIROOT_MODE", "default"));
 572       Int_t ret = gProof->EnablePackage(AliROOTParName(), params, true);
 573       if (ret < 0) {
 574         Error("ProofRailway::EnableAliROOT", "Failed to enable AliROOT PAR %s",
 575               AliROOTParName());
 576         return false;
 577       }
 578     }
 579
 580     // --- Make PAR file of Aux Files --------------------------------
 581     if (fAuxFiles.GetEntries() > 0) {
 582       TString name = TString::Format("%s_auxfiles", mgr->GetName());
 583       ParUtilities::MakeAuxFilePAR(fAuxFiles, name);
 584
 585       if (gProof->UploadPackage(name.Data(), TProof::kRemoveOld) < 0)
 586         Error("ProofRailway::PostSetup", "Failed to upload PAR file %s",
 587               name.Data());
 588       else
 589         fExtraPars.Append(Form(":%s", name.Data()));
 590     }
 591
 592     // --- Load par files --------------------------------------------
 593     TString    tmp  = fExtraPars.Strip(TString::kBoth,':');
 594     TObjArray* pars = tmp.Tokenize(":");
 595     TObject*   obj  = 0;
 596     TIter      next(pars);
 597     while ((obj = next())) {
 598       // Enable the package, but do not build on client - already done
 599       Int_t ret = gProof->EnablePackage(obj->GetName(), true);
 600       if (ret < 0) {
 601         Error("ProofRailway::PostSetup", "Failed to enable PAR %s",
 602               obj->GetName());
 603         return false;
 604       }
 605     }
 606
 607     // --- Load extra sources ----------------------------------------
 608     TString    tmp2 = fExtraSrcs.Strip(TString::kBoth, ':');
 609     TObjArray* srcs = tmp2.Tokenize(":");
 610     TIter      next2(srcs);
 611     while ((obj = next())) {
 612       Int_t ret = gProof->Load(Form("%s++g", obj->GetName()), true);
 613       if (ret < 0) {
 614         Error("ProofRailway::PostSetup", "Failed to compile %s", obj->GetName());
 615         return false;
 616       }
 617     }
 618     return true;
 619   }
 620   /**
 621    * Start the analysis
 622    *
 623    * @param nEvents Number of events to analyse
 624    *
 625    * @return The return value of AliAnalysisManager::StartAnalysis
 626    */
 627   virtual Long64_t Run(Long64_t nEvents=-1)
 628   {
 629     AliAnalysisManager* mgr = AliAnalysisManager::GetAnalysisManager();
 630     gProof->SetLogLevel(TMath::Max(fVerbose-2,0),
 631                         /* TProofDebug::kPacketizer| */
 632                         TProofDebug::kLoop|
 633                         /* TProofDebug::kSelector|
 634                         TProofDebug::kOutput|
 635                         TProofDebug::kInput|
 636                         TProofDebug::kGlobal|*/
 637                         TProofDebug::kPackage);
 638     TString dsName(fUrl.GetFile());
 639     // if (fUrl.GetAnchor() && fUrl.GetAnchor()[0] != '\0')
 640     //   dsName.Append(Form("#%s", fUrl.GetAnchor()));
 641     // Info("Run", "Output objects registered with PROOF:");
 642     // gProof->GetOutputList()->ls();
 643     Long64_t off = fOptions.AsLong("offset", 0);
 644     if (nEvents > 0 && nEvents < off) {
 645       Warning("Run", "Number of events %lld < offset (%lld), stopping",
 646               nEvents, off);
 647       return 0;
 648     }
 649     Long64_t ret = mgr->StartAnalysis(fUrl.GetProtocol(), dsName, nEvents, off);
 650
 651     if (fVerbose > 10)
 652       TProof::Mgr(fUrl.GetUrl())->GetSessionLogs()->Save("*","proof.log");
 653     return ret;
 654   }
 655 #if 0
 656   Bool_t AddMonitor(const TString& path)
 657   {
 658     if (path.IsNull()) return true;
 659
 660     TObjArray* tokens  = path.Tokenize("/");
 661     Int_t      nTokens = tokens->GetEntries();
 662     if (nTokens < 2) {
 663       Error("AddMonitor", "Monitors must be of the form:\n"
 664             "  <task>[:<slot>]/<name>\n"
 665             "  <task>[:<slot>]/<path>/<name>");
 666       return false;
 667     }
 668     // --- Get the manager
 669     AliAnalysisManager* mgr = AliAnalysisManager::GetAnalysisManager();
 670
 671     // --- Extract task and possibly slot number
 672     TString& sTask  = static_cast<TObjString*>(tokens->At(0))->String();
 673     Int_t    slotNo = 0;
 674     Ssiz_t   colon  = sTask.Index(":");
 675     if (colon != kNPOS) {
 676       TString sSlot = sTask(colon+1, sTask.Length()-colon-1);
 677       if (!sSlot.IsNull()) slotNo = sSlot.Atoi();
 678       sTask.Remove(colon, sTask.Length()-colon);
 679     }
 680
 681     AliAnalysisTask* task = mgr->GetTask(sTask);
 682     if (!task) {
 683       Error("AddMonitor", "Task \"%s\" not registered with manager",
 684             sTask.Data());
 685       return false;
 686     }
 687     AliAnalysisDataSlot* slot = task->GetOutputSlot(slotNo);
 688     if (!slot) {
 689       Error("AddMonitor", "Task \"%s\" does not have an output slot at %d",
 690             task->GetName(), slotNo);
 691       return false;
 692     }
 693     AliAnalysisDataContainer* cont = slot->GetContainer();
 694     if (!cont) {
 695       Error("AddMonitor", "Output slot %d of task \"%s\" has no container",
 696             slotNo, task->GetName());
 697       return false;
 698     }
 699     Int_t    idx   = 1;
 700     TString& first = static_cast<TObjString*>(tokens->At(idx))->String();
 701     if (first.EqualTo(cont->GetName())) {
 702       idx++;
 703     }
 704     TObject* data = cont->GetData();
 705     TObject* obj  = data;
 706     for (; idx < nTokens; idx++) {
 707     }
 708     return true;
 709   }
 710 #endif
 711   /**
 712    * Print information to standard output
 713    *
 714    * @param option
 715    */
 716   virtual void Print(Option_t* option="") const
 717   {
 718     Railway::Print(option);
 719     std::cout << std::boolalpha
 720               << "  --- Other settings -------\n"
 721               << "  Extra libraries  : " << fExtraLibs << "\n"
 722               << "  Extra PARs       : " << fExtraPars << "\n"
 723               << "  Extra sources    : " << fExtraSrcs << "\n"
 724               << "  Use PARs of tasks: " << fUsePars   << "\n"
 725               << "  Use PARs of base : " << fBasePars
 726               << std::noboolalpha << std::endl;
 727   }
 728   /**
 729    * Link an auxilary file to working directory
 730    *
 731    * @param name Name of the file
 732    * @param copy Copy rather than link
 733    *
 734    * @return true on success
 735    */
 736   virtual Bool_t AuxFile(const TString& name, bool copy=false)
 737   {
 738     Bool_t ret = Railway::AuxFile(name, copy);
 739     if (!name.BeginsWith("/")) {
 740       fAuxFiles.Add(new TObjString(name));
 741     }
 742 #if 0
 743     if (ret && name.EndsWith(".root")) {
 744       TFile* file = TFile::Open(name, "READ");
 745       if (file) {
 746         Info("AuxFile", "Adding input file %s", name.Data());
 747         gProof->AddInputData(file, true);
 748       }
 749     }
 750 #endif
 751     return ret;
 752   }
 753   Int_t SendFile(const TString& fileName)
 754   {
 755     Int_t    bufSize = 32768;
 756     Char_t   buf[bufSize];
 757     Long64_t size = 0;
 758     Long_t   id = 0, flags = 0, modtime = 0;
 759     if (gSystem->GetPathInfo(fileName.Data(), &id, &size, &flags, &modtime)==1
 760         || size <= 0) {
 761       Error("SendFile", "Cannot stat %s", fileName.Data());
 762       return -1;
 763     }
 764     TString fn(gSystem->BaseName(fileName.Data()));
 765     TList*  slaves = 0; // gProof->GetListOfActiveSlaves(); - protected
 766     TIter   next(slaves);
 767     TSlave* sl   = 0;
 768     Int_t   ret  = 0;
 769     Int_t   fd = open(fileName.Data(), O_RDONLY);
 770     while ((sl = static_cast<TSlave*>(next()))) {
 771       if (!sl->IsValid()) continue;
 772       if (sl->GetSlaveType() != TSlave::kSlave) continue;
 773
 774       // Always binary (first 1), never forward (last 0).
 775       snprintf(buf,bufSize,"%s %d %lld %d", fn.Data(), 1, size, 0);
 776       if (sl->GetSocket()->Send(buf, kPROOF_SENDFILE) == -1) {
 777         Warning("SendFile", "Could not send kPROOF_SENDFILE request");
 778         continue;
 779       }
 780
 781       // Go to the beginning of the file
 782       lseek(fd, 0, SEEK_SET);
 783       Int_t len = 0;
 784       do {
 785         while ((len = read(fd, buf, bufSize)) < 0 &&
 786                TSystem::GetErrno() == EINTR)
 787           TSystem::ResetErrno();
 788         if (len < 0) {
 789           Error("SendFile", "error reading input");
 790           close(fd);
 791           return -1;
 792         }
 793         if (len > 0 && sl->GetSocket()->SendRaw(buf, len) == -1) {
 794           Error("SendFile", "error writing to slave");
 795           sl = 0;
 796           break;
 797         }
 798       } while (len > 0);
 799       ret ++;
 800
 801       // Wait for slave - private
 802       // if (sl) gProof->Collect(sl,gEnv->GetValue("Proof.CollectTimeout",-1));
 803     }
 804
 805     // Close the file
 806     close(fd);
 807
 808     return ret;
 809   }
 810   /**
 811    * Path of output
 812    *
 813    * @return Path to output - possibly a data set
 814    */
 815   virtual TString OutputPath() const
 816   {
 817     TString ret;
 818     if (fOptions.Has("dsname")) {
 819       ret = Form("/%s/%s/", gProof->GetGroup(), gProof->GetUser());
 820       ret.Append(OutputUtilities::RegisteredDataset());
 821     }
 822     return ret;
 823   }
 824   /**
 825    * @return URL help string
 826    */
 827   virtual const Char_t* UrlHelp() const
 828   {
 829     return "proof://<host>[:<port>]/[<dataset>|<path>][?<options>][#<treeName>]";
 830   }
 831   /**
 832    * @return Short description
 833    */
 834   virtual const char* Desc() const { return "PROOF"; }
 835   TString fExtraLibs;
 836   TString fExtraPars;
 837   TString fExtraSrcs;
 838   Bool_t  fUsePars;
 839   Bool_t  fBasePars;
 840   Bool_t  fTestBuild;
 841   TList   fAuxFiles;
 842 };
 843 #endif
 844 //
 845 // EOF
 846 //