#include "TProofDraw.h"
#include "TProofPlayer.h"
#include "THashList.h"
#include "TEnv.h"
#include "TEventIter.h"
#include "TVirtualPacketizer.h"
#include "TSelector.h"
#include "TSocket.h"
#include "TProofServ.h"
#include "TProof.h"
#include "TProofOutputFile.h"
#include "TProofSuperMaster.h"
#include "TSlave.h"
#include "TClass.h"
#include "TROOT.h"
#include "TError.h"
#include "TException.h"
#include "MessageTypes.h"
#include "TMessage.h"
#include "TDSetProxy.h"
#include "TString.h"
#include "TSystem.h"
#include "TFile.h"
#include "TFileInfo.h"
#include "TFileMerger.h"
#include "TProofDebug.h"
#include "TTimer.h"
#include "TMap.h"
#include "TPerfStats.h"
#include "TStatus.h"
#include "TEventList.h"
#include "TProofLimitsFinder.h"
#include "TSortedList.h"
#include "TTree.h"
#include "TEntryList.h"
#include "TDSet.h"
#include "TDrawFeedback.h"
#include "TNamed.h"
#include "TObjString.h"
#include "TQueryResult.h"
#include "TMD5.h"
#include "TMethodCall.h"
#include "TObjArray.h"
#include "TMutex.h"
#ifndef R__TH1MERGEFIXED
#include "TH1.h"
#endif
#include "TVirtualMonitoring.h"
#include "TParameter.h"
#define kPEX_STOPPED 1001
#define kPEX_ABORTED 1002
static Bool_t gAbort = kFALSE;
class TAutoBinVal : public TNamed {
private:
Double_t fXmin, fXmax, fYmin, fYmax, fZmin, fZmax;
public:
TAutoBinVal(const char *name, Double_t xmin, Double_t xmax, Double_t ymin,
Double_t ymax, Double_t zmin, Double_t zmax) : TNamed(name,"")
{
fXmin = xmin; fXmax = xmax;
fYmin = ymin; fYmax = ymax;
fZmin = zmin; fZmax = zmax;
}
void GetAll(Double_t& xmin, Double_t& xmax, Double_t& ymin,
Double_t& ymax, Double_t& zmin, Double_t& zmax)
{
xmin = fXmin; xmax = fXmax;
ymin = fYmin; ymax = fYmax;
zmin = fZmin; zmax = fZmax;
}
};
class TDispatchTimer : public TTimer {
private:
TProofPlayer *fPlayer;
public:
TDispatchTimer(TProofPlayer *p) : TTimer(1000, kFALSE), fPlayer(p) { }
Bool_t Notify();
};
Bool_t TDispatchTimer::Notify()
{
if (gDebug > 0)
Info ("Notify","called!");
fPlayer->SetBit(TProofPlayer::kDispatchOneEvent);
Reset();
return kTRUE;
}
class TStopTimer : public TTimer {
private:
Bool_t fAbort;
TProofPlayer *fPlayer;
public:
TStopTimer(TProofPlayer *p, Bool_t abort, Int_t to);
Bool_t Notify();
};
TStopTimer::TStopTimer(TProofPlayer *p, Bool_t abort, Int_t to)
: TTimer(((to <= 0 || to > 864000) ? 10 : to * 1000), kFALSE)
{
if (gDebug > 0)
Info ("TStopTimer","enter: %d, timeout: %d", abort, to);
fPlayer = p;
fAbort = abort;
if (gDebug > 1)
Info ("TStopTimer","timeout set to %s ms", fTime.AsString());
}
Bool_t TStopTimer::Notify()
{
if (gDebug > 0)
Info ("Notify","called!");
if (fAbort)
Throw(kPEX_ABORTED);
else
Throw(kPEX_STOPPED);
return kTRUE;
}
ClassImp(TProofPlayer)
TProofPlayer::TProofPlayer(TProof *)
: fAutoBins(0), fOutput(0), fSelector(0), fSelectorClass(0),
fFeedbackTimer(0), fFeedbackPeriod(2000),
fEvIter(0), fSelStatus(0),
fTotalEvents(0), fQueryResults(0), fQuery(0), fDrawQueries(0),
fMaxDrawQueries(1), fStopTimer(0), fStopTimerMtx(0), fDispatchTimer(0)
{
fInput = new TList;
fExitStatus = kFinished;
fProgressStatus = new TProofProgressStatus();
SetProcessing(kFALSE);
static Bool_t initLimitsFinder = kFALSE;
if (!initLimitsFinder && gProofServ && !gProofServ->IsMaster()) {
THLimitsFinder::SetLimitsFinder(new TProofLimitsFinder);
initLimitsFinder = kTRUE;
}
}
TProofPlayer::~TProofPlayer()
{
fInput->Clear("nodelete");
SafeDelete(fInput);
SafeDelete(fSelector);
SafeDelete(fFeedbackTimer);
SafeDelete(fEvIter);
SafeDelete(fQueryResults);
SafeDelete(fDispatchTimer);
SafeDelete(fStopTimer);
}
void TProofPlayer::SetProcessing(Bool_t on)
{
if (on)
SetBit(TProofPlayer::kIsProcessing);
else
ResetBit(TProofPlayer::kIsProcessing);
}
void TProofPlayer::StopProcess(Bool_t abort, Int_t timeout)
{
if (gDebug > 0)
Info ("StopProcess","abort: %d, timeout: %d", abort, timeout);
if (fEvIter != 0)
fEvIter->StopProcess(abort);
Long_t to = 1;
if (abort == kTRUE) {
fExitStatus = kAborted;
} else {
fExitStatus = kStopped;
to = timeout;
}
if (to > 0)
SetStopTimer(kTRUE, abort, to);
}
void TProofPlayer::SetDispatchTimer(Bool_t on)
{
SafeDelete(fDispatchTimer);
ResetBit(TProofPlayer::kDispatchOneEvent);
if (on) {
fDispatchTimer = new TDispatchTimer(this);
fDispatchTimer->Start();
}
}
void TProofPlayer::SetStopTimer(Bool_t on, Bool_t abort, Int_t timeout)
{
fStopTimerMtx = (fStopTimerMtx) ? fStopTimerMtx : new TMutex(kTRUE);
R__LOCKGUARD(fStopTimerMtx);
SafeDelete(fStopTimer);
if (on) {
fStopTimer = new TStopTimer(this, abort, timeout);
fStopTimer->Start();
if (gDebug > 0)
Info ("SetStopTimer", "%s timer STARTED (timeout: %d)",
(abort ? "ABORT" : "STOP"), timeout);
} else {
if (gDebug > 0)
Info ("SetStopTimer", "timer STOPPED");
}
}
void TProofPlayer::AddQueryResult(TQueryResult *q)
{
if (!q) {
Warning("AddQueryResult","query undefined - do nothing");
return;
}
if (!(q->IsDraw())) {
if (!fQueryResults) {
fQueryResults = new TList;
fQueryResults->Add(q);
} else {
TIter nxr(fQueryResults);
TQueryResult *qr = 0;
TQueryResult *qp = 0;
while ((qr = (TQueryResult *) nxr())) {
if (*qr == *q) {
fQueryResults->Remove(qr);
delete qr;
break;
}
if (qr->GetEndTime().Convert() < q->GetEndTime().Convert())
qp = qr;
}
if (!qp) {
fQueryResults->AddFirst(q);
} else {
fQueryResults->AddAfter(qp, q);
}
}
} else if (IsClient()) {
if (fDrawQueries == fMaxDrawQueries && fMaxDrawQueries > 0) {
TIter nxr(fQueryResults);
TQueryResult *qr = 0;
while ((qr = (TQueryResult *) nxr())) {
if (qr->IsDraw()) {
fDrawQueries--;
fQueryResults->Remove(qr);
delete qr;
break;
}
}
}
if (fDrawQueries >= 0 && fDrawQueries < fMaxDrawQueries) {
fDrawQueries++;
if (!fQueryResults)
fQueryResults = new TList;
fQueryResults->Add(q);
}
}
}
void TProofPlayer::RemoveQueryResult(const char *ref)
{
if (fQueryResults) {
TIter nxq(fQueryResults);
TQueryResult *qr = 0;
while ((qr = (TQueryResult *) nxq())) {
if (qr->Matches(ref)) {
fQueryResults->Remove(qr);
delete qr;
}
}
}
}
TQueryResult *TProofPlayer::GetQueryResult(const char *ref)
{
if (fQueryResults) {
if (ref && strlen(ref) > 0) {
TIter nxq(fQueryResults);
TQueryResult *qr = 0;
while ((qr = (TQueryResult *) nxq())) {
if (qr->Matches(ref))
return qr;
}
} else {
return (TQueryResult *) fQueryResults->Last();
}
}
return (TQueryResult *)0;
}
void TProofPlayer::SetCurrentQuery(TQueryResult *q)
{
fPreviousQuery = fQuery;
fQuery = q;
}
void TProofPlayer::AddInput(TObject *inp)
{
fInput->Add(inp);
}
void TProofPlayer::ClearInput()
{
fInput->Clear();
}
TObject *TProofPlayer::GetOutput(const char *name) const
{
if (fOutput)
return fOutput->FindObject(name);
return 0;
}
TList *TProofPlayer::GetOutputList() const
{
TList *ol = fOutput;
if (!ol && fQuery)
ol = fQuery->GetOutputList();
return ol;
}
Int_t TProofPlayer::ReinitSelector(TQueryResult *qr)
{
Int_t rc = 0;
if (!qr) {
Info("ReinitSelector", "query undefined - do nothing");
return -1;
}
TString selec = qr->GetSelecImp()->GetName();
if (selec.Length() <= 0) {
Info("ReinitSelector", "selector name undefined - do nothing");
return -1;
}
Bool_t stdselec = TSelector::IsStandardDraw(selec);
Bool_t compselec = (selec.Contains(".") || stdselec) ? kFALSE : kTRUE;
TString ipathold;
if (!stdselec && !compselec) {
Bool_t expandselec = kTRUE;
TString dir, ipath;
char *selc = gSystem->Which(TROOT::GetMacroPath(), selec, kReadPermission);
if (selc) {
TMD5 *md5icur = 0, *md5iold = 0, *md5hcur = 0, *md5hold = 0;
md5icur = TMD5::FileChecksum(selc);
md5iold = qr->GetSelecImp()->Checksum();
char *selh = StrDup(selc);
char *p = (char *) strrchr(selh,'.');
if (p) strcpy(p+1,"h");
if (!gSystem->AccessPathName(selh, kReadPermission))
md5hcur = TMD5::FileChecksum(selh);
md5hold = qr->GetSelecHdr()->Checksum();
if (*md5hcur == *md5hold && *md5icur == *md5iold)
expandselec = kFALSE;
SafeDelete(md5icur);
SafeDelete(md5hcur);
SafeDelete(md5iold);
SafeDelete(md5hold);
if (selc) delete [] selc;
if (selh) delete [] selh;
}
Bool_t ok = kTRUE;
if (expandselec) {
ok = kFALSE;
TUUID u;
dir = Form("%s/%s",gSystem->TempDirectory(),u.AsString());
if (!(gSystem->MakeDirectory(dir))) {
selec = Form("%s/%s",dir.Data(),selec.Data());
qr->GetSelecImp()->SaveSource(selec);
TString seleh = Form("%s/%s",dir.Data(),qr->GetSelecHdr()->GetName());
qr->GetSelecHdr()->SaveSource(seleh);
ipathold = gSystem->GetIncludePath();
ipath = Form("-I%s %s", dir.Data(), gSystem->GetIncludePath());
gSystem->SetIncludePath(ipath.Data());
ok = kTRUE;
}
}
TString opt(qr->GetOptions());
Ssiz_t id = opt.Last('#');
if (id != kNPOS && id < opt.Length() - 1)
selec += opt(id + 1, opt.Length());
if (!ok) {
Info("ReinitSelector", "problems locating or exporting selector files");
return -1;
}
}
SafeDelete(fSelector);
fSelectorClass = 0;
Int_t iglevelsave = gErrorIgnoreLevel;
if (compselec)
gErrorIgnoreLevel = kBreak;
if ((fSelector = TSelector::GetSelector(selec))) {
if (compselec)
gErrorIgnoreLevel = iglevelsave;
fSelectorClass = fSelector->IsA();
fSelector->SetOption(qr->GetOptions());
} else {
if (compselec) {
gErrorIgnoreLevel = iglevelsave;
if (strlen(qr->GetLibList()) > 0) {
TString sl(qr->GetLibList());
TObjArray *oa = sl.Tokenize(" ");
if (oa) {
Bool_t retry = kFALSE;
TIter nxl(oa);
TObjString *os = 0;
while ((os = (TObjString *) nxl())) {
TString lib = gSystem->BaseName(os->GetName());
if (lib != "lib") {
lib.ReplaceAll("-l", "lib");
if (gSystem->Load(lib) == 0)
retry = kTRUE;
}
}
if (retry)
fSelector = TSelector::GetSelector(selec);
}
}
}
if (!fSelector) {
if (compselec)
Info("ReinitSelector", "compiled selector re-init failed:"
" automatic reload unsuccessful:"
" please load manually the correct library");
rc = -1;
}
}
if (fSelector) {
fSelector->SetInputList(qr->GetInputList());
if (stdselec) {
((TProofDraw *)fSelector)->DefVar();
} else {
fSelector->Begin(0);
}
}
if (ipathold.Length() > 0)
gSystem->SetIncludePath(ipathold.Data());
return rc;
}
Int_t TProofPlayer::AddOutputObject(TObject *)
{
MayNotUse("AddOutputObject");
return -1;
}
void TProofPlayer::AddOutput(TList *)
{
MayNotUse("AddOutput");
}
void TProofPlayer::StoreOutput(TList *)
{
MayNotUse("StoreOutput");
}
void TProofPlayer::StoreFeedback(TObject *, TList *)
{
MayNotUse("StoreFeedback");
}
void TProofPlayer::Progress(Long64_t , Long64_t )
{
MayNotUse("Progress");
}
void TProofPlayer::Progress(Long64_t , Long64_t ,
Long64_t ,
Float_t , Float_t ,
Float_t , Float_t )
{
MayNotUse("Progress");
}
void TProofPlayer::Feedback(TList *)
{
MayNotUse("Feedback");
}
TDrawFeedback *TProofPlayer::CreateDrawFeedback(TProof *p)
{
return new TDrawFeedback(p);
}
void TProofPlayer::SetDrawFeedbackOption(TDrawFeedback *f, Option_t *opt)
{
if (f)
f->SetOption(opt);
}
void TProofPlayer::DeleteDrawFeedback(TDrawFeedback *f)
{
delete f;
}
Long64_t TProofPlayer::Process(TDSet *dset, const char *selector_file,
Option_t *option, Long64_t nentries,
Long64_t first)
{
PDB(kGlobal,1) Info("Process","Enter");
fExitStatus = kFinished;
fOutput = 0;
TCleanup clean(this);
SafeDelete(fSelector);
fSelectorClass = 0;
Int_t version = -1;
TRY {
if (gProofServ)
gProofServ->CopyFromCache(selector_file, 1);
if (!(fSelector = TSelector::GetSelector(selector_file))) {
Error("Process", "cannot load: %s", selector_file );
return -1;
}
if (gProofServ)
gProofServ->CopyToCache(selector_file, 1);
fSelectorClass = fSelector->IsA();
version = fSelector->Version();
fOutput = fSelector->GetOutputList();
if (gProofServ)
TPerfStats::Start(fInput, fOutput);
fSelStatus = new TStatus;
fOutput->Add(fSelStatus);
fSelector->SetOption(option);
fSelector->SetInputList(fInput);
fTotalEvents = nentries;
if (fTotalEvents < 0 && gProofServ &&
gProofServ->IsMaster() && !gProofServ->IsParallel()) {
dset->Validate();
dset->Reset();
TDSetElement *e = 0;
while ((e = dset->Next())) {
fTotalEvents += e->GetNum();
}
}
dset->Reset();
fEvIter = TEventIter::Create(dset, fSelector, first, nentries);
if (version == 0) {
PDB(kLoop,1) Info("Process","Call Begin(0)");
fSelector->Begin(0);
} else {
if (IsClient()) {
PDB(kLoop,1) Info("Process","Call Begin(0)");
fSelector->Begin(0);
}
if (fSelStatus->IsOk()) {
PDB(kLoop,1) Info("Process","Call SlaveBegin(0)");
fSelector->SlaveBegin(0);
}
}
} CATCH(excode) {
SetProcessing(kFALSE);
Error("Process","exception %d caught", excode);
return -1;
} ENDTRY;
SetupFeedback();
if (gMonitoringWriter)
gMonitoringWriter->SendProcessingStatus("STARTED",kTRUE);
PDB(kLoop,1)
Info("Process","Looping over Process()");
Long64_t readbytesatstart = 0;
readbytesatstart = TFile::GetFileBytesRead();
if (gMonitoringWriter)
gMonitoringWriter->SendProcessingProgress(0,0,kTRUE);
SetDispatchTimer(kTRUE);
gAbort = kFALSE;
Long64_t entry;
fProgressStatus->Reset();
if (gProofServ)
gProofServ->GetSocket()->Send(kPROOF_ENDINIT);
TParameter<Long64_t> *par = (TParameter<Long64_t>*)fInput->FindObject("PROOF_MemLogFreq");
volatile Long64_t memlogfreq = (par) ? par->GetVal() : 100;
volatile Long_t memlim = (gProofServ) ? gProofServ->GetVirtMemHWM() : -1;
TRY {
while ((entry = fEvIter->GetNextEvent()) >= 0 && fSelStatus->IsOk()) {
SetProcessing(kTRUE);
if (version == 0) {
PDB(kLoop,3)
Info("Process","Call ProcessCut(%lld)", entry);
if (fSelector->ProcessCut(entry)) {
PDB(kLoop,3)
Info("Process","Call ProcessFill(%lld)", entry);
fSelector->ProcessFill(entry);
}
} else {
PDB(kLoop,3)
Info("Process","Call Process(%lld)", entry);
fSelector->Process(entry);
if (fSelector->GetAbort() == TSelector::kAbortProcess) {
SetProcessing(kFALSE);
break;
}
}
if (fSelStatus->IsOk()) {
fProgressStatus->IncEntries();
fProgressStatus->SetBytesRead(TFile::GetFileBytesRead()-readbytesatstart);
if (gMonitoringWriter)
gMonitoringWriter->SendProcessingProgress(fProgressStatus->GetEntries(),
TFile::GetFileBytesRead()-readbytesatstart, kFALSE);
if (memlogfreq > 0 && GetEventsProcessed()%memlogfreq == 0) {
ProcInfo_t pi;
if (!gSystem->GetProcInfo(&pi)){
Info("Process|Svc", "Memory %ld virtual %ld resident event %d",
pi.fMemVirtual, pi.fMemResident, GetEventsProcessed());
TString wmsg;
if (memlim > 0) {
if (pi.fMemVirtual > 0.95 * memlim) {
wmsg.Form("using more than 95% of allowed memory (%ld kB) - STOP processing", pi.fMemVirtual);
Error("Process", wmsg.Data());
wmsg.Insert(0, "ERROR: ");
if (gProofServ) gProofServ->SendAsynMessage(wmsg.Data());
fExitStatus = kStopped;
SetProcessing(kFALSE);
break;
} else if (pi.fMemVirtual > 0.80 * memlim) {
memlogfreq = 1;
wmsg.Form("using more than 80% of allowed memory (%ld kB)", pi.fMemVirtual);
Warning("Process", wmsg.Data());
wmsg.Insert(0, "WARNING: ");
if (gProofServ) gProofServ->SendAsynMessage(wmsg.Data());
}
}
}
}
}
if (TestBit(TProofPlayer::kDispatchOneEvent)) {
gSystem->DispatchOneEvent(kTRUE);
ResetBit(TProofPlayer::kDispatchOneEvent);
}
SetProcessing(kFALSE);
if (!fSelStatus->IsOk() || gROOT->IsInterrupted()) break;
}
} CATCH(excode) {
if (excode == kPEX_STOPPED) {
Info("Process","received stop-process signal");
fExitStatus = kStopped;
} else if (excode == kPEX_ABORTED) {
gAbort = kTRUE;
Info("Process","received abort-process signal");
fExitStatus = kAborted;
} else {
Error("Process","exception %d caught", excode);
gAbort = kTRUE;
fExitStatus = kAborted;
}
SetProcessing(kFALSE);
} ENDTRY;
PDB(kGlobal,2)
Info("Process","%lld events processed", fProgressStatus->GetEntries());
if (gMonitoringWriter) {
gMonitoringWriter->SendProcessingProgress(fProgressStatus->GetEntries(), TFile::GetFileBytesRead()-readbytesatstart, kFALSE);
gMonitoringWriter->SendProcessingStatus("DONE");
}
SetDispatchTimer(kFALSE);
if (fStopTimer != 0)
SetStopTimer(kFALSE, gAbort);
if (fFeedbackTimer != 0)
HandleTimer(0);
StopFeedback();
SafeDelete(fEvIter);
if (fExitStatus != kAborted) {
TIter nxo(GetOutputList());
TObject *o = 0;
while ((o = nxo())) {
if (o->IsA() == TProofOutputFile::Class()) {
((TProofOutputFile *)o)->SetWorkerOrdinal(gProofServ->GetOrdinal());
if (!strcmp(((TProofOutputFile *)o)->GetDir(),""))
((TProofOutputFile *)o)->SetDir(gProofServ->GetSessionDir());
}
}
if (fSelStatus->IsOk()) {
if (version == 0) {
PDB(kLoop,1) Info("Process","Call Terminate()");
fSelector->Terminate();
} else {
PDB(kLoop,1) Info("Process","Call SlaveTerminate()");
fSelector->SlaveTerminate();
if (IsClient() && fSelStatus->IsOk()) {
PDB(kLoop,1) Info("Process","Call Terminate()");
fSelector->Terminate();
}
}
}
if (gProofServ && !gProofServ->IsParallel()) {
TIter nxc(gROOT->GetListOfCanvases());
while (TObject *c = nxc())
fOutput->Add(c);
}
}
if (gProofServ)
TPerfStats::Stop();
return 0;
}
Long64_t TProofPlayer::Finalize(Bool_t, Bool_t)
{
MayNotUse("Finalize");
return -1;
}
Long64_t TProofPlayer::Finalize(TQueryResult *)
{
MayNotUse("Finalize");
return -1;
}
void TProofPlayer::UpdateAutoBin(const char *name,
Double_t& xmin, Double_t& xmax,
Double_t& ymin, Double_t& ymax,
Double_t& zmin, Double_t& zmax)
{
if ( fAutoBins == 0 ) {
fAutoBins = new THashList;
}
TAutoBinVal *val = (TAutoBinVal*) fAutoBins->FindObject(name);
if ( val == 0 ) {
if (gProofServ && !gProofServ->IsTopMaster()) {
TString key = name;
TProofLimitsFinder::AutoBinFunc(key,xmin,xmax,ymin,ymax,zmin,zmax);
}
val = new TAutoBinVal(name,xmin,xmax,ymin,ymax,zmin,zmax);
fAutoBins->Add(val);
} else {
val->GetAll(xmin,xmax,ymin,ymax,zmin,zmax);
}
}
TDSetElement *TProofPlayer::GetNextPacket(TSlave *, TMessage *)
{
MayNotUse("GetNextPacket");
return 0;
}
void TProofPlayer::SetupFeedback()
{
MayNotUse("SetupFeedback");
}
void TProofPlayer::StopFeedback()
{
MayNotUse("StopFeedback");
}
Long64_t TProofPlayer::DrawSelect(TDSet * , const char * ,
const char * , Option_t * ,
Long64_t , Long64_t )
{
MayNotUse("DrawSelect");
return -1;
}
void TProofPlayer::HandleGetTreeHeader(TMessage *)
{
MayNotUse("HandleGetTreeHeader|");
}
void TProofPlayer::HandleRecvHisto(TMessage *mess)
{
TObject *obj = mess->ReadObject(mess->GetClass());
if (obj->InheritsFrom(TH1::Class())) {
TH1 *h = (TH1*)obj;
h->SetDirectory(0);
TH1 *horg = (TH1*)gDirectory->GetList()->FindObject(h->GetName());
if (horg)
horg->Add(h);
else
h->SetDirectory(gDirectory);
}
}
Int_t TProofPlayer::DrawCanvas(TObject *obj)
{
static Int_t (*gDrawCanvasHook)(TObject *) = 0;
if (!gDrawCanvasHook) {
TString drawlib = "libProofDraw";
char *p = 0;
if ((p = gSystem->DynamicPathName(drawlib, kTRUE))) {
delete[] p;
if (gSystem->Load(drawlib) != -1) {
Func_t f = 0;
if ((f = gSystem->DynFindSymbol(drawlib,"DrawCanvas")))
gDrawCanvasHook = (Int_t (*)(TObject *))(f);
else
Warning("DrawCanvas", "can't find DrawCanvas");
} else
Warning("DrawCanvas", "can't load %s", drawlib.Data());
} else
Warning("DrawCanvas", "can't locate %s", drawlib.Data());
}
if (gDrawCanvasHook && obj)
return (*gDrawCanvasHook)(obj);
return 1;
}
Int_t TProofPlayer::GetDrawArgs(const char *var, const char *sel, Option_t *opt,
TString &selector, TString &objname)
{
static Int_t (*gGetDrawArgsHook)(const char *, const char *, Option_t *,
TString &, TString &) = 0;
if (!gGetDrawArgsHook) {
TString drawlib = "libProofDraw";
char *p = 0;
if ((p = gSystem->DynamicPathName(drawlib, kTRUE))) {
delete[] p;
if (gSystem->Load(drawlib) != -1) {
Func_t f = 0;
if ((f = gSystem->DynFindSymbol(drawlib,"GetDrawArgs")))
gGetDrawArgsHook = (Int_t (*)(const char *, const char *, Option_t *,
TString &, TString &))(f);
else
Warning("GetDrawArgs", "can't find GetDrawArgs");
} else
Warning("GetDrawArgs", "can't load %s", drawlib.Data());
} else
Warning("GetDrawArgs", "can't locate %s", drawlib.Data());
}
if (gGetDrawArgsHook)
return (*gGetDrawArgsHook)(var, sel, opt, selector, objname);
return 1;
}
ClassImp(TProofPlayerLocal)
ClassImp(TProofPlayerRemote)
TProofPlayerRemote::~TProofPlayerRemote()
{
SafeDelete(fOutput);
SafeDelete(fOutputLists);
SafeDelete(fFeedbackLists);
SafeDelete(fPacketizer);
}
Int_t TProofPlayerRemote::InitPacketizer(TDSet *dset, Long64_t nentries,
Long64_t first, const char *defpackunit,
const char *defpackdata)
{
SafeDelete(fPacketizer);
PDB(kGlobal,1) Info("Process","Enter");
fDSet = dset;
fExitStatus = kFinished;
Bool_t noData = dset->TestBit(TDSet::kEmpty) ? kTRUE : kFALSE;
TString packetizer;
TList *listOfMissingFiles = 0;
TMethodCall callEnv;
TClass *cl;
noData = dset->TestBit(TDSet::kEmpty) ? kTRUE : kFALSE;
if (noData) {
if (TProof::GetParameter(fInput, "PROOF_Packetizer", packetizer) != 0)
packetizer = defpackunit;
else
Info("InitPacketizer", "using alternate packetizer: %s", packetizer.Data());
cl = TClass::GetClass(packetizer);
if (cl == 0) {
Error("InitPacketizer", "class '%s' not found", packetizer.Data());
fExitStatus = kAborted;
return -1;
}
callEnv.InitWithPrototype(cl, cl->GetName(),"TList*,Long64_t,TList*,TProofProgressStatus*");
if (!callEnv.IsValid()) {
Error("InitPacketizer",
"cannot find correct constructor for '%s'", cl->GetName());
fExitStatus = kAborted;
return -1;
}
callEnv.ResetParam();
callEnv.SetParam((Long_t) fProof->GetListOfActiveSlaves());
callEnv.SetParam((Long64_t) nentries);
callEnv.SetParam((Long_t) fInput);
callEnv.SetParam((Long_t) fProgressStatus);
} else {
if ((listOfMissingFiles = (TList *)fInput->FindObject("MissingFiles"))) {
fInput->Remove(listOfMissingFiles);
} else {
listOfMissingFiles = new TList;
}
dset->Lookup(kTRUE, &listOfMissingFiles);
if (fProof->GetRunStatus() != TProof::kRunning) {
Error("InitPacketizer", "received stop/abort request");
fExitStatus = kAborted;
return -1;
}
if (!(dset->GetListOfElements()) ||
!(dset->GetListOfElements()->GetSize())) {
if (gProofServ)
gProofServ->SendAsynMessage("InitPacketizer: No files from the data set were found - Aborting");
Error("InitPacketizer", "No files from the data set were found - Aborting");
fExitStatus = kAborted;
if (listOfMissingFiles) {
listOfMissingFiles->SetOwner();
fOutput->Remove(listOfMissingFiles);
SafeDelete(listOfMissingFiles);
}
return -1;
}
if (TProof::GetParameter(fInput, "PROOF_Packetizer", packetizer) != 0)
packetizer = defpackdata;
else
Info("InitPacketizer", "using alternate packetizer: %s", packetizer.Data());
cl = TClass::GetClass(packetizer);
if (cl == 0) {
Error("InitPacketizer", "class '%s' not found", packetizer.Data());
fExitStatus = kAborted;
return -1;
}
callEnv.InitWithPrototype(cl, cl->GetName(),"TDSet*,TList*,Long64_t,Long64_t,TList*,TProofProgressStatus*");
if (!callEnv.IsValid()) {
Error("InitPacketizer", "cannot find correct constructor for '%s'", cl->GetName());
fExitStatus = kAborted;
return -1;
}
callEnv.ResetParam();
callEnv.SetParam((Long_t) dset);
callEnv.SetParam((Long_t) fProof->GetListOfActiveSlaves());
callEnv.SetParam((Long64_t) first);
callEnv.SetParam((Long64_t) nentries);
callEnv.SetParam((Long_t) fInput);
callEnv.SetParam((Long_t) fProgressStatus);
dset->SetBit(TDSet::kValidityChecked);
dset->ResetBit(TDSet::kSomeInvalid);
}
Long_t ret = 0;
callEnv.Execute(ret);
if ((fPacketizer = (TVirtualPacketizer *)ret) == 0) {
Error("InitPacketizer", "cannot construct '%s'", cl->GetName());
fExitStatus = kAborted;
return -1;
}
if (!fPacketizer->IsValid()) {
Error("InitPacketizer",
"instantiated packetizer object '%s' is invalid", cl->GetName());
fExitStatus = kAborted;
SafeDelete(fPacketizer);
return -1;
}
if (!noData) {
TDSetElement *elem = 0;
if (dset->TestBit(TDSet::kSomeInvalid)) {
TIter nxe(dset->GetListOfElements());
while ((elem = (TDSetElement *)nxe())) {
if (!elem->GetValid()) {
if (!listOfMissingFiles)
listOfMissingFiles = new TList;
listOfMissingFiles->Add(elem->GetFileInfo(dset->GetType()));
dset->Remove(elem, kFALSE);
}
}
dset->ResetBit(TDSet::kSomeInvalid);
}
if (listOfMissingFiles && listOfMissingFiles->GetSize() > 0) {
TIter missingFiles(listOfMissingFiles);
TString msg;
if (gDebug > 0) {
TFileInfo *fi = 0;
while ((fi = (TFileInfo *) missingFiles.Next())) {
if (fi->GetCurrentUrl()) {
msg = Form("File not found: %s - skipping!",
fi->GetCurrentUrl()->GetUrl());
} else {
msg = Form("File not found: %s - skipping!", fi->GetName());
}
if (gProofServ) gProofServ->SendAsynMessage(msg.Data());
}
}
if (!GetOutput("MissingFiles")) {
listOfMissingFiles->SetName("MissingFiles");
AddOutputObject(listOfMissingFiles);
}
TStatus *tmpStatus = (TStatus *)GetOutput("PROOF_Status");
if (!tmpStatus) {
tmpStatus = new TStatus();
AddOutputObject(tmpStatus);
}
Int_t ngood = dset->GetListOfElements()->GetSize();
Int_t nbad = listOfMissingFiles->GetSize();
Double_t xb = Double_t(nbad) / Double_t(ngood + nbad);
msg = Form(" About %.2f %c of the requested files (%d out of %d) were missing; details in"
" the 'missingFiles' list", xb * 100., '%', nbad, nbad + ngood);
tmpStatus->Add(msg.Data());
msg = Form(" +++\n"
" +++ About %.2f %c of the requested files (%d out of %d) are missing; details in"
" the 'MissingFiles' list\n"
" +++", xb * 100., '%', nbad, nbad + ngood);
if (gProofServ) gProofServ->SendAsynMessage(msg.Data());
} else {
SafeDelete(listOfMissingFiles);
}
}
return 0;
}
Long64_t TProofPlayerRemote::Process(TDSet *dset, const char *selector_file,
Option_t *option, Long64_t nentries,
Long64_t first)
{
PDB(kGlobal,1) Info("Process","Enter");
fDSet = dset;
fExitStatus = kFinished;
if (!fProgressStatus) {
Error("Process", "No progress status");
return -1;
}
fProgressStatus->Reset();
if (!fOutput)
fOutput = new TList;
else
fOutput->Clear();
SafeDelete(fFeedbackLists);
if (fProof->IsMaster()){
TPerfStats::Start(fInput, fOutput);
} else {
TPerfStats::Setup(fInput);
}
if(!SendSelector(selector_file)) return -1;
TMessage mesg(kPROOF_PROCESS);
TString fn(gSystem->BaseName(selector_file));
Bool_t sync = (fProof->GetQueryMode(option) == TProof::kSync);
TDSet *set = dset;
if (fProof->IsMaster()) {
PDB(kPacketizer,1) Info("Process","Create Proxy TDSet");
set = new TDSetProxy( dset->GetType(), dset->GetObjName(),
dset->GetDirectory() );
if (dset->TestBit(TDSet::kEmpty))
set->SetBit(TDSet::kEmpty);
const char *datapack = (fProof->IsLite()) ? "TPacketizer" : "TPacketizerAdaptive";
if (InitPacketizer(dset, nentries, first, "TPacketizerUnit", datapack) != 0) {
Error("Process", "cannot init the packetizer");
fExitStatus = kAborted;
return -1;
}
first = 0;
if (!fProof->GetParameter("PROOF_MemLogFreq")){
Long64_t memlogfreq = fPacketizer->GetTotalEntries()/(fProof->GetParallel()*100);
memlogfreq = (memlogfreq > 0) ? memlogfreq : 1;
fProof->SetParameter("PROOF_MemLogFreq", memlogfreq);
}
TString emsg;
if (TProof::SendInputData(fQuery, fProof, emsg) != 0)
Warning("Process", "could not forward input data: %s", emsg.Data());
} else {
if (fOutputLists) {
fOutputLists->Delete();
delete fOutputLists;
fOutputLists = 0;
}
if (!sync) {
gSystem->RedirectOutput(fProof->fLogFileName);
Printf(" ");
Info("Process","starting new query");
}
SafeDelete(fSelector);
fSelectorClass = 0;
if (!(fSelector = TSelector::GetSelector(selector_file))) {
if (!sync)
gSystem->RedirectOutput(0);
return -1;
}
fSelectorClass = fSelector->IsA();
fSelector->SetInputList(fInput);
fSelector->SetOption(option);
PDB(kLoop,1) Info("Process","Call Begin(0)");
fSelector->Begin(0);
fProof->SendInputDataFile();
if (!sync)
gSystem->RedirectOutput(0);
}
TCleanup clean(this);
SetupFeedback();
TString opt = option;
if (fProof->fProtocol < 13)
dset->SetWriteV3(kTRUE);
Long64_t num = (gProofServ && gProofServ->IsMaster() && gProofServ->IsParallel()) ? -1 : nentries;
Long64_t fst = (gProofServ && gProofServ->IsMaster() && gProofServ->IsParallel()) ? -1 : first;
TEntryList *enl = (!fProof->IsMaster()) ? dynamic_cast<TEntryList *>(set->GetEntryList())
: (TEntryList *)0;
TEventList *evl = (!fProof->IsMaster() && !enl) ? dynamic_cast<TEventList *>(set->GetEntryList())
: (TEventList *)0;
if (fProof->fProtocol > 14) {
mesg << set << fn << fInput << opt << num << fst << evl << sync << enl;
} else {
mesg << set << fn << fInput << opt << num << fst << evl << sync;
if (enl)
Warning("Process","entry lists not supported by the server");
}
fProof->ResetMergePrg();
PDB(kGlobal,1) Info("Process","Calling Broadcast");
fProof->Broadcast(mesg);
if (fProof->fProtocol < 13)
dset->SetWriteV3(kFALSE);
if (IsClient())
fProof->fRedirLog = kTRUE;
if (!IsClient()){
Info("Process|Svc", "Start merging Memory information");
}
if (!sync) {
if (IsClient()) {
PDB(kGlobal,1) Info("Process","Asynchronous processing:"
" activating CollectInputFrom");
fProof->Activate();
fProof->Collect();
return fProof->fSeqNum;
} else {
PDB(kGlobal,1) Info("Process","Calling Collect");
fProof->Collect();
HandleTimer(0);
StopFeedback();
return Finalize(kFALSE,sync);
}
} else {
PDB(kGlobal,1) Info("Process","Synchronous processing: calling Collect");
fProof->Collect();
if (!(fProof->IsSync())) {
Info("Process", "switching to the asynchronous mode ...");
return fProof->fSeqNum;
}
if (IsClient())
fProof->fRedirLog = kFALSE;
if (!IsClient()) {
HandleTimer(0);
if (fPacketizer && fQuery)
fQuery->SetProcessInfo(0, 0., fPacketizer->GetBytesRead(),
fPacketizer->GetInitTime(),
fPacketizer->GetProcTime());
}
StopFeedback();
if (!IsClient() || GetExitStatus() != TProofPlayer::kAborted)
return Finalize(kFALSE,sync);
else
return -1;
}
}
Bool_t TProofPlayerRemote::MergeOutputFiles()
{
if (fMergeFiles) {
TIter nxo(fOutput);
TObject *o = 0;
TProofOutputFile *pf = 0;
while ((o = nxo())) {
if ((pf = dynamic_cast<TProofOutputFile*>(o))) {
TFileMerger *filemerger = pf->GetFileMerger();
if (!filemerger) {
Error("MergeOutputFiles", "file merger is null in TProofOutputFile! Protocol error?");
pf->Print();
continue;
}
if (!filemerger->Merge()) {
Error("MergeOutputFiles", "cannot merge the output files");
continue;
}
TList *fileList = filemerger->GetMergeList();
if (fileList) {
TIter next(fileList);
TObjString *url = 0;
while((url = (TObjString*)next())) {
gSystem->Unlink(url->GetString());
}
}
filemerger->Reset();
}
}
}
return kTRUE;
}
Long64_t TProofPlayerRemote::Finalize(Bool_t force, Bool_t sync)
{
if (IsClient()) {
if (fOutputLists == 0) {
if (force)
if (fQuery)
return fProof->Finalize(Form("%s:%s", fQuery->GetTitle(),
fQuery->GetName()), force);
} else {
PDB(kGlobal,1) Info("Finalize","Calling Merge Output to finalize the output list");
MergeOutput();
}
}
Long64_t rv = 0;
if (fProof->IsMaster()) {
TPerfStats::Stop();
PDB(kOutput,1) Info("Finalize","Calling Merge Output");
MergeOutput();
MergeOutputFiles();
fOutput->SetOwner();
SafeDelete(fSelector);
} else {
if (fExitStatus != kAborted) {
if (!sync) {
if (ReinitSelector(fQuery) == -1) {
Info("Finalize", "problems reinitializing selector \"%s\"",
fQuery->GetSelecImp()->GetName());
return -1;
}
}
if (fPacketizer)
if (TList *failedPackets = fPacketizer->GetFailedPackets()) {
fPacketizer->SetFailedPackets(0);
failedPackets->SetName("FailedPackets");
AddOutputObject(failedPackets);
TStatus *status = (TStatus *)GetOutput("PROOF_Status");
if (!status) {
status = new TStatus();
AddOutputObject(status);
}
status->Add("Some packets were not processed! Check the the"
" 'FailedPackets' list in the output list");
}
fSelector->SetInputList(fInput);
TIter next(fOutput);
TList *output = fSelector->GetOutputList();
while(TObject* obj = next()) {
if (fProof->IsParallel() || DrawCanvas(obj) == 1)
output->Add(obj);
}
PDB(kLoop,1) Info("Finalize","Call Terminate()");
fOutput->Clear("nodelete");
fSelector->Terminate();
rv = fSelector->GetStatus();
TIter it(output);
while(TObject* o = it()) {
fOutput->Add(o);
}
if (fQuery) {
fQuery->SetOutputList(fOutput);
fQuery->SetFinalized();
} else {
Warning("Finalize","current TQueryResult object is undefined!");
}
output->SetOwner(kFALSE);
SafeDelete(fSelector);
fOutput->SetOwner(kFALSE);
SafeDelete(fOutput);
}
}
PDB(kGlobal,1) Info("Process","exit");
return rv;
}
Long64_t TProofPlayerRemote::Finalize(TQueryResult *qr)
{
PDB(kGlobal,1) Info("Finalize(TQueryResult *)","Enter");
if (!IsClient()) {
Info("Finalize(TQueryResult *)",
"method to be executed only on the clients");
return -1;
}
if (!qr) {
Info("Finalize(TQueryResult *)", "query undefined");
return -1;
}
if (qr->IsFinalized()) {
Info("Finalize(TQueryResult *)", "query already finalized");
return -1;
}
if (!fOutput)
fOutput = new TList;
else
fOutput->Clear();
if (fOutputLists) {
fOutputLists->Delete();
delete fOutputLists;
fOutputLists = 0;
}
gSystem->RedirectOutput(fProof->fLogFileName);
TList *tmp = (TList *) qr->GetOutputList();
if (!tmp) {
gSystem->RedirectOutput(0);
Info("Finalize(TQueryResult *)", "ouputlist is empty");
return -1;
}
TList *out = fOutput;
if (fProof->fProtocol < 11)
out = new TList;
TIter nxo(tmp);
TObject *o = 0;
while ((o = nxo()))
out->Add(o->Clone());
if (fProof->fProtocol < 11) {
out->SetOwner();
StoreOutput(out);
}
gSystem->RedirectOutput(0);
SetCurrentQuery(qr);
Long64_t rc = Finalize();
RestorePreviousQuery();
return rc;
}
Bool_t TProofPlayerRemote::SendSelector(const char* selector_file)
{
if (!selector_file) {
Info("SendSelector", "Invalid input: selector (file) name undefined");
return kFALSE;
}
if (!strchr(gSystem->BaseName(selector_file), '.')) {
if (gDebug > 1)
Info("SendSelector", "selector name '%s' does not contain a '.':"
" nothing to send, it will be loaded from a library", selector_file);
return kTRUE;
}
const char *cext[3] = { ".C", ".cxx", ".cc" };
Int_t e = 0;
for ( ; e < 3; e++)
if (strstr(selector_file, cext[e]))
break;
if (e >= 3) {
Info("SendSelector",
"Invalid extension: %s (supportd extensions: .C, .cxx, .cc", selector_file);
return kFALSE;
}
Int_t l = strlen(cext[e]);
TString selec = selector_file;
TString aclicMode;
TString arguments;
TString io;
selec = gSystem->SplitAclicMode(selec, aclicMode, arguments, io);
gSystem->ExpandPathName(selec);
TString mp(TROOT::GetMacroPath());
TString np(gSystem->DirName(selec));
if (!np.IsNull()) {
np += ":";
Int_t ip = (mp.BeginsWith(".:")) ? 2 : 0;
mp.Insert(ip, np);
}
TROOT::SetMacroPath(mp);
if (gDebug > 0)
Info("SendSelector", "macro path set to '%s'", TROOT::GetMacroPath());
TString header = selec;
header.Replace(header.Length()-l, l,".h");
if (gSystem->AccessPathName(header, kReadPermission)) {
TString h = header;
header = selec;
header.Replace(header.Length()-l, l,".hh");
if (gSystem->AccessPathName(header, kReadPermission)) {
Info("SendSelector",
"header file not found: tried: %s %s", h.Data(), header.Data());
return kFALSE;
}
}
if (fProof->SendFile(selec, (TProof::kBinary | TProof::kForward | TProof::kCp)) == -1) {
Info("SendSelector", "problems sending implementation file %s", selec.Data());
return kFALSE;
}
if (fProof->SendFile(header, (TProof::kBinary | TProof::kForward | TProof::kCp)) == -1) {
Info("SendSelector", "problems sending header file %s", header.Data());
return kFALSE;
}
return kTRUE;
}
void TProofPlayerRemote::MergeOutput()
{
PDB(kOutput,1) Info("MergeOutput","Enter");
if (fOutputLists == 0) {
PDB(kOutput,1) Info("MergeOutput","Leave (no output)");
return;
}
TIter next(fOutputLists);
TList *list;
while ( (list = (TList *) next()) ) {
TObject *obj = fOutput->FindObject(list->GetName());
if (obj == 0) {
obj = list->First();
list->Remove(obj);
fOutput->Add(obj);
}
if ( list->IsEmpty() ) continue;
TMethodCall callEnv;
if (obj->IsA())
callEnv.InitWithPrototype(obj->IsA(), "Merge", "TCollection*");
if (callEnv.IsValid()) {
callEnv.SetParam((Long_t) list);
callEnv.Execute(obj);
} else {
while ( (obj = list->First()) ) {
fOutput->Add(obj);
list->Remove(obj);
}
}
}
SafeDelete(fOutputLists);
PDB(kOutput,1) Info("MergeOutput","Leave (%d object(s))", fOutput->GetSize());
}
void TProofPlayerRemote::Progress(Long64_t total, Long64_t processed)
{
if (IsClient()) {
fProof->Progress(total, processed);
} else {
TMessage m(kPROOF_PROGRESS);
m << total << processed;
gProofServ->GetSocket()->Send(m);
}
}
void TProofPlayerRemote::Progress(Long64_t total, Long64_t processed,
Long64_t bytesread,
Float_t initTime, Float_t procTime,
Float_t evtrti, Float_t mbrti)
{
PDB(kGlobal,1)
Info("Progress","%lld %lld %lld %f %f %f %f", total, processed, bytesread,
initTime, procTime, evtrti, mbrti);
if (IsClient()) {
fProof->Progress(total, processed, bytesread, initTime, procTime, evtrti, mbrti);
} else {
TMessage m(kPROOF_PROGRESS);
m << total << processed << bytesread << initTime << procTime << evtrti << mbrti;
gProofServ->GetSocket()->Send(m);
}
}
void TProofPlayerRemote::Feedback(TList *objs)
{
fProof->Feedback(objs);
}
void TProofPlayerRemote::StopProcess(Bool_t abort, Int_t)
{
if (fPacketizer != 0)
fPacketizer->StopProcess(abort);
if (abort == kTRUE)
fExitStatus = kAborted;
else
fExitStatus = kStopped;
}
Int_t TProofPlayerRemote::AddOutputObject(TObject *obj)
{
PDB(kOutput,1)
Info("AddOutputObject","Enter: %p", obj);
if (!obj) {
PDB(kOutput,1) Info("AddOutputObject","Invalid input (obj == 0x0)");
return -1;
}
if (!fOutput)
fOutput = new TList;
Bool_t merged = kTRUE;
TList *elists = dynamic_cast<TList *> (obj);
if (elists && !strcmp(elists->GetName(), "PROOF_EventListsList")) {
TEventList *evlist = new TEventList("PROOF_EventList");
TIter nxevl(elists);
TEventList *evl = 0;
while ((evl = dynamic_cast<TEventList *> (nxevl()))) {
TIter nxelem(fDSet->GetListOfElements());
TDSetElement *elem = 0;
while ((elem = dynamic_cast<TDSetElement *> (nxelem()))) {
if (!strcmp(elem->GetFileName(), evl->GetName()))
break;
}
if (!elem) {
Error("AddOutputObject", "Found an event list for %s, but no object with"
" the same name in the TDSet", evl->GetName());
continue;
}
Long64_t offset = elem->GetTDSetOffset();
Long64_t *arr = evl->GetList();
Int_t num = evl->GetN();
if (arr && offset > 0)
for (Int_t i = 0; i < num; i++)
arr[i] += offset;
evlist->Add(evl);
}
Incorporate(evlist, fOutput, merged);
if (merged)
SafeDelete(evlist);
return 1;
}
TProofOutputFile *pf = dynamic_cast<TProofOutputFile*>(obj);
if (pf) {
if (!strcmp(pf->GetMode(),"CENTRAL"))
fMergeFiles = kTRUE;
if (!IsClient()) {
if (strlen(pf->GetOutputFileName()) <= 0) {
TString of(Form("root://%s", gSystem->HostName()));
if (gSystem->Getenv("XRDPORT")) {
TString sp(gSystem->Getenv("XRDPORT"));
if (sp.IsDigit())
of += Form(":%s", sp.Data());
}
TString sessionPath(gProofServ->GetSessionDir());
TString pfx = gEnv->GetValue("Path.Localroot","");
if (!pfx.IsNull())
sessionPath.Remove(0, pfx.Length());
of += Form("/%s/%s", sessionPath.Data(), pf->GetFileName());
pf->SetOutputFileName(of);
}
if (gDebug > 0)
pf->Print();
} else {
Printf("Output file: %s", pf->GetOutputFileName());
}
}
Incorporate(obj, fOutput, merged);
if (fProof && (!IsClient() || fProof->IsLite())){
ProcInfo_t pi;
if (!gSystem->GetProcInfo(&pi)){
RedirectOutput(fProof->IsLite());
Info("AddOutputObject|Svc", "Memory %ld virtual %ld resident after merging object %s",
pi.fMemVirtual, pi.fMemResident, obj->GetName());
RedirectOutput(0);
}
}
return (merged ? 1 : 0);
}
void TProofPlayerRemote::RedirectOutput(Bool_t on)
{
if (on && fProof && fProof->fLogFileW) {
TProofServ::SetErrorHandlerFile(fProof->fLogFileW);
fErrorHandler = SetErrorHandler(TProofServ::ErrorHandler);
} else if (!on) {
if (fErrorHandler) {
TProofServ::SetErrorHandlerFile(0);
SetErrorHandler(fErrorHandler);
}
}
}
void TProofPlayerRemote::AddOutput(TList *out)
{
PDB(kOutput,1) Info("AddOutput","Enter");
if (!out) {
PDB(kOutput,1) Info("AddOutput","Invalid input (out == 0x0)");
return;
}
if (!fOutput)
fOutput = new TList;
Bool_t merged = kTRUE;
TList *elists = dynamic_cast<TList *> (out->FindObject("PROOF_EventListsList"));
if (elists) {
TEventList *evlist = new TEventList("PROOF_EventList");
TIter nxevl(elists);
TEventList *evl = 0;
while ((evl = dynamic_cast<TEventList *> (nxevl()))) {
TIter nxelem(fDSet->GetListOfElements());
TDSetElement *elem = 0;
while ((elem = dynamic_cast<TDSetElement *> (nxelem()))) {
if (!strcmp(elem->GetFileName(), evl->GetName()))
break;
}
if (!elem) {
Error("AddOutput", "Found an event list for %s, but no object with"
" the same name in the TDSet", evl->GetName());
continue;
}
Long64_t offset = elem->GetTDSetOffset();
Long64_t *arr = evl->GetList();
Int_t num = evl->GetN();
if (arr && offset > 0)
for (Int_t i = 0; i < num; i++)
arr[i] += offset;
evlist->Add(evl);
}
out->Remove(elists);
delete elists;
Incorporate(evlist, fOutput, merged);
}
TIter nxo(out);
TObject *obj = 0;
while ((obj = nxo())) {
Incorporate(obj, fOutput, merged);
if (!merged)
out->Remove(obj);
}
return;
}
Int_t TProofPlayerRemote::Incorporate(TObject *newobj, TList *outlist, Bool_t &merged)
{
merged = kTRUE;
PDB(kOutput,1) Info("Incorporate", "enter: obj: %p, list: %p", newobj, outlist);
if (!newobj || !outlist) {
Error("Incorporate","Invalid inputs: obj: %p, list: %p", newobj, outlist);
return -1;
}
Bool_t specialH = !fProof->TestBit(TProof::kIsClient) || fProof->IsLite();
if (specialH && newobj->InheritsFrom("TH1")) {
if (!HandleHistogram(newobj)) {
PDB(kOutput,1) Info("Incorporate", "histogram object '%s' added to the"
" appropriate list for delayed merging", newobj->GetName());
merged = kFALSE;
return 0;
}
}
TObject *obj = outlist->FindObject(newobj->GetName());
if (!obj) {
outlist->Add(newobj);
merged = kFALSE;
return 0;
}
TMethodCall callEnv;
if (obj->IsA())
callEnv.InitWithPrototype(obj->IsA(), "Merge", "TCollection*");
if (callEnv.IsValid()) {
static TList *xlist = new TList;
xlist->Add(newobj);
callEnv.SetParam((Long_t) xlist);
callEnv.Execute(obj);
xlist->Clear();
} else {
outlist->Add(newobj);
merged = kFALSE;
}
return 0;
}
TObject *TProofPlayerRemote::HandleHistogram(TObject *obj)
{
TH1 *h = dynamic_cast<TH1 *>(obj);
if (!h) {
return obj;
}
Bool_t tobebinned = (h->GetBuffer()) ? kTRUE : kFALSE;
Int_t nent = h->GetBufferLength();
PDB(kOutput,2) Info("HandleHistogram", "h:%s ent:%d, buffer size: %d",
h->GetName(), nent, h->GetBufferSize());
TList *list = 0;
if (!fOutputLists) {
PDB(kOutput,2) Info("HandleHistogram", "Create fOutputLists");
fOutputLists = new TList;
fOutputLists->SetOwner();
}
list = (TList *) fOutputLists->FindObject(h->GetName());
TH1 *href = 0;
if (tobebinned) {
if (!list) {
list = new TList;
list->SetName(h->GetName());
list->SetOwner();
fOutputLists->Add(list);
if (fOutput && (href = (TH1 *) fOutput->FindObject(h->GetName()))) {
fOutput->Remove(href);
list->Add(href);
}
}
TIter nxh(list);
while ((href = (TH1 *) nxh())) {
if (href->GetBuffer() && href->GetBufferLength() < nent) break;
}
if (href) {
list->AddBefore(href, h);
} else {
list->Add(h);
}
return (TObject *)0;
} else {
if (list) {
TIter nxh(list);
while ((href = (TH1 *) nxh())) {
if (href->GetBuffer() || href->GetEntries() < nent) break;
}
if (href) {
list->AddBefore(href, h);
} else {
list->Add(h);
}
return (TObject *)0;
} else {
Int_t hsz = h->GetNbinsX() * h->GetNbinsY() * h->GetNbinsZ();
if (gProofServ && hsz > gProofServ->GetMsgSizeHWM()) {
return obj;
} else {
list = new TList;
list->SetName(h->GetName());
list->SetOwner();
fOutputLists->Add(list);
list->Add(h);
return (TObject *)0;
}
}
}
PDB(kOutput,1) Info("HandleHistogram", "Leaving");
}
void TProofPlayerRemote::StoreOutput(TList *out)
{
PDB(kOutput,1) Info("StoreOutput","Enter");
if ( out == 0 ) {
PDB(kOutput,1) Info("StoreOutput","Leave (empty)");
return;
}
TIter next(out);
out->SetOwner(kFALSE);
if (fOutputLists == 0) {
PDB(kOutput,2) Info("StoreOutput","Create fOutputLists");
fOutputLists = new TList;
fOutputLists->SetOwner();
}
TList* lists = dynamic_cast<TList*> (out->FindObject("PROOF_EventListsList"));
if (lists) {
out->Remove(lists);
TEventList *mainList = new TEventList("PROOF_EventList");
out->Add(mainList);
TIter it(lists);
TEventList *aList;
while ( (aList = dynamic_cast<TEventList*> (it())) ) {
TIter nxe(fDSet->GetListOfElements());
TDSetElement *elem;
while ( (elem = dynamic_cast<TDSetElement*> (nxe())) ) {
if (strcmp(elem->GetFileName(), aList->GetName()) == 0)
break;
}
if (!elem) {
Error("StoreOutput",Form("Found the EventList for %s, but no object with that name "
"in the TDSet", aList->GetName()));
continue;
}
Long64_t offset = elem->GetTDSetOffset();
Long64_t *arr = aList->GetList();
Int_t num = aList->GetN();
if (arr && offset)
for (int i = 0; i < num; i++)
arr[i] += offset;
mainList->Add(aList);
}
delete lists;
}
TObject *obj;
while( (obj = next()) ) {
PDB(kOutput,2) Info("StoreOutput","Find '%s'", obj->GetName() );
TList *list = (TList *) fOutputLists->FindObject( obj->GetName() );
if ( list == 0 ) {
PDB(kOutput,2) Info("StoreOutput","List not Found (creating)", obj->GetName() );
list = new TList;
list->SetName( obj->GetName() );
list->SetOwner();
fOutputLists->Add( list );
}
list->Add( obj );
}
delete out;
PDB(kOutput,1) Info("StoreOutput","Leave");
}
TList *TProofPlayerRemote::MergeFeedback()
{
PDB(kFeedback,1)
Info("MergeFeedback","Enter");
if ( fFeedbackLists == 0 ) {
PDB(kFeedback,1)
Info("MergeFeedback","Leave (no output)");
return 0;
}
TList *fb = new TList;
fb->SetOwner();
TIter next(fFeedbackLists);
TMap *map;
while ( (map = (TMap*) next()) ) {
PDB(kFeedback,2)
Info("MergeFeedback", "map %s size: %d", map->GetName(), map->GetSize());
TList *list = new TList;
TIter keys(map);
#ifndef R__TH1MERGEFIXED
Int_t nbmx = -1;
TObject *oref = 0;
#endif
while ( TObject *key = keys() ) {
TObject *o = map->GetValue(key);
TH1 *h = dynamic_cast<TH1 *>(o);
#ifndef R__TH1MERGEFIXED
if (h && !strncmp(o->GetName(),"PROOF_",6)) {
if (h->GetNbinsX() > nbmx) {
nbmx= h->GetNbinsX();
oref = o;
}
}
#endif
if (h) {
TIter nxh(list);
TH1 *href= 0;
while ((href = (TH1 *)nxh())) {
if (h->GetBuffer()) {
if (href->GetBuffer() && href->GetBufferLength() < h->GetBufferLength()) break;
} else {
if (href->GetBuffer() || href->GetEntries() < h->GetEntries()) break;
}
}
if (href) {
list->AddBefore(href, h);
} else {
list->Add(h);
}
} else {
list->Add(o);
}
}
#ifdef R__TH1MERGEFIXED
TObject *obj = list->First();
#else
TObject *obj = (oref) ? oref : list->First();
#endif
list->Remove(obj);
obj = obj->Clone();
fb->Add(obj);
if ( list->IsEmpty() ) {
delete list;
continue;
}
TMethodCall callEnv;
if (obj->IsA())
callEnv.InitWithPrototype(obj->IsA(), "Merge", "TCollection*");
if (callEnv.IsValid()) {
callEnv.SetParam((Long_t) list);
callEnv.Execute(obj);
} else {
while ( (obj = list->First()) ) {
fb->Add(obj->Clone());
list->Remove(obj);
}
}
delete list;
}
PDB(kFeedback,1)
Info("MergeFeedback","Leave (%d object(s))", fb->GetSize());
return fb;
}
void TProofPlayerRemote::StoreFeedback(TObject *slave, TList *out)
{
PDB(kFeedback,1)
Info("StoreFeedback","Enter");
if ( out == 0 ) {
PDB(kFeedback,1)
Info("StoreFeedback","Leave (empty)");
return;
}
if ( IsClient() ) {
Feedback(out);
delete out;
return;
}
if (fFeedbackLists == 0) {
PDB(kFeedback,2) Info("StoreFeedback","Create fFeedbackLists");
fFeedbackLists = new TList;
fFeedbackLists->SetOwner();
}
TIter next(out);
out->SetOwner(kFALSE);
const char *ord = ((TSlave*) slave)->GetOrdinal();
TObject *obj;
while( (obj = next()) ) {
PDB(kFeedback,2)
Info("StoreFeedback","%s: Find '%s'", ord, obj->GetName() );
TMap *map = (TMap*) fFeedbackLists->FindObject(obj->GetName());
if ( map == 0 ) {
PDB(kFeedback,2)
Info("StoreFeedback","%s: Map not Found (creating)", ord, obj->GetName() );
map = new TMap;
map->SetName(obj->GetName());
fFeedbackLists->Add(map);
} else {
PDB(kFeedback,2)
Info("StoreFeedback","%s: removing previous value", ord);
if (map->GetValue(slave))
delete map->GetValue(slave);
map->Remove(slave);
}
map->Add(slave, obj);
PDB(kFeedback,2)
Info("StoreFeedback","%s: %s, size: %d", ord, obj->GetName(), map->GetSize());
}
delete out;
PDB(kFeedback,1)
Info("StoreFeedback","Leave");
}
void TProofPlayerRemote::SetupFeedback()
{
if (IsClient()) return;
fFeedback = (TList*) fInput->FindObject("FeedbackList");
PDB(kFeedback,1) Info("SetupFeedback","\"FeedbackList\" %sfound",
fFeedback == 0 ? "NOT ":"");
if (fFeedback == 0 || fFeedback->GetSize() == 0) return;
SafeDelete(fFeedbackTimer);
fFeedbackPeriod = 2000;
TProof::GetParameter(fInput, "PROOF_FeedbackPeriod", fFeedbackPeriod);
fFeedbackTimer = new TTimer;
fFeedbackTimer->SetObject(this);
fFeedbackTimer->Start(fFeedbackPeriod, kTRUE);
}
void TProofPlayerRemote::StopFeedback()
{
if (fFeedbackTimer == 0) return;
PDB(kFeedback,1) Info("StopFeedback","Stop Timer");
SafeDelete(fFeedbackTimer);
}
Bool_t TProofPlayerRemote::HandleTimer(TTimer *)
{
PDB(kFeedback,2) Info("HandleTimer","Entry");
if (fFeedbackTimer == 0) return kFALSE;
TList *fb = new TList;
fb->SetOwner();
TIter next(fFeedback);
while( TObjString *name = (TObjString*) next() ) {
TObject *o = fOutput->FindObject(name->GetName());
if (o != 0) {
fb->Add(o->Clone());
TMap *m = 0;
if (fFeedbackLists &&
(m = (TMap *) fFeedbackLists->FindObject(name->GetName()))) {
fFeedbackLists->Remove(m);
m->DeleteValues();
delete m;
}
}
}
if (fb->GetSize() > 0) {
StoreFeedback(this, fb);
} else {
delete fb;
}
if (fFeedbackLists == 0) {
fFeedbackTimer->Start(fFeedbackPeriod, kTRUE);
return kFALSE;
}
fb = MergeFeedback();
PDB(kFeedback,2) Info("HandleTimer","Sending %d objects", fb->GetSize());
TMessage m(kPROOF_FEEDBACK);
m << fb;
gProofServ->GetSocket()->Send(m);
delete fb;
fFeedbackTimer->Start(fFeedbackPeriod, kTRUE);
return kFALSE;
}
TDSetElement *TProofPlayerRemote::GetNextPacket(TSlave *slave, TMessage *r)
{
TDSetElement *e = fPacketizer->GetNextPacket( slave, r );
if (e == 0) {
PDB(kPacketizer,2) Info("GetNextPacket","Done");
} else if (e == (TDSetElement*) -1) {
PDB(kPacketizer,2) Info("GetNextPacket","Waiting");
} else {
PDB(kPacketizer,2)
Info("GetNextPacket","To slave-%s (%s): '%s' '%s' '%s' %lld %lld",
slave->GetOrdinal(), slave->GetName(), e->GetFileName(),
e->GetDirectory(), e->GetObjName(), e->GetFirst(), e->GetNum());
}
return e;
}
Bool_t TProofPlayerRemote::IsClient() const
{
return fProof->TestBit(TProof::kIsClient);
}
Long64_t TProofPlayerRemote::DrawSelect(TDSet *set, const char *varexp,
const char *selection, Option_t *option,
Long64_t nentries, Long64_t firstentry)
{
TString selector, objname;
if (GetDrawArgs(varexp, selection, option, selector, objname) != 0) {
Error("DrawSelect", "parsing arguments");
return -1;
}
TNamed *varexpobj = new TNamed("varexp", varexp);
TNamed *selectionobj = new TNamed("selection", selection);
TObject *o = 0;
TList *savedInput = new TList;
TIter nxi(fInput);
while ((o = nxi()))
savedInput->Add(o);
TList *fb = (TList*) fInput->FindObject("FeedbackList");
if (fb) fInput->Remove(fb);
fInput->Clear();
if (fb) fInput->Add(fb);
fInput->Add(varexpobj);
fInput->Add(selectionobj);
if (objname == "")
objname = "htemp";
fProof->AddFeedback(objname);
Long64_t r = Process(set, selector, option, nentries, firstentry);
fProof->RemoveFeedback(objname);
fInput->Remove(varexpobj);
fInput->Remove(selectionobj);
if (TNamed *opt = dynamic_cast<TNamed*> (fInput->FindObject("PROOF_OPTIONS"))) {
fInput->Remove(opt);
delete opt;
}
delete varexpobj;
delete selectionobj;
fInput->Clear();
TIter nxsi(savedInput);
while ((o = nxsi()))
fInput->Add(o);
savedInput->SetOwner(kFALSE);
delete savedInput;
return r;
}
void TProofPlayerRemote::SetInitTime()
{
if (fPacketizer)
fPacketizer->SetInitTime();
}
ClassImp(TProofPlayerSlave)
void TProofPlayerSlave::SetupFeedback()
{
TList *fb = (TList*) fInput->FindObject("FeedbackList");
if (fb) {
PDB(kFeedback,1)
Info("SetupFeedback","\"FeedbackList\" found: %d objects", fb->GetSize());
} else {
PDB(kFeedback,1)
Info("SetupFeedback","\"FeedbackList\" NOT found");
}
if (fb == 0 || fb->GetSize() == 0) return;
SafeDelete(fFeedbackTimer);
fFeedbackPeriod = 2000;
TProof::GetParameter(fInput, "PROOF_FeedbackPeriod", fFeedbackPeriod);
fFeedbackTimer = new TTimer;
fFeedbackTimer->SetObject(this);
fFeedbackTimer->Start(fFeedbackPeriod, kTRUE);
fFeedback = fb;
}
void TProofPlayerSlave::StopFeedback()
{
if (fFeedbackTimer == 0) return;
PDB(kFeedback,1) Info("StopFeedback","Stop Timer");
SafeDelete(fFeedbackTimer);
}
Bool_t TProofPlayerSlave::HandleTimer(TTimer *)
{
PDB(kFeedback,2) Info("HandleTimer","Entry");
if (gProofServ && gProofServ->IsMaster() && !gProofServ->IsParallel()) {
TMessage m(kPROOF_PROGRESS);
m << fTotalEvents << GetEventsProcessed();
gProofServ->GetSocket()->Send(m);
}
if (fFeedback == 0) return kFALSE;
TList *fb = new TList;
fb->SetOwner(kFALSE);
if (fOutput == 0) {
fOutput = fSelector->GetOutputList();
}
if (fOutput) {
TIter next(fFeedback);
while( TObjString *name = (TObjString*) next() ) {
TObject *o = fOutput->FindObject(name->GetName());
if (o != 0) fb->Add(o);
}
}
PDB(kFeedback,2) Info("HandleTimer","Sending %d objects", fb->GetSize());
TMessage m(kPROOF_FEEDBACK);
m << fb;
gProofServ->GetSocket()->Send(m);
delete fb;
fFeedbackTimer->Start(fFeedbackPeriod, kTRUE);
return kFALSE;
}
void TProofPlayerSlave::HandleGetTreeHeader(TMessage *mess)
{
TMessage answ(kPROOF_GETTREEHEADER);
TDSet *dset;
(*mess) >> dset;
dset->Reset();
TDSetElement *e = dset->Next();
Long64_t entries = 0;
TFile *f = 0;
TTree *t = 0;
if (!e) {
PDB(kGlobal, 1) Info("HandleGetTreeHeader", "empty TDSet");
} else {
f = TFile::Open(e->GetFileName());
t = 0;
if (f) {
t = (TTree*) f->Get(e->GetObjName());
if (t) {
t->SetMaxVirtualSize(0);
t->DropBaskets();
entries = t->GetEntries();
while ((e = dset->Next()) != 0) {
TFile *f1 = TFile::Open(e->GetFileName());
if (f1) {
TTree *t1 = (TTree*) f1->Get(e->GetObjName());
if (t1) {
entries += t1->GetEntries();
delete t1;
}
delete f1;
}
}
t->SetMaxEntryLoop(entries);
}
}
}
if (t)
answ << TString("Success") << t;
else
answ << TString("Failed") << t;
fSocket->Send(answ);
SafeDelete(t);
SafeDelete(f);
}
ClassImp(TProofPlayerSuperMaster)
Long64_t TProofPlayerSuperMaster::Process(TDSet *dset, const char *selector_file,
Option_t *option, Long64_t nentries,
Long64_t first)
{
fProgressStatus->Reset();
PDB(kGlobal,1) Info("Process","Enter");
TProofSuperMaster *proof = dynamic_cast<TProofSuperMaster*>(GetProof());
if (!proof) return -1;
delete fOutput;
fOutput = new TList;
TPerfStats::Start(fInput, fOutput);
if (!SendSelector(selector_file)) {
Error("Process", "sending selector %s", selector_file);
return -1;
}
TCleanup clean(this);
SetupFeedback();
if (proof->IsMaster()) {
if (!dset->ElementsValid()) {
proof->ValidateDSet(dset);
if (!dset->ElementsValid()) {
Error("Process", "could not validate TDSet");
return -1;
}
}
TList msds;
msds.SetOwner();
TList keyholder;
keyholder.SetOwner();
TList valueholder;
valueholder.SetOwner();
TIter nextslave(proof->GetListOfActiveSlaves());
while (TSlave *sl = dynamic_cast<TSlave*>(nextslave())) {
TList *submasters = 0;
TPair *msd = dynamic_cast<TPair*>(msds.FindObject(sl->GetMsd()));
if (!msd) {
submasters = new TList;
submasters->SetName(sl->GetMsd());
keyholder.Add(submasters);
TList *setelements = new TSortedList(kSortDescending);
setelements->SetName(TString(sl->GetMsd())+"_Elements");
valueholder.Add(setelements);
msds.Add(new TPair(submasters, setelements));
} else {
submasters = dynamic_cast<TList*>(msd->Key());
}
submasters->Add(sl);
}
Long64_t cur = 0;
TIter nextelement(dset->GetListOfElements());
while (TDSetElement *elem = dynamic_cast<TDSetElement*>(nextelement())) {
if (elem->GetNum()<1) continue;
if (nentries !=-1 && cur>=first+nentries) {
break;
}
if (cur+elem->GetNum()-1<first) {
cur+=elem->GetNum();
continue;
}
if (cur<first) {
elem->SetNum(elem->GetNum()-(first-cur));
elem->SetFirst(elem->GetFirst()+first-cur);
cur=first;
}
if (nentries==-1 || cur+elem->GetNum()<=first+nentries) {
cur+=elem->GetNum();
} else {
elem->SetNum(first+nentries-cur);
cur=first+nentries;
}
TPair *msd = dynamic_cast<TPair*>(msds.FindObject(elem->GetMsd()));
if (!msd) {
Error("Process", "data requires mass storage domain '%s'"
" which is not accessible in this proof session",
elem->GetMsd());
return -1;
} else {
TList *elements = dynamic_cast<TList*>(msd->Value());
elements->Add(elem);
}
}
TList usedmasters;
TIter nextmsd(msds.MakeIterator());
while (TPair *msd = dynamic_cast<TPair*>(nextmsd())) {
TList *submasters = dynamic_cast<TList*>(msd->Key());
TList *setelements = dynamic_cast<TList*>(msd->Value());
Int_t nmasters = submasters->GetSize();
Int_t nelements = setelements->GetSize();
for (Int_t i=0; i<nmasters; i++) {
Long64_t nent = 0;
TDSet set(dset->GetType(), dset->GetObjName(),
dset->GetDirectory());
for (Int_t j = (i*nelements)/nmasters;
j < ((i+1)*nelements)/nmasters;
j++) {
TDSetElement *elem =
dynamic_cast<TDSetElement*>(setelements->At(j));
set.Add(elem->GetFileName(), elem->GetObjName(),
elem->GetDirectory(), elem->GetFirst(),
elem->GetNum(), elem->GetMsd());
nent += elem->GetNum();
}
if (set.GetListOfElements()->GetSize()>0) {
TMessage mesg(kPROOF_PROCESS);
TString fn(gSystem->BaseName(selector_file));
TString opt = option;
mesg << &set << fn << fInput << opt << Long64_t(-1) << Long64_t(0);
TSlave *sl = dynamic_cast<TSlave*>(submasters->At(i));
PDB(kGlobal,1) Info("Process",
"Sending TDSet with %d elements to submaster %s",
set.GetListOfElements()->GetSize(),
sl->GetOrdinal());
sl->GetSocket()->Send(mesg);
usedmasters.Add(sl);
fSlaves.AddLast(sl);
fSlaveProgress.Set(fSlaveProgress.GetSize()+1);
fSlaveProgress[fSlaveProgress.GetSize()-1] = 0;
fSlaveTotals.Set(fSlaveTotals.GetSize()+1);
fSlaveTotals[fSlaveTotals.GetSize()-1] = nent;
fSlaveBytesRead.Set(fSlaveBytesRead.GetSize()+1);
fSlaveBytesRead[fSlaveBytesRead.GetSize()-1] = 0;
fSlaveInitTime.Set(fSlaveInitTime.GetSize()+1);
fSlaveInitTime[fSlaveInitTime.GetSize()-1] = -1.;
fSlaveProcTime.Set(fSlaveProcTime.GetSize()+1);
fSlaveProcTime[fSlaveProcTime.GetSize()-1] = -1.;
fSlaveEvtRti.Set(fSlaveEvtRti.GetSize()+1);
fSlaveEvtRti[fSlaveEvtRti.GetSize()-1] = -1.;
fSlaveMBRti.Set(fSlaveMBRti.GetSize()+1);
fSlaveMBRti[fSlaveMBRti.GetSize()-1] = -1.;
}
}
}
if ( !IsClient() ) HandleTimer(0);
PDB(kGlobal,1) Info("Process","Calling Collect");
proof->Collect(&usedmasters);
HandleTimer(0);
}
StopFeedback();
PDB(kGlobal,1) Info("Process","Calling Merge Output");
MergeOutput();
TPerfStats::Stop();
return 0;
}
void TProofPlayerSuperMaster::Progress(TSlave *sl, Long64_t total, Long64_t processed)
{
Int_t idx = fSlaves.IndexOf(sl);
fSlaveProgress[idx] = processed;
if (fSlaveTotals[idx] != total)
Warning("Progress", "total events has changed for slave %s", sl->GetName());
fSlaveTotals[idx] = total;
Long64_t tot = 0;
Int_t i;
for (i = 0; i < fSlaveTotals.GetSize(); i++) tot += fSlaveTotals[i];
Long64_t proc = 0;
for (i = 0; i < fSlaveProgress.GetSize(); i++) proc += fSlaveProgress[i];
Progress(tot, proc);
}
void TProofPlayerSuperMaster::Progress(TSlave *sl, Long64_t total,
Long64_t processed, Long64_t bytesread,
Float_t initTime, Float_t procTime,
Float_t evtrti, Float_t mbrti)
{
PDB(kGlobal,2)
Info("Progress","%s: %lld %lld %f %f %f %f", sl->GetName(),
processed, bytesread, initTime, procTime, evtrti, mbrti);
Int_t idx = fSlaves.IndexOf(sl);
if (fSlaveTotals[idx] != total)
Warning("Progress", "total events has changed for slave %s", sl->GetName());
fSlaveTotals[idx] = total;
fSlaveProgress[idx] = processed;
fSlaveBytesRead[idx] = bytesread;
fSlaveInitTime[idx] = (initTime > -1.) ? initTime : fSlaveInitTime[idx];
fSlaveProcTime[idx] = (procTime > -1.) ? procTime : fSlaveProcTime[idx];
fSlaveEvtRti[idx] = (evtrti > -1.) ? evtrti : fSlaveEvtRti[idx];
fSlaveMBRti[idx] = (mbrti > -1.) ? mbrti : fSlaveMBRti[idx];
Int_t i;
Long64_t tot = 0;
Long64_t proc = 0;
Long64_t bytes = 0;
Float_t init = -1.;
Float_t ptime = -1.;
Float_t erti = 0.;
Float_t srti = 0.;
Int_t nerti = 0;
Int_t nsrti = 0;
for (i = 0; i < fSlaveTotals.GetSize(); i++) {
tot += fSlaveTotals[i];
if (i < fSlaveProgress.GetSize())
proc += fSlaveProgress[i];
if (i < fSlaveBytesRead.GetSize())
bytes += fSlaveBytesRead[i];
if (i < fSlaveInitTime.GetSize())
if (fSlaveInitTime[i] > -1. && (init < 0. || fSlaveInitTime[i] < init))
init = fSlaveInitTime[i];
if (i < fSlaveProcTime.GetSize())
if (fSlaveProcTime[i] > -1. && (ptime < 0. || fSlaveProcTime[i] > ptime))
ptime = fSlaveProcTime[i];
if (i < fSlaveEvtRti.GetSize())
if (fSlaveEvtRti[i] > -1.) {
erti += fSlaveEvtRti[i];
nerti++;
}
if (i < fSlaveMBRti.GetSize())
if (fSlaveMBRti[i] > -1.) {
srti += fSlaveMBRti[i];
nsrti++;
}
}
erti = (nerti > 0) ? erti / nerti : 0.;
srti = (nsrti > 0) ? srti / nerti : 0.;
Progress(tot, proc, bytes, init, ptime, erti, srti);
}
Bool_t TProofPlayerSuperMaster::HandleTimer(TTimer *)
{
if (fFeedbackTimer == 0) return kFALSE;
Int_t i;
Long64_t tot = 0;
Long64_t proc = 0;
Long64_t bytes = 0;
Float_t init = -1.;
Float_t ptime = -1.;
Float_t erti = 0.;
Float_t srti = 0.;
Int_t nerti = 0;
Int_t nsrti = 0;
for (i = 0; i < fSlaveTotals.GetSize(); i++) {
tot += fSlaveTotals[i];
if (i < fSlaveProgress.GetSize())
proc += fSlaveProgress[i];
if (i < fSlaveBytesRead.GetSize())
bytes += fSlaveBytesRead[i];
if (i < fSlaveInitTime.GetSize())
if (fSlaveInitTime[i] > -1. && (init < 0. || fSlaveInitTime[i] < init))
init = fSlaveInitTime[i];
if (i < fSlaveProcTime.GetSize())
if (fSlaveProcTime[i] > -1. && (ptime < 0. || fSlaveProcTime[i] > ptime))
ptime = fSlaveProcTime[i];
if (i < fSlaveEvtRti.GetSize())
if (fSlaveEvtRti[i] > -1.) {
erti += fSlaveEvtRti[i];
nerti++;
}
if (i < fSlaveMBRti.GetSize())
if (fSlaveMBRti[i] > -1.) {
srti += fSlaveMBRti[i];
nsrti++;
}
}
erti = (nerti > 0) ? erti / nerti : 0.;
srti = (nsrti > 0) ? srti / nerti : 0.;
TMessage m(kPROOF_PROGRESS);
m << tot << proc << bytes << init << ptime << erti << srti;
gProofServ->GetSocket()->Send(m);
if (fReturnFeedback)
return TProofPlayerRemote::HandleTimer(0);
else
return kFALSE;
}
void TProofPlayerSuperMaster::SetupFeedback()
{
if (IsClient()) return;
TProofPlayerRemote::SetupFeedback();
if (fFeedbackTimer) {
fReturnFeedback = kTRUE;
return;
} else {
fReturnFeedback = kFALSE;
}
SafeDelete(fFeedbackTimer);
fFeedbackPeriod = 2000;
TProof::GetParameter(fInput, "PROOF_FeedbackPeriod", fFeedbackPeriod);
fFeedbackTimer = new TTimer;
fFeedbackTimer->SetObject(this);
fFeedbackTimer->Start(fFeedbackPeriod, kTRUE);
}