TPacketizerAdaptive.cxx

Go to the documentation of this file.

// @(#)root/proofplayer:$Id$
// Author: Jan Iwaszkiewicz   11/12/06

/*************************************************************************
 * Copyright (C) 1995-2002, Rene Brun and Fons Rademakers.               *
 * All rights reserved.                                                  *
 *                                                                       *
 * For the licensing terms see $ROOTSYS/LICENSE.                         *
 * For the list of contributors see $ROOTSYS/README/CREDITS.             *
*************************************************************************/

//////////////////////////////////////////////////////////////////////////
//                                                                      //
// TPacketizerAdaptive                                                  //
//                                                                      //
// This packetizer is based on TPacketizer but uses different           //
// load-balancing algorithms and data structures.                       //
// Two main improvements in the load-balancing strategy:                //
// - First one was to change the order in which the files are assigned  //
//   to the computing nodes in such a way that network transfers are    //
//   evenly distributed in the query time. Transfer of the remote files //
//   was often becoming a bottleneck at the end of a query.             //
// - The other improvement is the use of time-based packet size. We     //
//   measure the processing rate of all the nodes and calculate the     //
//   packet size, so that it takes certain amount of time. In this way  //
//   packetizer prevents the situation where the query can't finish     //
//   because of one slow node.                                          //
//                                                                      //
// The data structures: TFileStat, TFileNode and TSlaveStat are         //
// enriched + changed and TFileNode::Compare method is changed.         //
//                                                                      //
//////////////////////////////////////////////////////////////////////////


#include "TPacketizerAdaptive.h"

#include "Riostream.h"
#include "TDSet.h"
#include "TError.h"
#include "TEnv.h"
#include "TEntryList.h"
#include "TEventList.h"
#include "TMap.h"
#include "TMessage.h"
#include "TMonitor.h"
#include "TNtupleD.h"
#include "TObject.h"
#include "TParameter.h"
#include "TPerfStats.h"
#include "TProofDebug.h"
#include "TProof.h"
#include "TProofServ.h"
#include "TSlave.h"
#include "TSocket.h"
#include "TSortedList.h"
#include "TUrl.h"
#include "TClass.h"
#include "TRandom.h"
#include "TMath.h"
#include "TObjString.h"
#include "TList.h"

//
// The following three utility classes manage the state of the
// work to be performed and the slaves involved in the process.
// A list of TFileNode(s) describes the hosts with files, each
// has a list of TFileStat(s) keeping the state for each TDSet
// element (file).
//
// The list of TSlaveStat(s) keep track of the work (being) done
// by each slave
//


//------------------------------------------------------------------------------

class TPacketizerAdaptive::TFileStat : public TObject {

private:
   Bool_t         fIsDone;       // is this element processed
   TFileNode     *fNode;         // my FileNode
   TDSetElement  *fElement;      // location of the file and its range
   Long64_t       fNextEntry;    // cursor in the range, -1 when done // needs changing

public:
   TFileStat(TFileNode *node, TDSetElement *elem, TList *file);

   Bool_t         IsDone() const {return fIsDone;}
   Bool_t         IsSortable() const { return kTRUE; }
   void           SetDone() {fIsDone = kTRUE;}
   TFileNode     *GetNode() const {return fNode;}
   TDSetElement  *GetElement() const {return fElement;}
   Long64_t       GetNextEntry() const {return fNextEntry;}
   void           MoveNextEntry(Long64_t step) {fNextEntry += step;}

   // This method is used to keep a sorted list of remaining files to be processed
   Int_t          Compare(const TObject* obj) const
   {
      // Return -1 if elem.entries < obj.elem.entries, 0 if elem.entries equal
      // and 1 if elem.entries < obj.elem.entries.
      const TFileStat *fst = dynamic_cast<const TFileStat*>(obj);
      if (fst && GetElement() && fst->GetElement()) {
         Long64_t ent = GetElement()->GetNum();
         Long64_t entfst = fst->GetElement()->GetNum();
         if (ent > 0 && entfst > 0) {
            if (ent > entfst) {
               return 1;
            } else if (ent < entfst) {
               return -1;
            } else {
               return 0;
            }
         }
      }
      // No info: assume equal (no change in order)
      return 0;
   }
   void Print(Option_t * = 0) const
   {  // Notify file name and entries
      Printf("TFileStat: %s %lld", fElement ? fElement->GetName() : "---",
                                   fElement ? fElement->GetNum() : -1);
   }
};

TPacketizerAdaptive::TFileStat::TFileStat(TFileNode *node, TDSetElement *elem, TList *files)
   : fIsDone(kFALSE), fNode(node), fElement(elem), fNextEntry(elem->GetFirst())
{
   // Constructor: add to the global list
   if (files) files->Add(this);
}

//------------------------------------------------------------------------------

// a class describing a file node as a part of a session
class TPacketizerAdaptive::TFileNode : public TObject {

private:
   TString        fNodeName;        // FQDN of the node
   TList         *fFiles;           // TDSetElements (files) stored on this node
   TObject       *fUnAllocFileNext; // cursor in fFiles
   TList         *fActFiles;        // files with work remaining
   TObject       *fActFileNext;     // cursor in fActFiles
   Int_t          fMySlaveCnt;      // number of slaves running on this node
                                    // (which can process remote files)
   Int_t          fExtSlaveCnt;     // number of external slaves processing
                                    // files on this node
   Int_t          fRunSlaveCnt;     // total number of slaves processing files
                                    // on this node
   Long64_t       fProcessed;       // number of events processed on this node
   Long64_t       fEvents;          // number of entries in files on this node

   Int_t          fStrategy;        // 0 means the classic and 1 (default) - the adaptive strategy

   TSortedList   *fFilesToProcess;  // Global list of files (TFileStat) to be processed (owned by TPacketizer)

public:
   TFileNode(const char *name, Int_t strategy, TSortedList *files);
   ~TFileNode() { delete fFiles; delete fActFiles; }

   void        IncMySlaveCnt() { fMySlaveCnt++; }
   Int_t       GetMySlaveCnt() const { return fMySlaveCnt; }
   void        IncExtSlaveCnt(const char *slave) { if (fNodeName != slave) fExtSlaveCnt++; }
   void        DecExtSlaveCnt(const char *slave) { if (fNodeName != slave) fExtSlaveCnt--; R__ASSERT(fExtSlaveCnt >= 0); }
   Int_t       GetSlaveCnt() const { return fMySlaveCnt + fExtSlaveCnt; }
   void        IncRunSlaveCnt() { fRunSlaveCnt++; }
   void        DecRunSlaveCnt() { fRunSlaveCnt--; R__ASSERT(fRunSlaveCnt >= 0); }
   Int_t       GetRunSlaveCnt() const { return fRunSlaveCnt; }
   Int_t       GetExtSlaveCnt() const { return fExtSlaveCnt; }
   Int_t       GetNumberOfActiveFiles() const { return fActFiles->GetSize(); }
   Bool_t      IsSortable() const { return kTRUE; }
   Int_t       GetNumberOfFiles() { return fFiles->GetSize(); }
   void        IncProcessed(Long64_t nEvents)
                  { fProcessed += nEvents; }
   Long64_t    GetProcessed() const { return fProcessed; }
   void        DecreaseProcessed(Long64_t nEvents) { fProcessed -= nEvents; }
   // this method is used by Compare() it adds 1, so it returns a number that
   // would be true if one more slave is added.
   Long64_t    GetEventsLeftPerSlave() const
      { return ((fEvents - fProcessed)/(fRunSlaveCnt + 1)); }
   void        IncEvents(Long64_t nEvents) { fEvents += nEvents; }
   const char *GetName() const { return fNodeName.Data(); }
   Long64_t    GetNEvents() const { return fEvents; }

   void Print(Option_t * = 0) const
   {
      TFileStat *fs = 0;
      TDSetElement *e = 0;
      Int_t nn = 0;
      Printf("++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++");
      Printf("+++ TFileNode: %s +++", fNodeName.Data());
      Printf("+++ Evts: %lld (total: %lld) ", fProcessed, fEvents);
      Printf("+++ Worker count: int:%d, ext: %d, tot:%d ", fMySlaveCnt, fExtSlaveCnt, fRunSlaveCnt);
      Printf("+++ Files: %d ", fFiles ? fFiles->GetSize() : 0);
      if (fFiles && fFiles->GetSize() > 0) {
         TIter nxf(fFiles);
         while ((fs = (TFileStat *) nxf())) {
            if ((e = fs->GetElement())) {
               Printf("+++  #%d: %s  %lld - %lld (%lld) - next: %lld ", ++nn, e->GetName(),
                     e->GetFirst(), e->GetFirst() + e->GetNum() - 1, e->GetNum(), fs->GetNextEntry());
            } else {
               Printf("+++  #%d: no element! ", ++nn);
            }
         }
      }
      Printf("+++ Active files: %d ", fActFiles ? fActFiles->GetSize() : 0);
      if (fActFiles && fActFiles->GetSize() > 0) {
         TIter nxaf(fActFiles);
         while ((fs = (TFileStat *) nxaf())) {
            if ((e = fs->GetElement())) {
               Printf("+++  #%d: %s  %lld - %lld (%lld) - next: %lld", ++nn, e->GetName(),
                      e->GetFirst(), e->GetFirst() + e->GetNum() - 1, e->GetNum(), fs->GetNextEntry());
            } else {
               Printf("+++  #%d: no element! ", ++nn);
            }
         }
      }
      Printf("++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++");
   }

   void Add(TDSetElement *elem, Bool_t tolist)
   {
      TList *files = tolist ? (TList *)fFilesToProcess : (TList *)0;
      TFileStat *f = new TFileStat(this, elem, files);
      fFiles->Add(f);
      if (fUnAllocFileNext == 0) fUnAllocFileNext = fFiles->First();
   }

   TFileStat *GetNextUnAlloc()
   {
      TObject *next = fUnAllocFileNext;

      if (next != 0) {
         // make file active
         fActFiles->Add(next);
         if (fActFileNext == 0) fActFileNext = fActFiles->First();

         // move cursor
         fUnAllocFileNext = fFiles->After(fUnAllocFileNext);
      }
      return (TFileStat *) next;
   }

   TFileStat *GetNextActive()
   {
      TObject *next = fActFileNext;

      if (fActFileNext != 0) {
         fActFileNext = fActFiles->After(fActFileNext);
         if (fActFileNext == 0) fActFileNext = fActFiles->First();
      }

      return (TFileStat *) next;
   }

   void RemoveActive(TFileStat *file)
   {
      if (fActFileNext == file) fActFileNext = fActFiles->After(file);
      fActFiles->Remove(file);
      if (fFilesToProcess) fFilesToProcess->Remove(file);
      if (fActFileNext == 0) fActFileNext = fActFiles->First();
   }

   Int_t Compare(const TObject *other) const
   {
      // Must return -1 if this is smaller than obj, 0 if objects are equal
      // and 1 if this is larger than obj.
      // smaller means more needing a new worker.
      // Two cases are considered depending on
      // relation between harddrive speed and network bandwidth.

      const TFileNode *obj = dynamic_cast<const TFileNode*>(other);
      if (!obj) {
         Error("Compare", "input is not a TPacketizer::TFileNode object");
         return 0;
      }

      // how many more events it has than obj

      if (fStrategy == 1) {
         // The default adaptive strategy.
         Int_t myVal = GetRunSlaveCnt();
         Int_t otherVal = obj->GetRunSlaveCnt();
         if (myVal < otherVal) {
            return -1;
         } else if (myVal > otherVal) {
            return 1;
         } else {
            // if this has more events to process than obj
            if ((fEvents - fProcessed) >
                (obj->GetNEvents() - obj->GetProcessed())) {
               return -1;
            } else {
               return 1;
            }
         }
      } else {
         Int_t myVal = GetSlaveCnt();
         Int_t otherVal = obj->GetSlaveCnt();
         if (myVal < otherVal) {
            return -1;
         } else if (myVal > otherVal) {
            return 1;
         } else {
            return 0;
         }
      }
   }

   void Reset()
   {
      fUnAllocFileNext = fFiles->First();
      fActFiles->Clear();
      fActFileNext = 0;
      fExtSlaveCnt = 0;
      fMySlaveCnt = 0;
      fRunSlaveCnt = 0;
   }
};


TPacketizerAdaptive::TFileNode::TFileNode(const char *name, Int_t strategy, TSortedList *files)
   : fNodeName(name), fFiles(new TList), fUnAllocFileNext(0),
     fActFiles(new TList), fActFileNext(0), fMySlaveCnt(0),
     fExtSlaveCnt(0), fRunSlaveCnt(0), fProcessed(0), fEvents(0),
     fStrategy(strategy), fFilesToProcess(files)
{
   // Constructor

   fFiles->SetOwner();
   fActFiles->SetOwner(kFALSE);
}

//------------------------------------------------------------------------------

class TPacketizerAdaptive::TSlaveStat : public TVirtualPacketizer::TVirtualSlaveStat {

friend class TPacketizerAdaptive;

private:
   TFileNode     *fFileNode;     // corresponding node or 0
   TFileStat     *fCurFile;      // file currently being processed
   TDSetElement  *fCurElem;      // TDSetElement currently being processed
   Long64_t       fCurProcessed; // events processed in the current file
   Float_t        fCurProcTime;  // proc time spent on the current file
   TList         *fDSubSet;      // packets processed by this worker

public:
   TSlaveStat(TSlave *slave);
   ~TSlaveStat();
   TFileNode  *GetFileNode() const { return fFileNode; }
   Long64_t    GetEntriesProcessed() const { return fStatus?fStatus->GetEntries():-1; }
   Double_t    GetProcTime() const { return fStatus?fStatus->GetProcTime():-1; }
   TFileStat  *GetCurFile() { return fCurFile; }
   void        SetFileNode(TFileNode *node) { fFileNode = node; }
   void        UpdateRates(TProofProgressStatus *st);
   Float_t     GetAvgRate() { return fStatus->GetRate(); }
   Float_t     GetCurRate() {
      return (fCurProcTime?fCurProcessed/fCurProcTime:0); }
   Int_t       GetLocalEventsLeft() {
      return fFileNode?(fFileNode->GetEventsLeftPerSlave()):0; }
   TList      *GetProcessedSubSet() { return fDSubSet; }
   TProofProgressStatus *GetProgressStatus() { return fStatus; }
   TProofProgressStatus *AddProcessed(TProofProgressStatus *st = 0);
};

////////////////////////////////////////////////////////////////////////////////
/// Constructor

TPacketizerAdaptive::TSlaveStat::TSlaveStat(TSlave *slave)
   : fFileNode(0), fCurFile(0), fCurElem(0),
     fCurProcessed(0), fCurProcTime(0)
{
   fDSubSet = new TList();
   fDSubSet->SetOwner();
   fSlave = slave;
   fStatus = new TProofProgressStatus();
   // The slave name is a special one in PROOF-Lite: avoid blocking on the DNS
   // for non existing names
   fWrkFQDN = slave->GetName();
   if (strcmp(slave->ClassName(), "TSlaveLite")) {
      fWrkFQDN = TUrl(fWrkFQDN).GetHostFQDN();
      // Get full name for local hosts
      if (fWrkFQDN.Contains("localhost") || fWrkFQDN == "127.0.0.1")
         fWrkFQDN = TUrl(gSystem->HostName()).GetHostFQDN();
   }
   PDB(kPacketizer, 2)
      Info("TSlaveStat", "wrk FQDN: %s", fWrkFQDN.Data());
}

////////////////////////////////////////////////////////////////////////////////
/// Cleanup

TPacketizerAdaptive::TSlaveStat::~TSlaveStat()
{
   SafeDelete(fDSubSet);
   SafeDelete(fStatus);
}

////////////////////////////////////////////////////////////////////////////////
/// Update packetizer rates

void TPacketizerAdaptive::TSlaveStat::UpdateRates(TProofProgressStatus *st)
{
   if (!st) {
      Error("UpdateRates", "no status object!");
      return;
   }
   if (fCurFile->IsDone()) {
      fCurProcTime = 0;
      fCurProcessed = 0;
   } else {
      fCurProcTime += st->GetProcTime() - GetProcTime();
      fCurProcessed += st->GetEntries() - GetEntriesProcessed();
   }
   fCurFile->GetNode()->IncProcessed(st->GetEntries() - GetEntriesProcessed());
   st->SetLastEntries(st->GetEntries() - fStatus->GetEntries());
   SafeDelete(fStatus);
   fStatus = st;
}

////////////////////////////////////////////////////////////////////////////////
/// Add the current element to the fDSubSet (subset processed by this worker)
/// and if the status arg is given, then change the size of the packet.
/// return the difference (*st - *fStatus)

TProofProgressStatus *TPacketizerAdaptive::TSlaveStat::AddProcessed(TProofProgressStatus *st)
{
   if (st && fDSubSet && fCurElem) {
      if (fCurElem->GetNum() != st->GetEntries() - GetEntriesProcessed())
         fCurElem->SetNum(st->GetEntries() - GetEntriesProcessed());
      fDSubSet->Add(fCurElem);
      TProofProgressStatus *diff = new TProofProgressStatus(*st - *fStatus);
      return diff;
   } else {
      Error("AddProcessed", "processed subset of current elem undefined");
      return 0;
   }
}

//------------------------------------------------------------------------------

ClassImp(TPacketizerAdaptive)

////////////////////////////////////////////////////////////////////////////////
/// Constructor

TPacketizerAdaptive::TPacketizerAdaptive(TDSet *dset, TList *slaves,
                                         Long64_t first, Long64_t num,
                                         TList *input, TProofProgressStatus *st)
                    : TVirtualPacketizer(input, st)
{
   PDB(kPacketizer,1) Info("TPacketizerAdaptive",
                           "enter (first %lld, num %lld)", first, num);

   // Init pointer members
   fSlaveStats = 0;
   fUnAllocated = 0;
   fActive = 0;
   fFileNodes = 0;
   fMaxPerfIdx = 1;
   fCachePacketSync = kTRUE;
   fMaxEntriesRatio = 2.;

   fMaxSlaveCnt = -1;
   fPacketAsAFraction = 4;
   fStrategy = 1;
   fFilesToProcess = new TSortedList;
   fFilesToProcess->SetOwner(kFALSE);

   if (!fProgressStatus) {
      Error("TPacketizerAdaptive", "No progress status");
      return;
   }

   // Attempt to synchronize the packet size with the tree cache size
   Int_t cpsync = -1;
   if (TProof::GetParameter(input, "PROOF_PacketizerCachePacketSync", cpsync) != 0) {
      // Check if there is a global cache-packet sync setting
      cpsync = gEnv->GetValue("Packetizer.CachePacketSync", 1);
   }
   if (cpsync >= 0) fCachePacketSync = (cpsync > 0) ? kTRUE : kFALSE;

   // Max file entries to avg allowed ratio for cache-to-packet synchronization
   // (applies only if fCachePacketSync is true; -1. disables the bound)
   if (TProof::GetParameter(input, "PROOF_PacketizerMaxEntriesRatio", fMaxEntriesRatio) != 0) {
      // Check if there is a global ratio setting
      fMaxEntriesRatio = gEnv->GetValue("Packetizer.MaxEntriesRatio", 2.);
   }

   // The possibility to change packetizer strategy to the basic TPacketizer's
   // one (in which workers always process their local data first).
   Int_t strategy = -1;
   if (TProof::GetParameter(input, "PROOF_PacketizerStrategy", strategy) != 0) {
      // Check if there is a global strategy setting
      strategy = gEnv->GetValue("Packetizer.Strategy", 1);
   }
   if (strategy == 0) {
      fStrategy = 0;
      Info("TPacketizerAdaptive", "using the basic strategy of TPacketizer");
   } else if (strategy != 1) {
      Warning("TPacketizerAdaptive", "unsupported strategy index (%d): ignore", strategy);
   }

   Long_t maxSlaveCnt = 0;
   if (TProof::GetParameter(input, "PROOF_MaxSlavesPerNode", maxSlaveCnt) == 0) {
      if (maxSlaveCnt < 0) {
         Info("TPacketizerAdaptive",
              "The value of PROOF_MaxSlavesPerNode must be positive");
         maxSlaveCnt = 0;
      }
   } else {
      // Try also with Int_t (recently supported in TProof::SetParameter)
      Int_t mxslcnt = -1;
      if (TProof::GetParameter(input, "PROOF_MaxSlavesPerNode", mxslcnt) == 0) {
         if (mxslcnt < 0) {
            Info("TPacketizerAdaptive",
                 "The value of PROOF_MaxSlavesPerNode must be positive");
            mxslcnt = 0;
         }
         maxSlaveCnt = (Long_t) mxslcnt;
      }
   }

   if (!maxSlaveCnt)
      maxSlaveCnt = gEnv->GetValue("Packetizer.MaxWorkersPerNode", 0);
   if (maxSlaveCnt > 0) {
      fMaxSlaveCnt = maxSlaveCnt;
      Info("TPacketizerAdaptive", "Setting max number of workers per node to %ld",
           fMaxSlaveCnt);
   }

   // if forceLocal parameter is set to 1 then eliminate the cross-worker
   // processing;
   // This minimizes the network usage on the PROOF cluser at the expense of
   // longer jobs processing times.
   // To process successfully the session must have workers with all the data!
   fForceLocal = kFALSE;
   Int_t forceLocal = 0;
   if (TProof::GetParameter(input, "PROOF_ForceLocal", forceLocal) == 0) {
      if (forceLocal == 1)
         fForceLocal = kTRUE;
      else
         Info("TPacketizerAdaptive",
            "The only accepted value of PROOF_ForceLocal parameter is 1 !");
   }

   // Below we provide a possibility to change the way packet size is
   // calculated or define the packet time directly.
   // fPacketAsAFraction can be interpreted as follows:
   // packet time is (expected job proc. time) / fPacketSizeAsAFraction.
   // It substitutes 20 in the old formula to calculate the fPacketSize:
   // fPacketSize = fTotalEntries / (20 * nslaves)
   Int_t packetAsAFraction = 0;
   if (TProof::GetParameter(input, "PROOF_PacketAsAFraction", packetAsAFraction) == 0) {
      if (packetAsAFraction > 0) {
         fPacketAsAFraction = packetAsAFraction;
         Info("TPacketizerAdaptive",
              "using alternate fraction of query time as a packet size: %d",
              packetAsAFraction);
      } else
         Info("TPacketizerAdaptive", "packetAsAFraction parameter must be higher than 0");
   }

   // Packet re-assignement
   fTryReassign = 0;
   Int_t tryReassign = 0;
   if (TProof::GetParameter(input, "PROOF_TryReassign", tryReassign) != 0)
      tryReassign = gEnv->GetValue("Packetizer.TryReassign", 0);
   fTryReassign = tryReassign;
   if (fTryReassign != 0)
      Info("TPacketizerAdaptive", "failed packets will be re-assigned");

   // Save the config parameters in the dedicated list so that they will be saved
   // in the outputlist and therefore in the relevant TQueryResult
   fConfigParams->Add(new TParameter<Int_t>("PROOF_PacketizerCachePacketSync", (Int_t)fCachePacketSync));
   fConfigParams->Add(new TParameter<Double_t>("PROOF_PacketizerMaxEntriesRatio", fMaxEntriesRatio));
   fConfigParams->Add(new TParameter<Int_t>("PROOF_PacketizerStrategy", fStrategy));
   fConfigParams->Add(new TParameter<Int_t>("PROOF_MaxWorkersPerNode", (Int_t)fMaxSlaveCnt));
   fConfigParams->Add(new TParameter<Int_t>("PROOF_ForceLocal", (Int_t)fForceLocal));
   fConfigParams->Add(new TParameter<Int_t>("PROOF_PacketAsAFraction", fPacketAsAFraction));

   Double_t baseLocalPreference = 1.2;
   fBaseLocalPreference = (Float_t)baseLocalPreference;
   if (TProof::GetParameter(input, "PROOF_BaseLocalPreference", baseLocalPreference) == 0)
      fBaseLocalPreference = (Float_t)baseLocalPreference;

   fFileNodes = new TList;
   fFileNodes->SetOwner();
   fUnAllocated = new TList;
   fUnAllocated->SetOwner(kFALSE);
   fActive = new TList;
   fActive->SetOwner(kFALSE);

   fValid = kTRUE;

   // Resolve end-point urls to optmize distribution
   // dset->Lookup(); // moved to TProofPlayerRemote::Process

   // Read list of mounted disks
   TObjArray *partitions = 0;
   TString partitionsStr;
   if (TProof::GetParameter(input, "PROOF_PacketizerPartitions", partitionsStr) != 0)
      partitionsStr = gEnv->GetValue("Packetizer.Partitions", "");
   if (!partitionsStr.IsNull()) {
      Info("TPacketizerAdaptive", "Partitions: %s", partitionsStr.Data());
      partitions = partitionsStr.Tokenize(",");
   }

   // Split into per host and disk entries
   dset->Reset();
   TDSetElement *e;
   while ((e = (TDSetElement*)dset->Next())) {

      if (e->GetValid()) continue;

      // The dataset name, if any
      if (fDataSet.IsNull() && e->GetDataSet() && strlen(e->GetDataSet()))
         fDataSet = e->GetDataSet();

      TUrl url = e->GetFileName();
      PDB(kPacketizer,2)
         Info("TPacketizerAdaptive", "element name: %s (url: %s)", e->GetFileName(), url.GetUrl());

      // Map non URL filenames to dummy host
      TString host;
      if ( !url.IsValid() ||
          (strncmp(url.GetProtocol(),"root", 4) &&
           strncmp(url.GetProtocol(),"rfio", 4) &&
           strncmp(url.GetProtocol(),"file", 4)) ) {
         host = "no-host";
      } else if ( url.IsValid() && !strncmp(url.GetProtocol(),"file", 4)) {
         host = "localhost";
         url.SetProtocol("root");
      } else {
         host = url.GetHostFQDN();
      }
      // Get full name for local hosts
      if (host.Contains("localhost") || host == "127.0.0.1") {
         url.SetHost(gSystem->HostName());
         host = url.GetHostFQDN();
      }

      // Find on which disk is the file, if any
      TString disk;
      if (partitions) {
         TIter iString(partitions);
         TObjString* os = 0;
         while ((os = (TObjString *)iString())) {
            // Compare begining of the url with disk mountpoint
            if (strncmp(url.GetFile(), os->GetName(), os->GetString().Length()) == 0) {
               disk = os->GetName();
               break;
            }
         }
      }
      // Node's url
      TString nodeStr;
      if (disk.IsNull())
         nodeStr.Form("%s://%s", url.GetProtocol(), host.Data());
      else
         nodeStr.Form("%s://%s/%s", url.GetProtocol(), host.Data(), disk.Data());
      TFileNode *node = (TFileNode *) fFileNodes->FindObject(nodeStr);

      if (node == 0) {
         node = new TFileNode(nodeStr, fStrategy, fFilesToProcess);
         fFileNodes->Add(node);
         PDB(kPacketizer,2)
            Info("TPacketizerAdaptive", "creating new node '%s' or the element", nodeStr.Data());
      } else {
         PDB(kPacketizer,2)
            Info("TPacketizerAdaptive", "adding element to existing node '%s'", nodeStr.Data());
      }

      node->Add(e, kFALSE);
   }

   fSlaveStats = new TMap;
   fSlaveStats->SetOwner(kFALSE);

   TSlave *slave;
   TIter si(slaves);
   while ((slave = (TSlave*) si.Next())) {
      fSlaveStats->Add( slave, new TSlaveStat(slave) );
      fMaxPerfIdx = slave->GetPerfIdx() > fMaxPerfIdx ?
         slave->GetPerfIdx() : fMaxPerfIdx;
   }

   // Setup file & filenode structure
   Reset();
   // Optimize the number of files to be open when running on subsample
   Int_t validateMode = 0;
   Int_t gprc = TProof::GetParameter(input, "PROOF_ValidateByFile", validateMode);
   Bool_t byfile = (gprc == 0 && validateMode > 0 && num > -1) ? kTRUE : kFALSE;
   if (num > -1)
      PDB(kPacketizer,2)
         Info("TPacketizerAdaptive",
              "processing subset of entries: validating by file? %s", byfile ? "yes": "no");
   ValidateFiles(dset, slaves, num, byfile);


   if (!fValid) return;

   // apply global range (first,num) to dset and rebuild structure
   // ommitting TDSet elements that are not needed

   Int_t files = 0;
   fTotalEntries = 0;
   fUnAllocated->Clear();  // avoid dangling pointers
   fActive->Clear();
   fFileNodes->Clear();    // then delete all objects
   PDB(kPacketizer,2)
      Info("TPacketizerAdaptive",
           "processing range: first %lld, num %lld", first, num);

   dset->Reset();
   Long64_t cur = 0;
   while (( e = (TDSetElement*)dset->Next())) {

      // Skip invalid or missing file; It will be moved
      // from the dset to the 'MissingFiles' list in the player.
      if (!e->GetValid()) continue;

      TUrl url = e->GetFileName();
      Long64_t eFirst = e->GetFirst();
      Long64_t eNum = e->GetNum();
      PDB(kPacketizer,2)
         Info("TPacketizerAdaptive", "processing element '%s'", e->GetFileName());
      PDB(kPacketizer,2)
         Info("TPacketizerAdaptive",
              " --> first %lld, elenum %lld (cur %lld) (entrylist: %p)", eFirst, eNum, cur, e->GetEntryList());

      if (!e->GetEntryList()) {
         // This element is before the start of the global range, skip it
         if (cur + eNum < first) {
            cur += eNum;
            PDB(kPacketizer,2)
               Info("TPacketizerAdaptive", " --> skip element cur %lld", cur);
            continue;
         }

         // This element is after the end of the global range, skip it
         if (num != -1 && (first+num <= cur)) {
            cur += eNum;
            PDB(kPacketizer,2)
               Info("TPacketizerAdaptive", " --> drop element cur %lld", cur);
            continue; // break ??
         }

         Bool_t inRange = kFALSE;
         if (cur <= first || (num != -1 && (first+num <= cur+eNum))) {

            if (cur <= first) {
               // If this element contains the start of the global range
               // adjust its start and number of entries
               e->SetFirst( eFirst + (first - cur) );
               e->SetNum( e->GetNum() - (first - cur) );
               PDB(kPacketizer,2)
                  Info("TPacketizerAdaptive", " --> adjust start %lld and end %lld",
                       eFirst + (first - cur), first + num - cur);
               inRange = kTRUE;
            }
            if (num != -1 && (first+num <= cur+eNum)) {
               // If this element contains the end of the global range
               // adjust its number of entries
               e->SetNum( first + num - e->GetFirst() - cur );
               PDB(kPacketizer,2)
                  Info("TPacketizerAdaptive", " --> adjust end %lld", first + num - cur);
               inRange = kTRUE;
            }

         } else {
            // Increment the counter ...
            PDB(kPacketizer,2)
               Info("TPacketizerAdaptive", " --> increment 'cur' by %lld", eNum);
            cur += eNum;
         }
         // Re-adjust eNum and cur, if needed
         if (inRange) {
            cur += eNum;
            eNum = e->GetNum();
         }

      } else {
         TEntryList *enl = dynamic_cast<TEntryList *>(e->GetEntryList());
         if (enl) {
            eNum = enl->GetN();
            PDB(kPacketizer,2)
               Info("TPacketizerAdaptive", " --> entry-list element: %lld entries", eNum);
         } else {
            TEventList *evl = dynamic_cast<TEventList *>(e->GetEntryList());
            eNum = evl ? evl->GetN() : eNum;
            PDB(kPacketizer,2)
               Info("TPacketizerAdaptive", " --> event-list element: %lld entries (evl:%p)", eNum, evl);
         }
         if (!eNum) {
            PDB(kPacketizer,2)
               Info("TPacketizerAdaptive", " --> empty entry- or event-list element!");
            continue;
         }
      }
      PDB(kPacketizer,2)
         Info("TPacketizerAdaptive", " --> next cur %lld", cur);

      // Map non URL filenames to dummy host
      TString host;
      if ( !url.IsValid() ||
          (strncmp(url.GetProtocol(),"root", 4) &&
           strncmp(url.GetProtocol(),"rfio", 4) &&
           strncmp(url.GetProtocol(),"file", 4)) ) {
         host = "no-host";
      } else if ( url.IsValid() && !strncmp(url.GetProtocol(),"file", 4)) {
         host = "localhost";
         url.SetProtocol("root");
      } else {
         host = url.GetHostFQDN();
      }
      // Get full name for local hosts
      if (host.Contains("localhost") || host == "127.0.0.1") {
         url.SetHost(gSystem->HostName());
         host = url.GetHostFQDN();
      }

      // Find, on which disk is the file
      TString disk;
      if (partitions) {
         TIter iString(partitions);
         TObjString* os = 0;
         while ((os = (TObjString *)iString())) {
            // Compare begining of the url with disk mountpoint
            if (strncmp(url.GetFile(), os->GetName(), os->GetString().Length()) == 0) {
               disk = os->GetName();
               break;
            }
         }
      }
      // Node's url
      TString nodeStr;
      if (disk.IsNull())
         nodeStr.Form("%s://%s", url.GetProtocol(), host.Data());
      else
         nodeStr.Form("%s://%s/%s", url.GetProtocol(), host.Data(), disk.Data());
      TFileNode *node = (TFileNode*) fFileNodes->FindObject(nodeStr);


      if (node == 0) {
         node = new TFileNode(nodeStr, fStrategy, fFilesToProcess);
         fFileNodes->Add( node );
         PDB(kPacketizer, 2)
            Info("TPacketizerAdaptive", " --> creating new node '%s' for element", nodeStr.Data());
      } else {
         PDB(kPacketizer, 2)
            Info("TPacketizerAdaptive", " --> adding element to exiting node '%s'", nodeStr.Data());
      }

      ++files;
      fTotalEntries += eNum;
      node->Add(e, kTRUE);
      node->IncEvents(eNum);
      PDB(kPacketizer,2) e->Print("a");
   }
   PDB(kPacketizer,1)
      Info("TPacketizerAdaptive", "processing %lld entries in %d files on %d hosts",
                                  fTotalEntries, files, fFileNodes->GetSize());

   // Set the total number for monitoring
   if (gPerfStats)
      gPerfStats->SetNumEvents(fTotalEntries);

   Reset();

   InitStats();

   if (!fValid)
      SafeDelete(fProgress);

   PDB(kPacketizer,1) Info("TPacketizerAdaptive", "return");
}

////////////////////////////////////////////////////////////////////////////////
/// Destructor.

TPacketizerAdaptive::~TPacketizerAdaptive()
{
   if (fSlaveStats) {
      fSlaveStats->DeleteValues();
   }

   SafeDelete(fSlaveStats);
   SafeDelete(fUnAllocated);
   SafeDelete(fActive);
   SafeDelete(fFileNodes);
   SafeDelete(fFilesToProcess);
}

////////////////////////////////////////////////////////////////////////////////
/// (re)initialise the statistics
/// called at the begining or after a worker dies.

void TPacketizerAdaptive::InitStats()
{
   // calculating how many files from TDSet are not cached on
   // any slave
   Int_t noRemoteFiles = 0;
   fNEventsOnRemLoc = 0;
   Int_t totalNumberOfFiles = 0;
   TIter next(fFileNodes);
   while (TFileNode *fn = (TFileNode*)next()) {
      totalNumberOfFiles += fn->GetNumberOfFiles();
      if (fn->GetMySlaveCnt() == 0) {
         noRemoteFiles += fn->GetNumberOfFiles();
         fNEventsOnRemLoc += (fn->GetNEvents() - fn->GetProcessed());
      }
   }

   if (totalNumberOfFiles == 0) {
      Info("InitStats", "no valid or non-empty file found: setting invalid");
      // No valid files: set invalid and return
      fValid = kFALSE;
      return;
   }

   fFractionOfRemoteFiles = (1.0 * noRemoteFiles) / totalNumberOfFiles;
   Info("InitStats",
        "fraction of remote files %f", fFractionOfRemoteFiles);

   if (!fValid)
      SafeDelete(fProgress);

   PDB(kPacketizer,1) Info("InitStats", "return");
}

////////////////////////////////////////////////////////////////////////////////
/// Get next unallocated file from 'node' or other nodes:
/// First try 'node'. If there is no more files, keep trying to
/// find an unallocated file on other nodes.

TPacketizerAdaptive::TFileStat *TPacketizerAdaptive::GetNextUnAlloc(TFileNode *node, const char *nodeHostName)
{
   TFileStat *file = 0;

   if (node != 0) {
      PDB(kPacketizer, 2)
         Info("GetNextUnAlloc", "looking for file on node %s", node->GetName());
      file = node->GetNextUnAlloc();
      if (file == 0) RemoveUnAllocNode(node);
   } else {
      if (nodeHostName && strlen(nodeHostName) > 0) {

         TFileNode *fn;
         // Make sure that they are in the corrected order
         fUnAllocated->Sort();
         PDB(kPacketizer,2) fUnAllocated->Print();

         // Loop over unallocated fileNode list
         for (int i = 0; i < fUnAllocated->GetSize(); i++) {

            if ((fn = (TFileNode *) fUnAllocated->At(i))) {
               TUrl uu(fn->GetName());
               PDB(kPacketizer, 2)
                  Info("GetNextUnAlloc", "comparing %s with %s...", nodeHostName, uu.GetHost());

               // Check, whether node's hostname is matching with current fileNode (fn)
               if (!strcmp(nodeHostName, uu.GetHost())) {
                  node = fn;

                  // Fetch next unallocated file from this node
                  if ((file = node->GetNextUnAlloc()) == 0) {
                     RemoveUnAllocNode(node);
                     node = 0;
                  } else {
                     PDB(kPacketizer, 2)
                        Info("GetNextUnAlloc", "found! (host: %s)", uu.GetHost());
                     break;
                  }
               }
            } else {
               Warning("GetNextUnAlloc", "unallocate entry %d is empty!", i);
            }
         }

         if (node != 0 && fMaxSlaveCnt > 0 && node->GetExtSlaveCnt() >= fMaxSlaveCnt) {
            // Unlike in TPacketizer we look at the number of ext slaves only.
            PDB(kPacketizer,1)
               Info("GetNextUnAlloc", "reached Workers-per-Node Limit (%ld)", fMaxSlaveCnt);
            node = 0;
         }
      }

      if (node == 0) {
         while (file == 0 && ((node = NextNode()) != 0)) {
            PDB(kPacketizer, 2)
               Info("GetNextUnAlloc", "looking for file on node %s", node->GetName());
            if ((file = node->GetNextUnAlloc()) == 0) RemoveUnAllocNode(node);
         }
      }
   }

   if (file != 0) {
      // if needed make node active
      if (fActive->FindObject(node) == 0) {
         fActive->Add(node);
      }
   }

   PDB(kPacketizer, 2) {
      if (!file) {
         Info("GetNextUnAlloc", "no file found!");
      } else {
         file->Print();
      }
   }

   return file;
}

////////////////////////////////////////////////////////////////////////////////
/// Get next node which has unallocated files.
/// the order is determined by TFileNode::Compare

TPacketizerAdaptive::TFileNode *TPacketizerAdaptive::NextNode()
{
   fUnAllocated->Sort();
   PDB(kPacketizer,2) {
      fUnAllocated->Print();
   }

   TFileNode *fn = (TFileNode*) fUnAllocated->First();
   if (fn != 0 && fMaxSlaveCnt > 0 && fn->GetExtSlaveCnt() >= fMaxSlaveCnt) {
      // unlike in TPacketizer we look at the number of ext slaves only.
      PDB(kPacketizer,1)
         Info("NextNode", "reached Workers-per-Node Limit (%ld)", fMaxSlaveCnt);
      fn = 0;
   }

   return fn;
}

////////////////////////////////////////////////////////////////////////////////
/// Remove unallocated node.

void TPacketizerAdaptive::RemoveUnAllocNode(TFileNode * node)
{
   fUnAllocated->Remove(node);
}

////////////////////////////////////////////////////////////////////////////////
/// Get next active file.

TPacketizerAdaptive::TFileStat *TPacketizerAdaptive::GetNextActive()
{
   TFileNode *node;
   TFileStat *file = 0;

   while (file == 0 && ((node = NextActiveNode()) != 0)) {
         file = node->GetNextActive();
         if (file == 0) RemoveActiveNode(node);
   }

   return file;
}


////////////////////////////////////////////////////////////////////////////////
/// Get next active node.

TPacketizerAdaptive::TFileNode *TPacketizerAdaptive::NextActiveNode()
{
   fActive->Sort();
   PDB(kPacketizer,2) {
      Info("NextActiveNode", "enter");
      fActive->Print();
   }

   TFileNode *fn = (TFileNode*) fActive->First();
   // look at only ext slaves
   if (fn != 0 && fMaxSlaveCnt > 0 && fn->GetExtSlaveCnt() >= fMaxSlaveCnt) {
      PDB(kPacketizer,1)
         Info("NextActiveNode","reached Workers-per-Node limit (%ld)", fMaxSlaveCnt);
      fn = 0;
   }

   return fn;
}

////////////////////////////////////////////////////////////////////////////////
/// Remove file from the list of actives.

void TPacketizerAdaptive::RemoveActive(TFileStat *file)
{
   TFileNode *node = file->GetNode();

   node->RemoveActive(file);
   if (node->GetNumberOfActiveFiles() == 0) RemoveActiveNode(node);
}

////////////////////////////////////////////////////////////////////////////////
/// Remove node from the list of actives.

void TPacketizerAdaptive::RemoveActiveNode(TFileNode *node)
{
   fActive->Remove(node);
}

////////////////////////////////////////////////////////////////////////////////
/// Reset the internal data structure for packet distribution.

void TPacketizerAdaptive::Reset()
{
   fUnAllocated->Clear();
   fUnAllocated->AddAll(fFileNodes);

   fActive->Clear();

   TIter files(fFileNodes);
   TFileNode *fn;
   while ((fn = (TFileNode*) files.Next()) != 0) {
      fn->Reset();
   }

   TIter slaves(fSlaveStats);
   TObject *key;
   while ((key = slaves.Next()) != 0) {
      TSlaveStat *slstat = (TSlaveStat*) fSlaveStats->GetValue(key);
      if (!slstat) {
         Warning("Reset", "TSlaveStat associated to key '%s' is NULL", key->GetName());
         continue;
      }
      // Find out which file nodes are on the worker machine and assign the
      // one with less workers assigned
      TFileNode *fnmin = 0;
      Int_t fncnt = fSlaveStats->GetSize();
      files.Reset();
      while ((fn = (TFileNode*) files.Next()) != 0) {
         if (!strcmp(slstat->GetName(), TUrl(fn->GetName()).GetHost())) {
            if (fn->GetMySlaveCnt() < fncnt) {
               fnmin = fn;
               fncnt = fn->GetMySlaveCnt();
            }
         }
      }
      if (fnmin != 0 ) {
         slstat->SetFileNode(fnmin);
         fnmin->IncMySlaveCnt();
         PDB(kPacketizer, 2)
            Info("Reset","assigning node '%s' to '%s' (cnt: %d)",
                         fnmin->GetName(), slstat->GetName(), fnmin->GetMySlaveCnt());
      }
      slstat->fCurFile = 0;
   }
}

////////////////////////////////////////////////////////////////////////////////
/// Check existence of file/dir/tree an get number of entries.
/// Assumes the files have been setup.

void TPacketizerAdaptive::ValidateFiles(TDSet *dset, TList *slaves,
                                        Long64_t maxent, Bool_t byfile)
{
   TMap     slaves_by_sock;
   TMonitor mon;
   TList    workers;


   // Setup the communication infrastructure

   workers.AddAll(slaves);
   TIter    si(slaves);
   TSlave   *slm;
   while ((slm = (TSlave*)si.Next()) != 0) {
      PDB(kPacketizer,3)
      Info("ValidateFiles","socket added to monitor: %p (%s)",
          slm->GetSocket(), slm->GetName());
      mon.Add(slm->GetSocket());
      slaves_by_sock.Add(slm->GetSocket(), slm);
   }

   mon.DeActivateAll();

   ((TProof*)gProof)->DeActivateAsyncInput();

   // Some monitoring systems (TXSocketHandler) need to know this
   ((TProof*)gProof)->fCurrentMonitor = &mon;

   // Identify the type
   if (!strcmp(dset->GetType(), "TTree")) SetBit(TVirtualPacketizer::kIsTree);

   // Preparing for client notification
   TString msg("Validating files");
   UInt_t n = 0;
   UInt_t tot = dset->GetListOfElements()->GetSize();
   Bool_t st = kTRUE;

   Long64_t totent = 0, nopenf = 0;
   while (kTRUE) {

      // send work
      while (TSlave *s = (TSlave *)workers.First()) {

         workers.Remove(s);

         // find a file

         TSlaveStat *slstat = (TSlaveStat*)fSlaveStats->GetValue(s);
         if (!slstat) {
            Error("ValidateFiles", "TSlaveStat associated to slave '%s' is NULL", s->GetName());
            continue;
         }

         TFileNode *node = 0;
         TFileStat *file = 0;

         // try its own node first
         if ((node = slstat->GetFileNode()) != 0) {
            PDB(kPacketizer,3) node->Print();
            file = GetNextUnAlloc(node);
            if (file == 0)
               slstat->SetFileNode(0);
         }

         // look for a file on any other node if necessary
         if (file == 0)
            file = GetNextUnAlloc();

         if (file != 0) {
            // files are done right away
            RemoveActive(file);

            slstat->fCurFile = file;
            TDSetElement *elem = file->GetElement();
            Long64_t entries = elem->GetEntries(kTRUE, kFALSE);
            if (entries < 0 || strlen(elem->GetTitle()) <= 0) {
               // This is decremented when we get the reply
               file->GetNode()->IncExtSlaveCnt(slstat->GetName());
               TMessage m(kPROOF_GETENTRIES);
               m << dset->IsTree()
               << TString(elem->GetFileName())
               << TString(elem->GetDirectory())
               << TString(elem->GetObjName());

               s->GetSocket()->Send( m );
               mon.Activate(s->GetSocket());
               PDB(kPacketizer,2)
                  Info("ValidateFiles",
                       "sent to worker-%s (%s) via %p GETENTRIES on %s %s %s %s",
                       s->GetOrdinal(), s->GetName(), s->GetSocket(),
                       dset->IsTree() ? "tree" : "objects", elem->GetFileName(),
                       elem->GetDirectory(), elem->GetObjName());
            } else {
               // Fill the info
               elem->SetTDSetOffset(entries);
               if (entries > 0) {
                  // Most likely valid
                  elem->SetValid();
                  if (!elem->GetEntryList()) {
                     if (elem->GetFirst() > entries) {
                        Error("ValidateFiles",
                              "first (%lld) higher then number of entries (%lld) in %s",
                               elem->GetFirst(), entries, elem->GetFileName());
                        // disable element
                        slstat->fCurFile->SetDone();
                        elem->Invalidate();
                        dset->SetBit(TDSet::kSomeInvalid);
                     }
                     if (elem->GetNum() == -1) {
                        elem->SetNum(entries - elem->GetFirst());
                     } else if (elem->GetFirst() + elem->GetNum() > entries) {
                        Warning("ValidateFiles", "num (%lld) + first (%lld) larger then number of"
                                 " keys/entries (%lld) in %s", elem->GetNum(), elem->GetFirst(),
                                 entries, elem->GetFileName());
                        elem->SetNum(entries - elem->GetFirst());
                     }
                     PDB(kPacketizer,2)
                        Info("ValidateFiles",
                             "found elem '%s' with %lld entries", elem->GetFileName(), entries);
                  }
               }
               // Count
               totent += entries;
               nopenf++;
               // Notify the client
               n++;
               gProof->SendDataSetStatus(msg, n, tot, st);

               // This worker is ready for the next validation
               workers.Add(s);
            }
         }
      }

      // Check if there is anything to wait for
      if (mon.GetActive() == 0) {
         if (byfile && maxent > 0) {
            // How many files do we still need ?
            Long64_t nrestf = (maxent - totent) * nopenf / totent ;
            if (nrestf <= 0 && maxent > totent) nrestf = 1;
            if (nrestf > 0) {
               PDB(kPacketizer,3)
                  Info("ValidateFiles", "{%lld, %lld, %lld}: needs to validate %lld more files",
                                         maxent, totent, nopenf, nrestf);
               si.Reset();
               while ((slm = (TSlave *) si.Next()) && nrestf--) {
                  workers.Add(slm);
               }
               continue;
            } else {
               PDB(kPacketizer,3)
                  Info("ValidateFiles", "no need to validate more files");
               break;
            }
         } else {
            break;
         }
      }

      PDB(kPacketizer,3) {
         Info("ValidateFiles", "waiting for %d slaves:", mon.GetActive());
         TList *act = mon.GetListOfActives();
         TIter next(act);
         while (TSocket *s = (TSocket*) next()) {
            TSlave *sl = (TSlave *) slaves_by_sock.GetValue(s);
            if (sl)
               Info("ValidateFiles", "   worker-%s (%s)",
                    sl->GetOrdinal(), sl->GetName());
         }
         delete act;
      }

      TSocket *sock = mon.Select();
      // If we have been interrupted break
      if (!sock) {
         Error("ValidateFiles", "selection has been interrupted - STOP");
         mon.DeActivateAll();
         fValid = kFALSE;
         break;
      }
      mon.DeActivate(sock);

      PDB(kPacketizer,3) Info("ValidateFiles", "select returned: %p", sock);

      TSlave *slave = (TSlave *) slaves_by_sock.GetValue( sock );
      if (!sock->IsValid()) {
         // A socket got invalid during validation
         Error("ValidateFiles", "worker-%s (%s) got invalid - STOP",
               slave->GetOrdinal(), slave->GetName());
         ((TProof*)gProof)->MarkBad(slave, "socket got invalid during validation");
         fValid = kFALSE;
         break;
      }

      TMessage *reply;

      if (sock->Recv(reply) <= 0) {
         // Notify
         Error("ValidateFiles", "Recv failed! for worker-%s (%s)",
                                slave->GetOrdinal(), slave->GetName());
         // Help! lost a slave? ('slave' is deleted inside here ...)
         ((TProof*)gProof)->MarkBad(slave, "receive failed during validation");
         fValid = kFALSE;
         continue;
      }

      if (reply->What() != kPROOF_GETENTRIES) {
         // Not what we want: handover processing to the central machinery
         Int_t what = reply->What();
         ((TProof*)gProof)->HandleInputMessage(slave, reply);
         if (what == kPROOF_FATAL) {
             Error("ValidateFiles", "kPROOF_FATAL from worker-%s (%s)",
                                    slave->GetOrdinal(), slave->GetName());
             fValid = kFALSE;
         } else {
            // Reactivate the socket
            mon.Activate(sock);
         }
         // Get next message
         continue;
      }

      TSlaveStat *slavestat = (TSlaveStat*) fSlaveStats->GetValue( slave );
      TDSetElement *e = slavestat->fCurFile->GetElement();
      slavestat->fCurFile->GetNode()->DecExtSlaveCnt(slavestat->GetName());
      Long64_t entries;

      (*reply) >> entries;

      // Extract object name, if there
      if ((reply->BufferSize() > reply->Length())) {
         TString objname;
         (*reply) >> objname;
         e->SetTitle(objname);
      }

      e->SetTDSetOffset(entries);
      if (entries > 0) {

         // This dataset element is most likely valid
         e->SetValid();

         if (!e->GetEntryList()) {
            if (e->GetFirst() > entries) {
               Error("ValidateFiles",
                     "first (%lld) higher then number of entries (%lld) in %s",
                      e->GetFirst(), entries, e->GetFileName());

               // Invalidate the element
               slavestat->fCurFile->SetDone();
               e->Invalidate();
               dset->SetBit(TDSet::kSomeInvalid);
            }

            if (e->GetNum() == -1) {
               e->SetNum(entries - e->GetFirst());
            } else if (e->GetFirst() + e->GetNum() > entries) {
               Error("ValidateFiles",
                     "num (%lld) + first (%lld) larger then number of keys/entries (%lld) in %s",
                      e->GetNum(), e->GetFirst(), entries, e->GetFileName());
               e->SetNum(entries - e->GetFirst());
            }
         }

         // Count
         totent += entries;
         nopenf++;

         // Notify the client
         n++;
         gProof->SendDataSetStatus(msg, n, tot, st);

      } else {

         Error("ValidateFiles", "cannot get entries for file: %s - skipping", e->GetFileName() );
         //
         // Need to fix this with a user option to allow incomplete file sets (rdm)
         //
         //fValid = kFALSE; // all element must be readable!
         if (gProofServ) {
            TMessage m(kPROOF_MESSAGE);
            m << TString(Form("Cannot get entries for file: %s - skipping",
                              e->GetFileName()));
            gProofServ->GetSocket()->Send(m);
         }

         // invalidate element
         e->Invalidate();
         dset->SetBit(TDSet::kSomeInvalid);
      }
      PDB(kPacketizer,3) Info("ValidateFiles", " %lld events validated", totent);

      // Ready for the next job, unless we have enough files
      if (maxent < 0 || ((totent < maxent) && !byfile))
         workers.Add(slave);
   }

   // report std. output from slaves??

   ((TProof*)gProof)->ActivateAsyncInput();

   // This needs to be reset
   ((TProof*)gProof)->fCurrentMonitor = 0;

   // No reason to continue if invalid
   if (!fValid)
      return;

   // compute the offset for each file element
   Long64_t offset = 0;
   Long64_t newOffset = 0;
   TIter next(dset->GetListOfElements());
   TDSetElement *el;
   while ( (el = dynamic_cast<TDSetElement*> (next())) ) {
      if (el->GetValid()) {
         newOffset = offset + el->GetTDSetOffset();
         el->SetTDSetOffset(offset);
         offset = newOffset;
      }
   }
}

////////////////////////////////////////////////////////////////////////////////
/// The result depends on the fStrategy

Int_t TPacketizerAdaptive::CalculatePacketSize(TObject *slStatPtr, Long64_t cachesz, Int_t learnent)
{
   Long64_t num;
   if (fStrategy == 0) {
      // TPacketizer's heuristic for starting packet size
      // Constant packet size;
      Int_t nslaves = fSlaveStats->GetSize();
      if (nslaves > 0) {
         num = fTotalEntries / (fPacketAsAFraction * nslaves);
      } else {
         num = 1;
      }
   } else {
      // The dynamic heuristic for setting the packet size (default)
      // Calculates the packet size based on performance of this slave
      // and estimated time left until the end of the query.
      TSlaveStat* slstat = (TSlaveStat*)slStatPtr;
      Float_t rate = slstat->GetCurRate();
      if (!rate)
         rate = slstat->GetAvgRate();
      if (rate) {

         // Global average rate
         Float_t avgProcRate = (GetEntriesProcessed()/(GetCumProcTime() / fSlaveStats->GetSize()));
         Float_t packetTime = ((fTotalEntries - GetEntriesProcessed())/avgProcRate)/fPacketAsAFraction;

         // Bytes-to-Event conversion
         Float_t bevt = (GetEntriesProcessed() > 0) ? GetBytesRead() / GetEntriesProcessed() : -1.;

         // Make sure it is not smaller then the cache, if the info is available and the size
         // synchronization is required. But apply the cache-packet size synchronization only if there
         // are enough left files to process and the files are all of similar sizes. Otherwise we risk
         // to not exploit optimally all potentially active workers.
         Bool_t cpsync = fCachePacketSync;
         if (fMaxEntriesRatio > 0. && cpsync) {
            if (fFilesToProcess && fFilesToProcess->GetSize() <= fSlaveStats->GetSize()) {
               Long64_t remEntries = fTotalEntries - GetEntriesProcessed();
               Long64_t maxEntries = -1;
               if (fFilesToProcess->Last()) {
                  TDSetElement *elem = (TDSetElement *) ((TPacketizerAdaptive::TFileStat *) fFilesToProcess->Last())->GetElement();
                  if (elem) maxEntries = elem->GetNum();
               }
               if (maxEntries > remEntries / fSlaveStats->GetSize() * fMaxEntriesRatio) {
                  PDB(kPacketizer,3) {
                     Info("CalculatePacketSize", "%s: switching off synchronization of packet and cache sizes:", slstat->GetOrdinal());
                     Info("CalculatePacketSize", "%s: few files (%d) remaining of very different sizes (max/avg = %.2f > %.2f)",
                                                 slstat->GetOrdinal(), fFilesToProcess->GetSize(),
                                                (Double_t)maxEntries / remEntries * fSlaveStats->GetSize(), fMaxEntriesRatio);
                  }
                  cpsync = kFALSE;
               }
            }
         }
         if (bevt > 0. && cachesz > 0 && cpsync) {
            if ((Long64_t)(rate * packetTime * bevt) < cachesz)
               packetTime = cachesz / bevt / rate;
         }

         // Apply min-max again, if required
         if (fMaxPacketTime > 0. && packetTime > fMaxPacketTime) packetTime = fMaxPacketTime;
         if (fMinPacketTime > 0. && packetTime < fMinPacketTime) packetTime = fMinPacketTime;

         // Translate the packet length in number of entries
         num = (Long64_t)(rate * packetTime);

         // Notify
         PDB(kPacketizer,2)
            Info("CalculatePacketSize","%s: avgr: %f, rate: %f, left: %lld, pacT: %f, sz: %f (csz: %f), num: %lld",
                 slstat->GetOrdinal(), avgProcRate, rate, fTotalEntries - GetEntriesProcessed(),
                 packetTime, ((bevt > 0) ? num*bevt/1048576. : -1.), cachesz/1048576., num);

      } else {
         // First packet for this worker in this query
         // Twice the learning phase
         num = (learnent > 0) ? 5 * learnent : 1000;

         // Notify
         PDB(kPacketizer,2)
            Info("CalculatePacketSize","%s: num: %lld", slstat->GetOrdinal(),  num);
      }
   }
   if (num < 1) num = 1;
   return num;
}

////////////////////////////////////////////////////////////////////////////////
/// To be used by GetNextPacket but also in reaction to kPROOF_STOPPROCESS
/// message (when the worker was asked to stop processing during a packet).
/// returns the #entries intended in the last packet - #processed entries

Int_t TPacketizerAdaptive::AddProcessed(TSlave *sl,
                                        TProofProgressStatus *status,
                                        Double_t latency,
                                        TList **listOfMissingFiles)
{
   // find slave
   TSlaveStat *slstat = (TSlaveStat*) fSlaveStats->GetValue( sl );
   if (!slstat) {
      Error("AddProcessed", "%s: TSlaveStat instance for worker %s not found!",
                            (sl ? sl->GetOrdinal() : "x.x"),
                            (sl ? sl->GetName() : "**undef**"));
      return -1;
   }

   // update stats & free old element

   if ( slstat->fCurElem != 0 ) {
      Long64_t expectedNumEv = slstat->fCurElem->GetNum();
      // Calculate the number of events processed in the last packet
      Long64_t numev;
      if (status && status->GetEntries() > 0)
         numev = status->GetEntries() - slstat->GetEntriesProcessed();
      else
         numev = 0;

      // Calculate the progress made in the last packet
      TProofProgressStatus *progress = 0;
      if (numev > 0) {
         // This also moves the pointer in the corrsponding TFileInfo
         progress = slstat->AddProcessed(status);
         if (progress) {
            (*fProgressStatus) += *progress;
            // update processing rate
            slstat->UpdateRates(status);
         }
      } else {
          progress = new TProofProgressStatus();
      }
      if (progress) {
         PDB(kPacketizer,2)
            Info("AddProcessed", "%s: %s: %lld %7.3lf %7.3lf %7.3lf %lld",
               sl->GetOrdinal(), sl->GetName(), progress->GetEntries(), latency,
               progress->GetProcTime(), progress->GetCPUTime(), progress->GetBytesRead());

         if (gPerfStats)
            gPerfStats->PacketEvent(sl->GetOrdinal(), sl->GetName(),
                                    slstat->fCurElem->GetFileName(),
                                    progress->GetEntries(),
                                    latency,
                                    progress->GetProcTime(),
                                    progress->GetCPUTime(),
                                    progress->GetBytesRead());
         delete progress;
      }
      if (numev != expectedNumEv) {
         // The last packet was not fully processed
         // and will be split in two:
         // - The completed part was marked as done.
         // - Create a new packet with the part to be resubmitted.
         TDSetElement *newPacket = new TDSetElement(*(slstat->fCurElem));
         if (newPacket && numev < expectedNumEv) {
            Long64_t first = newPacket->GetFirst();
            newPacket->SetFirst(first + numev);
            if (ReassignPacket(newPacket, listOfMissingFiles) == -1)
               SafeDelete(newPacket);
         } else
            Error("AddProcessed", "%s: processed too much? (%lld, %lld)",
                                  sl->GetOrdinal(), numev, expectedNumEv);

         // TODO: a signal handler which will send info from the worker
         // after a packet fails.
         /* Add it to the failed packets list.
         if (!fFailedPackets) {
            fFailedPackets = new TList();
         }
         fFailedPackets->Add(slstat->fCurElem);
         */
      }

      slstat->fCurElem = 0;
      return (expectedNumEv - numev);
   } else {
      // the kPROOF_STOPPRPOCESS message is send after the worker receives zero
      // as the reply to kPROOF_GETNEXTPACKET
      return -1;
   }
}

////////////////////////////////////////////////////////////////////////////////
/// Get next packet;
/// A meaningfull difference to TPacketizer is the fact that this
/// packetizer, for each worker, tries to predict whether the worker
/// will finish processing it's local files before the end of the query.
/// If yes, it allocates, to those workers, files from non-slave filenodes
/// or from slaves that are overloaded. The check is done every time a new
/// file needs to be assigned.

TDSetElement *TPacketizerAdaptive::GetNextPacket(TSlave *sl, TMessage *r)
{
   if ( !fValid ) {
      return 0;
   }

   // find slave

   TSlaveStat *slstat = (TSlaveStat*) fSlaveStats->GetValue( sl );
   if (!slstat) {
      Error("GetNextPacket", "TSlaveStat instance for worker %s not found!",
                            (sl ? sl->GetName() : "**undef**"));
      return 0;
   }

   // Attach to current file
   TFileStat *file = slstat->fCurFile;

   // Update stats & free old element

   Bool_t firstPacket = kFALSE;
   Long64_t cachesz = -1;
   Int_t learnent = -1;
   if ( slstat->fCurElem != 0 ) {

      Long64_t restEntries = 0;
      Double_t latency, proctime, proccpu;
      TProofProgressStatus *status = 0;
      Bool_t fileNotOpen = kFALSE, fileCorrupted = kFALSE;

      if (sl->GetProtocol() > 18) {

         (*r) >> latency;
         (*r) >> status;

         if (sl->GetProtocol() > 25) {
            (*r) >> cachesz >> learnent;
            if (r->BufferSize() > r->Length()) (*r) >> restEntries;
         }
         fileNotOpen = status->TestBit(TProofProgressStatus::kFileNotOpen) ? kTRUE : kFALSE;
         fileCorrupted = status->TestBit(TProofProgressStatus::kFileCorrupted) ? kTRUE : kFALSE;

      } else {

         Long64_t bytesRead = -1;

         (*r) >> latency >> proctime >> proccpu;
         // only read new info if available
         if (r->BufferSize() > r->Length()) (*r) >> bytesRead;
         if (r->BufferSize() > r->Length()) (*r) >> restEntries;
         Long64_t totev = 0;
         if (r->BufferSize() > r->Length()) (*r) >> totev;

         status = new TProofProgressStatus(totev, bytesRead, -1, proctime, proccpu);
         fileNotOpen = (restEntries < 0) ? kTRUE : kFALSE;
      }

      if (!fileNotOpen && !fileCorrupted) {
         if (AddProcessed(sl, status, latency) != 0)
            Error("GetNextPacket", "%s: the worker processed a different # of entries", sl->GetOrdinal());
         if (fProgressStatus->GetEntries() >= fTotalEntries) {
            if (fProgressStatus->GetEntries() > fTotalEntries)
               Error("GetNextPacket", "%s: processed too many entries! (%lld, %lld)",
                                      sl->GetOrdinal(), fProgressStatus->GetEntries(), fTotalEntries);
            // Send last timer message and stop the timer
            HandleTimer(0);
            SafeDelete(fProgress);
         }
      } else {
         if (file) {
            if (file->GetElement()) {
               if (fileCorrupted) {
                  Info("GetNextPacket", "%s: file '%s' turned corrupted: invalidating file (%lld)",
                                       sl->GetOrdinal(), file->GetElement()->GetName(), restEntries);
                  Int_t nunproc = AddProcessed(sl, status, latency);
                  PDB(kPacketizer,1)
                     Info("GetNextPacket", "%s: %d entries un-processed", sl->GetOrdinal(), nunproc);
                  // Remaining to be processed
                  Long64_t num = 0;
                  if (file->GetElement()->TestBit(TDSetElement::kCorrupted)) {
                     // Add the remainign entries in the packet to the ones already registered
                     num = file->GetElement()->GetEntries() + restEntries;
                  } else {
                     // First call: add the remaining entries in the packet and those of the file
                     // not yet assigned
                     Long64_t rest = file->GetElement()->GetEntries() - file->GetNextEntry();
                     num = restEntries + rest;
                  }
                  file->GetElement()->SetEntries(num);
                  PDB(kPacketizer,1)
                     Info("GetNextPacket", "%s: removed file: %s, entries left: %lld", sl->GetOrdinal(),
                                           file->GetElement()->GetName(), file->GetElement()->GetEntries());
                  // Flag as corrupted
                  file->GetElement()->SetBit(TDSetElement::kCorrupted);
               } else {
                  Info("GetNextPacket", "%s: file '%s' could not be open: invalidating related element",
                                        sl->GetOrdinal(), file->GetElement()->GetName());
               }
               // Invalidate the element
               file->GetElement()->Invalidate();
               // Add it to the failed packets list
               if (!fFailedPackets) fFailedPackets = new TList();
               if (!fFailedPackets->FindObject(file->GetElement()))
                  fFailedPackets->Add(file->GetElement());
            }
            // Deactivate this TFileStat
            file->SetDone();
            RemoveActive(file);
         } else {
            Info("GetNextPacket", "%s: error raised by worker, but TFileStat object invalid:"
                                  " protocol error?", sl->GetOrdinal());
         }
      }
   } else {
      firstPacket = kTRUE;
   }

   if ( fStop ) {
      HandleTimer(0);
      return 0;
   }

   TString nodeName;
   if (file != 0) nodeName = file->GetNode()->GetName();
   TString nodeHostName(slstat->GetName());

   PDB(kPacketizer,3)
      Info("GetNextPacket", "%s: entries processed: %lld - looking for a packet from node '%s'",
                            sl->GetOrdinal(), fProgressStatus->GetEntries(), nodeName.Data());

   // If current file is just finished
   if ( file != 0 && file->IsDone() ) {
      file->GetNode()->DecExtSlaveCnt(slstat->GetName());
      file->GetNode()->DecRunSlaveCnt();
      if (gPerfStats)
         gPerfStats->FileEvent(sl->GetOrdinal(), sl->GetName(), file->GetNode()->GetName(),
                               file->GetElement()->GetFileName(), kFALSE);
      file = 0;
   }
   // Reset the current file field
   slstat->fCurFile = file;

   Long64_t avgEventsLeftPerSlave =
      (fTotalEntries - fProgressStatus->GetEntries()) / fSlaveStats->GetSize();
   if (fTotalEntries == fProgressStatus->GetEntries())
      return 0;
   // Get a file if needed
   if ( file == 0) {
      // Needs a new file
      Bool_t openLocal;
      // Aiming for localPreference == 1 when #local == #remote events left
      Float_t localPreference = fBaseLocalPreference - (fNEventsOnRemLoc /
                                (0.4 *(fTotalEntries - fProgressStatus->GetEntries())));
      if ( slstat->GetFileNode() != 0 ) {
         // Local file node exists and has more events to process.
         fUnAllocated->Sort();
         TFileNode* firstNonLocalNode = (TFileNode*)fUnAllocated->First();
         Bool_t nonLocalNodePossible;
         if (fForceLocal)
            nonLocalNodePossible = 0;
         else
            nonLocalNodePossible = firstNonLocalNode ?
                    (fMaxSlaveCnt < 0 || (fMaxSlaveCnt > 0 && firstNonLocalNode->GetExtSlaveCnt() < fMaxSlaveCnt))
                                                     : 0;
         openLocal = !nonLocalNodePossible;
         Float_t slaveRate = slstat->GetAvgRate();
         if ( nonLocalNodePossible && fStrategy == 1) {
            // OpenLocal is set to kFALSE
            if ( slstat->GetFileNode()->GetRunSlaveCnt() >
                 slstat->GetFileNode()->GetMySlaveCnt() - 1 )
                // External slaves help slstat -> don't open nonlocal files
                // -1 because, at this point slstat is not running
                  openLocal = kTRUE;
            else if ( slaveRate == 0 ) { // first file for this slave
               // GetLocalEventsLeft() counts the potential slave
               // as running on its fileNode.
               if ( slstat->GetLocalEventsLeft() * localPreference
                   > (avgEventsLeftPerSlave))
                  openLocal = kTRUE;
               else if ( (firstNonLocalNode->GetEventsLeftPerSlave())
                     < slstat->GetLocalEventsLeft() * localPreference )
                  openLocal = kTRUE;
               else if ( firstNonLocalNode->GetExtSlaveCnt() > 1 )
                  openLocal = kTRUE;
               else if ( firstNonLocalNode->GetRunSlaveCnt() == 0 )
                  openLocal = kTRUE;
            } else {
               // At this point slstat has a non zero avg rate > 0
               Float_t slaveTime = slstat->GetLocalEventsLeft()/slaveRate;
               // And thus fCumProcTime, fProcessed > 0
               Float_t avgTime = avgEventsLeftPerSlave
                                 /(fProgressStatus->GetEntries()/GetCumProcTime());
               if (slaveTime * localPreference > avgTime)
                  openLocal = kTRUE;
               else if ((firstNonLocalNode->GetEventsLeftPerSlave())
                        < slstat->GetLocalEventsLeft() * localPreference)
                  openLocal = kTRUE;
            }
         }
         if (openLocal || fStrategy == 0) {
            // Try its own node
            file = slstat->GetFileNode()->GetNextUnAlloc();
            if (!file)
               file = slstat->GetFileNode()->GetNextActive();
            if ( file == 0 ) {
               // No more files on this worker
               slstat->SetFileNode(0);
            }
         }
      }

      // Try to find an unused filenode first
      if(file == 0 && !fForceLocal)
         file = GetNextUnAlloc(0, nodeHostName);

      // Then look at the active filenodes
      if(file == 0 && !fForceLocal)
         file = GetNextActive();

      if (file == 0) return 0;

      PDB(kPacketizer,3) if (fFilesToProcess) fFilesToProcess->Print();

      slstat->fCurFile = file;
      // if remote and unallocated file
      if (file->GetNode()->GetMySlaveCnt() == 0 &&
         file->GetElement()->GetFirst() == file->GetNextEntry()) {
         fNEventsOnRemLoc -= file->GetElement()->GetNum();
         if (fNEventsOnRemLoc < 0) {
            Error("GetNextPacket",
                  "inconsistent value for fNEventsOnRemLoc (%lld): stop delivering packets!",
                   fNEventsOnRemLoc);
            return 0;
         }
      }
      file->GetNode()->IncExtSlaveCnt(slstat->GetName());
      file->GetNode()->IncRunSlaveCnt();
      if (gPerfStats)
         gPerfStats->FileEvent(sl->GetOrdinal(), sl->GetName(),
                               file->GetNode()->GetName(),
                               file->GetElement()->GetFileName(), kTRUE);
   }

   Long64_t num = CalculatePacketSize(slstat, cachesz, learnent);

   // Get a packet

   TDSetElement *base = file->GetElement();
   Long64_t first = file->GetNextEntry();
   Long64_t last = base->GetFirst() + base->GetNum();

   // If the remaining part is smaller than the (packetsize * 1.5)
   // then increase the packetsize

   if ( first + num * 1.5 >= last ) {
      num = last - first;
      file->SetDone(); // done
      // Delete file from active list (unalloc list is single pass, no delete needed)
      RemoveActive(file);
   }

   // Update NextEntry in the file object
   file->MoveNextEntry(num);

   slstat->fCurElem = CreateNewPacket(base, first, num);
   if (base->GetEntryList())
      slstat->fCurElem->SetEntryList(base->GetEntryList(), first, num);

   // Flag the first packet of a new run (dataset)
   if (firstPacket)
      slstat->fCurElem->SetBit(TDSetElement::kNewRun);
   else
      slstat->fCurElem->ResetBit(TDSetElement::kNewRun);

   PDB(kPacketizer,2)
      Info("GetNextPacket","%s: %s %lld %lld (%lld)", sl->GetOrdinal(), base->GetFileName(), first, first + num - 1, num);

   return slstat->fCurElem;
}

////////////////////////////////////////////////////////////////////////////////
/// Return the number of workers still processing

Int_t TPacketizerAdaptive::GetActiveWorkers()
{
   Int_t actw = 0;
   TIter nxw(fSlaveStats);
   TObject *key;
   while ((key = nxw())) {
      TSlaveStat *wrkstat = (TSlaveStat *) fSlaveStats->GetValue(key);
      if (wrkstat && wrkstat->fCurFile) actw++;
   }
   // Done
   return actw;
}

////////////////////////////////////////////////////////////////////////////////
/// Get Estimation of the current rate; just summing the current rates of
/// the active workers

Float_t TPacketizerAdaptive::GetCurrentRate(Bool_t &all)
{
   all = kTRUE;
   // Loop over the workers
   Float_t currate = 0.;
   if (fSlaveStats && fSlaveStats->GetSize() > 0) {
      TIter nxw(fSlaveStats);
      TObject *key;
      while ((key = nxw()) != 0) {
         TSlaveStat *slstat = (TSlaveStat *) fSlaveStats->GetValue(key);
         if (slstat && slstat->GetProgressStatus() && slstat->GetEntriesProcessed() > 0) {
            // Sum-up the current rates
            currate += slstat->GetProgressStatus()->GetCurrentRate();
         } else {
            all = kFALSE;
         }
      }
   }
   // Done
   return currate;
}

////////////////////////////////////////////////////////////////////////////////
/// Get estimation for the number of processed entries and bytes read at time t,
/// based on the numbers already processed and the latests worker measured speeds.
/// If t <= 0 the current time is used.
/// Only the estimation for the entries is currently implemented.
/// This is needed to smooth the instantaneous rate plot.

Int_t TPacketizerAdaptive::GetEstEntriesProcessed(Float_t t, Long64_t &ent,
                                                  Long64_t &bytes, Long64_t &calls)
{
   // Default value
   ent = GetEntriesProcessed();
   bytes = GetBytesRead();
   calls = GetReadCalls();

   // Parse option
   if (fUseEstOpt == kEstOff)
      // Do not use estimation
      return 0;
   Bool_t current = (fUseEstOpt == kEstCurrent) ? kTRUE : kFALSE;

   TTime tnow = gSystem->Now();
   Double_t now = (t > 0) ? (Double_t)t : Long64_t(tnow) / (Double_t)1000.;
   Double_t dt = -1;

   // Loop over the workers
   Bool_t all = kTRUE;
   Float_t trate = 0.;
   if (fSlaveStats && fSlaveStats->GetSize() > 0) {
      ent = 0;
      TIter nxw(fSlaveStats);
      TObject *key;
      while ((key = nxw()) != 0) {
         TSlaveStat *slstat = (TSlaveStat *) fSlaveStats->GetValue(key);
         if (slstat) {
            // Those surely processed
            Long64_t e = slstat->GetEntriesProcessed();
            if (e <= 0) all = kFALSE;
            // Time elapsed since last update
            dt = now - slstat->GetProgressStatus()->GetLastUpdate();
            // Add estimated entries processed since last update
            Float_t rate = (current && slstat->GetCurRate() > 0) ? slstat->GetCurRate()
                                                                 : slstat->GetAvgRate();
            trate += rate;
            // Add estimated entries processed since last update
            e += (Long64_t) (dt * rate);
            // Add to the total
            ent += e;
            // Notify
            PDB(kPacketizer,3)
               Info("GetEstEntriesProcessed","%s: e:%lld rate:%f dt:%f e:%lld",
                                          slstat->fSlave->GetOrdinal(),
                                          slstat->GetEntriesProcessed(), rate, dt, e);
         }
      }
   }
   // Notify
   dt = now - fProgressStatus->GetLastUpdate();
   PDB(kPacketizer,2)
      Info("GetEstEntriesProcessed",
           "dt: %f, estimated entries: %lld (%lld), bytes read: %lld rate: %f (all: %d)",
                               dt, ent, GetEntriesProcessed(), bytes, trate, all);

   // Check values
   ent = (ent > 0) ? ent : fProgressStatus->GetEntries();
   ent = (ent <= fTotalEntries) ? ent : fTotalEntries;
   bytes = (bytes > 0) ? bytes : fProgressStatus->GetBytesRead();

   // Done
   return ((all) ? 0 : 1);
}

////////////////////////////////////////////////////////////////////////////////
/// This method can be called at any time during processing
/// as an effect of handling kPROOF_STOPPROCESS
/// If the output list from this worker is going to be sent back to the master,
/// the 'status' includes the number of entries processed by the slave.
/// From this we calculate the remaining part of the packet.
/// 0 indicates that the results from that worker were lost completely.
/// Assume that the filenodes for which we have a TFileNode object
/// are still up and running.

void TPacketizerAdaptive::MarkBad(TSlave *s, TProofProgressStatus *status,
                                  TList **listOfMissingFiles)
{
   TSlaveStat *slaveStat = (TSlaveStat *)(fSlaveStats->GetValue(s));
   if (!slaveStat) {
      Error("MarkBad", "Worker does not exist");
      return;
   }
   // Update worker counters
   if (slaveStat->fCurFile && slaveStat->fCurFile->GetNode()) {
      slaveStat->fCurFile->GetNode()->DecExtSlaveCnt(slaveStat->GetName());
      slaveStat->fCurFile->GetNode()->DecRunSlaveCnt();
   }

   // If status is defined, the remaining part of the last packet is
   // reassigned in AddProcessed called from handling kPROOF_STOPPROCESS
   if (!status) {
      // Get the subset processed by the bad worker.
      TList *subSet = slaveStat->GetProcessedSubSet();
      if (subSet) {
         // Take care of the current packet
         if (slaveStat->fCurElem) {
            subSet->Add(slaveStat->fCurElem);
         }
         // Merge overlapping or subsequent elements
         Int_t nmg = 0, ntries = 100;
         TDSetElement *e = 0, *enxt = 0;
         do {
            nmg = 0;
            e = (TDSetElement *) subSet->First();
            while ((enxt = (TDSetElement *) subSet->After(e))) {
               if (e->MergeElement(enxt) >= 0) {
                  nmg++;
                  subSet->Remove(enxt);
                  delete enxt;
               } else {
                  e = enxt;
               }
            }
         } while (nmg > 0 && --ntries > 0);
         // reassign the packets assigned to the bad slave and save the size;
         SplitPerHost(subSet, listOfMissingFiles);
         // the elements were reassigned so should not be deleted
         subSet->SetOwner(0);
      } else {
         Warning("MarkBad", "subset processed by bad worker not found!");
      }
      (*fProgressStatus) -= *(slaveStat->GetProgressStatus());
   }
   // remove slavestat from the map
   fSlaveStats->Remove(s);
   delete slaveStat;
   // recalculate fNEventsOnRemLoc and others
   InitStats();
}

////////////////////////////////////////////////////////////////////////////////
/// The file in the listOfMissingFiles can appear several times;
/// in order to fix that, a TDSetElement::Merge method is needed.

Int_t TPacketizerAdaptive::ReassignPacket(TDSetElement *e,
                                          TList **listOfMissingFiles)
{
   if (!e) {
      Error("ReassignPacket", "empty packet!");
      return -1;
   }
   // Check the old filenode
   TUrl url = e->GetFileName();
   // Check the host from which 'e' was previously read.
   // Map non URL filenames to dummy host
   TString host;
   if ( !url.IsValid() ||
       (strncmp(url.GetProtocol(),"root", 4) &&
        strncmp(url.GetProtocol(),"rfio", 4))) {
      host = "no-host";
   } else {
      host = url.GetHost();
   }

   // If accessible add it back to the old node
   // and do DecProcessed
   TFileNode *node = (TFileNode*) fFileNodes->FindObject( host );
   if (node && fTryReassign) {
      // The packet 'e' was processing data from this node.
      node->DecreaseProcessed(e->GetNum());
      // The file should be already in fFilesToProcess ...
      node->Add(e, kFALSE);
      if (!fUnAllocated->FindObject(node))
         fUnAllocated->Add(node);
      return 0;
   } else {
      // Add to the list of missing files
      TFileInfo *fi = e->GetFileInfo();
      if (listOfMissingFiles && *listOfMissingFiles)
         (*listOfMissingFiles)->Add((TObject *)fi);
      return -1;
   }
}

////////////////////////////////////////////////////////////////////////////////
/// Split into per host entries
/// The files in the listOfMissingFiles can appear several times;
/// in order to fix that, a TDSetElement::Merge method is needed.

void TPacketizerAdaptive::SplitPerHost(TList *elements,
                                       TList **listOfMissingFiles)
{
   if (!elements) {
      Error("SplitPerHost", "Empty list of packets!");
      return;
   }
   if (elements->GetSize() <= 0) {
      Error("SplitPerHost", "The input list contains no elements");
      return;
   }
   TIter subSetIter(elements);
   TDSetElement *e;
   while ((e = (TDSetElement*) subSetIter.Next())) {
      if (ReassignPacket(e, listOfMissingFiles) == -1) {
         // Remove from the list in order to delete it.
         if (elements->Remove(e))
            Error("SplitPerHost", "Error removing a missing file");
         delete e;
      }

   }
}