<commit_before>//
// Macro to run examples of analysis on PROOF.
// This macro uses an existing PROOF session or starts one at the indicated URL.
// In the case non existing PROOF session is found and no URL is given, the macro
// tries to start a local PROOF session.
//
// To run the macro:
//
//   root[] .L proof/runProof.C+
//   root[] runProof("<analysis>")
//
//   Currently available analysis are:
//
//   1. "simple"
//
//      root[] runProof("simple")
//
//      This will create a local PROOF session and run an analysis filling 100 histos
//      with 100000 gaussian random numbers, and displaying them in a canvas with 100
//      pads (10x10).
//      The number of histograms can be passed as 'arguments' to 'simple': e.g. to fill
//      16 histos with 1000000 entries use
//
//      root[] runProof("simple(nevt=1000000,nhist=16)")
//
//   2. "h1"
//
//      root[] runProof("h1")
//
//      This runs the 'famous' H1 analysis from $ROOTSYS/tree/h1analysis.C,.h .
//      The data are read from the HTTP server at root.cern.ch .
//
//  3. "event"
//
//      This is an example of using PROOF par files.
//      It runs event generation and simple analysis based on the 'Event' class found
//      under test.
//
//   root[] runProof("event")
//
//  4. "eventproc"
//
//      This is an example of using PROOF par files and process 'event' data from the root HTTP
//      server. It runs the ProofEventProc selector which is derived from the EventTree_Proc
//      one found under test/ProofBench.
//      The following specific arguments are available:
//      - 'readall'  to read the whole event; by default only the braches needed by the analysis
//                   are read (read 25% more bytes)
//      - 'datasrc=<dir-with-files>' to read the files from another server; the files must be named
//                                   'event_<num>.root' where <num>=1,2,...
//      - 'files=N' to change the number of files to be analysed (default is 10; max is 50 for
//                  the HTTP server).
//
//   root[] runProof("eventproc")
//
//  5. "pythia8"
//
//      This runs Pythia8 generation based on main03.cc example in Pythia 8.1 
//
//      To run this analysis ROOT must be configured with pythia8.
//
//      Note that before executing this analysis, the env variable PYTHIA8 must point
//      to the pythia8100 (or newer) directory; in particular, $PYTHIA8/xmldoc must
//      contain the file Index.xml; the tutorial assumes that the Pythia8 directory
//      is the same on all machines, i.e. local and worker ones
//
//   root[] runProof("pythia8")
//
//  6. "ntuple"
//
//      This is an example of final merging via files created on the workers, using
//      TProofOutputFile. The final file is called ProofNtuple.root and it is created
//      in the directory where the tutorial is run. If the PROOF cluster is remote, the
//      file is received by a local xrootd daemon started for the purpose. Because of
//      this, this example can be run only on unix clients.
//
//   root[] runProof("ntuple")
//
//  7. "dataset"
//
//      This is an example of automatic creation of a dataset from files created on the
//      workers, using TProofOutputFile. The dataset is called testNtuple and it is
//      automatically registered and verified. The files contain the same ntuple as in
//      previous example/tutorial 6 (the same selector ProofNTuple is used with a slightly
//      different configuration). The dataset is then used to produce the same plot
//      as in 5 but using the DrawSelect methods of PROOF, which also show how to set
//      style, color and other drawing attributes in PROOF. Depending on the relative
//      worker perforance, some of the produced files may result in having no entries;
//      if this happens, the file will be added to the missing (skipped) file list.
//      Increasing the number of events (via nevt=...) typically solves this issue.
//
//   root[] runProof("dataset")
//
//   General arguments
//   -----------------
//
//   The following arguments are valid for all examples (the ones specific to each
//   tutorial have been explained above)
//
//   1. nevt=N
//
//      Set the number of entries to N
//      E.g. runProof("simple(nevt=1000000000)") runs simple with 1000000000
//
//   2. asyn
//
//      Run in non blocking mode
//      E.g. root[] runProof("h1(asyn)")
//
//   3. nwrk=N
//
//      Set the number of active workers to N; usefull to test performance on a remote
//      cluster where control about the number of workers is not possible.
//      E.g. runProof("event(nwrk=2)") runs 'event' with 2 workers
//
//   4. punzip
//
//      Use parallel unzipping in reading files where relevant
//      E.g. root[] runProof("eventproc(punzip)")
//
//   5. cache=<bytes> (or <kbytes>K or <mbytes>M) 
//
//      Change the size of the tree cache; 0 or <0 disables the cache; value cane be in
//      bytes (no suffix), kilobytes (suffix 'K') or megabytes (suffix 'M')
//      E.g. root[] runProof("eventproc(cache=0)") 
//
//
//
//   In all cases, to run on a remote Proof cluster, the master URL must be passed as
//   second argument; e.g.
//
//   root[] runProof("simple","master.domain")
//
//   In the case of local running it is possible to specify the number of workers to
//   start as third argument (the default is the number of cores of the machine), e.g.
//
//   root[] runProof("simple",0,4)
//
//   will start 4 workers. Note that the real number of workers is changed only the
//   first time you call runProof into a ROOT session; following calls can reduce the
//   number of active workers, but not increase it. For example, in the same session of
//   the call above starting 4 workers, this
//
//   root[] runProof("simple",0,8)
//
//   will still use 4 workers, while this
//
//   root[] runProof("simple",0,2)
//
//   will disable 2 workers and use the other 2.
//


#include "TCanvas.h"
#include "TChain.h"
#include "TEnv.h"
#include "TFileCollection.h"
#include "TFrame.h"
#include "TProof.h"
#include "TString.h"
#include "TDrawFeedback.h"
#include "THashList.h"
#include "TList.h"
#include "TPad.h"
#include "TPaveText.h"

#include "getProof.C"
void plotNtuple(TProof *p, const char *ds, const char *ntptitle);

TDrawFeedback *fb = 0;

// Variable used to locate the Pythia8 directory for the Pythia8 example
const char *pythia8dir = 0;
const char *pythia8data = 0;

void runProof(const char *what = "simple",
              const char *url = "proof://localhost:11093",
              Int_t nwrks = -1)
{
   gEnv->SetValue("Proof.StatsHist",1);

   // Temp dir for PROOF tutorials
   // Force "/tmp/<user>" whenever possible to avoid length problems on MacOsX
   TString tmpdir("/tmp");
   if (gSystem->AccessPathName(tmpdir, kWritePermission)) tmpdir = gSystem->TempDirectory();
   TString us;
   UserGroup_t *ug = gSystem->GetUserInfo(gSystem->GetUid());
   if (!ug) {
      Printf("runProof: could not get user info");
      return;
   }
   us.Form("/%s", ug->fUser.Data());
   if (!tmpdir.EndsWith(us.Data())) tmpdir += us;
   gSystem->mkdir(tmpdir.Data(), kTRUE);
   if (gSystem->AccessPathName(tmpdir, kWritePermission)) {
      Printf("runProof: unable to get a writable tutorial directory (tried: %s)"
             " - cannot continue", tmpdir.Data());
      return;
   }
   TString tutdir = Form("%s/.proof-tutorial", tmpdir.Data());
   if (gSystem->AccessPathName(tutdir)) {
      Printf("runProof: creating the temporary directory"
                " for the tutorial (%s) ... ", tutdir.Data());
      if (gSystem->mkdir(tutdir, kTRUE) != 0) {
         Printf("runProof: could not assert / create the temporary directory"
                " for the tutorial (%s)", tutdir.Data());
         return;
      }
   }

   // For the Pythia8 example we need to set some environment variable;
   // This must be done BEFORE starting the PROOF session
   if (what && !strncmp(what, "pythia8", 7)) {
      // We assume that the remote location of Pythia8 is the same as the local one
      pythia8dir = gSystem->Getenv("PYTHIA8");
      if (!pythia8dir || strlen(pythia8dir) <= 0) {
         Printf("runProof: pythia8: environment variable PYTHIA8 undefined:"
                  " it must contain the path to pythia81xx root directory (local and remote) !");
         return;
      }
      pythia8data = gSystem->Getenv("PYTHIA8DATA");
      if (!pythia8data || strlen(pythia8data) <= 0) {
         gSystem->Setenv("PYTHIA8DATA", Form("%s/xmldoc", pythia8dir));
         pythia8data = gSystem->Getenv("PYTHIA8DATA");
         if (!pythia8data || strlen(pythia8data) <= 0) {
            Printf("runProof: pythia8: environment variable PYTHIA8DATA undefined:"
                   " it one must contain the path to pythia81xx/xmldoc"
                   " subdirectory (local and remote) !");
            return;
         }
      }
      TString env = Form("echo export PYTHIA8=%s; export PYTHIA8DATA=%s",
                         pythia8dir, pythia8data);
      TProof::AddEnvVar("PROOF_INITCMD", env.Data());
   }

   // Get the PROOF Session
   TProof *proof = getProof(url, nwrks, tutdir.Data(), "ask");
   if (!proof) {
      Printf("runProof: could not start/attach a PROOF session");
      return;
   }

   // Determine locality of this session
   Bool_t isProofLocal = kFALSE;
   TUrl uu(url);
   if (!strcmp(uu.GetHost(), "localhost") ||
       !strcmp(uu.GetHostFQDN(), TUrl(gSystem->HostName()).GetHostFQDN())) {
      isProofLocal = kTRUE;
   }
#ifdef WIN32
   if (isProofLocal && what && !strcmp(what, "ntuple", 6)) {
      // Not support on windows
      Printf("runProof: the 'ntuple' example needs to run xrootd to receive the output file, \n"
             "          but xrootd is not supported on Windows - cannot continue");
      return;
   }
#endif

   TString proofsessions(Form("%s/sessions",tutdir.Data()));
   // Save tag of the used session
   FILE *fs = fopen(proofsessions.Data(), "a");
   if (!fs) {
      Printf("runProof: could not create files for sessions tags");
   } else {
      fprintf(fs,"session-%s\n", proof->GetSessionTag());
      fclose(fs);
   }
   if (!proof) {
      Printf("runProof: could not start/attach a PROOF session");
      return;
   }

   // Set the number of workers (may only reduce the number of active workers
   // in the session)
   if (nwrks > 0)
      proof->SetParallel(nwrks);

   // Where is the code to run
   char *rootbin = gSystem->Which(gSystem->Getenv("PATH"), "root.exe", kExecutePermission);
   if (!rootbin) {
      Printf("runProof: root.exe not found: please check the environment!");
      return;
   }
   TString rootsys(gSystem->DirName(rootbin));
   rootsys = gSystem->DirName(rootsys);
   TString tutorials(Form("%s/tutorials", rootsys.Data()));
   delete[] rootbin;

   // Create feedback displayer
   if (!fb) {
      new TCanvas("PROOF_EventsHist_canvas", "Events per Worker", 100, 600, 600, 400);
      fb = new TDrawFeedback(proof);
   }
   if (!proof->GetFeedbackList() || !proof->GetFeedbackList()->FindObject("PROOF_EventsHist")) {
      // Number of events per worker
      proof->AddFeedback("PROOF_EventsHist");
   }

   // Have constant progress reporting based on estimated info
   proof->SetParameter("PROOF_RateEstimation", "average");

   // Parse 'what'; it is in the form 'analysis(arg1,arg2,...)'
   TString args(what);
   args.ReplaceAll("("," "); 
   args.ReplaceAll(")"," "); 
   args.ReplaceAll(","," "); 
   Ssiz_t from = 0;
   TString act, tok;
   if (!args.Tokenize(act, from, " ")) {
      // Cannot continue
      Printf("runProof: action not found: check your arguments (%s)", what);
      return;
   }

   // Parse out number of events and  'asyn' option, used almost by every test
   TString aNevt, aNwrk, opt, sel, punzip("off"), aCache;
   Long64_t suf = 1;
   while (args.Tokenize(tok, from, " ")) {
      // Number of events
      if (tok.BeginsWith("nevt=")) {
         aNevt = tok;
         aNevt.ReplaceAll("nevt=","");
         if (!aNevt.IsDigit()) {
            Printf("runProof: %s: error parsing the 'nevt=' option (%s) - ignoring", act.Data(), tok.Data());
            aNevt = "";
         }
      }
      // Sync or async ?
      if (tok.BeginsWith("asyn"))
         opt = "ASYN";
      // Number of workers
      if (tok.BeginsWith("nwrk=")) {
         aNwrk = tok;
         aNwrk.ReplaceAll("nwrk=","");
         if (!aNwrk.IsDigit()) {
            Printf("runProof: %s: error parsing the 'nwrk=' option (%s) - ignoring", act.Data(), tok.Data());
            aNwrk = "";
         }
      }
      // Parallel unzipping ?
      if (tok.BeginsWith("punzip"))
         punzip = "on";
      // Number of workers
      if (tok.BeginsWith("cache=")) {
         aCache = tok;
         aCache.ReplaceAll("cache=","");
         if (aCache.EndsWith("k")) { aCache.Remove(TString::kTrailing, 'k'); suf = 1024; }
         if (aCache.EndsWith("K")) { aCache.Remove(TString::kTrailing, 'K'); suf = 1024; }
         if (aCache.EndsWith("M")) { aCache.Remove(TString::kTrailing, 'M'); suf = 1024*1024; }
         if (!aCache.IsDigit()) {
            Printf("runProof: %s: error parsing the 'cache=' option (%s) - ignoring", act.Data(), tok.Data());
            aCache = "";
         }
      }
   }
   Long64_t nevt = (aNevt.IsNull()) ? -1 : aNevt.Atoi();
   Long64_t nwrk = (aNwrk.IsNull()) ? -1 : aNwrk.Atoi();
   from = 0;

   // Set number workers
   if (nwrk > 0) {
      if (proof->GetParallel() < nwrk) {
         Printf("runProof: %s: request for a number of workers larger then available - ignored", act.Data());
      } else {
         proof->SetParallel(nwrk);
      }
   }

   // Parallel unzip
   if (punzip == "on") {
      proof->SetParameter("PROOF_UseParallelUnzip", (Int_t)1);
      Printf("runProof: %s: parallel unzip enabled", act.Data());
   } else {
      proof->SetParameter("PROOF_UseParallelUnzip", (Int_t)0);
   }

   // Tree cache
   if (!aCache.IsNull()) {
      Long64_t cachesz = aCache.Atoi() * suf;
      if (cachesz <= 0) {
         proof->SetParameter("PROOF_UseTreeCache", (Int_t)0);
         Printf("runProof: %s: disabling tree cache", act.Data());
      } else {
         proof->SetParameter("PROOF_UseTreeCache", (Int_t)1);
         proof->SetParameter("PROOF_CacheSize", cachesz);
         Printf("runProof: %s: setting cache size to %lld", act.Data(), cachesz);
      }
   } else {
      // Use defaults
      proof->DeleteParameters("PROOF_UseTreeCache");
      proof->DeleteParameters("PROOF_CacheSize");
   }

   // Action
   if (act == "simple") {
      // ProofSimple is an example of non-data driven analysis; it
      // creates and fills with random numbers a given number of histos

      // Default 10000 events
      nevt = (nevt < 0) ? 100000 : nevt;
      // Find out the number of histograms
      TString aNhist;
      while (args.Tokenize(tok, from, " ")) {
         // Number of histos
         if (tok.BeginsWith("nhist=")) {
            aNhist = tok;
            aNhist.ReplaceAll("nhist=","");
            if (!aNhist.IsDigit()) {
               Printf("runProof: error parsing the 'nhist=' option (%s) - ignoring", tok.Data());
               aNhist = "";
            }
         }
      }
      Int_t nhist = (aNhist.IsNull()) ? 100 : aNhist.Atoi();
      Printf("\nrunProof: running \"simple\" with nhist= %d and nevt= %d\n", nhist, nevt);

      // The number of histograms is added as parameter in the input list
      proof->SetParameter("ProofSimple_NHist", (Long_t)nhist);
      // The selector string
      sel.Form("%s/proof/ProofSimple.C+", tutorials.Data());
      //
      // Run it for nevt times
      proof->Process(sel.Data(), nevt, opt);

   } else if (act == "h1") {
      // This is the famous 'h1' example analysis run on Proof reading the
      // data from the ROOT http server.

      // Create the chain
      TChain *chain = new TChain("h42");
      chain->Add("http://root.cern.ch/files/h1/dstarmb.root");
      chain->Add("http://root.cern.ch/files/h1/dstarp1a.root");
      chain->Add("http://root.cern.ch/files/h1/dstarp1b.root");
      chain->Add("http://root.cern.ch/files/h1/dstarp2.root");
      // We run on Proof
      chain->SetProof();
      // The selector
      sel.Form("%s/tree/h1analysis.C+", tutorials.Data());
      // Run it 
      Printf("\nrunProof: running \"h1\"\n");
      chain->Process(sel.Data(),opt);

   } else if (act == "pythia8") {

      TString path(Form("%s/Index.xml", pythia8data));
      gSystem->ExpandPathName(path);
      if (gSystem->AccessPathName(path)) {
         Printf("runProof: pythia8: PYTHIA8DATA directory (%s) must"
                " contain the Index.xml file !", pythia8data);
         return;
      }
      TString pythia8par("proof/pythia8");
      if (gSystem->AccessPathName(Form("%s.par", pythia8par.Data()))) {
         pythia8par = "pythia8";
         if (gSystem->AccessPathName(Form("%s.par", pythia8par.Data()))) {
            Printf("runProof: pythia8: par file not found: tried 'proof/pythia8.par'"
                   " and 'pythia8.par'");
            return;
         }
      }
      proof->UploadPackage(pythia8par);
      proof->EnablePackage("pythia8");
      // Show enabled packages
      proof->ShowEnabledPackages();
      Printf("runProof: pythia8: check settings:");
      proof->Exec(".!echo hostname = `hostname`; echo \"ls pythia8:\"; ls pythia8");
      // Loading libraries needed
      if (gSystem->Load("libEG.so") < 0) {
         Printf("runProof: pythia8: libEG not found \n");
         return;
      }
      if (gSystem->Load("libEGPythia8.so") < 0) {
         Printf("runProof: pythia8: libEGPythia8 not found \n");
         return;
      }
      // Setting the default number of events, if needed
      nevt = (nevt < 0) ? 100 : nevt;
      Printf("\nrunProof: running \"Pythia01\" nevt= %d\n", nevt);
      // The selector string
      sel.Form("%s/proof/ProofPythia.C+", tutorials.Data());
      // Run it for nevt times
      proof->Process(sel.Data(), nevt);

  } else if (act == "event") {

      TString eventpar("proof/event");
      if (gSystem->AccessPathName(Form("%s.par", eventpar.Data()))) {
         eventpar = "event";
         if (gSystem->AccessPathName(Form("%s.par", eventpar.Data()))) {
            Printf("runProof: event: par file not found: tried 'proof/event.par'"
                   " and 'event.par'");
            return;
         }
      }

      proof->UploadPackage(eventpar);
      proof->EnablePackage("event");
      Printf("Enabled packages...\n");
      proof->ShowEnabledPackages(); 

      // Setting the default number of events, if needed
      nevt = (nevt < 0) ? 100 : nevt;
      Printf("\nrunProof: running \"event\" nevt= %d\n", nevt);
      // The selector string
      sel.Form("%s/proof/ProofEvent.C+", tutorials.Data());
      // Run it for nevt times
      proof->Process(sel.Data(), nevt);

  } else if (act == "eventproc") {

      TString eventpar = TString::Format("%s/proof/event.par", tutorials.Data());
      gSystem->ExpandPathName(eventpar);
      if (gSystem->AccessPathName(eventpar.Data())) {
         eventpar = gSystem->BaseName(eventpar.Data());
         if (gSystem->AccessPathName(eventpar.Data())) {
            Printf("runProof: event: par file not found: tried 'proof/event.par'"
                   " and 'event.par'");
            return;
         }
      }

      proof->UploadPackage(eventpar);
      proof->EnablePackage("event");
      Printf("Enabled packages...\n");
      proof->ShowEnabledPackages(); 

      // Extract the number of files to process, data source and
      // other parameters controlling the run ... 
      TString aFiles, aDataSrc("http://root.cern.ch/files/data");
      proof->SetParameter("ProofEventProc_Read", "optimized");
      while (args.Tokenize(tok, from, " ")) {
         // Number of events
         if (tok.BeginsWith("files=")) {
            aFiles = tok;
            aFiles.ReplaceAll("files=","");
            if (!aFiles.IsDigit()) {
               Printf("runProof: error parsing the 'files=' option (%s) - ignoring", tok.Data());
               aFiles = "";
            }
         } else if (tok.BeginsWith("datasrc=")) {
            aDataSrc = tok;
            aDataSrc.ReplaceAll("datasrc=","");
            if (aDataSrc.IsDigit()) {
               Printf("runProof: error parsing the 'datasrc=' option (%s) - ignoring", tok.Data());
               aFiles = "";
            }
         } else if (tok == "readall") {
            proof->SetParameter("ProofEventProc_Read", "readall");
            Printf("runProof: eventproc: reading the full event");
         }
      }
      Int_t nFiles = (aFiles.IsNull()) ? 10 : aFiles.Atoi();
         Printf("runProof: found aFiles: '%s', nFiles: %d", aFiles.Data(), nFiles);
      if (nFiles > 50) {
         Printf("runProof: max number of files is 50 - resizing request");
         nFiles = 50;
      }

      // Create the chain
      TChain *c = new TChain("EventTree");
      Int_t i = 1;
      TString fn;
      for (i = 1; i <= nFiles; i++) {
         fn.Form("%s/event_%d.root", aDataSrc.Data(), i);
         c->AddFile(fn.Data());
      }
      c->SetProof();

      // The selector
      sel.Form("%s/proof/ProofEventProc.C+", tutorials.Data());
      // Run it
      Printf("\nrunProof: running \"eventproc\"\n");
      c->Process(sel.Data(), opt);<commit_msg>Notify the data source for the 'eventproc' tutorial<commit_after>//
// Macro to run examples of analysis on PROOF.
// This macro uses an existing PROOF session or starts one at the indicated URL.
// In the case non existing PROOF session is found and no URL is given, the macro
// tries to start a local PROOF session.
//
// To run the macro:
//
//   root[] .L proof/runProof.C+
//   root[] runProof("<analysis>")
//
//   Currently available analysis are:
//
//   1. "simple"
//
//      root[] runProof("simple")
//
//      This will create a local PROOF session and run an analysis filling 100 histos
//      with 100000 gaussian random numbers, and displaying them in a canvas with 100
//      pads (10x10).
//      The number of histograms can be passed as 'arguments' to 'simple': e.g. to fill
//      16 histos with 1000000 entries use
//
//      root[] runProof("simple(nevt=1000000,nhist=16)")
//
//   2. "h1"
//
//      root[] runProof("h1")
//
//      This runs the 'famous' H1 analysis from $ROOTSYS/tree/h1analysis.C,.h .
//      The data are read from the HTTP server at root.cern.ch .
//
//  3. "event"
//
//      This is an example of using PROOF par files.
//      It runs event generation and simple analysis based on the 'Event' class found
//      under test.
//
//   root[] runProof("event")
//
//  4. "eventproc"
//
//      This is an example of using PROOF par files and process 'event' data from the root HTTP
//      server. It runs the ProofEventProc selector which is derived from the EventTree_Proc
//      one found under test/ProofBench.
//      The following specific arguments are available:
//      - 'readall'  to read the whole event; by default only the braches needed by the analysis
//                   are read (read 25% more bytes)
//      - 'datasrc=<dir-with-files>' to read the files from another server; the files must be named
//                                   'event_<num>.root' where <num>=1,2,...
//      - 'files=N' to change the number of files to be analysed (default is 10; max is 50 for
//                  the HTTP server).
//
//   root[] runProof("eventproc")
//
//  5. "pythia8"
//
//      This runs Pythia8 generation based on main03.cc example in Pythia 8.1 
//
//      To run this analysis ROOT must be configured with pythia8.
//
//      Note that before executing this analysis, the env variable PYTHIA8 must point
//      to the pythia8100 (or newer) directory; in particular, $PYTHIA8/xmldoc must
//      contain the file Index.xml; the tutorial assumes that the Pythia8 directory
//      is the same on all machines, i.e. local and worker ones
//
//   root[] runProof("pythia8")
//
//  6. "ntuple"
//
//      This is an example of final merging via files created on the workers, using
//      TProofOutputFile. The final file is called ProofNtuple.root and it is created
//      in the directory where the tutorial is run. If the PROOF cluster is remote, the
//      file is received by a local xrootd daemon started for the purpose. Because of
//      this, this example can be run only on unix clients.
//
//   root[] runProof("ntuple")
//
//  7. "dataset"
//
//      This is an example of automatic creation of a dataset from files created on the
//      workers, using TProofOutputFile. The dataset is called testNtuple and it is
//      automatically registered and verified. The files contain the same ntuple as in
//      previous example/tutorial 6 (the same selector ProofNTuple is used with a slightly
//      different configuration). The dataset is then used to produce the same plot
//      as in 5 but using the DrawSelect methods of PROOF, which also show how to set
//      style, color and other drawing attributes in PROOF. Depending on the relative
//      worker perforance, some of the produced files may result in having no entries;
//      if this happens, the file will be added to the missing (skipped) file list.
//      Increasing the number of events (via nevt=...) typically solves this issue.
//
//   root[] runProof("dataset")
//
//   General arguments
//   -----------------
//
//   The following arguments are valid for all examples (the ones specific to each
//   tutorial have been explained above)
//
//   1. nevt=N
//
//      Set the number of entries to N
//      E.g. runProof("simple(nevt=1000000000)") runs simple with 1000000000
//
//   2. asyn
//
//      Run in non blocking mode
//      E.g. root[] runProof("h1(asyn)")
//
//   3. nwrk=N
//
//      Set the number of active workers to N; usefull to test performance on a remote
//      cluster where control about the number of workers is not possible.
//      E.g. runProof("event(nwrk=2)") runs 'event' with 2 workers
//
//   4. punzip
//
//      Use parallel unzipping in reading files where relevant
//      E.g. root[] runProof("eventproc(punzip)")
//
//   5. cache=<bytes> (or <kbytes>K or <mbytes>M) 
//
//      Change the size of the tree cache; 0 or <0 disables the cache; value cane be in
//      bytes (no suffix), kilobytes (suffix 'K') or megabytes (suffix 'M')
//      E.g. root[] runProof("eventproc(cache=0)") 
//
//
//
//   In all cases, to run on a remote Proof cluster, the master URL must be passed as
//   second argument; e.g.
//
//   root[] runProof("simple","master.domain")
//
//   In the case of local running it is possible to specify the number of workers to
//   start as third argument (the default is the number of cores of the machine), e.g.
//
//   root[] runProof("simple",0,4)
//
//   will start 4 workers. Note that the real number of workers is changed only the
//   first time you call runProof into a ROOT session; following calls can reduce the
//   number of active workers, but not increase it. For example, in the same session of
//   the call above starting 4 workers, this
//
//   root[] runProof("simple",0,8)
//
//   will still use 4 workers, while this
//
//   root[] runProof("simple",0,2)
//
//   will disable 2 workers and use the other 2.
//


#include "TCanvas.h"
#include "TChain.h"
#include "TEnv.h"
#include "TFileCollection.h"
#include "TFrame.h"
#include "TProof.h"
#include "TString.h"
#include "TDrawFeedback.h"
#include "THashList.h"
#include "TList.h"
#include "TPad.h"
#include "TPaveText.h"

#include "getProof.C"
void plotNtuple(TProof *p, const char *ds, const char *ntptitle);

TDrawFeedback *fb = 0;

// Variable used to locate the Pythia8 directory for the Pythia8 example
const char *pythia8dir = 0;
const char *pythia8data = 0;

void runProof(const char *what = "simple",
              const char *url = "proof://localhost:11093",
              Int_t nwrks = -1)
{
   gEnv->SetValue("Proof.StatsHist",1);

   // Temp dir for PROOF tutorials
   // Force "/tmp/<user>" whenever possible to avoid length problems on MacOsX
   TString tmpdir("/tmp");
   if (gSystem->AccessPathName(tmpdir, kWritePermission)) tmpdir = gSystem->TempDirectory();
   TString us;
   UserGroup_t *ug = gSystem->GetUserInfo(gSystem->GetUid());
   if (!ug) {
      Printf("runProof: could not get user info");
      return;
   }
   us.Form("/%s", ug->fUser.Data());
   if (!tmpdir.EndsWith(us.Data())) tmpdir += us;
   gSystem->mkdir(tmpdir.Data(), kTRUE);
   if (gSystem->AccessPathName(tmpdir, kWritePermission)) {
      Printf("runProof: unable to get a writable tutorial directory (tried: %s)"
             " - cannot continue", tmpdir.Data());
      return;
   }
   TString tutdir = Form("%s/.proof-tutorial", tmpdir.Data());
   if (gSystem->AccessPathName(tutdir)) {
      Printf("runProof: creating the temporary directory"
                " for the tutorial (%s) ... ", tutdir.Data());
      if (gSystem->mkdir(tutdir, kTRUE) != 0) {
         Printf("runProof: could not assert / create the temporary directory"
                " for the tutorial (%s)", tutdir.Data());
         return;
      }
   }

   // For the Pythia8 example we need to set some environment variable;
   // This must be done BEFORE starting the PROOF session
   if (what && !strncmp(what, "pythia8", 7)) {
      // We assume that the remote location of Pythia8 is the same as the local one
      pythia8dir = gSystem->Getenv("PYTHIA8");
      if (!pythia8dir || strlen(pythia8dir) <= 0) {
         Printf("runProof: pythia8: environment variable PYTHIA8 undefined:"
                  " it must contain the path to pythia81xx root directory (local and remote) !");
         return;
      }
      pythia8data = gSystem->Getenv("PYTHIA8DATA");
      if (!pythia8data || strlen(pythia8data) <= 0) {
         gSystem->Setenv("PYTHIA8DATA", Form("%s/xmldoc", pythia8dir));
         pythia8data = gSystem->Getenv("PYTHIA8DATA");
         if (!pythia8data || strlen(pythia8data) <= 0) {
            Printf("runProof: pythia8: environment variable PYTHIA8DATA undefined:"
                   " it one must contain the path to pythia81xx/xmldoc"
                   " subdirectory (local and remote) !");
            return;
         }
      }
      TString env = Form("echo export PYTHIA8=%s; export PYTHIA8DATA=%s",
                         pythia8dir, pythia8data);
      TProof::AddEnvVar("PROOF_INITCMD", env.Data());
   }

   // Get the PROOF Session
   TProof *proof = getProof(url, nwrks, tutdir.Data(), "ask");
   if (!proof) {
      Printf("runProof: could not start/attach a PROOF session");
      return;
   }

   // Determine locality of this session
   Bool_t isProofLocal = kFALSE;
   TUrl uu(url);
   if (!strcmp(uu.GetHost(), "localhost") ||
       !strcmp(uu.GetHostFQDN(), TUrl(gSystem->HostName()).GetHostFQDN())) {
      isProofLocal = kTRUE;
   }
#ifdef WIN32
   if (isProofLocal && what && !strcmp(what, "ntuple", 6)) {
      // Not support on windows
      Printf("runProof: the 'ntuple' example needs to run xrootd to receive the output file, \n"
             "          but xrootd is not supported on Windows - cannot continue");
      return;
   }
#endif

   TString proofsessions(Form("%s/sessions",tutdir.Data()));
   // Save tag of the used session
   FILE *fs = fopen(proofsessions.Data(), "a");
   if (!fs) {
      Printf("runProof: could not create files for sessions tags");
   } else {
      fprintf(fs,"session-%s\n", proof->GetSessionTag());
      fclose(fs);
   }
   if (!proof) {
      Printf("runProof: could not start/attach a PROOF session");
      return;
   }

   // Set the number of workers (may only reduce the number of active workers
   // in the session)
   if (nwrks > 0)
      proof->SetParallel(nwrks);

   // Where is the code to run
   char *rootbin = gSystem->Which(gSystem->Getenv("PATH"), "root.exe", kExecutePermission);
   if (!rootbin) {
      Printf("runProof: root.exe not found: please check the environment!");
      return;
   }
   TString rootsys(gSystem->DirName(rootbin));
   rootsys = gSystem->DirName(rootsys);
   TString tutorials(Form("%s/tutorials", rootsys.Data()));
   delete[] rootbin;

   // Create feedback displayer
   if (!fb) {
      new TCanvas("PROOF_EventsHist_canvas", "Events per Worker", 100, 600, 600, 400);
      fb = new TDrawFeedback(proof);
   }
   if (!proof->GetFeedbackList() || !proof->GetFeedbackList()->FindObject("PROOF_EventsHist")) {
      // Number of events per worker
      proof->AddFeedback("PROOF_EventsHist");
   }

   // Have constant progress reporting based on estimated info
   proof->SetParameter("PROOF_RateEstimation", "average");

   // Parse 'what'; it is in the form 'analysis(arg1,arg2,...)'
   TString args(what);
   args.ReplaceAll("("," "); 
   args.ReplaceAll(")"," "); 
   args.ReplaceAll(","," "); 
   Ssiz_t from = 0;
   TString act, tok;
   if (!args.Tokenize(act, from, " ")) {
      // Cannot continue
      Printf("runProof: action not found: check your arguments (%s)", what);
      return;
   }

   // Parse out number of events and  'asyn' option, used almost by every test
   TString aNevt, aNwrk, opt, sel, punzip("off"), aCache;
   Long64_t suf = 1;
   while (args.Tokenize(tok, from, " ")) {
      // Number of events
      if (tok.BeginsWith("nevt=")) {
         aNevt = tok;
         aNevt.ReplaceAll("nevt=","");
         if (!aNevt.IsDigit()) {
            Printf("runProof: %s: error parsing the 'nevt=' option (%s) - ignoring", act.Data(), tok.Data());
            aNevt = "";
         }
      }
      // Sync or async ?
      if (tok.BeginsWith("asyn"))
         opt = "ASYN";
      // Number of workers
      if (tok.BeginsWith("nwrk=")) {
         aNwrk = tok;
         aNwrk.ReplaceAll("nwrk=","");
         if (!aNwrk.IsDigit()) {
            Printf("runProof: %s: error parsing the 'nwrk=' option (%s) - ignoring", act.Data(), tok.Data());
            aNwrk = "";
         }
      }
      // Parallel unzipping ?
      if (tok.BeginsWith("punzip"))
         punzip = "on";
      // Number of workers
      if (tok.BeginsWith("cache=")) {
         aCache = tok;
         aCache.ReplaceAll("cache=","");
         if (aCache.EndsWith("k")) { aCache.Remove(TString::kTrailing, 'k'); suf = 1024; }
         if (aCache.EndsWith("K")) { aCache.Remove(TString::kTrailing, 'K'); suf = 1024; }
         if (aCache.EndsWith("M")) { aCache.Remove(TString::kTrailing, 'M'); suf = 1024*1024; }
         if (!aCache.IsDigit()) {
            Printf("runProof: %s: error parsing the 'cache=' option (%s) - ignoring", act.Data(), tok.Data());
            aCache = "";
         }
      }
   }
   Long64_t nevt = (aNevt.IsNull()) ? -1 : aNevt.Atoi();
   Long64_t nwrk = (aNwrk.IsNull()) ? -1 : aNwrk.Atoi();
   from = 0;

   // Set number workers
   if (nwrk > 0) {
      if (proof->GetParallel() < nwrk) {
         Printf("runProof: %s: request for a number of workers larger then available - ignored", act.Data());
      } else {
         proof->SetParallel(nwrk);
      }
   }

   // Parallel unzip
   if (punzip == "on") {
      proof->SetParameter("PROOF_UseParallelUnzip", (Int_t)1);
      Printf("runProof: %s: parallel unzip enabled", act.Data());
   } else {
      proof->SetParameter("PROOF_UseParallelUnzip", (Int_t)0);
   }

   // Tree cache
   if (!aCache.IsNull()) {
      Long64_t cachesz = aCache.Atoi() * suf;
      if (cachesz <= 0) {
         proof->SetParameter("PROOF_UseTreeCache", (Int_t)0);
         Printf("runProof: %s: disabling tree cache", act.Data());
      } else {
         proof->SetParameter("PROOF_UseTreeCache", (Int_t)1);
         proof->SetParameter("PROOF_CacheSize", cachesz);
         Printf("runProof: %s: setting cache size to %lld", act.Data(), cachesz);
      }
   } else {
      // Use defaults
      proof->DeleteParameters("PROOF_UseTreeCache");
      proof->DeleteParameters("PROOF_CacheSize");
   }

   // Action
   if (act == "simple") {
      // ProofSimple is an example of non-data driven analysis; it
      // creates and fills with random numbers a given number of histos

      // Default 10000 events
      nevt = (nevt < 0) ? 100000 : nevt;
      // Find out the number of histograms
      TString aNhist;
      while (args.Tokenize(tok, from, " ")) {
         // Number of histos
         if (tok.BeginsWith("nhist=")) {
            aNhist = tok;
            aNhist.ReplaceAll("nhist=","");
            if (!aNhist.IsDigit()) {
               Printf("runProof: error parsing the 'nhist=' option (%s) - ignoring", tok.Data());
               aNhist = "";
            }
         }
      }
      Int_t nhist = (aNhist.IsNull()) ? 100 : aNhist.Atoi();
      Printf("\nrunProof: running \"simple\" with nhist= %d and nevt= %d\n", nhist, nevt);

      // The number of histograms is added as parameter in the input list
      proof->SetParameter("ProofSimple_NHist", (Long_t)nhist);
      // The selector string
      sel.Form("%s/proof/ProofSimple.C+", tutorials.Data());
      //
      // Run it for nevt times
      proof->Process(sel.Data(), nevt, opt);

   } else if (act == "h1") {
      // This is the famous 'h1' example analysis run on Proof reading the
      // data from the ROOT http server.

      // Create the chain
      TChain *chain = new TChain("h42");
      chain->Add("http://root.cern.ch/files/h1/dstarmb.root");
      chain->Add("http://root.cern.ch/files/h1/dstarp1a.root");
      chain->Add("http://root.cern.ch/files/h1/dstarp1b.root");
      chain->Add("http://root.cern.ch/files/h1/dstarp2.root");
      // We run on Proof
      chain->SetProof();
      // The selector
      sel.Form("%s/tree/h1analysis.C+", tutorials.Data());
      // Run it 
      Printf("\nrunProof: running \"h1\"\n");
      chain->Process(sel.Data(),opt);

   } else if (act == "pythia8") {

      TString path(Form("%s/Index.xml", pythia8data));
      gSystem->ExpandPathName(path);
      if (gSystem->AccessPathName(path)) {
         Printf("runProof: pythia8: PYTHIA8DATA directory (%s) must"
                " contain the Index.xml file !", pythia8data);
         return;
      }
      TString pythia8par("proof/pythia8");
      if (gSystem->AccessPathName(Form("%s.par", pythia8par.Data()))) {
         pythia8par = "pythia8";
         if (gSystem->AccessPathName(Form("%s.par", pythia8par.Data()))) {
            Printf("runProof: pythia8: par file not found: tried 'proof/pythia8.par'"
                   " and 'pythia8.par'");
            return;
         }
      }
      proof->UploadPackage(pythia8par);
      proof->EnablePackage("pythia8");
      // Show enabled packages
      proof->ShowEnabledPackages();
      Printf("runProof: pythia8: check settings:");
      proof->Exec(".!echo hostname = `hostname`; echo \"ls pythia8:\"; ls pythia8");
      // Loading libraries needed
      if (gSystem->Load("libEG.so") < 0) {
         Printf("runProof: pythia8: libEG not found \n");
         return;
      }
      if (gSystem->Load("libEGPythia8.so") < 0) {
         Printf("runProof: pythia8: libEGPythia8 not found \n");
         return;
      }
      // Setting the default number of events, if needed
      nevt = (nevt < 0) ? 100 : nevt;
      Printf("\nrunProof: running \"Pythia01\" nevt= %d\n", nevt);
      // The selector string
      sel.Form("%s/proof/ProofPythia.C+", tutorials.Data());
      // Run it for nevt times
      proof->Process(sel.Data(), nevt);

  } else if (act == "event") {

      TString eventpar("proof/event");
      if (gSystem->AccessPathName(Form("%s.par", eventpar.Data()))) {
         eventpar = "event";
         if (gSystem->AccessPathName(Form("%s.par", eventpar.Data()))) {
            Printf("runProof: event: par file not found: tried 'proof/event.par'"
                   " and 'event.par'");
            return;
         }
      }

      proof->UploadPackage(eventpar);
      proof->EnablePackage("event");
      Printf("Enabled packages...\n");
      proof->ShowEnabledPackages(); 

      // Setting the default number of events, if needed
      nevt = (nevt < 0) ? 100 : nevt;
      Printf("\nrunProof: running \"event\" nevt= %d\n", nevt);
      // The selector string
      sel.Form("%s/proof/ProofEvent.C+", tutorials.Data());
      // Run it for nevt times
      proof->Process(sel.Data(), nevt);

  } else if (act == "eventproc") {

      TString eventpar = TString::Format("%s/proof/event.par", tutorials.Data());
      gSystem->ExpandPathName(eventpar);
      if (gSystem->AccessPathName(eventpar.Data())) {
         eventpar = gSystem->BaseName(eventpar.Data());
         if (gSystem->AccessPathName(eventpar.Data())) {
            Printf("runProof: event: par file not found: tried 'proof/event.par'"
                   " and 'event.par'");
            return;
         }
      }

      proof->UploadPackage(eventpar);
      proof->EnablePackage("event");
      Printf("Enabled packages...\n");
      proof->ShowEnabledPackages(); 

      // Extract the number of files to process, data source and
      // other parameters controlling the run ... 
      TString aFiles, aDataSrc("http://root.cern.ch/files/data");
      proof->SetParameter("ProofEventProc_Read", "optimized");
      while (args.Tokenize(tok, from, " ")) {
         // Number of events
         if (tok.BeginsWith("files=")) {
            aFiles = tok;
            aFiles.ReplaceAll("files=","");
            if (!aFiles.IsDigit()) {
               Printf("runProof: error parsing the 'files=' option (%s) - ignoring", tok.Data());
               aFiles = "";
            }
         } else if (tok.BeginsWith("datasrc=")) {
            tok.ReplaceAll("datasrc=","");
            if (tok.IsDigit()) {
               Printf("runProof: error parsing the 'datasrc=' option (%s) - ignoring", tok.Data());
            } else {
               aDataSrc = tok;
               Printf("runProof: reading files from: %s", aDataSrc.Data());
            }
         } else if (tok == "readall") {
            proof->SetParameter("ProofEventProc_Read", "readall");
            Printf("runProof: eventproc: reading the full event");
         }
      }
      Int_t nFiles = (aFiles.IsNull()) ? 10 : aFiles.Atoi();
         Printf("runProof: found aFiles: '%s', nFiles: %d", aFiles.Data(), nFiles);
      if (nFiles > 50) {
         Printf("runProof: max number of files is 50 - resizing request");
         nFiles = 50;
      }

      // Create the chain
      TChain *c = new TChain("EventTree");
      Int_t i = 1;
      TString fn;
      for (i = 1; i <= nFiles; i++) {
         fn.Form("%s/event_%d.root", aDataSrc.Data(), i);
         c->AddFile(fn.Data());
      }
      c->SetProof();

      // The selector
      sel.Form("%s/proof/ProofEventProc.C+", tutorials.Data());
      // Run it
      Printf("\nrunProof: running \"eventproc\"\n");
      c->Process(sel.Data(), opt);
{'fraction_non_alphanumeric': 0.14516656543475215, 'fraction_numerical': 0.013362612717843293, 'mean_word_length': 2.4000794281175537, 'pattern_counts': {'":': 0, '<': 43, '<?xml version=': 0, '>': 169, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 12}, 'pii_count': 0, 'substrings_counts': 4, 'word_list_counts': {'cursed_substrings.json': 2, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7190939', 'n_tokens_mistral': 13885, 'n_tokens_neox': 13322, 'n_words': 4022}