TProof.cxx

Go to the documentation of this file.

// @(#)root/proof:$Id: a2a50e759072c37ccbc65ecbcce735a76de86e95 $
// Author: Fons Rademakers   13/02/97
 
/*************************************************************************
 * Copyright (C) 1995-2000, Rene Brun and Fons Rademakers.               *
 * All rights reserved.                                                  *
 *                                                                       *
 * For the licensing terms see $ROOTSYS/LICENSE.                         *
 * For the list of contributors see $ROOTSYS/README/CREDITS.             *
 *************************************************************************/
/**
  \defgroup proof PROOF
 
  Classes defining the Parallel ROOT Facility, PROOF, a framework for parallel analysis of ROOT TTrees.
 
*/
 
/**
  \defgroup proofkernel PROOF kernel Libraries
  \ingroup proof
 
  The PROOF kernel libraries (libProof, libProofPlayer, libProofDraw) contain the classes defining
  the kernel of the PROOF facility, i.e. the protocol and the utilities to steer data processing
  and handling of results.
 
*/
 
/** \class TProof
\ingroup proofkernel
 
This class controls a Parallel ROOT Facility, PROOF, cluster.
It fires the worker servers, it keeps track of how many workers are
running, it keeps track of the workers running status, it broadcasts
messages to all workers, it collects results, etc.
 
*/
 
#include <stdlib.h>
#include <fcntl.h>
#include <errno.h>
#ifdef WIN32
#   include <io.h>
#   include <sys/stat.h>
#   include <sys/types.h>
#   include "snprintf.h"
#else
#   include <unistd.h>
#endif
#include <vector>
 
#include "RConfigure.h"
#include "Riostream.h"
#include "Getline.h"
#include "TBrowser.h"
#include "TChain.h"
#include "TCondor.h"
#include "TDSet.h"
#include "TError.h"
#include "TEnv.h"
#include "TEntryList.h"
#include "TEventList.h"
#include "TFile.h"
#include "TFileInfo.h"
#include "TFTP.h"
#include "THashList.h"
#include "TInterpreter.h"
#include "TKey.h"
#include "TMap.h"
#include "TMath.h"
#include "TMessage.h"
#include "TMonitor.h"
#include "TObjArray.h"
#include "TObjString.h"
#include "TParameter.h"
#include "TProof.h"
#include "TProofNodeInfo.h"
#include "TProofOutputFile.h"
#include "TVirtualProofPlayer.h"
#include "TVirtualPacketizer.h"
#include "TProofServ.h"
#include "TPluginManager.h"
#include "TQueryResult.h"
#include "TRandom.h"
#include "TRegexp.h"
#include "TROOT.h"
#include "TSlave.h"
#include "TSocket.h"
#include "TSortedList.h"
#include "TSystem.h"
#include "TTree.h"
#include "TUrl.h"
#include "TFileCollection.h"
#include "TDataSetManager.h"
#include "TDataSetManagerFile.h"
#include "TMacro.h"
#include "TSelector.h"
#include "TPRegexp.h"
#include "TPackMgr.h"
 
#include <mutex>
 
TProof *gProof = 0;
 
// Rotating indicator
char TProofMergePrg::fgCr[4] = {'-', '\\', '|', '/'};
 
TList   *TProof::fgProofEnvList = 0;          // List of env vars for proofserv
TPluginHandler *TProof::fgLogViewer = 0;      // Log viewer handler
 
ClassImp(TProof);
 
//----- PROOF Interrupt signal handler -----------------------------------------
////////////////////////////////////////////////////////////////////////////////
/// TProof interrupt handler.
 
Bool_t TProofInterruptHandler::Notify()
{
   if (!fProof->IsTty() || fProof->GetRemoteProtocol() < 22) {
 
      // Cannot ask the user : abort any remote processing
      fProof->StopProcess(kTRUE);
 
   } else {
      // Real stop or request to switch to asynchronous?
      const char *a = 0;
      if (fProof->GetRemoteProtocol() < 22) {
         a = Getline("\nSwitch to asynchronous mode not supported remotely:"
                     "\nEnter S/s to stop, Q/q to quit, any other key to continue: ");
      } else {
         a = Getline("\nEnter A/a to switch asynchronous, S/s to stop, Q/q to quit,"
                     " any other key to continue: ");
      }
      if (a[0] == 'Q' || a[0] == 'S' || a[0] == 'q' || a[0] == 's') {
 
         Info("Notify","Processing interrupt signal ... %c", a[0]);
 
         // Stop or abort any remote processing
         Bool_t abort = (a[0] == 'Q' || a[0] == 'q') ? kTRUE : kFALSE;
         fProof->StopProcess(abort);
 
      } else if ((a[0] == 'A' || a[0] == 'a') && fProof->GetRemoteProtocol() >= 22) {
         // Stop any remote processing
         fProof->GoAsynchronous();
      }
   }
 
   return kTRUE;
}
 
//----- Input handler for messages from TProofServ -----------------------------
////////////////////////////////////////////////////////////////////////////////
/// Constructor
 
TProofInputHandler::TProofInputHandler(TProof *p, TSocket *s)
                   : TFileHandler(s->GetDescriptor(),1),
                     fSocket(s), fProof(p)
{
}
 
////////////////////////////////////////////////////////////////////////////////
/// Handle input
 
Bool_t TProofInputHandler::Notify()
{
   fProof->CollectInputFrom(fSocket);
   return kTRUE;
}
 
 
//------------------------------------------------------------------------------
 
ClassImp(TSlaveInfo);
 
////////////////////////////////////////////////////////////////////////////////
/// Used to sort slaveinfos by ordinal.
 
Int_t TSlaveInfo::Compare(const TObject *obj) const
{
   if (!obj) return 1;
 
   const TSlaveInfo *si = dynamic_cast<const TSlaveInfo*>(obj);
 
   if (!si) return fOrdinal.CompareTo(obj->GetName());
 
   const char *myord = GetOrdinal();
   const char *otherord = si->GetOrdinal();
   while (myord && otherord) {
      Int_t myval = atoi(myord);
      Int_t otherval = atoi(otherord);
      if (myval < otherval) return 1;
      if (myval > otherval) return -1;
      myord = strchr(myord, '.');
      if (myord) myord++;
      otherord = strchr(otherord, '.');
      if (otherord) otherord++;
   }
   if (myord) return -1;
   if (otherord) return 1;
   return 0;
}
 
////////////////////////////////////////////////////////////////////////////////
/// Used to compare slaveinfos by ordinal.
 
Bool_t TSlaveInfo::IsEqual(const TObject* obj) const
{
   if (!obj) return kFALSE;
   const TSlaveInfo *si = dynamic_cast<const TSlaveInfo*>(obj);
   if (!si) return kFALSE;
   return (strcmp(GetOrdinal(), si->GetOrdinal()) == 0);
}
 
////////////////////////////////////////////////////////////////////////////////
/// Print slave info. If opt = "active" print only the active
/// slaves, if opt="notactive" print only the not active slaves,
/// if opt = "bad" print only the bad slaves, else
/// print all slaves.
 
void TSlaveInfo::Print(Option_t *opt) const
{
   TString stat = fStatus == kActive ? "active" :
                  fStatus == kBad ? "bad" :
                  "not active";
 
   Bool_t newfmt = kFALSE;
   TString oo(opt);
   if (oo.Contains("N")) {
      newfmt = kTRUE;
      oo.ReplaceAll("N","");
   }
   if (oo == "active" && fStatus != kActive) return;
   if (oo == "notactive" && fStatus != kNotActive) return;
   if (oo == "bad" && fStatus != kBad) return;
 
   if (newfmt) {
      TString msd, si, datadir;
      if (!(fMsd.IsNull())) msd.Form("| msd: %s ", fMsd.Data());
      if (!(fDataDir.IsNull())) datadir.Form("| datadir: %s ", fDataDir.Data());
      if (fSysInfo.fCpus > 0) {
         si.Form("| %s, %d cores, %d MB ram", fHostName.Data(),
               fSysInfo.fCpus, fSysInfo.fPhysRam);
      } else {
         si.Form("| %s", fHostName.Data());
      }
      Printf("Worker: %9s %s %s%s| %s", fOrdinal.Data(), si.Data(), msd.Data(), datadir.Data(), stat.Data());
 
   } else {
      TString msd  = fMsd.IsNull() ? "<null>" : fMsd.Data();
 
      std::cout << "Slave: "          << fOrdinal
         << "  hostname: "     << fHostName
         << "  msd: "          << msd
         << "  perf index: "   << fPerfIndex
         << "  "               << stat
         << std::endl;
   }
}
 
////////////////////////////////////////////////////////////////////////////////
/// Setter for fSysInfo
 
void TSlaveInfo::SetSysInfo(SysInfo_t si)
{
   fSysInfo.fOS       = si.fOS;          // OS
   fSysInfo.fModel    = si.fModel;       // computer model
   fSysInfo.fCpuType  = si.fCpuType;     // type of cpu
   fSysInfo.fCpus     = si.fCpus;        // number of cpus
   fSysInfo.fCpuSpeed = si.fCpuSpeed;    // cpu speed in MHz
   fSysInfo.fBusSpeed = si.fBusSpeed;    // bus speed in MHz
   fSysInfo.fL2Cache  = si.fL2Cache;     // level 2 cache size in KB
   fSysInfo.fPhysRam  = si.fPhysRam;     // Physical RAM
}
 
ClassImp(TProof);
 
//------------------------------------------------------------------------------
 
////////////////////////////////////////////////////////////////////////////////
/// Destructor
 
TMergerInfo::~TMergerInfo()
{
   // Just delete the list, the objects are owned by other list
   if (fWorkers) {
      fWorkers->SetOwner(kFALSE);
      SafeDelete(fWorkers);
   }
}
////////////////////////////////////////////////////////////////////////////////
/// Increase number of already merged workers by 1
 
void TMergerInfo::SetMergedWorker()
{
   if (AreAllWorkersMerged())
      Error("SetMergedWorker", "all workers have been already merged before!");
   else
      fMergedWorkers++;
}
 
////////////////////////////////////////////////////////////////////////////////
/// Add new worker to the list of workers to be merged by this merger
 
void TMergerInfo::AddWorker(TSlave *sl)
{
   if (!fWorkers)
      fWorkers = new TList();
   if (fWorkersToMerge == fWorkers->GetSize()) {
      Error("AddWorker", "all workers have been already assigned to this merger");
      return;
   }
   fWorkers->Add(sl);
}
 
////////////////////////////////////////////////////////////////////////////////
/// Return if merger has already merged all workers, i.e. if it has finished its merging job
 
Bool_t TMergerInfo::AreAllWorkersMerged()
{
   return (fWorkersToMerge == fMergedWorkers);
}
 
////////////////////////////////////////////////////////////////////////////////
/// Return if the determined number of workers has been already assigned to this merger
 
Bool_t TMergerInfo::AreAllWorkersAssigned()
{
      if (!fWorkers)
         return kFALSE;
 
      return (fWorkers->GetSize() == fWorkersToMerge);
}
 
////////////////////////////////////////////////////////////////////////////////
/// This a private API function.
/// It checks whether the connection string contains a PoD cluster protocol.
/// If it does, then the connection string will be changed to reflect
/// a real PROOF connection string for a PROOF cluster managed by PoD.
/// PoD: http://pod.gsi.de .
/// Return -1 if the PoD request failed; return 0 otherwise.
 
static Int_t PoDCheckUrl(TString *_cluster)
{
   if ( !_cluster )
      return 0;
 
   // trim spaces from both sides of the string
   *_cluster = _cluster->Strip( TString::kBoth );
   // PoD protocol string
   const TString pod_prot("pod");
 
   // URL test
   // TODO: The URL test is to support remote PoD servers (not managed by pod-remote)
   TUrl url( _cluster->Data() );
   if( pod_prot.CompareTo(url.GetProtocol(), TString::kIgnoreCase) )
      return 0;
 
   // PoD cluster is used
   // call pod-info in a batch mode (-b).
   // pod-info will find either a local PoD cluster or
   // a remote one, manged by pod-remote.
   *_cluster = gSystem->GetFromPipe("pod-info -c -b");
   if( 0 == _cluster->Length() ) {
      Error("PoDCheckUrl", "PoD server is not running");
      return -1;
   }
   return 0;
}
 
////////////////////////////////////////////////////////////////////////////////
/// Create a PROOF environment. Starting PROOF involves either connecting
/// to a master server, which in turn will start a set of slave servers, or
/// directly starting as master server (if master = ""). Masterurl is of
/// the form: [proof[s]://]host[:port]. Conffile is the name of the config
/// file describing the remote PROOF cluster (this argument alows you to
/// describe different cluster configurations).
/// The default is proof.conf. Confdir is the directory where the config
/// file and other PROOF related files are (like motd and noproof files).
/// Loglevel is the log level (default = 1). User specified custom config
/// files will be first looked for in $HOME/.conffile.
 
TProof::TProof(const char *masterurl, const char *conffile, const char *confdir,
               Int_t loglevel, const char *alias, TProofMgr *mgr)
       : fUrl(masterurl)
{
   // Default initializations
   InitMembers();
 
   // This may be needed during init
   fManager = mgr;
 
   // Default server type
   fServType = TProofMgr::kXProofd;
 
   // Default query mode
   fQueryMode = kSync;
 
   // Parse the main URL, adjusting the missing fields and setting the relevant
   // bits
   ResetBit(TProof::kIsClient);
   ResetBit(TProof::kIsMaster);
 
   // Protocol and Host
   if (!masterurl || strlen(masterurl) <= 0) {
      fUrl.SetProtocol("proof");
      fUrl.SetHost("__master__");
   } else if (!(strstr(masterurl, "://"))) {
      fUrl.SetProtocol("proof");
   }
   // Port
   if (fUrl.GetPort() == TUrl(" ").GetPort())
      fUrl.SetPort(TUrl("proof:// ").GetPort());
 
   // Make sure to store the FQDN, so to get a solid reference for subsequent checks
   if (!strcmp(fUrl.GetHost(), "__master__"))
      fMaster = fUrl.GetHost();
   else if (!strlen(fUrl.GetHost()))
      fMaster = gSystem->GetHostByName(gSystem->HostName()).GetHostName();
   else
      fMaster = gSystem->GetHostByName(fUrl.GetHost()).GetHostName();
 
   // Server type
   if (strlen(fUrl.GetOptions()) > 0) {
      TString opts(fUrl.GetOptions());
      if (!(strncmp(fUrl.GetOptions(),"std",3))) {
         fServType = TProofMgr::kProofd;
         opts.Remove(0,3);
         fUrl.SetOptions(opts.Data());
      } else if (!(strncmp(fUrl.GetOptions(),"lite",4))) {
         fServType = TProofMgr::kProofLite;
         opts.Remove(0,4);
         fUrl.SetOptions(opts.Data());
      }
   }
 
   // Instance type
   fMasterServ = kFALSE;
   SetBit(TProof::kIsClient);
   ResetBit(TProof::kIsMaster);
   if (fMaster == "__master__") {
      fMasterServ = kTRUE;
      ResetBit(TProof::kIsClient);
      SetBit(TProof::kIsMaster);
   } else if (fMaster == "prooflite") {
      // Client and master are merged
      fMasterServ = kTRUE;
      SetBit(TProof::kIsMaster);
   }
   // Flag that we are a client
   if (TestBit(TProof::kIsClient))
      if (!gSystem->Getenv("ROOTPROOFCLIENT")) gSystem->Setenv("ROOTPROOFCLIENT","");
 
   Init(masterurl, conffile, confdir, loglevel, alias);
 
   // If the user was not set, get it from the master
   if (strlen(fUrl.GetUser()) <= 0) {
      TString usr, emsg;
      if (Exec("gProofServ->GetUser()", "0", kTRUE) == 0) {
         TObjString *os = fMacroLog.GetLineWith("const char");
         if (os) {
            Ssiz_t fst =  os->GetString().First('\"');
            Ssiz_t lst =  os->GetString().Last('\"');
            usr = os->GetString()(fst+1, lst-fst-1);
         } else {
            emsg = "could not find 'const char *' string in macro log";
         }
      } else {
         emsg = "could not retrieve user info";
      }
      if (!emsg.IsNull()) {
         // Get user logon name
         UserGroup_t *pw = gSystem->GetUserInfo();
         if (pw) {
            usr = pw->fUser;
            delete pw;
         }
         Warning("TProof", "%s: using local default %s", emsg.Data(), usr.Data());
      }
      // Set the user name in the main URL
      fUrl.SetUser(usr.Data());
   }
 
   // If called by a manager, make sure it stays in last position
   // for cleaning
   if (mgr) {
      R__LOCKGUARD(gROOTMutex);
      gROOT->GetListOfSockets()->Remove(mgr);
      gROOT->GetListOfSockets()->Add(mgr);
   }
 
   // Old-style server type: we add this to the list and set the global pointer
   if (IsProofd() || TestBit(TProof::kIsMaster))
      if (!gROOT->GetListOfProofs()->FindObject(this))
         gROOT->GetListOfProofs()->Add(this);
 
   // Still needed by the packetizers: needs to be changed
   gProof = this;
}
 
////////////////////////////////////////////////////////////////////////////////
/// Protected constructor to be used by classes deriving from TProof
/// (they have to call Init themselves and override StartSlaves
/// appropriately).
///
/// This constructor simply closes any previous gProof and sets gProof
/// to this instance.
 
TProof::TProof() : fUrl(""), fServType(TProofMgr::kXProofd)
{
   // Default initializations
   InitMembers();
 
   if (!gROOT->GetListOfProofs()->FindObject(this))
      gROOT->GetListOfProofs()->Add(this);
 
   gProof = this;
}
 
////////////////////////////////////////////////////////////////////////////////
/// Default initializations
 
void TProof::InitMembers()
{
   fValid = kFALSE;
   fTty = kFALSE;
   fRecvMessages = 0;
   fSlaveInfo = 0;
   fMasterServ = kFALSE;
   fSendGroupView = kFALSE;
   fIsPollingWorkers = kFALSE;
   fLastPollWorkers_s = -1;
   fActiveSlaves = 0;
   fInactiveSlaves = 0;
   fUniqueSlaves = 0;
   fAllUniqueSlaves = 0;
   fNonUniqueMasters = 0;
   fActiveMonitor = 0;
   fUniqueMonitor = 0;
   fAllUniqueMonitor = 0;
   fCurrentMonitor = 0;
   fBytesRead = 0;
   fRealTime = 0;
   fCpuTime = 0;
   fIntHandler = 0;
   fProgressDialog = 0;
   fProgressDialogStarted = kFALSE;
   SetBit(kUseProgressDialog);
   fPlayer = 0;
   fFeedback = 0;
   fChains = 0;
   fDSet = 0;
   fNotIdle = 0;
   fSync = kTRUE;
   fRunStatus = kRunning;
   fIsWaiting = kFALSE;
   fRedirLog = kFALSE;
   fLogFileW = 0;
   fLogFileR = 0;
   fLogToWindowOnly = kFALSE;
   fSaveLogToMacro = kFALSE;
   fMacroLog.SetName("ProofLogMacro");
 
   fWaitingSlaves = 0;
   fQueries = 0;
   fOtherQueries = 0;
   fDrawQueries = 0;
   fMaxDrawQueries = 1;
   fSeqNum = 0;
 
   fSessionID = -1;
   fEndMaster = kFALSE;
 
   fPackMgr = 0;
   fEnabledPackagesOnCluster = 0;
 
   fInputData = 0;
 
   fPrintProgress = 0;
 
   fLoadedMacros = 0;
 
   fProtocol = -1;
   fSlaves = 0;
   fTerminatedSlaveInfos = 0;
   fBadSlaves = 0;
   fAllMonitor = 0;
   fDataReady = kFALSE;
   fBytesReady = 0;
   fTotalBytes = 0;
   fAvailablePackages = 0;
   fEnabledPackages = 0;
   fRunningDSets = 0;
 
   fCollectTimeout = -1;
 
   fManager = 0;
   fQueryMode = kSync;
   fDynamicStartup = kFALSE;
 
   fMergersSet = kFALSE;
   fMergersByHost = kFALSE;
   fMergers = 0;
   fMergersCount = -1;
   fLastAssignedMerger = 0;
   fWorkersToMerge = 0;
   fFinalizationRunning = kFALSE;
 
   fPerfTree = "";
 
   fWrksOutputReady = 0;
 
   fSelector = 0;
 
   fPrepTime = 0.;
 
   // Check if the user defined a list of environment variables to send over:
   // include them into the dedicated list
   if (gSystem->Getenv("PROOF_ENVVARS")) {
      TString envs(gSystem->Getenv("PROOF_ENVVARS")), env, envsfound;
      Int_t from = 0;
      while (envs.Tokenize(env, from, ",")) {
         if (!env.IsNull()) {
            if (!gSystem->Getenv(env)) {
               Warning("Init", "request for sending over undefined environemnt variable '%s' - ignoring", env.Data());
            } else {
               if (!envsfound.IsNull()) envsfound += ",";
               envsfound += env;
               TProof::DelEnvVar(env);
               TProof::AddEnvVar(env, gSystem->Getenv(env));
            }
         }
      }
      if (envsfound.IsNull()) {
         Warning("Init", "none of the requested env variables were found: '%s'", envs.Data());
      } else {
         Info("Init", "the following environment variables have been added to the list to be sent to the nodes: '%s'", envsfound.Data());
      }
   }
 
   // Done
   return;
}
 
////////////////////////////////////////////////////////////////////////////////
/// Clean up PROOF environment.
 
TProof::~TProof()
{
   if (fChains) {
      while (TChain *chain = dynamic_cast<TChain*> (fChains->First()) ) {
         // remove "chain" from list
         chain->SetProof(0);
         RemoveChain(chain);
      }
   }
 
   // remove links to packages enabled on the client
   if (TestBit(TProof::kIsClient)) {
      // iterate over all packages
      TList *epl = fPackMgr->GetListOfEnabled();
      if (epl) {
         TIter nxp(epl);
         while (TObjString *pck = (TObjString *)(nxp())) {
            FileStat_t stat;
            if (gSystem->GetPathInfo(pck->String(), stat) == 0) {
               // check if symlink, if so unlink
               // NOTE: GetPathInfo() returns 1 in case of symlink that does not point to
               // existing file or to a directory, but if fIsLink is true the symlink exists
               if (stat.fIsLink)
                  gSystem->Unlink(pck->String());
            }
         }
         epl->Delete();
         delete epl;
      }
   }
 
   Close();
   SafeDelete(fIntHandler);
   SafeDelete(fSlaves);
   SafeDelete(fActiveSlaves);
   SafeDelete(fInactiveSlaves);
   SafeDelete(fUniqueSlaves);
   SafeDelete(fAllUniqueSlaves);
   SafeDelete(fNonUniqueMasters);
   SafeDelete(fTerminatedSlaveInfos);
   SafeDelete(fBadSlaves);
   SafeDelete(fAllMonitor);
   SafeDelete(fActiveMonitor);
   SafeDelete(fUniqueMonitor);
   SafeDelete(fAllUniqueMonitor);
   SafeDelete(fSlaveInfo);
   SafeDelete(fChains);
   SafeDelete(fPlayer);
   SafeDelete(fFeedback);
   SafeDelete(fWaitingSlaves);
   SafeDelete(fAvailablePackages);
   SafeDelete(fEnabledPackages);
   SafeDelete(fLoadedMacros);
   SafeDelete(fPackMgr);
   SafeDelete(fRecvMessages);
   SafeDelete(fInputData);
   SafeDelete(fRunningDSets);
   SafeDelete(fEnabledPackagesOnCluster);
   if (fWrksOutputReady) {
      fWrksOutputReady->SetOwner(kFALSE);
      delete fWrksOutputReady;
   }
   if (fQueries) {
      fQueries->Delete();
      delete fQueries;
   }
 
   // remove file with redirected logs
   if (TestBit(TProof::kIsClient)) {
      if (fLogFileR)
         fclose(fLogFileR);
      if (fLogFileW)
         fclose(fLogFileW);
      if (fLogFileName.Length() > 0)
         gSystem->Unlink(fLogFileName);
   }
 
   // Remove for the global list
   gROOT->GetListOfProofs()->Remove(this);
   // ... and from the manager list
   if (fManager && fManager->IsValid())
      fManager->DiscardSession(this);
 
   if (gProof && gProof == this) {
      // Set previous as default
      TIter pvp(gROOT->GetListOfProofs(), kIterBackward);
      while ((gProof = (TProof *)pvp())) {
         if (gProof->InheritsFrom(TProof::Class()))
            break;
      }
   }
 
   // For those interested in our destruction ...
   Emit("~TProof()");
   Emit("CloseWindow()");
}
 
////////////////////////////////////////////////////////////////////////////////
/// Start the PROOF environment. Starting PROOF involves either connecting
/// to a master server, which in turn will start a set of slave servers, or
/// directly starting as master server (if master = ""). For a description
/// of the arguments see the TProof ctor. Returns the number of started
/// master or slave servers, returns 0 in case of error, in which case
/// fValid remains false.
 
Int_t TProof::Init(const char *, const char *conffile,
                   const char *confdir, Int_t loglevel, const char *alias)
{
   R__ASSERT(gSystem);
 
   fValid = kFALSE;
 
   // Connected to terminal?
   fTty = (isatty(0) == 0 || isatty(1) == 0) ? kFALSE : kTRUE;
 
   // If in attach mode, options is filled with additional info
   Bool_t attach = kFALSE;
   if (strlen(fUrl.GetOptions()) > 0) {
      attach = kTRUE;
      // A flag from the GUI
      TString opts = fUrl.GetOptions();
      if (opts.Contains("GUI")) {
         SetBit(TProof::kUsingSessionGui);
         opts.Remove(opts.Index("GUI"));
         fUrl.SetOptions(opts);
      }
   }
 
   if (TestBit(TProof::kIsMaster)) {
      // Fill default conf file and conf dir
      if (!conffile || !conffile[0])
         fConfFile = kPROOF_ConfFile;
      if (!confdir  || !confdir[0])
         fConfDir  = kPROOF_ConfDir;
      // The group; the client receives it in the kPROOF_SESSIONTAG message
      if (gProofServ) fGroup = gProofServ->GetGroup();
   } else {
      fConfDir     = confdir;
      fConfFile    = conffile;
   }
 
   // Analysise the conffile field
   if (fConfFile.Contains("workers=0")) fConfFile.ReplaceAll("workers=0", "masteronly");
   ParseConfigField(fConfFile);
 
   fWorkDir        = gSystem->WorkingDirectory();
   fLogLevel       = loglevel;
   fProtocol       = kPROOF_Protocol;
   fSendGroupView  = kTRUE;
   fImage          = fMasterServ ? "" : "<local>";
   fIntHandler     = 0;
   fStatus         = 0;
   fRecvMessages   = new TList;
   fRecvMessages->SetOwner(kTRUE);
   fSlaveInfo      = 0;
   fChains         = new TList;
   fAvailablePackages = 0;
   fEnabledPackages = 0;
   fRunningDSets   = 0;
   fEndMaster      = TestBit(TProof::kIsMaster) ? kTRUE : kFALSE;
   fInputData      = 0;
   ResetBit(TProof::kNewInputData);
   fPrintProgress  = 0;
 
   // Timeout for some collect actions
   fCollectTimeout = gEnv->GetValue("Proof.CollectTimeout", -1);
 
   // Should the workers be started dynamically; default: no
   fDynamicStartup = gEnv->GetValue("Proof.DynamicStartup", kFALSE);
 
   // Default entry point for the data pool is the master
   if (TestBit(TProof::kIsClient))
      fDataPoolUrl.Form("root://%s", fMaster.Data());
   else
      fDataPoolUrl = "";
 
   fProgressDialog        = 0;
   fProgressDialogStarted = kFALSE;
 
   // Default alias is the master name
   TString      al = (alias) ? alias : fMaster.Data();
   SetAlias(al);
 
   // Client logging of messages from the master and slaves
   fRedirLog = kFALSE;
   if (TestBit(TProof::kIsClient)) {
      fLogFileName.Form("%s/ProofLog_%d", gSystem->TempDirectory(), gSystem->GetPid());
      if ((fLogFileW = fopen(fLogFileName, "w")) == 0)
         Error("Init", "could not create temporary logfile");
      if ((fLogFileR = fopen(fLogFileName, "r")) == 0)
         Error("Init", "could not open temp logfile for reading");
   }
   fLogToWindowOnly = kFALSE;
 
   // Status of cluster
   fNotIdle = 0;
   // Query type
   fSync = (attach) ? kFALSE : kTRUE;
   // Not enqueued
   fIsWaiting = kFALSE;
 
   // Counters
   fBytesRead = 0;
   fRealTime = 0;
   fCpuTime = 0;
 
   // List of queries
   fQueries = 0;
   fOtherQueries = 0;
   fDrawQueries = 0;
   fMaxDrawQueries = 1;
   fSeqNum = 0;
 
   // Remote ID of the session
   fSessionID = -1;
 
   // Part of active query
   fWaitingSlaves = 0;
 
   // Make remote PROOF player
   fPlayer = 0;
   MakePlayer();
 
   fFeedback = new TList;
   fFeedback->SetOwner();
   fFeedback->SetName("FeedbackList");
   AddInput(fFeedback);
 
   // sort slaves by descending performance index
   fSlaves           = new TSortedList(kSortDescending);
   fActiveSlaves     = new TList;
   fInactiveSlaves   = new TList;
   fUniqueSlaves     = new TList;
   fAllUniqueSlaves  = new TList;
   fNonUniqueMasters = new TList;
   fBadSlaves        = new TList;
   fAllMonitor       = new TMonitor;
   fActiveMonitor    = new TMonitor;
   fUniqueMonitor    = new TMonitor;
   fAllUniqueMonitor = new TMonitor;
   fCurrentMonitor   = 0;
 
   fTerminatedSlaveInfos = new TList;
   fTerminatedSlaveInfos->SetOwner(kTRUE);
 
   fLoadedMacros            = 0;
   fPackMgr                 = 0;
 
   // Enable optimized sending of streamer infos to use embedded backward/forward
   // compatibility support between different ROOT versions and different versions of
   // users classes
   Bool_t enableSchemaEvolution = gEnv->GetValue("Proof.SchemaEvolution",1);
   if (enableSchemaEvolution) {
      TMessage::EnableSchemaEvolutionForAll();
   } else {
      Info("TProof", "automatic schema evolution in TMessage explicitly disabled");
   }
 
   if (IsMaster()) {
      // to make UploadPackage() method work on the master as well.
      if (gProofServ) fPackMgr = gProofServ->GetPackMgr();
   } else {
 
      TString sandbox;
      if (GetSandbox(sandbox, kTRUE) != 0) {
         Error("Init", "failure asserting sandbox directory %s", sandbox.Data());
         return 0;
      }
 
      // Package Dir
      TString packdir = gEnv->GetValue("Proof.PackageDir", "");
      if (packdir.IsNull())
         packdir.Form("%s/%s", sandbox.Data(), kPROOF_PackDir);
      if (AssertPath(packdir, kTRUE) != 0) {
         Error("Init", "failure asserting directory %s", packdir.Data());
         return 0;
      }
      fPackMgr = new TPackMgr(packdir);
      if (gDebug > 0)
         Info("Init", "package directory set to %s", packdir.Data());
   }
 
   if (!IsMaster()) {
      // List of directories where to look for global packages
      TString globpack = gEnv->GetValue("Proof.GlobalPackageDirs","");
      TProofServ::ResolveKeywords(globpack);
      Int_t nglb = TPackMgr::RegisterGlobalPath(globpack);
      if (gDebug > 0)
         Info("Init", " %d global package directories registered", nglb);
   }
 
   // Master may want dynamic startup
   if (fDynamicStartup) {
      if (!IsMaster()) {
         // If on client - start the master
         if (!StartSlaves(attach))
            return 0;
      }
   } else {
 
      // Master Only mode (for operations requiring only the master, e.g. dataset browsing,
      // result retrieving, ...)
      Bool_t masterOnly = gEnv->GetValue("Proof.MasterOnly", kFALSE);
      if (!IsMaster() || !masterOnly) {
         // Start slaves (the old, static, per-session way)
         if (!StartSlaves(attach))
            return 0;
         // Client: Is Master in dynamic startup mode?
         if (!IsMaster()) {
            Int_t dyn = 0;
            GetRC("Proof.DynamicStartup", dyn);
            if (dyn != 0) fDynamicStartup = kTRUE;
         }
      }
   }
   // we are now properly initialized
   fValid = kTRUE;
 
   // De-activate monitor (will be activated in Collect)
   fAllMonitor->DeActivateAll();
 
   // By default go into parallel mode
   Int_t nwrk = GetRemoteProtocol() > 35 ? -1 : 9999;
   TNamed *n = 0;
   if (TProof::GetEnvVars() &&
      (n = (TNamed *) TProof::GetEnvVars()->FindObject("PROOF_NWORKERS"))) {
      TString s(n->GetTitle());
      if (s.IsDigit()) nwrk = s.Atoi();
   }
   GoParallel(nwrk, attach);
 
   // Send relevant initial state to slaves
   if (!attach)
      SendInitialState();
   else if (!IsIdle())
      // redirect log
      fRedirLog = kTRUE;
 
   // Done at this point, the alias will be communicated to the coordinator, if any
   if (TestBit(TProof::kIsClient))
      SetAlias(al);
 
   SetActive(kFALSE);
 
   if (IsValid()) {
 
      // Activate input handler
      ActivateAsyncInput();
 
      R__LOCKGUARD(gROOTMutex);
      gROOT->GetListOfSockets()->Add(this);
   }
 
   AskParallel();
 
   return fActiveSlaves->GetSize();
}
 
////////////////////////////////////////////////////////////////////////////////
/// Set the sandbox path from ' Proof.Sandbox' or the alternative var 'rc'.
/// Use the existing setting or the default if nothing is found.
/// If 'assert' is kTRUE, make also sure that the path exists.
/// Return 0 on success, -1 on failure
 
Int_t TProof::GetSandbox(TString &sb, Bool_t assert, const char *rc)
{
   // Get it from 'rc', if defined
   if (rc && strlen(rc)) sb = gEnv->GetValue(rc, sb);
   // Or use the default 'rc'
   if (sb.IsNull()) sb = gEnv->GetValue("Proof.Sandbox", "");
   // If nothing found , use the default
   if (sb.IsNull()) sb.Form("~/%s", kPROOF_WorkDir);
   // Expand special settings
   if (sb == ".") {
      sb = gSystem->pwd();
   } else if (sb == "..") {
      sb = gSystem->GetDirName(gSystem->pwd());
   }
   gSystem->ExpandPathName(sb);
 
   // Assert the path, if required
   if (assert && AssertPath(sb, kTRUE) != 0) return -1;
   // Done
   return 0;
}
 
////////////////////////////////////////////////////////////////////////////////
/// The config file field may contain special instructions which need to be
/// parsed at the beginning, e.g. for debug runs with valgrind.
/// Several options can be given separated by a ','
 
void TProof::ParseConfigField(const char *config)
{
   TString sconf(config), opt;
   Ssiz_t from = 0;
   Bool_t cpuPin = kFALSE;
 
   // Analysise the field
   const char *cq = (IsLite()) ? "\"" : "";
   while (sconf.Tokenize(opt, from, ",")) {
      if (opt.IsNull()) continue;
 
      if (opt.BeginsWith("valgrind")) {
         // Any existing valgrind setting? User can give full settings, which we fully respect,
         // or pass additional options for valgrind by prefixing 'valgrind_opts:'. For example,
         //    TProof::AddEnvVar("PROOF_MASTER_WRAPPERCMD", "valgrind_opts:--time-stamp --leak-check=full"
         // will add option "--time-stamp --leak-check=full" to our default options
         TString mst, top, sub, wrk, all;
         TList *envs = fgProofEnvList;
         TNamed *n = 0;
         if (envs) {
            if ((n = (TNamed *) envs->FindObject("PROOF_WRAPPERCMD")))
               all = n->GetTitle();
            if ((n = (TNamed *) envs->FindObject("PROOF_MASTER_WRAPPERCMD")))
               mst = n->GetTitle();
            if ((n = (TNamed *) envs->FindObject("PROOF_TOPMASTER_WRAPPERCMD")))
               top = n->GetTitle();
            if ((n = (TNamed *) envs->FindObject("PROOF_SUBMASTER_WRAPPERCMD")))
               sub = n->GetTitle();
            if ((n = (TNamed *) envs->FindObject("PROOF_SLAVE_WRAPPERCMD")))
               wrk = n->GetTitle();
         }
         if (all != "" && mst == "") mst = all;
         if (all != "" && top == "") top = all;
         if (all != "" && sub == "") sub = all;
         if (all != "" && wrk == "") wrk = all;
         if (all != "" && all.BeginsWith("valgrind_opts:")) {
            // The field is used to add an option Reset the setting
            Info("ParseConfigField","valgrind run: resetting 'PROOF_WRAPPERCMD':"
                                    " must be set again for next run , if any");
            TProof::DelEnvVar("PROOF_WRAPPERCMD");
         }
         TString var, cmd;
         cmd.Form("%svalgrind -v --suppressions=<rootsys>/etc/valgrind-root.supp", cq);
         TString mstlab("NO"), wrklab("NO");
         Bool_t doMaster = (opt == "valgrind" || (opt.Contains("master") &&
                           !opt.Contains("topmaster") && !opt.Contains("submaster")))
                         ? kTRUE : kFALSE;
         if (doMaster) {
            if (!IsLite()) {
               // Check if we have to add a var
               if (mst == "" || mst.BeginsWith("valgrind_opts:")) {
                  mst.ReplaceAll("valgrind_opts:","");
                  var.Form("%s --log-file=<logfilemst>.valgrind.log %s", cmd.Data(), mst.Data());
                  TProof::AddEnvVar("PROOF_MASTER_WRAPPERCMD", var);
                  mstlab = "YES";
               } else if (mst != "") {
                  mstlab = "YES";
               }
            } else {
               if (opt.Contains("master")) {
                  Warning("ParseConfigField",
                        "master valgrinding does not make sense for PROOF-Lite: ignoring");
                  opt.ReplaceAll("master", "");
                  if (!opt.Contains("workers")) return;
               }
               if (opt == "valgrind" || opt == "valgrind=") opt = "valgrind=workers";
            }
         }
         if (opt.Contains("topmaster")) {
            // Check if we have to add a var
            if (top == "" || top.BeginsWith("valgrind_opts:")) {
               top.ReplaceAll("valgrind_opts:","");
               var.Form("%s --log-file=<logfilemst>.valgrind.log %s", cmd.Data(), top.Data());
               TProof::AddEnvVar("PROOF_TOPMASTER_WRAPPERCMD", var);
               mstlab = "YES";
            } else if (top != "") {
               mstlab = "YES";
            }
         }
         if (opt.Contains("submaster")) {
            // Check if we have to add a var
            if (sub == "" || sub.BeginsWith("valgrind_opts:")) {
               sub.ReplaceAll("valgrind_opts:","");
               var.Form("%s --log-file=<logfilemst>.valgrind.log %s", cmd.Data(), sub.Data());
               TProof::AddEnvVar("PROOF_SUBMASTER_WRAPPERCMD", var);
               mstlab = "YES";
            } else if (sub != "") {
               mstlab = "YES";
            }
         }
         if (opt.Contains("=workers") || opt.Contains("+workers")) {
            // Check if we have to add a var
            if (wrk == "" || wrk.BeginsWith("valgrind_opts:")) {
               wrk.ReplaceAll("valgrind_opts:","");
               var.Form("%s --log-file=<logfilewrk>.__valgrind__.log %s%s", cmd.Data(), wrk.Data(), cq);
               TProof::AddEnvVar("PROOF_SLAVE_WRAPPERCMD", var);
               TString nwrks("2");
               Int_t inw = opt.Index('#');
               if (inw != kNPOS) {
                  nwrks = opt(inw+1, opt.Length());
                  if (!nwrks.IsDigit()) nwrks = "2";
               }
               // Set the relevant variables
               if (!IsLite()) {
                  TProof::AddEnvVar("PROOF_NWORKERS", nwrks);
               } else {
                  gEnv->SetValue("ProofLite.Workers", nwrks.Atoi());
               }
               wrklab = nwrks;
               // Register the additional worker log in the session file
               // (for the master this is done automatically)
               TProof::AddEnvVar("PROOF_ADDITIONALLOG", "__valgrind__.log*");
            } else if (wrk != "") {
               wrklab = "ALL";
            }
         }
         // Increase the relevant timeouts
         if (!IsLite()) {
            TProof::AddEnvVar("PROOF_INTWAIT", "5000");
            gEnv->SetValue("Proof.SocketActivityTimeout", 6000);
         } else {
            gEnv->SetValue("ProofLite.StartupTimeOut", 5000);
         }
         // Warn for slowness
         Printf(" ");
         if (!IsLite()) {
            Printf(" ---> Starting a debug run with valgrind (master:%s, workers:%s)", mstlab.Data(), wrklab.Data());
         } else {
            Printf(" ---> Starting a debug run with valgrind (workers:%s)", wrklab.Data());
         }
         Printf(" ---> Please be patient: startup may be VERY slow ...");
         Printf(" ---> Logs will be available as special tags in the log window (from the progress dialog or TProof::LogViewer()) ");
         Printf(" ---> (Reminder: this debug run makes sense only if you are running a debug version of ROOT)");
         Printf(" ");
 
      } else if (opt.BeginsWith("igprof-pp")) {
 
         // IgProf profiling on master and worker. PROOF does not set the
         // environment for you: proper environment variables (like PATH and
         // LD_LIBRARY_PATH) should be set externally
 
         Printf("*** Requested IgProf performance profiling ***");
         TString addLogExt = "__igprof.pp__.log";
         TString addLogFmt = "igprof -pk -pp -t proofserv.exe -o %s.%s";
         TString tmp;
 
         if (IsLite()) {
            addLogFmt.Append("\"");
            addLogFmt.Prepend("\"");
         }
 
         tmp.Form(addLogFmt.Data(), "<logfilemst>", addLogExt.Data());
         TProof::AddEnvVar("PROOF_MASTER_WRAPPERCMD",  tmp.Data());
 
         tmp.Form(addLogFmt.Data(), "<logfilewrk>", addLogExt.Data());
         TProof::AddEnvVar("PROOF_SLAVE_WRAPPERCMD", tmp.Data() );
 
         TProof::AddEnvVar("PROOF_ADDITIONALLOG", addLogExt.Data());
 
      } else if (opt.BeginsWith("cpupin=")) {
         // Enable CPU pinning. Takes as argument the list of processor IDs
         // that will be used in order. Processor IDs are numbered from 0,
         // use likwid to see how they are organized. A possible parameter
         // format would be:
         //
         //   cpupin=3+4+0+9+10+22+7
         //
         // Only the specified processor IDs will be used in a round-robin
         // fashion, dealing with the fact that you can request more workers
         // than the number of processor IDs you have specified.
         //
         // To use all available processors in their order:
         //
         //   cpupin=*
 
         opt.Remove(0, 7);
 
         // Remove any char which is neither a number nor a plus '+'
         for (Ssiz_t i=0; i<opt.Length(); i++) {
            Char_t c = opt[i];
            if ((c != '+') && ((c < '0') || (c > '9')))
               opt[i] = '_';
         }
         opt.ReplaceAll("_", "");
         TProof::AddEnvVar("PROOF_SLAVE_CPUPIN_ORDER", opt);
         cpuPin = kTRUE;
      } else if (opt.BeginsWith("workers=")) {
 
         // Request for a given number of workers (within the max) or worker
         // startup combination:
         //      workers=5         start max 5 workers (or less, if less are assigned)
         //      workers=2x        start max 2 workers per node (or less, if less are assigned)
         opt.ReplaceAll("workers=","");
         TProof::AddEnvVar("PROOF_NWORKERS", opt);
      }
   }
 
   // In case of PROOF-Lite, enable CPU pinning when requested (Linux only)
   #ifdef R__LINUX
   if (IsLite() && cpuPin) {
      Printf("*** Requested CPU pinning ***");
      const TList *ev = GetEnvVars();
      const char *pinCmd = "taskset -c <cpupin>";
      TString val;
      TNamed *p;
      if (ev && (p = dynamic_cast<TNamed *>(ev->FindObject("PROOF_SLAVE_WRAPPERCMD")))) {
         val = p->GetTitle();
         val.Insert(val.Length()-1, " ");
         val.Insert(val.Length()-1, pinCmd);
      }
      else {
         val.Form("\"%s\"", pinCmd);
      }
      TProof::AddEnvVar("PROOF_SLAVE_WRAPPERCMD", val.Data());
   }
   #endif
}
 
////////////////////////////////////////////////////////////////////////////////
/// Make sure that 'path' exists; if 'writable' is kTRUE, make also sure
/// that the path is writable
 
Int_t TProof::AssertPath(const char *inpath, Bool_t writable)
{
   if (!inpath || strlen(inpath) <= 0) {
      Error("AssertPath", "undefined input path");
      return -1;
   }
 
   TString path(inpath);
   gSystem->ExpandPathName(path);
 
   if (gSystem->AccessPathName(path, kFileExists)) {
      if (gSystem->mkdir(path, kTRUE) != 0) {
         Error("AssertPath", "could not create path %s", path.Data());
         return -1;
      }
   }
   // It must be writable
   if (gSystem->AccessPathName(path, kWritePermission) && writable) {
      if (gSystem->Chmod(path, 0666) != 0) {
         Error("AssertPath", "could not make path %s writable", path.Data());
         return -1;
      }
   }
 
   // Done
   return 0;
}
 
////////////////////////////////////////////////////////////////////////////////
/// Set manager and schedule its destruction after this for clean
/// operations.
 
void TProof::SetManager(TProofMgr *mgr)
{
   fManager = mgr;
 
   if (mgr) {
      R__LOCKGUARD(gROOTMutex);
      gROOT->GetListOfSockets()->Remove(mgr);
      gROOT->GetListOfSockets()->Add(mgr);
   }
}
 
////////////////////////////////////////////////////////////////////////////////
/// Works on the master node only.
/// It starts workers on the machines in workerList and sets the paths,
/// packages and macros as on the master.
/// It is a subbstitute for StartSlaves(...)
/// The code is mostly the master part of StartSlaves,
/// with the parallel startup removed.
 
Int_t TProof::AddWorkers(TList *workerList)
{
   if (!IsMaster()) {
      Error("AddWorkers", "AddWorkers can only be called on the master!");
      return -1;
   }
 
   if (!workerList || !(workerList->GetSize())) {
      Error("AddWorkers", "empty list of workers!");
      return -2;
   }
 
   // Code taken from master part of StartSlaves with the parllel part removed
 
   fImage = gProofServ->GetImage();
   if (fImage.IsNull())
      fImage.Form("%s:%s", TUrl(gSystem->HostName()).GetHostFQDN(), gProofServ->GetWorkDir());
 
   // Get all workers
   UInt_t nSlaves = workerList->GetSize();
   UInt_t nSlavesDone = 0;
   Int_t ord = 0;
 
   // Loop over all new workers and start them (if we had already workers it means we are
   // increasing parallelism or that is not the first time we are called)
   Bool_t goMoreParallel = (fSlaves->GetEntries() > 0) ? kTRUE : kFALSE;
 
   // A list of TSlave objects for workers that are being added
   TList *addedWorkers = new TList();
   if (!addedWorkers) {
      // This is needed to silence Coverity ...
      Error("AddWorkers", "cannot create new list for the workers to be added");
      return -2;
   }
   addedWorkers->SetOwner(kFALSE);
   TListIter next(workerList);
   TObject *to;
   TProofNodeInfo *worker;
   TSlaveInfo *dummysi = new TSlaveInfo();
   while ((to = next())) {
      // Get the next worker from the list
      worker = (TProofNodeInfo *)to;
 
      // Read back worker node info
      const Char_t *image = worker->GetImage().Data();
      const Char_t *workdir = worker->GetWorkDir().Data();
      Int_t perfidx = worker->GetPerfIndex();
      Int_t sport = worker->GetPort();
      if (sport == -1)
         sport = fUrl.GetPort();
 
      // Create worker server
      TString fullord;
      if (worker->GetOrdinal().Length() > 0) {
         fullord.Form("%s.%s", gProofServ->GetOrdinal(), worker->GetOrdinal().Data());
      } else {
         fullord.Form("%s.%d", gProofServ->GetOrdinal(), ord);
      }
 
      // Remove worker from the list of workers terminated gracefully
      dummysi->SetOrdinal(fullord);
      TSlaveInfo *rmsi = (TSlaveInfo *)fTerminatedSlaveInfos->Remove(dummysi);
      SafeDelete(rmsi);
 
      // Create worker server
      TString wn(worker->GetNodeName());
      if (wn == "localhost" || wn.BeginsWith("localhost.")) wn = gSystem->HostName();
      TUrl u(TString::Format("%s:%d", wn.Data(), sport));
      // Add group info in the password firdl, if any
      if (strlen(gProofServ->GetGroup()) > 0) {
         // Set also the user, otherwise the password is not exported
         if (strlen(u.GetUser()) <= 0)
            u.SetUser(gProofServ->GetUser());
         u.SetPasswd(gProofServ->GetGroup());
      }
      TSlave *slave = 0;
      if (worker->IsWorker()) {
         slave = CreateSlave(u.GetUrl(), fullord, perfidx, image, workdir);
      } else {
         slave = CreateSubmaster(u.GetUrl(), fullord,
                                 image, worker->GetMsd(), worker->GetNWrks());
      }
 
      // Add to global list (we will add to the monitor list after
      // finalizing the server startup)
      Bool_t slaveOk = kTRUE;
      fSlaves->Add(slave);
      if (slave->IsValid()) {
         addedWorkers->Add(slave);
      } else {
         slaveOk = kFALSE;
         fBadSlaves->Add(slave);
         Warning("AddWorkers", "worker '%s' is invalid", slave->GetOrdinal());
      }
 
      PDB(kGlobal,3)
         Info("AddWorkers", "worker on host %s created"
              " and added to list (ord: %s)", worker->GetName(), slave->GetOrdinal());
 
      // Notify opening of connection
      nSlavesDone++;
      TMessage m(kPROOF_SERVERSTARTED);
      m << TString("Opening connections to workers") << nSlaves
        << nSlavesDone << slaveOk;
      gProofServ->GetSocket()->Send(m);
 
      ord++;
   } //end of the worker loop
   SafeDelete(dummysi);
 
   // Cleanup
   SafeDelete(workerList);
 
   nSlavesDone = 0;
 
   // Here we finalize the server startup: in this way the bulk
   // of remote operations are almost parallelized
   TIter nxsl(addedWorkers);
   TSlave *sl = 0;
   while ((sl = (TSlave *) nxsl())) {
 
      // Finalize setup of the server
      if (sl->IsValid())
          sl->SetupServ(TSlave::kSlave, 0);
 
      // Monitor good slaves
      Bool_t slaveOk = kTRUE;
      if (sl->IsValid()) {
         fAllMonitor->Add(sl->GetSocket());
      PDB(kGlobal,3)
         Info("AddWorkers", "worker on host %s finalized"
              " and added to list", sl->GetOrdinal());
      } else {
         slaveOk = kFALSE;
         fBadSlaves->Add(sl);
      }
 
      // Notify end of startup operations
      nSlavesDone++;
      TMessage m(kPROOF_SERVERSTARTED);
      m << TString("Setting up worker servers") << nSlaves
        << nSlavesDone << slaveOk;
      gProofServ->GetSocket()->Send(m);
   }
 
   // Now set new state on the added workers (on all workers for simplicity)
   // use fEnabledPackages, fLoadedMacros,
   // gSystem->GetDynamicPath() and gSystem->GetIncludePath()
   // no need to load packages that are only loaded and not enabled (dyn mode)
   Int_t nwrk = GetRemoteProtocol() > 35 ? -1 : 9999;
   TNamed *n = 0;
   if (TProof::GetEnvVars() &&
      (n = (TNamed *) TProof::GetEnvVars()->FindObject("PROOF_NWORKERS"))) {
      TString s(n->GetTitle());
      if (s.IsDigit()) nwrk = s.Atoi();
   }
 
   if (fDynamicStartup && goMoreParallel) {
 
      PDB(kGlobal, 3)
         Info("AddWorkers", "will invoke GoMoreParallel()");
      Int_t nw = GoMoreParallel(nwrk);
      PDB(kGlobal, 3)
         Info("AddWorkers", "GoMoreParallel()=%d", nw);
 
   }
   else {
      // Not in Dynamic Workers mode
      PDB(kGlobal, 3)
         Info("AddWorkers", "will invoke GoParallel()");
      GoParallel(nwrk, kFALSE, 0);
   }
 
   // Set worker processing environment
   SetupWorkersEnv(addedWorkers, goMoreParallel);
 
   // Update list of current workers
   PDB(kGlobal, 3)
      Info("AddWorkers", "will invoke SaveWorkerInfo()");
   SaveWorkerInfo();
 
   // Inform the client that the number of workers has changed
   if (fDynamicStartup && gProofServ) {
      PDB(kGlobal, 3)
         Info("AddWorkers", "will invoke SendParallel()");
      gProofServ->SendParallel(kTRUE);
 
      if (goMoreParallel && fPlayer) {
         // In case we are adding workers dynamically to an existing process, we
         // should invoke a special player's Process() to set only added workers
         // to the proper state
         PDB(kGlobal, 3)
            Info("AddWorkers", "will send the PROCESS message to selected workers");
         fPlayer->JoinProcess(addedWorkers);
         // Update merger counters (new workers are not yet active)
         fMergePrg.SetNWrks(fActiveSlaves->GetSize() + addedWorkers->GetSize());
      }
   }
 
   // Cleanup
   delete addedWorkers;
 
   return 0;
}
 
////////////////////////////////////////////////////////////////////////////////
/// Set up packages, loaded macros, include and lib paths ...
 
void TProof::SetupWorkersEnv(TList *addedWorkers, Bool_t increasingWorkers)
{
   TList *server_packs = gProofServ ? gProofServ->GetEnabledPackages() : nullptr;
   // Packages
   TList *packs = server_packs ? server_packs : GetEnabledPackages();
   if (packs && packs->GetSize() > 0) {
      TIter nxp(packs);
      TPair *pck = 0;
      while ((pck = (TPair *) nxp())) {
         // Upload and Enable methods are intelligent and avoid
         // re-uploading or re-enabling of a package to a node that has it.
         if (fDynamicStartup && increasingWorkers) {
            // Upload only on added workers
            PDB(kGlobal, 3)
               Info("SetupWorkersEnv", "will invoke UploadPackage() and EnablePackage() on added workers");
            if (UploadPackage(pck->GetName(), kUntar, addedWorkers) >= 0)
               EnablePackage(pck->GetName(), (TList *) pck->Value(), kTRUE, addedWorkers);
         } else {
            PDB(kGlobal, 3)
               Info("SetupWorkersEnv", "will invoke UploadPackage() and EnablePackage() on all workers");
            if (UploadPackage(pck->GetName()) >= 0)
               EnablePackage(pck->GetName(), (TList *) pck->Value(), kTRUE);
         }
      }
   }
 
   if (server_packs) {
      server_packs->Delete();
      delete server_packs;
   }
 
   // Loaded macros
   if (fLoadedMacros) {
      TIter nxp(fLoadedMacros);
      TObjString *os = 0;
      while ((os = (TObjString *) nxp())) {
         PDB(kGlobal, 3) {
            Info("SetupWorkersEnv", "will invoke Load() on selected workers");
            Printf("Loading a macro : %s", os->GetName());
         }
         Load(os->GetName(), kTRUE, kTRUE, addedWorkers);
      }
   }
 
   // Dynamic path
   TString dyn = gSystem->GetDynamicPath();
   dyn.ReplaceAll(":", " ");
   dyn.ReplaceAll("\"", " ");
   PDB(kGlobal, 3)
      Info("SetupWorkersEnv", "will invoke AddDynamicPath() on selected workers");
   AddDynamicPath(dyn, kFALSE, addedWorkers, kFALSE); // Do not Collect
 
   // Include path
   TString inc = gSystem->GetIncludePath();
   inc.ReplaceAll("-I", " ");
   inc.ReplaceAll("\"", " ");
   PDB(kGlobal, 3)
      Info("SetupWorkersEnv", "will invoke AddIncludePath() on selected workers");
   AddIncludePath(inc, kFALSE, addedWorkers, kFALSE);  // Do not Collect
 
   // Done
   return;
}
 
////////////////////////////////////////////////////////////////////////////////
/// Used for shuting down the workres after a query is finished.
/// Sends each of the workers from the workerList, a kPROOF_STOP message.
/// If the workerList == 0, shutdown all the workers.
 
Int_t TProof::RemoveWorkers(TList *workerList)
{
   if (!IsMaster()) {
      Error("RemoveWorkers", "RemoveWorkers can only be called on the master!");
      return -1;
   }
 
   fFileMap.clear(); // This could be avoided if CopyFromCache was used in SendFile
 
   if (!workerList) {
      // shutdown all the workers
      TIter nxsl(fSlaves);
      TSlave *sl = 0;
      while ((sl = (TSlave *) nxsl())) {
         // Shut down the worker assumig that it is not processing
         TerminateWorker(sl);
      }
 
   } else {
      if (!(workerList->GetSize())) {
         Error("RemoveWorkers", "The list of workers should not be empty!");
         return -2;
      }
 
      // Loop over all the workers and stop them
      TListIter next(workerList);
      TObject *to;
      TProofNodeInfo *worker;
      while ((to = next())) {
         TSlave *sl = 0;
         if (!strcmp(to->ClassName(), "TProofNodeInfo")) {
            // Get the next worker from the list
            worker = (TProofNodeInfo *)to;
            TIter nxsl(fSlaves);
            while ((sl = (TSlave *) nxsl())) {
               // Shut down the worker assumig that it is not processing
               if (sl->GetName() == worker->GetNodeName())
                  break;
            }
         } else if (to->InheritsFrom(TSlave::Class())) {
            sl = (TSlave *) to;
         } else {
            Warning("RemoveWorkers","unknown object type: %s - it should be"
                  " TProofNodeInfo or inheriting from TSlave", to->ClassName());
         }
         // Shut down the worker assumig that it is not processing
         if (sl) {
            if (gDebug > 0)
               Info("RemoveWorkers","terminating worker %s", sl->GetOrdinal());
            TerminateWorker(sl);
         }
      }
   }
 
   // Update also the master counter
   if (gProofServ && fSlaves->GetSize() <= 0) gProofServ->ReleaseWorker("master");
 
   return 0;
}
 
////////////////////////////////////////////////////////////////////////////////
/// Start up PROOF slaves.
 
Bool_t TProof::StartSlaves(Bool_t attach)
{
   // If this is a master server, find the config file and start slave
   // servers as specified in the config file
   if (TestBit(TProof::kIsMaster)) {
 
      Int_t pc = 0;
      TList *workerList = new TList;
      // Get list of workers
      if (gProofServ->GetWorkers(workerList, pc) == TProofServ::kQueryStop) {
         TString emsg("no resource currently available for this session: please retry later");
         if (gDebug > 0) Info("StartSlaves", "%s", emsg.Data());
         gProofServ->SendAsynMessage(emsg.Data());
         return kFALSE;
      }
      // Setup the workers
      if (AddWorkers(workerList) < 0)
         return kFALSE;
 
   } else {
 
      // create master server
      Printf("Starting master: opening connection ...");
      TSlave *slave = CreateSubmaster(fUrl.GetUrl(), "0", "master", 0);
 
      if (slave->IsValid()) {
 
         // Notify
         fprintf(stderr,"Starting master:"
                        " connection open: setting up server ...             \r");
         StartupMessage("Connection to master opened", kTRUE, 1, 1);
 
         if (!attach) {
 
            // Set worker interrupt handler
            slave->SetInterruptHandler(kTRUE);
 
            // Finalize setup of the server
            slave->SetupServ(TSlave::kMaster, fConfFile);
 
            if (slave->IsValid()) {
 
               // Notify
               Printf("Starting master: OK                                     ");
               StartupMessage("Master started", kTRUE, 1, 1);
 
               // check protocol compatibility
               // protocol 1 is not supported anymore
               if (fProtocol == 1) {
                  Error("StartSlaves",
                        "client and remote protocols not compatible (%d and %d)",
                        kPROOF_Protocol, fProtocol);
                  slave->Close("S");
                  delete slave;
                  return kFALSE;
               }
 
               fSlaves->Add(slave);
               fAllMonitor->Add(slave->GetSocket());
 
               // Unset worker interrupt handler
               slave->SetInterruptHandler(kFALSE);
 
               // Set interrupt PROOF handler from now on
               fIntHandler = new TProofInterruptHandler(this);
 
               // Give-up after 5 minutes
               Int_t rc = Collect(slave, 300);
               Int_t slStatus = slave->GetStatus();
               if (slStatus == -99 || slStatus == -98 || rc == 0) {
                  fSlaves->Remove(slave);
                  fAllMonitor->Remove(slave->GetSocket());
                  if (slStatus == -99)
                     Error("StartSlaves", "no resources available or problems setting up workers (check logs)");
                  else if (slStatus == -98)
                     Error("StartSlaves", "could not setup output redirection on master");
                  else
                     Error("StartSlaves", "setting up master");
                  slave->Close("S");
                  delete slave;
                  return 0;
               }
 
               if (!slave->IsValid()) {
                  fSlaves->Remove(slave);
                  fAllMonitor->Remove(slave->GetSocket());
                  slave->Close("S");
                  delete slave;
                  Error("StartSlaves",
                        "failed to setup connection with PROOF master server");
                  return kFALSE;
               }
 
               if (!gROOT->IsBatch() && TestBit(kUseProgressDialog)) {
                  if ((fProgressDialog =
                     gROOT->GetPluginManager()->FindHandler("TProofProgressDialog")))
                     if (fProgressDialog->LoadPlugin() == -1)
                        fProgressDialog = 0;
               }
            } else {
               // Notify
               Printf("Starting master: failure");
            }
         } else {
 
            // Notify
            Printf("Starting master: OK                                     ");
            StartupMessage("Master attached", kTRUE, 1, 1);
 
            if (!gROOT->IsBatch() && TestBit(kUseProgressDialog)) {
               if ((fProgressDialog =
                  gROOT->GetPluginManager()->FindHandler("TProofProgressDialog")))
                  if (fProgressDialog->LoadPlugin() == -1)
                     fProgressDialog = 0;
            }
 
            fSlaves->Add(slave);
            fIntHandler = new TProofInterruptHandler(this);
         }
 
      } else {
         delete slave;
         // Notify only if verbosity is on: most likely the failure has already been notified
         if (gDebug > 0)
            Error("StartSlaves", "failed to create (or connect to) the PROOF master server");
         return kFALSE;
      }
   }
 
   return kTRUE;
}
 
////////////////////////////////////////////////////////////////////////////////
/// Close all open slave servers.
/// Client can decide to shutdown the remote session by passing option is 'S'
/// or 's'. Default for clients is detach, if supported. Masters always
/// shutdown the remote counterpart.
 
void TProof::Close(Option_t *opt)
{
   {  std::lock_guard<std::recursive_mutex> lock(fCloseMutex);
 
      fValid = kFALSE;
      if (fSlaves) {
         if (fIntHandler)
            fIntHandler->Remove();
 
         TIter nxs(fSlaves);
         TSlave *sl = 0;
         while ((sl = (TSlave *)nxs()))
            sl->Close(opt);
 
         fActiveSlaves->Clear("nodelete");
         fUniqueSlaves->Clear("nodelete");
         fAllUniqueSlaves->Clear("nodelete");
         fNonUniqueMasters->Clear("nodelete");
         fBadSlaves->Clear("nodelete");
         fInactiveSlaves->Clear("nodelete");
         fSlaves->Delete();
      }
   }
 
   {  R__LOCKGUARD(gROOTMutex);
      gROOT->GetListOfSockets()->Remove(this);
 
      if (fChains) {
         while (TChain *chain = dynamic_cast<TChain*> (fChains->First()) ) {
            // remove "chain" from list
            chain->SetProof(0);
            RemoveChain(chain);
         }
      }
 
      if (IsProofd()) {
 
         gROOT->GetListOfProofs()->Remove(this);
         if (gProof && gProof == this) {
            // Set previous proofd-related as default
            TIter pvp(gROOT->GetListOfProofs(), kIterBackward);
            while ((gProof = (TProof *)pvp())) {
               if (gProof->IsProofd())
                  break;
            }
         }
      }
   }
}
 
////////////////////////////////////////////////////////////////////////////////
/// Create a new TSlave of type TSlave::kSlave.
/// Note: creation of TSlave is private with TProof as a friend.
/// Derived classes must use this function to create slaves.
 
TSlave *TProof::CreateSlave(const char *url, const char *ord,
                            Int_t perf, const char *image, const char *workdir)
{
   TSlave* sl = TSlave::Create(url, ord, perf, image,
                               this, TSlave::kSlave, workdir, 0);
 
   if (sl->IsValid()) {
      sl->SetInputHandler(new TProofInputHandler(this, sl->GetSocket()));
      // must set fParallel to 1 for slaves since they do not
      // report their fParallel with a LOG_DONE message
      sl->fParallel = 1;
   }
 
   return sl;
}
 
 
////////////////////////////////////////////////////////////////////////////////
/// Create a new TSlave of type TSlave::kMaster.
/// Note: creation of TSlave is private with TProof as a friend.
/// Derived classes must use this function to create slaves.
 
TSlave *TProof::CreateSubmaster(const char *url, const char *ord,
                                const char *image, const char *msd, Int_t nwk)
{
   TSlave *sl = TSlave::Create(url, ord, 100, image, this,
                               TSlave::kMaster, 0, msd, nwk);
 
   if (sl->IsValid()) {
      sl->SetInputHandler(new TProofInputHandler(this, sl->GetSocket()));
   }
 
   return sl;
}
 
////////////////////////////////////////////////////////////////////////////////
/// Find slave that has TSocket s. Returns 0 in case slave is not found.
 
TSlave *TProof::FindSlave(TSocket *s) const
{
   TSlave *sl;
   TIter   next(fSlaves);
 
   while ((sl = (TSlave *)next())) {
      if (sl->IsValid() && sl->GetSocket() == s)
         return sl;
   }
   return 0;
}
 
////////////////////////////////////////////////////////////////////////////////
/// Add to the fUniqueSlave list the active slaves that have a unique
/// (user) file system image. This information is used to transfer files
/// only once to nodes that share a file system (an image). Submasters
/// which are not in fUniqueSlaves are put in the fNonUniqueMasters
/// list. That list is used to trigger the transferring of files to
/// the submaster's unique slaves without the need to transfer the file
/// to the submaster.
 
void TProof::FindUniqueSlaves()
{
   fUniqueSlaves->Clear();
   fUniqueMonitor->RemoveAll();
   fAllUniqueSlaves->Clear();
   fAllUniqueMonitor->RemoveAll();
   fNonUniqueMasters->Clear();
 
   TIter next(fActiveSlaves);
 
   while (TSlave *sl = dynamic_cast<TSlave*>(next())) {
      if (fImage == sl->fImage) {
         if (sl->GetSlaveType() == TSlave::kMaster) {
            fNonUniqueMasters->Add(sl);
            fAllUniqueSlaves->Add(sl);
            fAllUniqueMonitor->Add(sl->GetSocket());
         }
         continue;
      }
 
      TIter next2(fUniqueSlaves);
      TSlave *replace_slave = 0;
      Bool_t add = kTRUE;
      while (TSlave *sl2 = dynamic_cast<TSlave*>(next2())) {
         if (sl->fImage == sl2->fImage) {
            add = kFALSE;
            if (sl->GetSlaveType() == TSlave::kMaster) {
               if (sl2->GetSlaveType() == TSlave::kSlave) {
                  // give preference to master
                  replace_slave = sl2;
                  add = kTRUE;
               } else if (sl2->GetSlaveType() == TSlave::kMaster) {
                  fNonUniqueMasters->Add(sl);
                  fAllUniqueSlaves->Add(sl);
                  fAllUniqueMonitor->Add(sl->GetSocket());
               } else {
                  Error("FindUniqueSlaves", "TSlave is neither Master nor Slave");
                  R__ASSERT(0);
               }
            }
            break;
         }
      }
 
      if (add) {
         fUniqueSlaves->Add(sl);
         fAllUniqueSlaves->Add(sl);
         fUniqueMonitor->Add(sl->GetSocket());
         fAllUniqueMonitor->Add(sl->GetSocket());
         if (replace_slave) {
            fUniqueSlaves->Remove(replace_slave);
            fAllUniqueSlaves->Remove(replace_slave);
            fUniqueMonitor->Remove(replace_slave->GetSocket());
            fAllUniqueMonitor->Remove(replace_slave->GetSocket());
         }
      }
   }
 
   // will be actiavted in Collect()
   fUniqueMonitor->DeActivateAll();
   fAllUniqueMonitor->DeActivateAll();
}
 
////////////////////////////////////////////////////////////////////////////////
/// Return number of slaves as described in the config file.
 
Int_t TProof::GetNumberOfSlaves() const
{
   return fSlaves->GetSize();
}
 
////////////////////////////////////////////////////////////////////////////////
/// Return number of active slaves, i.e. slaves that are valid and in
/// the current computing group.
 
Int_t TProof::GetNumberOfActiveSlaves() const
{
   return fActiveSlaves->GetSize();
}
 
////////////////////////////////////////////////////////////////////////////////
/// Return number of inactive slaves, i.e. slaves that are valid but not in
/// the current computing group.
 
Int_t TProof::GetNumberOfInactiveSlaves() const
{
   return fInactiveSlaves->GetSize();
}
 
////////////////////////////////////////////////////////////////////////////////
/// Return number of unique slaves, i.e. active slaves that have each a
/// unique different user files system.
 
Int_t TProof::GetNumberOfUniqueSlaves() const
{
   return fUniqueSlaves->GetSize();
}
 
////////////////////////////////////////////////////////////////////////////////
/// Return number of bad slaves. This are slaves that we in the config
/// file, but refused to startup or that died during the PROOF session.
 
Int_t TProof::GetNumberOfBadSlaves() const
{
   return fBadSlaves->GetSize();
}
 
////////////////////////////////////////////////////////////////////////////////
/// Ask the for the statistics of the slaves.
 
void TProof::AskStatistics()
{
   if (!IsValid()) return;
 
   Broadcast(kPROOF_GETSTATS, kActive);
   Collect(kActive, fCollectTimeout);
}
 
////////////////////////////////////////////////////////////////////////////////
/// Get statistics about CPU time, real time and bytes read.
/// If verbose, print the resuls (always available via GetCpuTime(), GetRealTime()
/// and GetBytesRead()
 
void TProof::GetStatistics(Bool_t verbose)
{
   if (fProtocol > 27) {
      // This returns the correct result
      AskStatistics();
   } else {
      // AskStatistics is buggy: parse the output of Print()
      RedirectHandle_t rh;
      gSystem->RedirectOutput(fLogFileName, "a", &rh);
      Print();
      gSystem->RedirectOutput(0, 0, &rh);
      TMacro *mp = GetLastLog();
      if (mp) {
         // Look for global directories
         TIter nxl(mp->GetListOfLines());
         TObjString *os = 0;
         while ((os = (TObjString *) nxl())) {
            TString s(os->GetName());
            if (s.Contains("Total MB's processed:")) {
               s.ReplaceAll("Total MB's processed:", "");
               if (s.IsFloat()) fBytesRead = (Long64_t) s.Atof() * (1024*1024);
            } else if (s.Contains("Total real time used (s):")) {
               s.ReplaceAll("Total real time used (s):", "");
               if (s.IsFloat()) fRealTime = s.Atof();
            } else if (s.Contains("Total CPU time used (s):")) {
               s.ReplaceAll("Total CPU time used (s):", "");
               if (s.IsFloat()) fCpuTime = s.Atof();
            }
         }
         delete mp;
      }
   }
 
   if (verbose) {
      Printf(" Real/CPU time (s): %.3f / %.3f; workers: %d; processed: %.2f MBs",
             GetRealTime(), GetCpuTime(), GetParallel(), float(GetBytesRead())/(1024*1024));
   }
}
 
////////////////////////////////////////////////////////////////////////////////
/// Ask the for the number of parallel slaves.
 
void TProof::AskParallel()
{
   if (!IsValid()) return;
 
   Broadcast(kPROOF_GETPARALLEL, kActive);
   Collect(kActive, fCollectTimeout);
}
 
////////////////////////////////////////////////////////////////////////////////
/// Ask the master for the list of queries.
 
TList *TProof::GetListOfQueries(Option_t *opt)
{
   if (!IsValid() || TestBit(TProof::kIsMaster)) return (TList *)0;
 
   Bool_t all = ((strchr(opt,'A') || strchr(opt,'a'))) ? kTRUE : kFALSE;
   TMessage m(kPROOF_QUERYLIST);
   m << all;
   Broadcast(m, kActive);
   Collect(kActive, fCollectTimeout);
 
   // This should have been filled by now
   return fQueries;
}
 
////////////////////////////////////////////////////////////////////////////////
/// Number of queries processed by this session
 
Int_t TProof::GetNumberOfQueries()
{
   if (fQueries)
      return fQueries->GetSize() - fOtherQueries;
   return 0;
}
 
////////////////////////////////////////////////////////////////////////////////
/// Set max number of draw queries whose results are saved
 
void TProof::SetMaxDrawQueries(Int_t max)
{
   if (max > 0) {
      if (fPlayer)
         fPlayer->SetMaxDrawQueries(max);
      fMaxDrawQueries = max;
   }
}
 
////////////////////////////////////////////////////////////////////////////////
/// Get max number of queries whose full results are kept in the
/// remote sandbox
 
void TProof::GetMaxQueries()
{
   TMessage m(kPROOF_MAXQUERIES);
   m << kFALSE;
   Broadcast(m, kActive);
   Collect(kActive, fCollectTimeout);
}
 
////////////////////////////////////////////////////////////////////////////////
/// Return pointer to the list of query results in the player
 
TList *TProof::GetQueryResults()
{
   return (fPlayer ? fPlayer->GetListOfResults() : (TList *)0);
}
 
////////////////////////////////////////////////////////////////////////////////
/// Return pointer to the full TQueryResult instance owned by the player
/// and referenced by 'ref'. If ref = 0 or "", return the last query result.
 
TQueryResult *TProof::GetQueryResult(const char *ref)
{
   return (fPlayer ? fPlayer->GetQueryResult(ref) : (TQueryResult *)0);
}
 
////////////////////////////////////////////////////////////////////////////////
/// Ask the master for the list of queries.
/// Options:
///           "A"     show information about all the queries known to the
///                   server, i.e. even those processed by other sessions
///           "L"     show only information about queries locally available
///                   i.e. already retrieved. If "L" is specified, "A" is
///                   ignored.
///           "F"     show all details available about queries
///           "H"     print help menu
/// Default ""
 
void TProof::ShowQueries(Option_t *opt)
{
   Bool_t help = ((strchr(opt,'H') || strchr(opt,'h'))) ? kTRUE : kFALSE;
   if (help) {
 
      // Help
 
      Printf("+++");
      Printf("+++ Options: \"A\" show all queries known to server");
      Printf("+++          \"L\" show retrieved queries");
      Printf("+++          \"F\" full listing of query info");
      Printf("+++          \"H\" print this menu");
      Printf("+++");
      Printf("+++ (case insensitive)");
      Printf("+++");
      Printf("+++ Use Retrieve(<#>) to retrieve the full"
             " query results from the master");
      Printf("+++     e.g. Retrieve(8)");
 
      Printf("+++");
 
      return;
   }
 
   if (!IsValid()) return;
 
   Bool_t local = ((strchr(opt,'L') || strchr(opt,'l'))) ? kTRUE : kFALSE;
 
   TObject *pq = 0;
   if (!local) {
      GetListOfQueries(opt);
 
      if (!fQueries) return;
 
      TIter nxq(fQueries);
 
      // Queries processed by other sessions
      if (fOtherQueries > 0) {
         Printf("+++");
         Printf("+++ Queries processed during other sessions: %d", fOtherQueries);
         Int_t nq = 0;
         while (nq++ < fOtherQueries && (pq = nxq()))
            pq->Print(opt);
      }
 
      // Queries processed by this session
      Printf("+++");
      Printf("+++ Queries processed during this session: selector: %d, draw: %d",
              GetNumberOfQueries(), fDrawQueries);
      while ((pq = nxq()))
         pq->Print(opt);
 
   } else {
 
      // Queries processed by this session
      Printf("+++");
      Printf("+++ Queries processed during this session: selector: %d, draw: %d",
              GetNumberOfQueries(), fDrawQueries);
 
      // Queries available locally
      TList *listlocal = fPlayer ? fPlayer->GetListOfResults() : (TList *)0;
      if (listlocal) {
         Printf("+++");
         Printf("+++ Queries available locally: %d", listlocal->GetSize());
         TIter nxlq(listlocal);
         while ((pq = nxlq()))
            pq->Print(opt);
      }
   }
   Printf("+++");
}
 
////////////////////////////////////////////////////////////////////////////////
/// See if the data is ready to be analyzed.
 
Bool_t TProof::IsDataReady(Long64_t &totalbytes, Long64_t &bytesready)
{
   if (!IsValid()) return kFALSE;
 
   TList submasters;
   TIter nextSlave(GetListOfActiveSlaves());
   while (TSlave *sl = dynamic_cast<TSlave*>(nextSlave())) {
      if (sl->GetSlaveType() == TSlave::kMaster) {
         submasters.Add(sl);
      }
   }
 
   fDataReady = kTRUE; //see if any submasters set it to false
   fBytesReady = 0;
   fTotalBytes = 0;
   //loop over submasters and see if data is ready
   if (submasters.GetSize() > 0) {
      Broadcast(kPROOF_DATA_READY, &submasters);
      Collect(&submasters);
   }
 
   bytesready = fBytesReady;
   totalbytes = fTotalBytes;
 
   EmitVA("IsDataReady(Long64_t,Long64_t)", 2, totalbytes, bytesready);
 
   PDB(kGlobal,2)
      Info("IsDataReady", "%lld / %lld (%s)",
           bytesready, totalbytes, fDataReady?"READY":"NOT READY");
 
   return fDataReady;
}
 
////////////////////////////////////////////////////////////////////////////////
/// Send interrupt to master or slave servers.
 
void TProof::Interrupt(EUrgent type, ESlaves list)
{
   if (!IsValid()) return;
 
   TList *slaves = 0;
   if (list == kAll)       slaves = fSlaves;
   if (list == kActive)    slaves = fActiveSlaves;
   if (list == kUnique)    slaves = fUniqueSlaves;
   if (list == kAllUnique) slaves = fAllUniqueSlaves;
 
   if (slaves->GetSize() == 0) return;
 
   TSlave *sl;
   TIter   next(slaves);
 
   while ((sl = (TSlave *)next())) {
      if (sl->IsValid()) {
 
         // Ask slave to progate the interrupt request
         sl->Interrupt((Int_t)type);
      }
   }
}
 
////////////////////////////////////////////////////////////////////////////////
/// Returns number of slaves active in parallel mode. Returns 0 in case
/// there are no active slaves. Returns -1 in case of error.
 
Int_t TProof::GetParallel() const
{
   if (!IsValid()) return -1;
 
   // iterate over active slaves and return total number of slaves
   TIter nextSlave(GetListOfActiveSlaves());
   Int_t nparallel = 0;
   while (TSlave* sl = dynamic_cast<TSlave*>(nextSlave()))
      if (sl->GetParallel() >= 0)
         nparallel += sl->GetParallel();
 
   return nparallel;
}
 
////////////////////////////////////////////////////////////////////////////////
/// Returns list of TSlaveInfo's. In case of error return 0.
 
TList *TProof::GetListOfSlaveInfos()
{
   if (!IsValid()) return 0;
 
   if (fSlaveInfo == 0) {
      fSlaveInfo = new TSortedList(kSortDescending);
      fSlaveInfo->SetOwner();
   } else {
      fSlaveInfo->Delete();
   }
 
   TList masters;
   TIter next(GetListOfSlaves());
   TSlave *slave;
 
   while ((slave = (TSlave *) next()) != 0) {
      if (slave->GetSlaveType() == TSlave::kSlave) {
         const char *name = IsLite() ? gSystem->HostName() : slave->GetName();
         TSlaveInfo *slaveinfo = new TSlaveInfo(slave->GetOrdinal(),
                                                name,
                                                slave->GetPerfIdx());
         fSlaveInfo->Add(slaveinfo);
 
         TIter nextactive(GetListOfActiveSlaves());
         TSlave *activeslave;
         while ((activeslave = (TSlave *) nextactive())) {
            if (TString(slaveinfo->GetOrdinal()) == activeslave->GetOrdinal()) {
               slaveinfo->SetStatus(TSlaveInfo::kActive);
               break;
            }
         }
 
         TIter nextbad(GetListOfBadSlaves());
         TSlave *badslave;
         while ((badslave = (TSlave *) nextbad())) {
            if (TString(slaveinfo->GetOrdinal()) == badslave->GetOrdinal()) {
               slaveinfo->SetStatus(TSlaveInfo::kBad);
               break;
            }
         }
         // Get system info if supported
         if (slave->IsValid()) {
            if (slave->GetSocket()->Send(kPROOF_GETSLAVEINFO) == -1)
               MarkBad(slave, "could not send kPROOF_GETSLAVEINFO message");
            else
               masters.Add(slave);
         }
 
      } else if (slave->GetSlaveType() == TSlave::kMaster) {
         if (slave->IsValid()) {
            if (slave->GetSocket()->Send(kPROOF_GETSLAVEINFO) == -1)
               MarkBad(slave, "could not send kPROOF_GETSLAVEINFO message");
            else
               masters.Add(slave);
         }
      } else {
         Error("GetSlaveInfo", "TSlave is neither Master nor Slave");
         R__ASSERT(0);
      }
   }
   if (masters.GetSize() > 0) Collect(&masters);
 
   return fSlaveInfo;
}
 
////////////////////////////////////////////////////////////////////////////////
/// Activate slave server list.
 
void TProof::Activate(TList *slaves)
{
   TMonitor *mon = fAllMonitor;
   mon->DeActivateAll();
 
   slaves = !slaves ? fActiveSlaves : slaves;
 
   TIter next(slaves);
   TSlave *sl;
   while ((sl = (TSlave*) next())) {
      if (sl->IsValid())
         mon->Activate(sl->GetSocket());
   }
}
 
////////////////////////////////////////////////////////////////////////////////
/// Activate (on == TRUE) or deactivate (on == FALSE) all sockets
/// monitored by 'mon'.
 
void TProof::SetMonitor(TMonitor *mon, Bool_t on)
{
   TMonitor *m = (mon) ? mon : fCurrentMonitor;
   if (m) {
      if (on)
         m->ActivateAll();
      else
         m->DeActivateAll();
   }
}
 
////////////////////////////////////////////////////////////////////////////////
/// Broadcast the group priority to all workers in the specified list. Returns
/// the number of workers the message was successfully sent to.
/// Returns -1 in case of error.
 
Int_t TProof::BroadcastGroupPriority(const char *grp, Int_t priority, TList *workers)
{
   if (!IsValid()) return -1;
 
   if (workers->GetSize() == 0) return 0;
 
   int   nsent = 0;
   TIter next(workers);
 
   TSlave *wrk;
   while ((wrk = (TSlave *)next())) {
      if (wrk->IsValid()) {
         if (wrk->SendGroupPriority(grp, priority) == -1)
            MarkBad(wrk, "could not send group priority");
         else
            nsent++;
      }
   }
 
   return nsent;
}
 
////////////////////////////////////////////////////////////////////////////////
/// Broadcast the group priority to all workers in the specified list. Returns
/// the number of workers the message was successfully sent to.
/// Returns -1 in case of error.
 
Int_t TProof::BroadcastGroupPriority(const char *grp, Int_t priority, ESlaves list)
{
   TList *workers = 0;
   if (list == kAll)       workers = fSlaves;
   if (list == kActive)    workers = fActiveSlaves;
   if (list == kUnique)    workers = fUniqueSlaves;
   if (list == kAllUnique) workers = fAllUniqueSlaves;
 
   return BroadcastGroupPriority(grp, priority, workers);
}
 
////////////////////////////////////////////////////////////////////////////////
/// Reset the merge progress notificator
 
void TProof::ResetMergePrg()
{
   fMergePrg.Reset(fActiveSlaves->GetSize());
}
 
////////////////////////////////////////////////////////////////////////////////
/// Broadcast a message to all slaves in the specified list. Returns
/// the number of slaves the message was successfully sent to.
/// Returns -1 in case of error.
 
Int_t TProof::Broadcast(const TMessage &mess, TList *slaves)
{
   if (!IsValid()) return -1;
 
   if (!slaves || slaves->GetSize() == 0) return 0;
 
   int   nsent = 0;
   TIter next(slaves);
 
   TSlave *sl;
   while ((sl = (TSlave *)next())) {
      if (sl->IsValid()) {
         if (sl->GetSocket()->Send(mess) == -1)
            MarkBad(sl, "could not broadcast request");
         else
            nsent++;
      }
   }
 
   return nsent;
}
 
////////////////////////////////////////////////////////////////////////////////
/// Broadcast a message to all slaves in the specified list (either
/// all slaves or only the active slaves). Returns the number of slaves
/// the message was successfully sent to. Returns -1 in case of error.
 
Int_t TProof::Broadcast(const TMessage &mess, ESlaves list)
{
   TList *slaves = 0;
   if (list == kAll)       slaves = fSlaves;
   if (list == kActive)    slaves = fActiveSlaves;
   if (list == kUnique)    slaves = fUniqueSlaves;
   if (list == kAllUnique) slaves = fAllUniqueSlaves;
 
   return Broadcast(mess, slaves);
}
 
////////////////////////////////////////////////////////////////////////////////
/// Broadcast a character string buffer to all slaves in the specified
/// list. Use kind to set the TMessage what field. Returns the number of
/// slaves the message was sent to. Returns -1 in case of error.
 
Int_t TProof::Broadcast(const char *str, Int_t kind, TList *slaves)
{
   TMessage mess(kind);
   if (str) mess.WriteString(str);
   return Broadcast(mess, slaves);
}
 
////////////////////////////////////////////////////////////////////////////////
/// Broadcast a character string buffer to all slaves in the specified
/// list (either all slaves or only the active slaves). Use kind to
/// set the TMessage what field. Returns the number of slaves the message
/// was sent to. Returns -1 in case of error.
 
Int_t TProof::Broadcast(const char *str, Int_t kind, ESlaves list)
{
   TMessage mess(kind);
   if (str) mess.WriteString(str);
   return Broadcast(mess, list);
}
 
////////////////////////////////////////////////////////////////////////////////
/// Broadcast an object to all slaves in the specified list. Use kind to
/// set the TMEssage what field. Returns the number of slaves the message
/// was sent to. Returns -1 in case of error.
 
Int_t TProof::BroadcastObject(const TObject *obj, Int_t kind, TList *slaves)
{
   TMessage mess(kind);
   mess.WriteObject(obj);
   return Broadcast(mess, slaves);
}
 
////////////////////////////////////////////////////////////////////////////////
/// Broadcast an object to all slaves in the specified list. Use kind to
/// set the TMEssage what field. Returns the number of slaves the message
/// was sent to. Returns -1 in case of error.
 
Int_t TProof::BroadcastObject(const TObject *obj, Int_t kind, ESlaves list)
{
   TMessage mess(kind);
   mess.WriteObject(obj);
   return Broadcast(mess, list);
}
 
////////////////////////////////////////////////////////////////////////////////
/// Broadcast a raw buffer of specified length to all slaves in the
/// specified list. Returns the number of slaves the buffer was sent to.
/// Returns -1 in case of error.
 
Int_t TProof::BroadcastRaw(const void *buffer, Int_t length, TList *slaves)
{
   if (!IsValid()) return -1;
 
   if (slaves->GetSize() == 0) return 0;
 
   int   nsent = 0;
   TIter next(slaves);
 
   TSlave *sl;
   while ((sl = (TSlave *)next())) {
      if (sl->IsValid()) {
         if (sl->GetSocket()->SendRaw(buffer, length) == -1)
            MarkBad(sl, "could not send broadcast-raw request");
         else
            nsent++;
      }
   }
 
   return nsent;
}
 
////////////////////////////////////////////////////////////////////////////////
/// Broadcast a raw buffer of specified length to all slaves in the
/// specified list. Returns the number of slaves the buffer was sent to.
/// Returns -1 in case of error.
 
Int_t TProof::BroadcastRaw(const void *buffer, Int_t length, ESlaves list)
{
   TList *slaves = 0;
   if (list == kAll)       slaves = fSlaves;
   if (list == kActive)    slaves = fActiveSlaves;
   if (list == kUnique)    slaves = fUniqueSlaves;
   if (list == kAllUnique) slaves = fAllUniqueSlaves;
 
   return BroadcastRaw(buffer, length, slaves);
}
 
////////////////////////////////////////////////////////////////////////////////
/// Broadcast file to all workers in the specified list. Returns the number of workers
/// the buffer was sent to.
/// Returns -1 in case of error.
 
Int_t TProof::BroadcastFile(const char *file, Int_t opt, const char *rfile, TList *wrks)
{
   if (!IsValid()) return -1;
 
   if (wrks->GetSize() == 0) return 0;
 
   int   nsent = 0;
   TIter next(wrks);
 
   TSlave *wrk;
   while ((wrk = (TSlave *)next())) {
      if (wrk->IsValid()) {
         if (SendFile(file, opt, rfile, wrk) < 0)
            Error("BroadcastFile",
                  "problems sending file to worker %s (%s)",
                  wrk->GetOrdinal(), wrk->GetName());
         else
            nsent++;
      }
   }
 
   return nsent;
}
 
////////////////////////////////////////////////////////////////////////////////
/// Broadcast file to all workers in the specified list. Returns the number of workers
/// the buffer was sent to.
/// Returns -1 in case of error.
 
Int_t TProof::BroadcastFile(const char *file, Int_t opt, const char *rfile, ESlaves list)
{
   TList *wrks = 0;
   if (list == kAll)       wrks = fSlaves;
   if (list == kActive)    wrks = fActiveSlaves;
   if (list == kUnique)    wrks = fUniqueSlaves;
   if (list == kAllUnique) wrks = fAllUniqueSlaves;
 
   return BroadcastFile(file, opt, rfile, wrks);
}
 
////////////////////////////////////////////////////////////////////////////////
/// Release the used monitor to be used, making sure to delete newly created
/// monitors.
 
void TProof::ReleaseMonitor(TMonitor *mon)
{
   if (mon && (mon != fAllMonitor) && (mon != fActiveMonitor)
           && (mon != fUniqueMonitor) && (mon != fAllUniqueMonitor)) {
      delete mon;
   }
}
 
////////////////////////////////////////////////////////////////////////////////
/// Collect responses from slave sl. Returns the number of slaves that
/// responded (=1).
/// If timeout >= 0, wait at most timeout seconds (timeout = -1 by default,
/// which means wait forever).
/// If defined (>= 0) endtype is the message that stops this collection.
 
Int_t TProof::Collect(const TSlave *sl, Long_t timeout, Int_t endtype, Bool_t deactonfail)
{
   Int_t rc = 0;
 
   TMonitor *mon = 0;
   if (!sl->IsValid()) return 0;
 
   if (fCurrentMonitor == fAllMonitor) {
      mon = new TMonitor;
   } else {
      mon = fAllMonitor;
      mon->DeActivateAll();
   }
   mon->Activate(sl->GetSocket());
 
   rc = Collect(mon, timeout, endtype, deactonfail);
   ReleaseMonitor(mon);
   return rc;
}
 
////////////////////////////////////////////////////////////////////////////////
/// Collect responses from the slave servers. Returns the number of slaves
/// that responded.
/// If timeout >= 0, wait at most timeout seconds (timeout = -1 by default,
/// which means wait forever).
/// If defined (>= 0) endtype is the message that stops this collection.
 
Int_t TProof::Collect(TList *slaves, Long_t timeout, Int_t endtype, Bool_t deactonfail)
{
   Int_t rc = 0;
 
   TMonitor *mon = 0;
 
   if (fCurrentMonitor == fAllMonitor) {
      mon = new TMonitor;
   } else {
      mon = fAllMonitor;
      mon->DeActivateAll();
   }
   TIter next(slaves);
   TSlave *sl;
   while ((sl = (TSlave*) next())) {
      if (sl->IsValid())
         mon->Activate(sl->GetSocket());
   }
 
   rc = Collect(mon, timeout, endtype, deactonfail);
   ReleaseMonitor(mon);
   return rc;
}
 
////////////////////////////////////////////////////////////////////////////////
/// Collect responses from the slave servers. Returns the number of slaves
/// that responded.
/// If timeout >= 0, wait at most timeout seconds (timeout = -1 by default,
/// which means wait forever).
/// If defined (>= 0) endtype is the message that stops this collection.
 
Int_t TProof::Collect(ESlaves list, Long_t timeout, Int_t endtype, Bool_t deactonfail)
{
   Int_t rc = 0;
   TMonitor *mon = 0;
 
   if (list == kAll)       mon = fAllMonitor;
   if (list == kActive)    mon = fActiveMonitor;
   if (list == kUnique)    mon = fUniqueMonitor;
   if (list == kAllUnique) mon = fAllUniqueMonitor;
   if (fCurrentMonitor == mon) {
      // Get a copy
      mon = new TMonitor(*mon);
   }
   mon->ActivateAll();
 
   rc = Collect(mon, timeout, endtype, deactonfail);
   ReleaseMonitor(mon);
   return rc;
}
 
////////////////////////////////////////////////////////////////////////////////
/// Collect responses from the slave servers. Returns the number of messages
/// received. Can be 0 if there are no active slaves.
/// If timeout >= 0, wait at most timeout seconds (timeout = -1 by default,
/// which means wait forever).
/// If defined (>= 0) endtype is the message that stops this collection.
/// Collect also stops its execution from time to time to check for new
/// workers in Dynamic Startup mode.
 
Int_t TProof::Collect(TMonitor *mon, Long_t timeout, Int_t endtype, Bool_t deactonfail)
{
   Int_t collectId = gRandom->Integer(9999);
 
   PDB(kCollect, 3)
      Info("Collect", ">>>>>> Entering collect responses #%04d", collectId);
 
   // Reset the status flag and clear the messages in the list, if any
   fStatus = 0;
   fRecvMessages->Clear();
 
   Long_t actto = (Long_t)(gEnv->GetValue("Proof.SocketActivityTimeout", -1) * 1000);
 
   if (!mon->GetActive(actto)) return 0;
 
   DeActivateAsyncInput();
 
   // Used by external code to know what we are monitoring
   TMonitor *savedMonitor = 0;
   if (fCurrentMonitor) {
      savedMonitor = fCurrentMonitor;
      fCurrentMonitor = mon;
   } else {
      fCurrentMonitor = mon;
      fBytesRead = 0;
      fRealTime  = 0.0;
      fCpuTime   = 0.0;
   }
 
   // We want messages on the main window during synchronous collection,
   // but we save the present status to restore it at the end
   Bool_t saveRedirLog = fRedirLog;
   if (!IsIdle() && !IsSync())
      fRedirLog = kFALSE;
 
   int cnt = 0, rc = 0;
 
   // Timeout counter
   Long_t nto = timeout;
   PDB(kCollect, 2)
      Info("Collect","#%04d: active: %d", collectId, mon->GetActive());
 
   // On clients, handle Ctrl-C during collection
   if (fIntHandler)
      fIntHandler->Add();
 
   // Sockets w/o activity during the last 'sto' millisecs are deactivated
   Int_t nact = 0;
   Long_t sto = -1;
   Int_t nsto = 60;
   Int_t pollint = gEnv->GetValue("Proof.DynamicStartupPollInt", (Int_t) kPROOF_DynWrkPollInt_s);
   mon->ResetInterrupt();
   while ((nact = mon->GetActive(sto)) && (nto < 0 || nto > 0)) {
 
      // Dump last waiting sockets, if in debug mode
      PDB(kCollect, 2) {
         if (nact < 4) {
            TList *al = mon->GetListOfActives();
            if (al && al->GetSize() > 0) {
               Info("Collect"," %d node(s) still active:", al->GetSize());
               TIter nxs(al);
               TSocket *xs = 0;
               while ((xs = (TSocket *)nxs())) {
                  TSlave *wrk = FindSlave(xs);
                  if (wrk)
                     Info("Collect","   %s (%s)", wrk->GetName(), wrk->GetOrdinal());
                  else
                     Info("Collect","   %p: %s:%d", xs, xs->GetInetAddress().GetHostName(),
                                                        xs->GetInetAddress().GetPort());
               }
            }
            delete al;
         }
      }
 
      // Preemptive poll for new workers on the master only in Dynamic Mode and only
      // during processing (TODO: should work on Top Master only)
      if (TestBit(TProof::kIsMaster) && !IsIdle() && fDynamicStartup && !fIsPollingWorkers &&
         ((fLastPollWorkers_s == -1) || (time(0)-fLastPollWorkers_s >= pollint))) {
         fIsPollingWorkers = kTRUE;
         if (PollForNewWorkers() > 0) DeActivateAsyncInput();
         fLastPollWorkers_s = time(0);
         fIsPollingWorkers = kFALSE;
         PDB(kCollect, 1)
            Info("Collect","#%04d: now active: %d", collectId, mon->GetActive());
      }
 
      // Wait for a ready socket
      PDB(kCollect, 3)
         Info("Collect", "Will invoke Select() #%04d", collectId);
      TSocket *s = mon->Select(1000);
 
      if (s && s != (TSocket *)(-1)) {
         // Get and analyse the info it did receive
         rc = CollectInputFrom(s, endtype, deactonfail);
         if (rc  == 1 || (rc == 2 && !savedMonitor)) {
            // Deactivate it if we are done with it
            mon->DeActivate(s);
            TList *al = mon->GetListOfActives();
            PDB(kCollect, 2)
               Info("Collect","#%04d: deactivating %p (active: %d, %p)", collectId,
                              s, mon->GetActive(),
                              al->First());
            delete al;
         } else if (rc == 2) {
            // This end message was for the saved monitor
            // Deactivate it if we are done with it
            if (savedMonitor) {
               savedMonitor->DeActivate(s);
               TList *al = mon->GetListOfActives();
               PDB(kCollect, 2)
                  Info("Collect","save monitor: deactivating %p (active: %d, %p)",
                                 s, savedMonitor->GetActive(),
                                 al->First());
               delete al;
            }
         }
 
         // Update counter (if no error occured)
         if (rc >= 0)
            cnt++;
      } else {
         // If not timed-out, exit if not stopped or not aborted
         // (player exits status is finished in such a case); otherwise,
         // we still need to collect the partial output info
         if (!s)
            if (fPlayer && (fPlayer->GetExitStatus() == TVirtualProofPlayer::kFinished))
               mon->DeActivateAll();
         // Decrease the timeout counter if requested
         if (s == (TSocket *)(-1) && nto > 0)
            nto--;
      }
 
      // Check if there are workers with ready output to be sent and ask the first to send it
      if (IsMaster() && fWrksOutputReady && fWrksOutputReady->GetSize() > 0) {
         // Maximum number of concurrent sendings
         Int_t mxws = gEnv->GetValue("Proof.ControlSendOutput", 1);
         if (TProof::GetParameter(fPlayer->GetInputList(), "PROOF_ControlSendOutput", mxws) != 0)
            mxws = gEnv->GetValue("Proof.ControlSendOutput", 1);
         TIter nxwr(fWrksOutputReady);
         TSlave *wrk = 0;
         while (mxws && (wrk = (TSlave *) nxwr())) {
            if (!wrk->TestBit(TSlave::kOutputRequested)) {
               // Ask worker for output
               TMessage sendoutput(kPROOF_SENDOUTPUT);
               PDB(kCollect, 2)
                  Info("Collect", "worker %s was asked to send its output to master",
                                                wrk->GetOrdinal());
               if (wrk->GetSocket()->Send(sendoutput) != 1) {
                  wrk->SetBit(TSlave::kOutputRequested);
                  mxws--;
               }
            } else {
               // Count
               mxws--;
            }
         }
      }
 
      // Check if we need to check the socket activity (we do it every 10 cycles ~ 10 sec)
      sto = -1;
      if (--nsto <= 0) {
         sto = (Long_t) actto;
         nsto = 60;
      }
 
   } // end loop over active monitors
 
   // If timed-out, deactivate the remaining sockets
   if (nto == 0) {
      TList *al = mon->GetListOfActives();
      if (al && al->GetSize() > 0) {
         // Notify the name of those which did timeout
         Info("Collect"," %d node(s) went in timeout:", al->GetSize());
         TIter nxs(al);
         TSocket *xs = 0;
         while ((xs = (TSocket *)nxs())) {
            TSlave *wrk = FindSlave(xs);
            if (wrk)
               Info("Collect","   %s", wrk->GetName());
            else
               Info("Collect","   %p: %s:%d", xs, xs->GetInetAddress().GetHostName(),
                                                  xs->GetInetAddress().GetPort());
         }
      }
      delete al;
      mon->DeActivateAll();
   }
 
   // Deactivate Ctrl-C special handler
   if (fIntHandler)
      fIntHandler->Remove();
 
   // make sure group view is up to date
   SendGroupView();
 
   // Restore redirection setting
   fRedirLog = saveRedirLog;
 
   // Restore the monitor
   fCurrentMonitor = savedMonitor;
 
   ActivateAsyncInput();
 
   PDB(kCollect, 3)
      Info("Collect", "<<<<<< Exiting collect responses #%04d", collectId);
 
   return cnt;
}
 
////////////////////////////////////////////////////////////////////////////////
/// Asks the PROOF Serv for new workers in Dynamic Startup mode and activates
/// them. Returns the number of new workers found, or <0 on errors.
 
Int_t TProof::PollForNewWorkers()
{
   // Requests for worker updates
   Int_t dummy = 0;
   TList *reqWorkers = new TList();
   reqWorkers->SetOwner(kFALSE);
 
   if (!TestBit(TProof::kIsMaster)) {
      Error("PollForNewWorkers", "Can't invoke: not on a master -- should not happen!");
      return -1;
   }
   if (!gProofServ) {
      Error("PollForNewWorkers", "No ProofServ available -- should not happen!");
      return -1;
   }
 
   gProofServ->GetWorkers(reqWorkers, dummy, kTRUE);  // last 2 are dummy
 
   // List of new workers only (TProofNodeInfo)
   TList *newWorkers = new TList();
   newWorkers->SetOwner(kTRUE);
 
   TIter next(reqWorkers);
   TProofNodeInfo *ni;
   TString fullOrd;
   while (( ni = dynamic_cast<TProofNodeInfo *>(next()) )) {
 
      // Form the full ordinal
      fullOrd.Form("%s.%s", gProofServ->GetOrdinal(), ni->GetOrdinal().Data());
 
      TIter nextInner(fSlaves);
      TSlave *sl;
      Bool_t found = kFALSE;
      while (( sl = dynamic_cast<TSlave *>(nextInner()) )) {
         if ( strcmp(sl->GetOrdinal(), fullOrd.Data()) == 0 ) {
            found = kTRUE;
            break;
         }
      }
 
      if (found) delete ni;
      else {
         newWorkers->Add(ni);
         PDB(kGlobal, 1)
            Info("PollForNewWorkers", "New worker found: %s:%s",
               ni->GetNodeName().Data(), fullOrd.Data());
      }
   }
 
   delete reqWorkers;  // not owner
 
   Int_t nNewWorkers = newWorkers->GetEntries();
 
   // Add the new workers
   if (nNewWorkers > 0) {
      PDB(kGlobal, 1)
         Info("PollForNewWorkers", "Requesting to add %d new worker(s)", newWorkers->GetEntries());
      Int_t rv = AddWorkers(newWorkers);
      if (rv < 0) {
         Error("PollForNewWorkers", "Call to AddWorkers() failed (got %d < 0)", rv);
         return -1;
      }
      // Don't delete newWorkers: AddWorkers() will do that
   }
   else {
      PDB(kGlobal, 2)
         Info("PollForNewWorkers", "No new worker found");
      delete newWorkers;
   }
 
   return nNewWorkers;
}
 
////////////////////////////////////////////////////////////////////////////////
/// Remove links to objects in list 'ol' from gDirectory
 
void TProof::CleanGDirectory(TList *ol)
{
   if (ol) {
      TIter nxo(ol);
      TObject *o = 0;
      while ((o = nxo()))
         gDirectory->RecursiveRemove(o);
   }
}
 
////////////////////////////////////////////////////////////////////////////////
/// Collect and analyze available input from socket s.
/// Returns 0 on success, -1 if any failure occurs.
 
Int_t TProof::CollectInputFrom(TSocket *s, Int_t endtype, Bool_t deactonfail)
{
   TMessage *mess;
 
   Int_t recvrc = 0;
   if ((recvrc = s->Recv(mess)) < 0) {
      PDB(kCollect,2)
         Info("CollectInputFrom","%p: got %d from Recv()", s, recvrc);
      Bool_t bad = kTRUE;
      if (recvrc == -5) {
         // Broken connection: try reconnection
         if (fCurrentMonitor) fCurrentMonitor->Remove(s);
         if (s->Reconnect() == 0) {
            if (fCurrentMonitor) fCurrentMonitor->Add(s);
            bad = kFALSE;
         }
      }
      if (bad)
         MarkBad(s, "problems receiving a message in TProof::CollectInputFrom(...)");
      // Ignore this wake up
      return -1;
   }
   if (!mess) {
      // we get here in case the remote server died
      MarkBad(s, "undefined message in TProof::CollectInputFrom(...)");
      return -1;
   }
   Int_t rc = 0;
 
   Int_t what = mess->What();
   TSlave *sl = FindSlave(s);
   rc = HandleInputMessage(sl, mess, deactonfail);
   if (rc == 1 && (endtype >= 0) && (what != endtype))
      // This message was for the base monitor in recursive case
      rc = 2;
 
   // We are done successfully
   return rc;
}
 
////////////////////////////////////////////////////////////////////////////////
/// Analyze the received message.
/// Returns 0 on success (1 if this the last message from this socket), -1 if
/// any failure occurs.
 
Int_t TProof::HandleInputMessage(TSlave *sl, TMessage *mess, Bool_t deactonfail)
{
   char str[512];
   TObject *obj;
   Int_t rc = 0;
 
   if (!mess || !sl) {
      Warning("HandleInputMessage", "given an empty message or undefined worker");
      return -1;
   }
   Bool_t delete_mess = kTRUE;
   TSocket *s = sl->GetSocket();
   if (!s) {
      Warning("HandleInputMessage", "worker socket is undefined");
      return -1;
   }
 
   // The message type
   Int_t what = mess->What();
 
   PDB(kCollect,3)
      Info("HandleInputMessage", "got type %d from '%s'", what, sl->GetOrdinal());
 
   switch (what) {
 
      case kMESS_OK:
         // Add the message to the list
         fRecvMessages->Add(mess);
         delete_mess = kFALSE;
         break;
 
      case kMESS_OBJECT:
         if (fPlayer) fPlayer->HandleRecvHisto(mess);
         break;
 
      case kPROOF_FATAL:
         {  TString msg;
            if ((mess->BufferSize() > mess->Length()))
               (*mess) >> msg;
            if (msg.IsNull()) {
               MarkBad(s, "received kPROOF_FATAL");
            } else {
               MarkBad(s, msg);
            }
         }
         if (fProgressDialogStarted) {
            // Finalize the progress dialog
            Emit("StopProcess(Bool_t)", kTRUE);
         }
         break;
 
      case kPROOF_STOP:
         // Stop collection from this worker
         Info("HandleInputMessage", "received kPROOF_STOP from %s: disabling any further collection this worker",
              sl->GetOrdinal());
         rc = 1;
         break;
 
      case kPROOF_GETTREEHEADER:
         // Add the message to the list
         fRecvMessages->Add(mess);
         delete_mess = kFALSE;
         rc = 1;
         break;
 
      case kPROOF_TOUCH:
         // send a request for touching the remote admin file
         {
            sl->Touch();
         }
         break;
 
      case kPROOF_GETOBJECT:
         // send slave object it asks for
         mess->ReadString(str, sizeof(str));
         obj = gDirectory->Get(str);
         if (obj)
            s->SendObject(obj);
         else
            s->Send(kMESS_NOTOK);
         break;
 
      case kPROOF_GETPACKET:
         {
            PDB(kGlobal,2)
               Info("HandleInputMessage","%s: kPROOF_GETPACKET", sl->GetOrdinal());
            TDSetElement *elem = 0;
            elem = fPlayer ? fPlayer->GetNextPacket(sl, mess) : 0;
 
            if (elem != (TDSetElement*) -1) {
               TMessage answ(kPROOF_GETPACKET);
               answ << elem;
               s->Send(answ);
 
               while (fWaitingSlaves != 0 && fWaitingSlaves->GetSize()) {
                  TPair *p = (TPair*) fWaitingSlaves->First();
                  s = (TSocket*) p->Key();
                  TMessage *m = (TMessage*) p->Value();
 
                  elem = fPlayer ? fPlayer->GetNextPacket(sl, m) : 0;
                  if (elem != (TDSetElement*) -1) {
                     TMessage a(kPROOF_GETPACKET);
                     a << elem;
                     s->Send(a);
                     // remove has to happen via Links because TPair does not have
                     // a Compare() function and therefore RemoveFirst() and
                     // Remove(TObject*) do not work
                     fWaitingSlaves->Remove(fWaitingSlaves->FirstLink());
                     delete p;
                     delete m;
                  } else {
                     break;
                  }
               }
            } else {
               if (fWaitingSlaves == 0) fWaitingSlaves = new TList;
               fWaitingSlaves->Add(new TPair(s, mess));
               delete_mess = kFALSE;
            }
         }
         break;
 
      case kPROOF_LOGFILE:
         {
            Int_t size;
            (*mess) >> size;
            PDB(kGlobal,2)
               Info("HandleInputMessage","%s: kPROOF_LOGFILE: size: %d", sl->GetOrdinal(), size);
            RecvLogFile(s, size);
         }
         break;
 
      case kPROOF_LOGDONE:
         (*mess) >> sl->fStatus >> sl->fParallel;
         PDB(kCollect,2)
            Info("HandleInputMessage","%s: kPROOF_LOGDONE: status %d  parallel %d",
                 sl->GetOrdinal(), sl->fStatus, sl->fParallel);
         if (sl->fStatus != 0) {
            // Return last nonzero status
            fStatus = sl->fStatus;
            // Deactivate the worker, if required
            if (deactonfail) DeactivateWorker(sl->fOrdinal);
         }
         // Remove from the workers-ready list
         if (fWrksOutputReady && fWrksOutputReady->FindObject(sl)) {
            sl->ResetBit(TSlave::kOutputRequested);
            fWrksOutputReady->Remove(sl);
         }
         rc = 1;
         break;
 
      case kPROOF_GETSTATS:
         {
            (*mess) >> sl->fBytesRead >> sl->fRealTime >> sl->fCpuTime
                  >> sl->fWorkDir >> sl->fProofWorkDir;
            PDB(kCollect,2)
               Info("HandleInputMessage", "kPROOF_GETSTATS: %s", sl->fWorkDir.Data());
            TString img;
            if ((mess->BufferSize() > mess->Length()))
               (*mess) >> img;
            // Set image
            if (img.IsNull()) {
               if (sl->fImage.IsNull())
                  sl->fImage.Form("%s:%s", TUrl(sl->fName).GetHostFQDN(),
                                           sl->fProofWorkDir.Data());
            } else {
               sl->fImage = img;
            }
            PDB(kGlobal,2)
               Info("HandleInputMessage",
                        "kPROOF_GETSTATS:%s image: %s", sl->GetOrdinal(), sl->GetImage());
 
            fBytesRead += sl->fBytesRead;
            fRealTime  += sl->fRealTime;
            fCpuTime   += sl->fCpuTime;
            rc = 1;
         }
         break;
 
      case kPROOF_GETPARALLEL:
         {
            Bool_t async = kFALSE;
            (*mess) >> sl->fParallel;
            if ((mess->BufferSize() > mess->Length()))
               (*mess) >> async;
            rc = (async) ? 0 : 1;
         }
         break;
 
      case kPROOF_CHECKFILE:
         {  // New servers (>= 5.22) send the status
            if ((mess->BufferSize() > mess->Length())) {
               (*mess) >> fCheckFileStatus;
            } else {
               // Form old servers this meant success (failure was signaled with the
               // dangerous kPROOF_FATAL)
               fCheckFileStatus = 1;
            }
            rc = 1;
         }
         break;
 
      case kPROOF_SENDFILE:
         {  // New server: signals ending of sendfile operation
            rc = 1;
         }
         break;
 
      case kPROOF_PACKAGE_LIST:
         {
            PDB(kGlobal,2) Info("HandleInputMessage","kPROOF_PACKAGE_LIST: enter");
            Int_t type = 0;
            (*mess) >> type;
            switch (type) {
            case TProof::kListEnabledPackages:
               SafeDelete(fEnabledPackages);
               fEnabledPackages = (TList *) mess->ReadObject(TList::Class());
               if (fEnabledPackages) {
                  fEnabledPackages->SetOwner();
               } else {
                  Error("HandleInputMessage",
                        "kPROOF_PACKAGE_LIST: kListEnabledPackages: TList not found in message!");
               }
               break;
            case TProof::kListPackages:
               SafeDelete(fAvailablePackages);
               fAvailablePackages = (TList *) mess->ReadObject(TList::Class());
               if (fAvailablePackages) {
                  fAvailablePackages->SetOwner();
               } else {
                  Error("HandleInputMessage",
                        "kPROOF_PACKAGE_LIST: kListPackages: TList not found in message!");
               }
               break;
            default:
               Error("HandleInputMessage", "kPROOF_PACKAGE_LIST: unknown type: %d", type);
            }
         }
         break;
 
      case kPROOF_SENDOUTPUT:
         {
            // We start measuring the merging time
            fPlayer->SetMerging();
 
            // Worker is ready to send output: make sure the relevant bit is reset
            sl->ResetBit(TSlave::kOutputRequested);
            PDB(kGlobal,2)
               Info("HandleInputMessage","kPROOF_SENDOUTPUT: enter (%s)", sl->GetOrdinal());
            // Create the list if not yet done
            if (!fWrksOutputReady) {
               fWrksOutputReady = new TList;
               fWrksOutputReady->SetOwner(kFALSE);
            }
            fWrksOutputReady->Add(sl);
         }
         break;
 
      case kPROOF_OUTPUTOBJECT:
         {
            // We start measuring the merging time
            fPlayer->SetMerging();
 
            PDB(kGlobal,2)
               Info("HandleInputMessage","kPROOF_OUTPUTOBJECT: enter");
            Int_t type = 0;
            const char *prefix = gProofServ ? gProofServ->GetPrefix() : "Lite-0";
            if (!TestBit(TProof::kIsClient) && !fMergersSet && !fFinalizationRunning) {
               Info("HandleInputMessage", "finalization on %s started ...", prefix);
               fFinalizationRunning = kTRUE;
            }
 
            while ((mess->BufferSize() > mess->Length())) {
               (*mess) >> type;
               // If a query result header, add it to the player list
               if (fPlayer) {
                  if (type == 0) {
                     // Retrieve query result instance (output list not filled)
                     TQueryResult *pq =
                        (TQueryResult *) mess->ReadObject(TQueryResult::Class());
                     if (pq) {
                        // Add query to the result list in TProofPlayer
                        fPlayer->AddQueryResult(pq);
                        fPlayer->SetCurrentQuery(pq);
                        // And clear the output list, as we start merging a new set of results
                        if (fPlayer->GetOutputList())
                           fPlayer->GetOutputList()->Clear();
                        // Add the unique query tag as TNamed object to the input list
                        // so that it is available in TSelectors for monitoring
                        TString qid = TString::Format("%s:%s",pq->GetTitle(),pq->GetName());
                        if (fPlayer->GetInputList()->FindObject("PROOF_QueryTag"))
                           fPlayer->GetInputList()->Remove(fPlayer->GetInputList()->FindObject("PROOF_QueryTag"));
                        fPlayer->AddInput(new TNamed("PROOF_QueryTag", qid.Data()));
                     } else {
                        Warning("HandleInputMessage","kPROOF_OUTPUTOBJECT: query result missing");
                     }
                  } else if (type > 0) {
                     // Read object
                     TObject *o = mess->ReadObject(TObject::Class());
                     // Increment counter on the client side
                     fMergePrg.IncreaseIdx();
                     TString msg;
                     Bool_t changed = kFALSE;
                     msg.Form("%s: merging output objects ... %s", prefix, fMergePrg.Export(changed));
                     if (gProofServ) {
                        gProofServ->SendAsynMessage(msg.Data(), kFALSE);
                     } else if (IsTty() || changed) {
                        fprintf(stderr, "%s\r", msg.Data());
                     }
                     // Add or merge it
                     if ((fPlayer->AddOutputObject(o) == 1)) {
                        // Remove the object if it has been merged
                        SafeDelete(o);
                     }
                     if (type > 1) {
                        // Update the merger progress info
                        fMergePrg.DecreaseNWrks();
                        if (TestBit(TProof::kIsClient) && !IsLite()) {
                           // In PROOFLite this has to be done once only in TProofLite::Process
                           TQueryResult *pq = fPlayer->GetCurrentQuery();
                           if (pq) {
                              pq->SetOutputList(fPlayer->GetOutputList(), kFALSE);
                              // Add input objects (do not override remote settings, if any)
                              TObject *xo = 0;
                              TIter nxin(fPlayer->GetInputList());
                              // Servers prior to 5.28/00 do not create the input list in the TQueryResult
                              if (!pq->GetInputList()) pq->SetInputList(new TList());
                              while ((xo = nxin()))
                                 if (!pq->GetInputList()->FindObject(xo->GetName()))
                                    pq->AddInput(xo->Clone());
                              // If the last object, notify the GUI that the result arrived
                              QueryResultReady(TString::Format("%s:%s", pq->GetTitle(), pq->GetName()));
                           }
                           // Processing is over
                           UpdateDialog();
                        }
                     }
                  }
               } else {
                  Warning("HandleInputMessage", "kPROOF_OUTPUTOBJECT: player undefined!");
               }
            }
         }
         break;
 
      case kPROOF_OUTPUTLIST:
         {
            // We start measuring the merging time
 
            PDB(kGlobal,2)
               Info("HandleInputMessage","%s: kPROOF_OUTPUTLIST: enter", sl->GetOrdinal());
            TList *out = 0;
            if (fPlayer) {
               fPlayer->SetMerging();
               if (TestBit(TProof::kIsMaster) || fProtocol < 7) {
                  out = (TList *) mess->ReadObject(TList::Class());
               } else {
                  TQueryResult *pq =
                     (TQueryResult *) mess->ReadObject(TQueryResult::Class());
                  if (pq) {
                     // Add query to the result list in TProofPlayer
                     fPlayer->AddQueryResult(pq);
                     fPlayer->SetCurrentQuery(pq);
                     // To avoid accidental cleanups from anywhere else
                     // remove objects from gDirectory and clone the list
                     out = pq->GetOutputList();
                     CleanGDirectory(out);
                     out = (TList *) out->Clone();
                     // Notify the GUI that the result arrived
                     QueryResultReady(TString::Format("%s:%s", pq->GetTitle(), pq->GetName()));
                  } else {
                     PDB(kGlobal,2)
                        Info("HandleInputMessage",
                             "%s: kPROOF_OUTPUTLIST: query result missing", sl->GetOrdinal());
                  }
               }
               if (out) {
                  out->SetOwner();
                  fPlayer->AddOutput(out); // Incorporate the list
                  SafeDelete(out);
               } else {
                  PDB(kGlobal,2)
                     Info("HandleInputMessage",
                          "%s: kPROOF_OUTPUTLIST: outputlist is empty", sl->GetOrdinal());
               }
            } else {
               Warning("HandleInputMessage",
                       "%s: kPROOF_OUTPUTLIST: player undefined!", sl->GetOrdinal());
            }
            // On clients at this point processing is over
            if (TestBit(TProof::kIsClient) && !IsLite())
               UpdateDialog();
         }
         break;
 
      case kPROOF_QUERYLIST:
         {
            PDB(kGlobal,2) Info("HandleInputMessage","kPROOF_QUERYLIST: enter");
            (*mess) >> fOtherQueries >> fDrawQueries;
            if (fQueries) {
               fQueries->Delete();
               delete fQueries;
               fQueries = 0;
            }
            fQueries = (TList *) mess->ReadObject(TList::Class());
         }
         break;
 
      case kPROOF_RETRIEVE:
         {
            PDB(kGlobal,2) Info("HandleInputMessage","kPROOF_RETRIEVE: enter");
            TQueryResult *pq =
               (TQueryResult *) mess->ReadObject(TQueryResult::Class());
            if (pq && fPlayer) {
               fPlayer->AddQueryResult(pq);
               // Notify the GUI that the result arrived
               QueryResultReady(TString::Format("%s:%s", pq->GetTitle(), pq->GetName()));
            } else {
               PDB(kGlobal,2)
                  Info("HandleInputMessage",
                       "kPROOF_RETRIEVE: query result missing or player undefined");
            }
         }
         break;
 
      case kPROOF_MAXQUERIES:
         {
            PDB(kGlobal,2) Info("HandleInputMessage","kPROOF_MAXQUERIES: enter");
            Int_t max = 0;
 
            (*mess) >> max;
            Printf("Number of queries fully kept remotely: %d", max);
         }
         break;
 
      case kPROOF_SERVERSTARTED:
         {
            PDB(kGlobal,2) Info("HandleInputMessage","kPROOF_SERVERSTARTED: enter");
 
            UInt_t tot = 0, done = 0;
            TString action;
            Bool_t st = kTRUE;
 
            (*mess) >> action >> tot >> done >> st;
 
            if (TestBit(TProof::kIsClient)) {
               if (tot) {
                  TString type = (action.Contains("submas")) ? "submasters"
                                                             : "workers";
                  Int_t frac = (Int_t) (done*100.)/tot;
                  char msg[512] = {0};
                  if (frac >= 100) {
                     snprintf(msg, 512, "%s: OK (%d %s)                 \n",
                             action.Data(),tot, type.Data());
                  } else {
                     snprintf(msg, 512, "%s: %d out of %d (%d %%)\r",
                             action.Data(), done, tot, frac);
                  }
                  if (fSync)
                     fprintf(stderr,"%s", msg);
                  else
                     NotifyLogMsg(msg, 0);
               }
               // Notify GUIs
               StartupMessage(action.Data(), st, (Int_t)done, (Int_t)tot);
            } else {
 
               // Just send the message one level up
               TMessage m(kPROOF_SERVERSTARTED);
               m << action << tot << done << st;
               gProofServ->GetSocket()->Send(m);
            }
         }
         break;
 
      case kPROOF_DATASET_STATUS:
         {
            PDB(kGlobal,2) Info("HandleInputMessage","kPROOF_DATASET_STATUS: enter");
 
            UInt_t tot = 0, done = 0;
            TString action;
            Bool_t st = kTRUE;
 
            (*mess) >> action >> tot >> done >> st;
 
            if (TestBit(TProof::kIsClient)) {
               if (tot) {
                  TString type = "files";
                  Int_t frac = (Int_t) (done*100.)/tot;
                  char msg[512] = {0};
                  if (frac >= 100) {
                     snprintf(msg, 512, "%s: OK (%d %s)                 \n",
                             action.Data(),tot, type.Data());
                  } else {
                     snprintf(msg, 512, "%s: %d out of %d (%d %%)\r",
                             action.Data(), done, tot, frac);
                  }
                  if (fSync)
                     fprintf(stderr,"%s", msg);
                  else
                     NotifyLogMsg(msg, 0);
               }
               // Notify GUIs
               DataSetStatus(action.Data(), st, (Int_t)done, (Int_t)tot);
            } else {
 
               // Just send the message one level up
               TMessage m(kPROOF_DATASET_STATUS);
               m << action << tot << done << st;
               gProofServ->GetSocket()->Send(m);
            }
         }
         break;
 
      case kPROOF_STARTPROCESS:
         {
            PDB(kGlobal,2) Info("HandleInputMessage","kPROOF_STARTPROCESS: enter");
 
            // For Proof-Lite this variable is the number of workers and is set
            // by the player
            if (!IsLite()) {
               fNotIdle = 1;
               fIsWaiting = kFALSE;
            }
 
            // Redirect the output, if needed
            fRedirLog = (fSync) ? fRedirLog : kTRUE;
 
            // The signal is used on masters by XrdProofdProtocol to catch
            // the start of processing; on clients it allows to update the
            // progress dialog
            if (!TestBit(TProof::kIsMaster)) {
 
               // This is the end of preparation
               fQuerySTW.Stop();
               fPrepTime = fQuerySTW.RealTime();
               PDB(kGlobal,2) Info("HandleInputMessage","Preparation time: %f s", fPrepTime);
 
               TString selec;
               Int_t dsz = -1;
               Long64_t first = -1, nent = -1;
               (*mess) >> selec >> dsz >> first >> nent;
               // Start or reset the progress dialog
               if (!gROOT->IsBatch()) {
                  if (fProgressDialog &&
                      !TestBit(kUsingSessionGui) && TestBit(kUseProgressDialog)) {
                     if (!fProgressDialogStarted) {
                        fProgressDialog->ExecPlugin(5, this,
                                                   selec.Data(), dsz, first, nent);
                        fProgressDialogStarted = kTRUE;
                     } else {
                        ResetProgressDialog(selec, dsz, first, nent);
                     }
                  }
                  ResetBit(kUsingSessionGui);
               }
            }
         }
         break;
 
      case kPROOF_ENDINIT:
         {
            PDB(kGlobal,2) Info("HandleInputMessage","kPROOF_ENDINIT: enter");
 
            if (TestBit(TProof::kIsMaster)) {
               if (fPlayer)
                  fPlayer->SetInitTime();
            }
         }
         break;
 
      case kPROOF_SETIDLE:
         {
            PDB(kGlobal,2)
               Info("HandleInputMessage","kPROOF_SETIDLE from '%s': enter (%d)", sl->GetOrdinal(), fNotIdle);
 
            // The session is idle
            if (IsLite()) {
               if (fNotIdle > 0) {
                  fNotIdle--;
                  PDB(kGlobal,2)
                     Info("HandleInputMessage", "%s: got kPROOF_SETIDLE", sl->GetOrdinal());
               } else {
                  Warning("HandleInputMessage",
                          "%s: got kPROOF_SETIDLE but no running workers ! protocol error?",
                          sl->GetOrdinal());
               }
            } else {
               fNotIdle = 0;
               // Check if the query has been enqueued
               if ((mess->BufferSize() > mess->Length()))
                  (*mess) >> fIsWaiting;
            }
         }
         break;
 
      case kPROOF_QUERYSUBMITTED:
         {
            PDB(kGlobal,2) Info("HandleInputMessage","kPROOF_QUERYSUBMITTED: enter");
 
            // We have received the sequential number
            (*mess) >> fSeqNum;
            Bool_t sync = fSync;
            if ((mess->BufferSize() > mess->Length()))
               (*mess) >> sync;
            if (sync !=  fSync && fSync) {
               // The server required to switch to asynchronous mode
               Activate();
               fSync = kFALSE;
            }
            DisableGoAsyn();
            // Check if the query has been enqueued
            fIsWaiting = kTRUE;
            // For Proof-Lite this variable is the number of workers and is set by the player
            if (!IsLite())
               fNotIdle = 1;
 
            rc = 1;
         }
         break;
 
      case kPROOF_SESSIONTAG:
         {
            PDB(kGlobal,2) Info("HandleInputMessage","kPROOF_SESSIONTAG: enter");
 
            // We have received the unique tag and save it as name of this object
            TString stag;
            (*mess) >> stag;
            SetName(stag);
            // In the TSlave object
            sl->SetSessionTag(stag);
            // Server may have also sent the group
            if ((mess->BufferSize() > mess->Length()))
               (*mess) >> fGroup;
            // Server may have also sent the user
            if ((mess->BufferSize() > mess->Length())) {
               TString usr;
               (*mess) >> usr;
               if (!usr.IsNull()) fUrl.SetUser(usr.Data());
            }
         }
         break;
 
      case kPROOF_FEEDBACK:
         {
            PDB(kGlobal,2)
               Info("HandleInputMessage","kPROOF_FEEDBACK: enter");
            TList *out = (TList *) mess->ReadObject(TList::Class());
            out->SetOwner();
            if (fPlayer)
               fPlayer->StoreFeedback(sl, out); // Adopts the list
            else
               // Not yet ready: stop collect asap
               rc = 1;
         }
         break;
 
      case kPROOF_AUTOBIN:
         {
            PDB(kGlobal,2) Info("HandleInputMessage","kPROOF_AUTOBIN: enter");
 
            TString name;
            Double_t xmin, xmax, ymin, ymax, zmin, zmax;
 
            (*mess) >> name >> xmin >> xmax >> ymin >> ymax >> zmin >> zmax;
 
            if (fPlayer) fPlayer->UpdateAutoBin(name,xmin,xmax,ymin,ymax,zmin,zmax);
 
            TMessage answ(kPROOF_AUTOBIN);
 
            answ << name << xmin << xmax << ymin << ymax << zmin << zmax;
 
            s->Send(answ);
         }
         break;
 
      case kPROOF_PROGRESS:
         {
            PDB(kGlobal,2) Info("HandleInputMessage","kPROOF_PROGRESS: enter");
 
            if (GetRemoteProtocol() > 25) {
               // New format
               TProofProgressInfo *pi = 0;
               (*mess) >> pi;
               fPlayer->Progress(sl,pi);
            } else if (GetRemoteProtocol() > 11) {
               Long64_t total, processed, bytesread;
               Float_t initTime, procTime, evtrti, mbrti;
               (*mess) >> total >> processed >> bytesread
                     >> initTime >> procTime
                     >> evtrti >> mbrti;
               if (fPlayer)
                  fPlayer->Progress(sl, total, processed, bytesread,
                                    initTime, procTime, evtrti, mbrti);
 
            } else {
               // Old format
               Long64_t total, processed;
               (*mess) >> total >> processed;
               if (fPlayer)
                  fPlayer->Progress(sl, total, processed);
            }
         }
         break;
 
      case kPROOF_STOPPROCESS:
         {
            // This message is sent from a worker that finished processing.
            // We determine whether it was asked to finish by the
            // packetizer or stopped during processing a packet
            // (by TProof::RemoveWorkers() or by an external signal).
            // In the later case call packetizer->MarkBad.
            PDB(kGlobal,2) Info("HandleInputMessage","kPROOF_STOPPROCESS: enter");
 
            Long64_t events = 0;
            Bool_t abort = kFALSE;
            TProofProgressStatus *status = 0;
 
            if ((mess->BufferSize() > mess->Length()) && (fProtocol > 18)) {
               (*mess) >> status >> abort;
            } else if ((mess->BufferSize() > mess->Length()) && (fProtocol > 8)) {
               (*mess) >> events >> abort;
            } else {
               (*mess) >> events;
            }
            if (fPlayer) {
               if (fProtocol > 18) {
                  TList *listOfMissingFiles = 0;
                  if (!(listOfMissingFiles = (TList *)GetOutput("MissingFiles"))) {
                     listOfMissingFiles = new TList();
                     listOfMissingFiles->SetName("MissingFiles");
                     if (fPlayer)
                        fPlayer->AddOutputObject(listOfMissingFiles);
                  }
                  if (fPlayer->GetPacketizer()) {
                     Int_t ret =
                        fPlayer->GetPacketizer()->AddProcessed(sl, status, 0, &listOfMissingFiles);
                     if (ret > 0)
                        fPlayer->GetPacketizer()->MarkBad(sl, status, &listOfMissingFiles);
                     // This object is now owned by the packetizer
                     status = 0;
                  }
                  if (status) fPlayer->AddEventsProcessed(status->GetEntries());
               } else {
                  fPlayer->AddEventsProcessed(events);
               }
            }
            SafeDelete(status);
            if (!TestBit(TProof::kIsMaster))
               Emit("StopProcess(Bool_t)", abort);
            break;
         }
 
      case kPROOF_SUBMERGER:
         {
            PDB(kGlobal,2) Info("HandleInputMessage", "kPROOF_SUBMERGER: enter");
            HandleSubmerger(mess, sl);
         }
         break;
 
      case kPROOF_GETSLAVEINFO:
         {
            PDB(kGlobal,2) Info("HandleInputMessage", "kPROOF_GETSLAVEINFO: enter");
 
            Bool_t active = (GetListOfActiveSlaves()->FindObject(sl) != 0);
            Bool_t bad = (GetListOfBadSlaves()->FindObject(sl) != 0);
            TList* tmpinfo = 0;
            (*mess) >> tmpinfo;
            if (tmpinfo == 0) {
               Error("HandleInputMessage", "kPROOF_GETSLAVEINFO: no list received!");
            } else {
               tmpinfo->SetOwner(kFALSE);
               Int_t nentries = tmpinfo->GetSize();
               for (Int_t i=0; i<nentries; i++) {
                  TSlaveInfo* slinfo =
                     dynamic_cast<TSlaveInfo*>(tmpinfo->At(i));
                  if (slinfo) {
                     // If PROOF-Lite
                     if (IsLite()) slinfo->fHostName = gSystem->HostName();
                     // Check if we have already a instance for this worker
                     TIter nxw(fSlaveInfo);
                     TSlaveInfo *ourwi = 0;
                     while ((ourwi = (TSlaveInfo *)nxw())) {
                        if (!strcmp(ourwi->GetOrdinal(), slinfo->GetOrdinal())) {
                           ourwi->SetSysInfo(slinfo->GetSysInfo());
                           ourwi->fHostName = slinfo->GetName();
                           if (slinfo->GetDataDir() && (strlen(slinfo->GetDataDir()) > 0))
                              ourwi->fDataDir = slinfo->GetDataDir();
                           break;
                        }
                     }
                     if (!ourwi) {
                        fSlaveInfo->Add(slinfo);
                     } else {
                        slinfo = ourwi;
                     }
                     if (slinfo->fStatus != TSlaveInfo::kBad) {
                        if (!active) slinfo->SetStatus(TSlaveInfo::kNotActive);
                        if (bad) slinfo->SetStatus(TSlaveInfo::kBad);
                     }
                     if (sl->GetMsd() && (strlen(sl->GetMsd()) > 0))
                        slinfo->fMsd = sl->GetMsd();
                  }
               }
               delete tmpinfo;
               rc = 1;
            }
         }
         break;
 
      case kPROOF_VALIDATE_DSET:
         {
            PDB(kGlobal,2)
               Info("HandleInputMessage", "kPROOF_VALIDATE_DSET: enter");
            TDSet* dset = 0;
            (*mess) >> dset;
            if (!fDSet)
               Error("HandleInputMessage", "kPROOF_VALIDATE_DSET: fDSet not set");
            else
               fDSet->Validate(dset);
            delete dset;
         }
         break;
 
      case kPROOF_DATA_READY:
         {
            PDB(kGlobal,2) Info("HandleInputMessage", "kPROOF_DATA_READY: enter");
            Bool_t dataready = kFALSE;
            Long64_t totalbytes, bytesready;
            (*mess) >> dataready >> totalbytes >> bytesready;
            fTotalBytes += totalbytes;
            fBytesReady += bytesready;
            if (dataready == kFALSE) fDataReady = dataready;
         }
         break;
 
      case kPROOF_PING:
         // do nothing (ping is already acknowledged)
         break;
 
      case kPROOF_MESSAGE:
         {
            PDB(kGlobal,2) Info("HandleInputMessage","kPROOF_MESSAGE: enter");
 
            // We have received the unique tag and save it as name of this object
            TString msg;
            (*mess) >> msg;
            Bool_t lfeed = kTRUE;
            if ((mess->BufferSize() > mess->Length()))
               (*mess) >> lfeed;
 
            if (TestBit(TProof::kIsClient)) {
 
               if (fSync) {
                  // Notify locally
                  fprintf(stderr,"%s%c", msg.Data(), (lfeed ? '\n' : '\r'));
               } else {
                  // Notify locally taking care of redirection, windows logs, ...
                  NotifyLogMsg(msg, (lfeed ? "\n" : "\r"));
               }
            } else {
 
               // The message is logged for debugging purposes.
               fprintf(stderr,"%s%c", msg.Data(), (lfeed ? '\n' : '\r'));
               if (gProofServ) {
                  // We hide it during normal operations
                  gProofServ->FlushLogFile();
 
                  // And send the message one level up
                  gProofServ->SendAsynMessage(msg, lfeed);
               }
            }
         }
         break;
 
      case kPROOF_VERSARCHCOMP:
         {
            TString vac;
            (*mess) >> vac;
            PDB(kGlobal,2) Info("HandleInputMessage","kPROOF_VERSARCHCOMP: %s", vac.Data());
            Int_t from = 0;
            TString vers, archcomp;
            if (vac.Tokenize(vers, from, "|"))
               vac.Tokenize(archcomp, from, "|");
            sl->SetArchCompiler(archcomp);
            vers.ReplaceAll(":","|");
            sl->SetROOTVersion(vers);
         }
         break;
 
      default:
         {
            Error("HandleInputMessage", "unknown command received from '%s' (what = %d)",
                                        sl->GetOrdinal(), what);
         }
         break;
   }
 
   // Cleanup
   if (delete_mess)
      delete mess;
 
   // We are done successfully
   return rc;
}
 
////////////////////////////////////////////////////////////////////////////////
/// Process a message of type kPROOF_SUBMERGER
 
void TProof::HandleSubmerger(TMessage *mess, TSlave *sl)
{
   // Message sub-type
   Int_t type = 0;
   (*mess) >> type;
   TSocket *s = sl->GetSocket();
 
   switch (type) {
      case kOutputSent:
         {
            if (IsEndMaster()) {
               Int_t  merger_id = -1;
               (*mess) >> merger_id;
 
               PDB(kSubmerger, 2)
                  Info("HandleSubmerger", "kOutputSent: Worker %s:%d:%s had sent its output to merger #%d",
                                          sl->GetName(), sl->GetPort(), sl->GetOrdinal(), merger_id);
 
               if (!fMergers || fMergers->GetSize() <= merger_id) {
                  Error("HandleSubmerger", "kOutputSize: #%d not in list ", merger_id);
                  break;
               }
               TMergerInfo * mi = (TMergerInfo *) fMergers->At(merger_id);
               mi->SetMergedWorker();
               if (mi->AreAllWorkersMerged()) {
                  mi->Deactivate();
                  if (GetActiveMergersCount() == 0) {
                     fMergers->Clear();
                     delete fMergers;
                     fMergersSet = kFALSE;
                     fMergersCount = -1;
                     fLastAssignedMerger = 0;
                     PDB(kSubmerger, 2) Info("HandleSubmerger", "all mergers removed ... ");
                  }
               }
            } else {
               PDB(kSubmerger, 2) Error("HandleSubmerger","kOutputSent: received not on endmaster!");
            }
         }
         break;
 
      case kMergerDown:
         {
            Int_t  merger_id = -1;
            (*mess) >> merger_id;
 
            PDB(kSubmerger, 2) Info("HandleSubmerger", "kMergerDown: #%d ", merger_id);
 
            if (!fMergers || fMergers->GetSize() <= merger_id) {
               Error("HandleSubmerger", "kMergerDown: #%d not in list ", merger_id);
               break;
            }
 
            TMergerInfo * mi = (TMergerInfo *) fMergers->At(merger_id);
            if (!mi->IsActive()) {
               break;
            } else {
               mi->Deactivate();
            }
 
            // Stop the invalid merger in the case it is still listening
            TMessage stop(kPROOF_SUBMERGER);
            stop << Int_t(kStopMerging);
            stop <<  0;
            s->Send(stop);
 
            // Ask for results from merger (only original results from this node as worker are returned)
            AskForOutput(mi->GetMerger());
 
            // Ask for results from all workers assigned to this merger
            TIter nxo(mi->GetWorkers());
            TObject * o = 0;
            while ((o = nxo())) {
               AskForOutput((TSlave *)o);
            }
            PDB(kSubmerger, 2) Info("HandleSubmerger", "kMergerDown:%d: exit", merger_id);
         }
         break;
 
      case kOutputSize:
         {
            if (IsEndMaster()) {
               PDB(kSubmerger, 2)
                  Info("HandleSubmerger", "worker %s reported as finished ", sl->GetOrdinal());
 
               const char *prefix = gProofServ ? gProofServ->GetPrefix() : "Lite-0";
               if (!fFinalizationRunning) {
                  Info("HandleSubmerger", "finalization on %s started ...", prefix);
                  fFinalizationRunning = kTRUE;
               }
 
               Int_t  output_size = 0;
               Int_t  merging_port = 0;
               (*mess) >> output_size >> merging_port;
 
               PDB(kSubmerger, 2) Info("HandleSubmerger",
                                       "kOutputSize: Worker %s:%d:%s reports %d output objects (+ available port %d)",
                                       sl->GetName(), sl->GetPort(), sl->GetOrdinal(), output_size, merging_port);
               TString msg;
               if (!fMergersSet) {
 
                  Int_t activeWorkers = fCurrentMonitor ? fCurrentMonitor->GetActive() : GetNumberOfActiveSlaves();
 
                  // First pass - setting number of mergers according to user or dynamically
                  fMergersCount = -1; // No mergers used if not set by user
                  TParameter<Int_t> *mc = dynamic_cast<TParameter<Int_t> *>(GetParameter("PROOF_UseMergers"));
                  if (mc) fMergersCount = mc->GetVal(); // Value set by user
                  TParameter<Int_t> *mh = dynamic_cast<TParameter<Int_t> *>(GetParameter("PROOF_MergersByHost"));
                  if (mh) fMergersByHost = (mh->GetVal() != 0) ? kTRUE : kFALSE; // Assign submergers by hostname
 
                  // Mergers count specified by user but not valid
                  if (fMergersCount < 0 || (fMergersCount > (activeWorkers/2) )) {
                     msg.Form("%s: Invalid request: cannot start %d mergers for %d workers",
                              prefix, fMergersCount, activeWorkers);
                     if (gProofServ)
                        gProofServ->SendAsynMessage(msg);
                     else
                        Printf("%s",msg.Data());
                     fMergersCount = 0;
                  }
                  // Mergers count will be set dynamically
                  if ((fMergersCount == 0) && (!fMergersByHost)) {
                     if (activeWorkers > 1) {
                        fMergersCount = TMath::Nint(TMath::Sqrt(activeWorkers));
                        if (activeWorkers / fMergersCount < 2)
                           fMergersCount = (Int_t) TMath::Sqrt(activeWorkers);
                     }
                     if (fMergersCount > 1)
                        msg.Form("%s: Number of mergers set dynamically to %d (for %d workers)",
                                 prefix, fMergersCount, activeWorkers);
                     else {
                        msg.Form("%s: No mergers will be used for %d workers",
                                 prefix, activeWorkers);
                        fMergersCount = -1;
                     }
                     if (gProofServ)
                        gProofServ->SendAsynMessage(msg);
                     else
                        Printf("%s",msg.Data());
                  } else if (fMergersByHost) {
                     // We force mergers at host level to minimize network traffic
                     if (activeWorkers > 1) {
                        fMergersCount = 0;
                        THashList hosts;
                        TIter nxwk(fSlaves);
                        TObject *wrk = 0;
                        while ((wrk = nxwk())) {
                           if (!hosts.FindObject(wrk->GetName())) {
                              hosts.Add(new TObjString(wrk->GetName()));
                              fMergersCount++;
                           }
                        }
                     }
                     if (fMergersCount > 1)
                        msg.Form("%s: Number of mergers set to %d (for %d workers), one for each slave host",
                                 prefix, fMergersCount, activeWorkers);
                     else {
                        msg.Form("%s: No mergers will be used for %d workers",
                                 prefix, activeWorkers);
                        fMergersCount = -1;
                     }
                     if (gProofServ)
                        gProofServ->SendAsynMessage(msg);
                     else
                        Printf("%s",msg.Data());
                  } else {
                     msg.Form("%s: Number of mergers set by user to %d (for %d workers)",
                              prefix, fMergersCount, activeWorkers);
                     if (gProofServ)
                        gProofServ->SendAsynMessage(msg);
                     else
                        Printf("%s",msg.Data());
                  }
 
                  // We started merging; we call it here because fMergersCount is still the original number
                  // and can be saved internally
                  fPlayer->SetMerging(kTRUE);
 
                  // Update merger counters (new workers are not yet active)
                  fMergePrg.SetNWrks(fMergersCount);
 
                  if (fMergersCount > 0) {
 
                     fMergers = new TList();
                     fLastAssignedMerger = 0;
                     // Total number of workers, which will not act as mergers ('pure workers')
                     fWorkersToMerge = (activeWorkers - fMergersCount);
                     // Establish the first merger
                     if (!CreateMerger(sl, merging_port)) {
                        // Cannot establish first merger
                        AskForOutput(sl);
                        fWorkersToMerge--;
                        fMergersCount--;
                     }
                     if (IsLite()) fMergePrg.SetNWrks(fMergersCount);
                  } else {
                     AskForOutput(sl);
                  }
                  fMergersSet = kTRUE;
               } else {
                  // Multiple pass
                  if (fMergersCount == -1) {
                     // No mergers. Workers send their outputs directly to master
                     AskForOutput(sl);
                  } else {
                     if ((fRedirectNext > 0 ) && (!fMergersByHost)) {
                        RedirectWorker(s, sl, output_size);
                        fRedirectNext--;
                     } else {
                        Bool_t newMerger = kTRUE;
                        if (fMergersByHost) {
                           TIter nxmg(fMergers);
                           TMergerInfo *mgi = 0;
                           while ((mgi = (TMergerInfo *) nxmg())) {
                              if (!strcmp(sl->GetName(), mgi->GetMerger()->GetName())) {
                                 newMerger = kFALSE;
                                 break;
                              }
                           }
                        }
                        if ((fMergersCount > fMergers->GetSize()) && newMerger) {
                           // Still not enough mergers established
                           if (!CreateMerger(sl, merging_port)) {
                              // Cannot establish a merger
                              AskForOutput(sl);
                              fWorkersToMerge--;
                              fMergersCount--;
                           }
                        } else
                           RedirectWorker(s, sl, output_size);
                     }
                  }
               }
            } else {
               Error("HandleSubMerger","kOutputSize received not on endmaster!");
            }
         }
      break;
   }
}
 
////////////////////////////////////////////////////////////////////////////////
/// Redirect output of worker sl to some merger
 
void TProof::RedirectWorker(TSocket *s, TSlave * sl, Int_t output_size)
{
   Int_t merger_id = -1;
 
   if (fMergersByHost) {
      for (Int_t i = 0; i < fMergers->GetSize(); i++) {
         TMergerInfo *mgi = (TMergerInfo *)fMergers->At(i);
         if (!strcmp(sl->GetName(), mgi->GetMerger()->GetName())) {
            merger_id = i;
            break;
         }
      }
   } else {
      merger_id  = FindNextFreeMerger();
   }
 
   if (merger_id == -1) {
      // No free merger (probably it had crashed before)
      AskForOutput(sl);
   } else {
      TMessage sendoutput(kPROOF_SUBMERGER);
      sendoutput << Int_t(kSendOutput);
      PDB(kSubmerger, 2)
         Info("RedirectWorker", "redirecting worker %s to merger %d", sl->GetOrdinal(), merger_id);
 
       PDB(kSubmerger, 2) Info("RedirectWorker", "redirecting output to merger #%d",  merger_id);
       if (!fMergers || fMergers->GetSize() <= merger_id) {
          Error("RedirectWorker", "#%d not in list ", merger_id);
          return;
       }
       TMergerInfo * mi = (TMergerInfo *) fMergers->At(merger_id);
 
       TString hname = (IsLite()) ? "localhost" : mi->GetMerger()->GetName();
       sendoutput <<  merger_id;
       sendoutput << hname;
       sendoutput << mi->GetPort();
       s->Send(sendoutput);
       mi->AddMergedObjects(output_size);
       mi->AddWorker(sl);
   }
}
 
////////////////////////////////////////////////////////////////////////////////
/// Return a merger, which is both active and still accepts some workers to be
/// assigned to it. It works on the 'round-robin' basis.
 
Int_t TProof::FindNextFreeMerger()
{
   while (fLastAssignedMerger < fMergers->GetSize() &&
         (!((TMergerInfo*)fMergers->At(fLastAssignedMerger))->IsActive() ||
           ((TMergerInfo*)fMergers->At(fLastAssignedMerger))->AreAllWorkersAssigned())) {
      fLastAssignedMerger++;
   }
 
   if (fLastAssignedMerger == fMergers->GetSize()) {
      fLastAssignedMerger = 0;
   } else {
      return fLastAssignedMerger++;
   }
 
   while (fLastAssignedMerger < fMergers->GetSize() &&
         (!((TMergerInfo*)fMergers->At(fLastAssignedMerger))->IsActive() ||
           ((TMergerInfo*)fMergers->At(fLastAssignedMerger))->AreAllWorkersAssigned())) {
      fLastAssignedMerger++;
   }
 
   if (fLastAssignedMerger == fMergers->GetSize()) {
      return -1;
   } else {
      return fLastAssignedMerger++;
   }
}
 
////////////////////////////////////////////////////////////////////////////////
/// Master asks for output from worker sl
 
void TProof::AskForOutput(TSlave *sl)
{
   TMessage sendoutput(kPROOF_SUBMERGER);
   sendoutput << Int_t(kSendOutput);
 
   PDB(kSubmerger, 2) Info("AskForOutput",
                           "worker %s was asked to send its output to master",
                            sl->GetOrdinal());
 
   sendoutput << -1;
   sendoutput << TString("master");
   sendoutput << -1;
   sl->GetSocket()->Send(sendoutput);
   if (IsLite()) fMergePrg.IncreaseNWrks();
}
 
////////////////////////////////////////////////////////////////////////////////
/// Final update of the progress dialog
 
void TProof::UpdateDialog()
{
   if (!fPlayer) return;
 
   // Handle abort ...
   if (fPlayer->GetExitStatus() == TVirtualProofPlayer::kAborted) {
      if (fSync)
         Info("UpdateDialog",
              "processing was aborted - %lld events processed",
              fPlayer->GetEventsProcessed());
 
      if (GetRemoteProtocol() > 11) {
         // New format
         Progress(-1, fPlayer->GetEventsProcessed(), -1, -1., -1., -1., -1.);
      } else {
         Progress(-1, fPlayer->GetEventsProcessed());
      }
      Emit("StopProcess(Bool_t)", kTRUE);
   }
 
   // Handle stop ...
   if (fPlayer->GetExitStatus() == TVirtualProofPlayer::kStopped) {
      if (fSync)
         Info("UpdateDialog",
              "processing was stopped - %lld events processed",
              fPlayer->GetEventsProcessed());
 
      if (GetRemoteProtocol() > 25) {
         // New format
         Progress(-1, fPlayer->GetEventsProcessed(), -1, -1., -1., -1., -1., -1, -1, -1.);
      } else if (GetRemoteProtocol() > 11) {
         Progress(-1, fPlayer->GetEventsProcessed(), -1, -1., -1., -1., -1.);
      } else {
         Progress(-1, fPlayer->GetEventsProcessed());
      }
      Emit("StopProcess(Bool_t)", kFALSE);
   }
 
   // Final update of the dialog box
   if (GetRemoteProtocol() > 25) {
      // New format
      EmitVA("Progress(Long64_t,Long64_t,Long64_t,Float_t,Float_t,Float_t,Float_t,Int_t,Int_t,Float_t)",
              10, (Long64_t)(-1), (Long64_t)(-1), (Long64_t)(-1),(Float_t)(-1.),(Float_t)(-1.),
                  (Float_t)(-1.),(Float_t)(-1.),(Int_t)(-1),(Int_t)(-1),(Float_t)(-1.));
   } else if (GetRemoteProtocol() > 11) {
      // New format
      EmitVA("Progress(Long64_t,Long64_t,Long64_t,Float_t,Float_t,Float_t,Float_t)",
               7, (Long64_t)(-1), (Long64_t)(-1), (Long64_t)(-1),
                  (Float_t)(-1.),(Float_t)(-1.),(Float_t)(-1.),(Float_t)(-1.));
   } else {
      EmitVA("Progress(Long64_t,Long64_t)", 2, (Long64_t)(-1), (Long64_t)(-1));
   }
}
 
////////////////////////////////////////////////////////////////////////////////
/// Activate the a-sync input handler.
 
void TProof::ActivateAsyncInput()
{
   TIter next(fSlaves);
   TSlave *sl;
 
   while ((sl = (TSlave*) next()))
      if (sl->GetInputHandler())
         sl->GetInputHandler()->Add();
}
 
////////////////////////////////////////////////////////////////////////////////
/// De-activate a-sync input handler.
 
void TProof::DeActivateAsyncInput()
{
   TIter next(fSlaves);
   TSlave *sl;
 
   while ((sl = (TSlave*) next()))
      if (sl->GetInputHandler())
         sl->GetInputHandler()->Remove();
}
 
////////////////////////////////////////////////////////////////////////////////
/// Get the active mergers count
 
Int_t TProof::GetActiveMergersCount()
{
   if (!fMergers) return 0;
 
   Int_t active_mergers = 0;
 
   TIter mergers(fMergers);
   TMergerInfo *mi = 0;
   while ((mi = (TMergerInfo *)mergers())) {
      if (mi->IsActive()) active_mergers++;
   }
 
   return active_mergers;
}
 
////////////////////////////////////////////////////////////////////////////////
/// Create a new merger
 
Bool_t TProof::CreateMerger(TSlave *sl, Int_t port)
{
   PDB(kSubmerger, 2)
      Info("CreateMerger", "worker %s will be merger ", sl->GetOrdinal());
 
   PDB(kSubmerger, 2) Info("CreateMerger","Begin");
 
   if (port <= 0) {
      PDB(kSubmerger,2)
         Info("CreateMerger", "cannot create merger on port %d - exit", port);
      return kFALSE;
   }
 
   Int_t workers = -1;
   if (!fMergersByHost) {
      Int_t mergersToCreate = fMergersCount - fMergers->GetSize();
      // Number of pure workers, which are not simply divisible by mergers
      Int_t rest = fWorkersToMerge % mergersToCreate;
      // We add one more worker for each of the first 'rest' mergers being established
      if (rest > 0 && fMergers->GetSize() < rest) {
         rest = 1;
      } else {
         rest = 0;
      }
      workers = (fWorkersToMerge / mergersToCreate) + rest;
   } else {
      Int_t workersOnHost = 0;
      for (Int_t i = 0; i < fActiveSlaves->GetSize(); i++) {
         if(!strcmp(sl->GetName(), fActiveSlaves->At(i)->GetName())) workersOnHost++;
      }
      workers = workersOnHost - 1;
   }
 
   TString msg;
   msg.Form("worker %s on host %s will be merger for %d additional workers", sl->GetOrdinal(), sl->GetName(), workers);
 
   if (gProofServ) {
      gProofServ->SendAsynMessage(msg);
   } else {
      Printf("%s",msg.Data());
   }
   TMergerInfo * merger = new TMergerInfo(sl, port, workers);
 
   TMessage bemerger(kPROOF_SUBMERGER);
   bemerger << Int_t(kBeMerger);
   bemerger <<  fMergers->GetSize();
   bemerger <<  workers;
   sl->GetSocket()->Send(bemerger);
 
   PDB(kSubmerger,2) Info("CreateMerger",
                          "merger #%d (port: %d) for %d workers started",
                          fMergers->GetSize(), port, workers);
 
   fMergers->Add(merger);
   fWorkersToMerge = fWorkersToMerge - workers;
 
   fRedirectNext = workers / 2;
 
   PDB(kSubmerger, 2) Info("CreateMerger", "exit");
   return kTRUE;
}
 
////////////////////////////////////////////////////////////////////////////////
/// Add a bad slave server to the bad slave list and remove it from
/// the active list and from the two monitor objects. Assume that the work
/// done by this worker was lost and ask packerizer to reassign it.
 
void TProof::MarkBad(TSlave *wrk, const char *reason)
{
   std::lock_guard<std::recursive_mutex> lock(fCloseMutex);
 
   // We may have been invalidated in the meanwhile: nothing to do in such a case
   if (!IsValid()) return;
 
   if (!wrk) {
      Error("MarkBad", "worker instance undefined: protocol error? ");
      return;
   }
 
   // Local URL
   static TString thisurl;
   if (thisurl.IsNull()) {
      if (IsMaster()) {
         Int_t port = gEnv->GetValue("ProofServ.XpdPort",-1);
         thisurl = TUrl(gSystem->HostName()).GetHostFQDN();
         if (port > 0) thisurl += TString::Format(":%d", port);
      } else {
         thisurl.Form("%s@%s:%d", fUrl.GetUser(), fUrl.GetHost(), fUrl.GetPort());
      }
   }
 
   if (!reason || (strcmp(reason, kPROOF_TerminateWorker) && strcmp(reason, kPROOF_WorkerIdleTO))) {
      // Message for notification
      const char *mastertype = (gProofServ && gProofServ->IsTopMaster()) ? "top master" : "master";
      TString src = IsMaster() ? Form("%s at %s", mastertype, thisurl.Data()) : "local session";
      TString msg;
      msg.Form("\n +++ Message from %s : marking %s:%d (%s) as bad\n +++ Reason: %s",
               src.Data(), wrk->GetName(), wrk->GetPort(), wrk->GetOrdinal(),
               (reason && strlen(reason)) ? reason : "unknown");
      Info("MarkBad", "%s", msg.Data());
      // Notify one level up, if the case
      // Add some hint for diagnostics
      if (gProofServ) {
         msg += TString::Format("\n\n +++ Most likely your code crashed on worker %s at %s:%d.\n",
                     wrk->GetOrdinal(), wrk->GetName(), wrk->GetPort());
      } else {
         msg += TString::Format("\n\n +++ Most likely your code crashed\n");
      }
      msg += TString::Format(" +++ Please check the session logs for error messages either using\n");
      msg += TString::Format(" +++ the 'Show logs' button or executing\n");
      msg += TString::Format(" +++\n");
      if (gProofServ) {
         msg += TString::Format(" +++ root [] TProof::Mgr(\"%s\")->GetSessionLogs()->"
                                "Display(\"%s\",0)\n\n", thisurl.Data(), wrk->GetOrdinal());
         gProofServ->SendAsynMessage(msg, kTRUE);
      } else {
         msg += TString::Format(" +++ root [] TProof::Mgr(\"%s\")->GetSessionLogs()->"
                                "Display(\"*\")\n\n", thisurl.Data());
         Printf("%s", msg.Data());
      }
   } else if (reason) {
      if (gDebug > 0 && strcmp(reason, kPROOF_WorkerIdleTO)) {
         Info("MarkBad", "worker %s at %s:%d asked to terminate",
                         wrk->GetOrdinal(), wrk->GetName(), wrk->GetPort());
      }
   }
 
   if (IsMaster() && reason) {
      if (strcmp(reason, kPROOF_TerminateWorker)) {
         // if the reason was not a planned termination
         TList *listOfMissingFiles = 0;
         if (!(listOfMissingFiles = (TList *)GetOutput("MissingFiles"))) {
            listOfMissingFiles = new TList();
            listOfMissingFiles->SetName("MissingFiles");
            if (fPlayer)
               fPlayer->AddOutputObject(listOfMissingFiles);
         }
         // If a query is being processed, assume that the work done by
         // the worker was lost and needs to be reassigned.
         TVirtualPacketizer *packetizer = fPlayer ? fPlayer->GetPacketizer() : 0;
         if (packetizer) {
            // the worker was lost so do resubmit the packets
            packetizer->MarkBad(wrk, 0, &listOfMissingFiles);
         }
      } else {
         // Tell the coordinator that we are gone
         if (gProofServ) {
            TString ord(wrk->GetOrdinal());
            Int_t id = ord.Last('.');
            if (id != kNPOS) ord.Remove(0, id+1);
            gProofServ->ReleaseWorker(ord.Data());
         }
      }
   } else if (TestBit(TProof::kIsClient) && reason && !strcmp(reason, kPROOF_WorkerIdleTO)) {
      // We are invalid after this
      fValid = kFALSE;
   }
 
   fActiveSlaves->Remove(wrk);
   FindUniqueSlaves();
 
   fAllMonitor->Remove(wrk->GetSocket());
   fActiveMonitor->Remove(wrk->GetSocket());
 
   fSendGroupView = kTRUE;
 
   if (IsMaster()) {
      if (reason && !strcmp(reason, kPROOF_TerminateWorker)) {
         // if the reason was a planned termination then delete the worker and
         // remove it from all the lists
         fSlaves->Remove(wrk);
         fBadSlaves->Remove(wrk);
         fActiveSlaves->Remove(wrk);
         fInactiveSlaves->Remove(wrk);
         fUniqueSlaves->Remove(wrk);
         fAllUniqueSlaves->Remove(wrk);
         fNonUniqueMasters->Remove(wrk);
 
         // we add it to the list of terminated slave infos instead, so that it
         // stays available in the .workers persistent file
         TSlaveInfo *si = new TSlaveInfo(
            wrk->GetOrdinal(),
            Form("%s@%s:%d", wrk->GetUser(), wrk->GetName(), wrk->GetPort()),
            0, "", wrk->GetWorkDir());
         if (!fTerminatedSlaveInfos->Contains(si)) fTerminatedSlaveInfos->Add(si);
         else delete si;
 
         delete wrk;
      } else {
         fBadSlaves->Add(wrk);
         fActiveSlaves->Remove(wrk);
         fUniqueSlaves->Remove(wrk);
         fAllUniqueSlaves->Remove(wrk);
         fNonUniqueMasters->Remove(wrk);
         if (fCurrentMonitor) fCurrentMonitor->DeActivate(wrk->GetSocket());
         wrk->Close();
         // Update the mergers count, if needed
         if (fMergersSet) {
            Int_t mergersCount = -1;
            TParameter<Int_t> *mc = dynamic_cast<TParameter<Int_t> *>(GetParameter("PROOF_UseMergers"));
            if (mc) mergersCount = mc->GetVal(); // Value set by user
            // Mergers count is set dynamically: recalculate it
            if (mergersCount == 0) {
               Int_t activeWorkers = fCurrentMonitor ? fCurrentMonitor->GetActive() : GetNumberOfActiveSlaves();
               if (activeWorkers > 1) {
                  fMergersCount = TMath::Nint(TMath::Sqrt(activeWorkers));
                  if (activeWorkers / fMergersCount < 2)
                     fMergersCount = (Int_t) TMath::Sqrt(activeWorkers);
               }
            }
         }
      }
 
      // Update session workers files
      SaveWorkerInfo();
   } else {
      // On clients the proof session should be removed from the lists
      // and deleted, since it is not valid anymore
      fSlaves->Remove(wrk);
      if (fManager)
         fManager->DiscardSession(this);
   }
}
 
////////////////////////////////////////////////////////////////////////////////
/// Add slave with socket s to the bad slave list and remove if from
/// the active list and from the two monitor objects.
 
void TProof::MarkBad(TSocket *s, const char *reason)
{
   std::lock_guard<std::recursive_mutex> lock(fCloseMutex);
 
   // We may have been invalidated in the meanwhile: nothing to do in such a case
   if (!IsValid()) return;
 
   TSlave *wrk = FindSlave(s);
   MarkBad(wrk, reason);
}
 
////////////////////////////////////////////////////////////////////////////////
/// Ask an active worker 'wrk' to terminate, i.e. to shutdown
 
void TProof::TerminateWorker(TSlave *wrk)
{
   if (!wrk) {
      Warning("TerminateWorker", "worker instance undefined: protocol error? ");
      return;
   }
 
   // Send stop message
   if (wrk->GetSocket() && wrk->GetSocket()->IsValid()) {
      TMessage mess(kPROOF_STOP);
      wrk->GetSocket()->Send(mess);
   } else {
      if (gDebug > 0)
         Info("TerminateWorker", "connection to worker is already down: cannot"
                                 " send termination message");
   }
 
   // This is a bad worker from now on
   MarkBad(wrk, kPROOF_TerminateWorker);
}
 
////////////////////////////////////////////////////////////////////////////////
/// Ask an active worker 'ord' to terminate, i.e. to shutdown
 
void TProof::TerminateWorker(const char *ord)
{
   if (ord && strlen(ord) > 0) {
      Bool_t all = (ord[0] == '*') ? kTRUE : kFALSE;
      if (IsMaster()) {
         TIter nxw(fSlaves);
         TSlave *wrk = 0;
         while ((wrk = (TSlave *)nxw())) {
            if (all || !strcmp(wrk->GetOrdinal(), ord)) {
               TerminateWorker(wrk);
               if (!all) break;
            }
         }
      } else {
         TMessage mess(kPROOF_STOP);
         mess << TString(ord);
         Broadcast(mess);
      }
   }
}
 
////////////////////////////////////////////////////////////////////////////////
/// Ping PROOF. Returns 1 if master server responded.
 
Int_t TProof::Ping()
{
   return Ping(kActive);
}
 
////////////////////////////////////////////////////////////////////////////////
/// Ping PROOF slaves. Returns the number of slaves that responded.
 
Int_t TProof::Ping(ESlaves list)
{
   TList *slaves = 0;
   if (list == kAll)       slaves = fSlaves;
   if (list == kActive)    slaves = fActiveSlaves;
   if (list == kUnique)    slaves = fUniqueSlaves;
   if (list == kAllUnique) slaves = fAllUniqueSlaves;
 
   if (slaves->GetSize() == 0) return 0;
 
   int   nsent = 0;
   TIter next(slaves);
 
   TSlave *sl;
   while ((sl = (TSlave *)next())) {
      if (sl->IsValid()) {
         if (sl->Ping() == -1) {
            MarkBad(sl, "ping unsuccessful");
         } else {
            nsent++;
         }
      }
   }
 
   return nsent;
}