Logo ROOT   6.30.04
Reference Guide
 All Namespaces Files Pages
TPerfStats.cxx
Go to the documentation of this file.
1 // @(#)root/proofplayer:$Id$
2 // Author: Kristjan Gulbrandsen 11/05/04
3 
4 /*************************************************************************
5  * Copyright (C) 1995-2004, Rene Brun and Fons Rademakers. *
6  * All rights reserved. *
7  * *
8  * For the licensing terms see $ROOTSYS/LICENSE. *
9  * For the list of contributors see $ROOTSYS/README/CREDITS. *
10  *************************************************************************/
11 
12 /** \class TPerfStats
13 \ingroup proofkernel
14 
15 Provides the interface for the PROOF internal performance measurement
16 and event tracing.
17 
18 */
19 
20 #include "TPerfStats.h"
21 
22 #include "Riostream.h"
23 #include "TCollection.h"
24 #include "TEnv.h"
25 #include "TError.h"
26 #include "TFile.h"
27 #include "TH1.h"
28 #include "TH2.h"
29 #include "TDSet.h"
30 #include "TProofDebug.h"
31 #include "TProof.h"
32 #include "TProofServ.h"
33 #include "TSlave.h"
34 #include "TStatus.h"
35 #include "TTree.h"
36 #include "TSQLServer.h"
37 #include "TSQLResult.h"
38 #include "TParameter.h"
39 #include "TPluginManager.h"
40 #include "TROOT.h"
41 #include "TTimeStamp.h"
42 #include "TProofMonSender.h"
43 
44 ClassImp(TPerfEvent);
45 ClassImp(TPerfStats);
46 
47 
48 //------------------------------------------------------------------------------
49 
50 ////////////////////////////////////////////////////////////////////////////////
51 /// Constructor
52 
53 TPerfEvent::TPerfEvent(TTimeStamp *offset)
54  : fEvtNode("-3"), fType(TVirtualPerfStats::kUnDefined), fSlave(),
55  fEventsProcessed(0), fBytesRead(0), fLen(0), fLatency(0.0), fProcTime(0.0), fCpuTime(0.0),
56  fIsStart(kFALSE), fIsOk(kFALSE)
57 {
58  if (gProofServ != 0) {
59  fEvtNode = gProofServ->GetOrdinal();
60  } else {
61  if (gProof && gProof->IsLite())
62  fEvtNode = "0";
63  else
64  fEvtNode = "-2"; // not on a PROOF server
65  }
66 
67  if (offset != 0) {
68  fTimeStamp = TTimeStamp(fTimeStamp.GetSec() - offset->GetSec(),
69  fTimeStamp.GetNanoSec() - offset->GetNanoSec());
70  }
71 }
72 
73 ////////////////////////////////////////////////////////////////////////////////
74 /// Compare method. Must return -1 if this is smaller than obj,
75 /// 0 if objects are equal and 1 if this is larger than obj.
76 
77 Int_t TPerfEvent::Compare(const TObject *obj) const
78 {
79  const TPerfEvent *pe = dynamic_cast<const TPerfEvent*>(obj);
80 
81  if (!pe) {
82  Error("Compare", "input is not a TPerfEvent object");
83  return 0;
84  }
85 
86  if (fTimeStamp < pe->fTimeStamp) {
87  return -1;
88  } else if (fTimeStamp == pe->fTimeStamp) {
89  return 0;
90  } else {
91  return 1;
92  }
93 }
94 
95 ////////////////////////////////////////////////////////////////////////////////
96 /// Dump content of this instance
97 
98 void TPerfEvent::Print(Option_t *) const
99 {
100  TString where;
101  if (fEvtNode == -2) {
102  where = "TPerfEvent: StandAlone ";
103  } else if ( fEvtNode == -1 ) {
104  where = "TPerfEvent: Master ";
105  } else {
106  where.Form("TPerfEvent: Worker %s ", fEvtNode.Data());
107  }
108  Printf("%s %s %f", where.Data(),
109  TVirtualPerfStats::EventType(fType), double(fTimeStamp));
110 }
111 
112 Long_t TPerfStats::fgVirtMemMax = -1;
113 Long_t TPerfStats::fgResMemMax = -1;
114 
115 ////////////////////////////////////////////////////////////////////////////////
116 /// Normal constructor.
117 
118 TPerfStats::TPerfStats(TList *input, TList *output)
119  : fTrace(0), fPerfEvent(0), fPacketsHist(0), fProcPcktHist(0),
120  fEventsHist(0), fNodeHist(0), fLatencyHist(0),
121  fProcTimeHist(0), fCpuTimeHist(0), fBytesRead(0),
122  fTotCpuTime(0.), fTotBytesRead(0), fTotEvents(0), fNumEvents(0),
123  fSlaves(0), fDoHist(kFALSE),
124  fDoTrace(kFALSE), fDoTraceRate(kFALSE), fDoSlaveTrace(kFALSE), fDoQuota(kFALSE),
125  fMonitorPerPacket(kFALSE), fMonSenders(3),
126  fDataSet("+++none+++"), fDataSetSize(-1), fOutput(output)
127 {
128  TProof *proof = (gProofServ) ? gProofServ->GetProof() : gProof;
129 
130  // Master flag
131  Bool_t isMaster = ((proof && proof->TestBit(TProof::kIsMaster)) ||
132  (gProofServ && gProofServ->IsMaster())) ? kTRUE : kFALSE;
133  Bool_t isEndMaster = ((gProofServ && gProofServ->IsEndMaster()) ||
134  (proof && proof->IsLite())) ? kTRUE : kFALSE;
135 
136  TList *l = 0;
137  Bool_t deletel = kFALSE;
138  TParameter<Int_t> *dyns = (TParameter<Int_t> *) input->FindObject("PROOF_DynamicStartup");
139  if (dyns) {
140  // When starring up dynamically the number of slots needs to be guessed from the
141  // maximum workers request. There is no way to change this later on.
142  Int_t nwrks = dyns->GetVal();
143  if (nwrks > 0) {
144  l = new TList;
145  for (Int_t i = 0; i < nwrks; i++) {
146  TSlaveInfo *wi = new TSlaveInfo(TString::Format("0.%d", i));
147  wi->SetStatus(TSlaveInfo::kActive);
148  l->Add(wi);
149  }
150  l->SetOwner(kTRUE);
151  deletel = kTRUE;
152  }
153  }
154  if (!l) l = proof ? proof->GetListOfSlaveInfos() : 0 ;
155 
156  TIter nextslaveinfo(l);
157  while (TSlaveInfo *si = dynamic_cast<TSlaveInfo*>(nextslaveinfo()))
158  if (si->fStatus == TSlaveInfo::kActive) fSlaves++;
159 
160  PDB(kMonitoring,1) Info("TPerfStats", "Statistics for %d slave(s)", fSlaves);
161 
162  fDoHist = (input->FindObject("PROOF_StatsHist") != 0);
163  fDoTrace = (input->FindObject("PROOF_StatsTrace") != 0);
164  fDoTraceRate = (input->FindObject("PROOF_RateTrace") != 0);
165  fDoSlaveTrace = (input->FindObject("PROOF_SlaveStatsTrace") != 0);
166  PDB(kMonitoring,1)
167  Info("TPerfStats", "master:%d hist:%d,trace:%d,rate:%d,wrktrace:%d",
168  isMaster, fDoHist, fDoTrace, fDoTraceRate, fDoSlaveTrace);
169 
170  // Check per packet monitoring
171  Int_t perpacket = -1;
172  if (TProof::GetParameter(input, "PROOF_MonitorPerPacket", perpacket) != 0) {
173  // Check if there is a global monitor-per-packet setting
174  perpacket = gEnv->GetValue("Proof.MonitorPerPacket", 0);
175  }
176  fMonitorPerPacket = (perpacket == 1) ? kTRUE : kFALSE;
177  if (fMonitorPerPacket)
178  Info("TPerfStats", "sending full information after each packet");
179 
180  // Extract the name of the dataset
181  TObject *o = 0;
182  TIter nxi(input);
183  while ((o = nxi()))
184  if (!strncmp(o->ClassName(), "TDSet", strlen("TDSet"))) break;
185  if (o) {
186  fDSet = (TDSet *) o;
187  fDataSetSize = fDSet->GetNumOfFiles();
188  if (fDataSetSize > 0) {
189  fDataSet = "";
190  TString grus = (gProofServ) ? TString::Format("/%s/%s/", gProofServ->GetGroup(),
191  gProofServ->GetUser()) : TString("");
192  TString dss = fDSet->GetName(), ds;
193  Ssiz_t fd = 0, nq = kNPOS;
194  while (dss.Tokenize(ds, fd, "[,| ]")) {
195  if ((nq = ds.Index("?")) != kNPOS) ds.Remove(nq);
196  ds.ReplaceAll(grus, "");
197  if (!fDataSet.IsNull()) fDataSet += ",";
198  fDataSet += ds;
199  }
200  }
201  }
202 
203  // Dataset string limited in length: get the authorized size
204  fDataSetLen = gEnv->GetValue("Proof.Monitor.DataSetLen", 512);
205  if (fDataSetLen != 512)
206  Info("TPerfStats", "dataset string length truncated to %d chars", fDataSetLen);
207  if (fDataSet.Length() > fDataSetLen) fDataSet.Resize(fDataSetLen);
208  //
209  PDB(kMonitoring,1)
210  Info("TPerfStats", "dataset: '%s', # files: %d", fDataSet.Data(), fDataSetSize);
211 
212  if ((isMaster && (fDoTrace || fDoTraceRate)) || (!isMaster && fDoSlaveTrace)) {
213  // Construct tree
214  gDirectory->RecursiveRemove(gDirectory->FindObject("PROOF_PerfStats"));
215  fTrace = new TTree("PROOF_PerfStats", "PROOF Statistics");
216  fTrace->SetDirectory(0);
217  fTrace->Bronch("PerfEvents", "TPerfEvent", &fPerfEvent, 64000, 0);
218  output->Add(fTrace);
219  PDB(kMonitoring,1)
220  Info("TPerfStats", "tree '%s' added to the output list", fTrace->GetName());
221  }
222 
223  if (fDoHist && isEndMaster) {
224  // Make Histograms
225  Double_t time_per_bin = 1e-3; // 10ms
226  Double_t min_time = 0;
227  Int_t ntime_bins = 1000;
228 
229  gDirectory->RecursiveRemove(gDirectory->FindObject("PROOF_PacketsHist"));
230  fPacketsHist = new TH1D("PROOF_PacketsHist", "Packets processed per Worker",
231  fSlaves, 0, fSlaves);
232  fPacketsHist->SetFillColor(kCyan);
233  fPacketsHist->SetDirectory(0);
234  fPacketsHist->SetMinimum(0);
235  output->Add(fPacketsHist);
236  PDB(kMonitoring,1)
237  Info("TPerfStats", "histo '%s' added to the output list", fPacketsHist->GetName());
238 
239  gDirectory->RecursiveRemove(gDirectory->FindObject("PROOF_ProcPcktHist"));
240  fProcPcktHist = new TH1I("PROOF_ProcPcktHist", "Packets being processed per Worker",
241  fSlaves, 0, fSlaves);
242  fProcPcktHist->SetFillColor(kRed);
243  fProcPcktHist->SetDirectory(0);
244  fProcPcktHist->SetMinimum(0);
245  output->Add(fProcPcktHist);
246  PDB(kMonitoring,1)
247  Info("TPerfStats", "histo '%s' added to the output list", fProcPcktHist->GetName());
248 
249  gDirectory->RecursiveRemove(gDirectory->FindObject("PROOF_EventsHist"));
250  fEventsHist = new TH1D("PROOF_EventsHist", "Events processed per Worker",
251  fSlaves, 0, fSlaves);
252  fEventsHist->SetFillColor(kGreen);
253  fEventsHist->SetDirectory(0);
254  fEventsHist->SetMinimum(0);
255  output->Add(fEventsHist);
256  PDB(kMonitoring,1)
257  Info("TPerfStats", "histo '%s' added to the output list", fEventsHist->GetName());
258 
259  gDirectory->RecursiveRemove(gDirectory->FindObject("PROOF_NodeHist"));
260  fNodeHist = new TH1D("PROOF_NodeHist", "Slaves per Fileserving Node",
261  fSlaves, 0, fSlaves);
262  fNodeHist->SetDirectory(0);
263  fNodeHist->SetMinimum(0);
264  fNodeHist->SetCanExtend(TH1::kAllAxes);
265  output->Add(fNodeHist);
266  PDB(kMonitoring,1)
267  Info("TPerfStats", "histo '%s' added to the output list", fNodeHist->GetName());
268 
269  gDirectory->RecursiveRemove(gDirectory->FindObject("PROOF_LatencyHist"));
270  fLatencyHist = new TH2D("PROOF_LatencyHist", "GetPacket Latency per Worker",
271  fSlaves, 0, fSlaves,
272  ntime_bins, min_time, time_per_bin);
273  fLatencyHist->SetDirectory(0);
274  fLatencyHist->SetMarkerStyle(4);
275  fLatencyHist->SetCanExtend(TH1::kAllAxes);
276  output->Add(fLatencyHist);
277  PDB(kMonitoring,1)
278  Info("TPerfStats", "histo '%s' added to the output list", fLatencyHist->GetName());
279 
280  gDirectory->RecursiveRemove(gDirectory->FindObject("PROOF_ProcTimeHist"));
281  fProcTimeHist = new TH2D("PROOF_ProcTimeHist", "Packet Processing Time per Worker",
282  fSlaves, 0, fSlaves,
283  ntime_bins, min_time, time_per_bin);
284  fProcTimeHist->SetDirectory(0);
285  fProcTimeHist->SetMarkerStyle(4);
286  fProcTimeHist->SetCanExtend(TH1::kAllAxes);
287  output->Add(fProcTimeHist);
288  PDB(kMonitoring,1)
289  Info("TPerfStats", "histo '%s' added to the output list", fProcTimeHist->GetName());
290 
291  gDirectory->RecursiveRemove(gDirectory->FindObject("PROOF_CpuTimeHist"));
292  fCpuTimeHist = new TH2D("PROOF_CpuTimeHist", "Packet CPU Time per Worker",
293  fSlaves, 0, fSlaves,
294  ntime_bins, min_time, time_per_bin);
295  fCpuTimeHist->SetDirectory(0);
296  fCpuTimeHist->SetMarkerStyle(4);
297  fCpuTimeHist->SetCanExtend(TH1::kAllAxes);
298  output->Add(fCpuTimeHist);
299  PDB(kMonitoring,1)
300  Info("TPerfStats", "histo '%s' added to the output list", fCpuTimeHist->GetName());
301 
302  nextslaveinfo.Reset();
303  Int_t slavebin=1;
304  while (TSlaveInfo *si = dynamic_cast<TSlaveInfo*>(nextslaveinfo())) {
305  if (si->fStatus == TSlaveInfo::kActive) {
306  fPacketsHist->GetXaxis()->SetBinLabel(slavebin, si->GetOrdinal());
307  fProcPcktHist->GetXaxis()->SetBinLabel(slavebin, si->GetOrdinal());
308  fEventsHist->GetXaxis()->SetBinLabel(slavebin, si->GetOrdinal());
309  fNodeHist->GetXaxis()->SetBinLabel(slavebin, si->GetOrdinal());
310  fLatencyHist->GetXaxis()->SetBinLabel(slavebin, si->GetOrdinal());
311  fProcTimeHist->GetXaxis()->SetBinLabel(slavebin, si->GetOrdinal());
312  fCpuTimeHist->GetXaxis()->SetBinLabel(slavebin, si->GetOrdinal());
313  slavebin++;
314  }
315  }
316  }
317  // Cleanup
318  if (deletel) delete(l);
319 
320  if (isMaster) {
321 
322  // Monitoring for query performances using monitoring system (e.g. Monalisa, SQL, ...)
323  //
324  // We support multiple specifications separated by ',' or '|' or '\' (the latter need
325  // top be escaped three times in the regular experession), e.g.
326  // ProofServ.Monitoring: Monalisa bla bla bla,
327  // +ProofServ.Monitoring: SQL blu blu blu
328 
329  TString mons = gEnv->GetValue("ProofServ.Monitoring", ""), mon;
330  Ssiz_t fmon = 0;
331  TProofMonSender *monSender = 0;
332  while (mons.Tokenize(mon, fmon, "[,|\\\\]")) {
333  if (mon != "") {
334  // Extract arguments (up to 9 'const char *')
335  TString a[10];
336  Int_t from = 0;
337  TString tok, sendopts;
338  Int_t na = 0;
339  while (mon.Tokenize(tok, from, " ")) {
340  if (tok.BeginsWith("sendopts:")) {
341  tok.ReplaceAll("sendopts:", "");
342  sendopts = tok;
343  } else {
344  a[na++] = tok;
345  }
346  }
347  na--;
348  // Get monitor object from the plugin manager
349  TPluginHandler *h = 0;
350  if ((h = gROOT->GetPluginManager()->FindHandler("TProofMonSender", a[0]))) {
351  if (h->LoadPlugin() != -1) {
352  monSender =
353  (TProofMonSender *) h->ExecPlugin(na, a[1].Data(), a[2].Data(), a[3].Data(),
354  a[4].Data(), a[5].Data(), a[6].Data(),
355  a[7].Data(), a[8].Data(), a[9].Data());
356  if (monSender && monSender->TestBit(TObject::kInvalidObject)) SafeDelete(monSender);
357  if (monSender && monSender->SetSendOptions(sendopts) != 0) SafeDelete(monSender);
358  }
359  }
360  }
361 
362  if (monSender) {
363  fMonSenders.Add(monSender);
364  PDB(kMonitoring,1)
365  Info("TPerfStats", "created monitoring object: %s - # of active monitors: %d",
366  mon.Data(), fMonSenders.GetEntries());
367  fDoQuota = kTRUE;
368  }
369  monSender = 0;
370  }
371  }
372 }
373 
374 ////////////////////////////////////////////////////////////////////////////////
375 /// Destructor
376 
377 TPerfStats::~TPerfStats()
378 {
379  // Shutdown the monitor writers, if any
380  fMonSenders.SetOwner(kTRUE);
381  fMonSenders.Delete();
382 }
383 
384 ////////////////////////////////////////////////////////////////////////////////
385 /// Simple event.
386 
387 void TPerfStats::SimpleEvent(EEventType type)
388 {
389  if (type == kStop && fPacketsHist != 0) {
390  fPacketsHist->LabelsDeflate("X");
391  fPacketsHist->LabelsOption("auv","X");
392  }
393 
394  if (type == kStop && fDoQuota)
395  WriteQueryLog();
396 
397  if (fTrace == 0) return;
398 
399  TPerfEvent pe(&fTzero);
400  pe.fType = type;
401 
402  fPerfEvent = &pe;
403  fTrace->SetBranchAddress("PerfEvents",&fPerfEvent);
404  fTrace->Fill();
405  fPerfEvent = 0;
406 }
407 
408 ////////////////////////////////////////////////////////////////////////////////
409 /// Packet event.
410 /// See WriteQueryLog for the descripition of the structure sent for monitoring
411 /// when fMonitorPerPacket is kTRUE.
412 
413 void TPerfStats::PacketEvent(const char *slave, const char* slavename, const char* filename,
414  Long64_t eventsprocessed, Double_t latency, Double_t proctime,
415  Double_t cputime, Long64_t bytesRead)
416 {
417  if (fDoTrace && fTrace != 0) {
418  TPerfEvent pe(&fTzero);
419 
420  pe.fType = kPacket;
421  pe.fSlaveName = slavename;
422  pe.fFileName = filename;
423  pe.fSlave = slave;
424  pe.fEventsProcessed = eventsprocessed;
425  pe.fBytesRead = bytesRead;
426  pe.fLatency = latency;
427  pe.fProcTime = proctime;
428  pe.fCpuTime = cputime;
429 
430  fPerfEvent = &pe;
431  fTrace->SetBranchAddress("PerfEvents",&fPerfEvent);
432  fTrace->Fill();
433  fPerfEvent = 0;
434  }
435 
436  PDB(kMonitoring,1)
437  Info("PacketEvent","%s: fDoHist: %d, fPacketsHist: %p, eventsprocessed: %lld",
438  slave, fDoHist, fPacketsHist, eventsprocessed);
439 
440  if (fDoHist && fPacketsHist != 0) {
441  fPacketsHist->Fill(slave, 1);
442  fEventsHist->Fill(slave, eventsprocessed);
443  fLatencyHist->Fill(slave, latency, 1);
444  fProcTimeHist->Fill(slave, proctime, 1);
445  fCpuTimeHist->Fill(slave, cputime, 1);
446  }
447 
448  if (fDoQuota) {
449  fTotCpuTime += cputime;
450  fTotBytesRead += bytesRead;
451  fTotEvents += eventsprocessed;
452  }
453 
454  // Write to monitoring system, if requested
455  if (!fMonSenders.IsEmpty() && fMonitorPerPacket) {
456  TQueryResult *qr = (gProofServ && gProofServ->GetProof()) ?
457  gProofServ->GetProof()->GetQueryResult() : 0;
458  if (!gProofServ || !gProofServ->GetSessionTag() || !gProofServ->GetProof() || !qr) {
459  Error("PacketEvent", "some required object are undefined (%p %p %p %p)",
460  gProofServ, (gProofServ ? gProofServ->GetSessionTag() : 0),
461  (gProofServ ? gProofServ->GetProof() : 0),
462  ((gProofServ && gProofServ->GetProof()) ? qr : 0));
463  return;
464  }
465 
466  TTimeStamp stop;
467  TString identifier;
468  identifier.Form("%s-q%d", gProofServ->GetSessionTag(), qr->GetSeqNum());
469 
470  TList values;
471  values.SetOwner();
472  values.Add(new TParameter<int>("id", 0));
473  values.Add(new TNamed("user", gProofServ->GetUser()));
474  values.Add(new TNamed("proofgroup", gProofServ->GetGroup()));
475  values.Add(new TNamed("begin", fTzero.AsString("s")));
476  values.Add(new TNamed("end", stop.AsString("s")));
477  values.Add(new TParameter<int>("walltime", stop.GetSec()-fTzero.GetSec()));
478  values.Add(new TParameter<Long64_t>("bytesread", fTotBytesRead));
479  values.Add(new TParameter<Long64_t>("events", fTotEvents));
480  values.Add(new TParameter<Long64_t>("totevents", fNumEvents));
481  values.Add(new TParameter<int>("workers", fSlaves));
482  values.Add(new TNamed("querytag", identifier.Data()));
483 
484  // Memory usage on workers
485  TStatus *pst = (fOutput) ? (TStatus *) fOutput->FindObject("PROOF_Status") : 0;
486  // This most likely will be always NULL when sending from GetNextPacket ...
487  Long64_t vmxw = (pst) ? (Long64_t) pst->GetVirtMemMax() : -1;
488  Long64_t rmxw = (pst) ? (Long64_t) pst->GetResMemMax() : -1;
489  values.Add(new TParameter<Long64_t>("vmemmxw", vmxw));
490  values.Add(new TParameter<Long64_t>("rmemmxw", rmxw));
491  // Memory usage on master
492  values.Add(new TParameter<Long64_t>("vmemmxm", (Long64_t) fgVirtMemMax));
493  values.Add(new TParameter<Long64_t>("rmemmxm", (Long64_t) fgResMemMax));
494  // Dataset information
495  values.Add(new TNamed("dataset", fDataSet.Data()));
496  values.Add(new TParameter<int>("numfiles", fDataSetSize));
497  // Missing files
498  TList *mfls = (fOutput) ? (TList *) fOutput->FindObject("MissingFiles") : 0;
499  Int_t nmiss = (mfls && mfls->GetSize() > 0) ? mfls->GetSize() : 0;
500  values.Add(new TParameter<int>("missfiles", nmiss));
501  // Query status
502  Int_t est = (pst) ? pst->GetExitStatus() : -1;
503  values.Add(new TParameter<int>("status", est));
504  // Root version
505  TString rver = TString::Format("%s|%s", gROOT->GetVersion(), gROOT->GetGitCommit());
506  values.Add(new TNamed("rootver", rver.Data()));
507 
508  for (Int_t i = 0; i < fMonSenders.GetEntries(); i++) {
509  TProofMonSender *m = (TProofMonSender *) fMonSenders[i];
510  if (m) {
511  // Send query summary
512  if (m->SendSummary(&values, identifier) != 0)
513  Error("PacketEvent", "sending of summary info failed (%s)", m->GetName());
514  } else {
515  Warning("PacketEvent", "undefined entry found in monitors array for id: %d", i);
516  }
517  }
518  }
519 }
520 
521 ////////////////////////////////////////////////////////////////////////////////
522 /// File event.
523 
524 void TPerfStats::FileEvent(const char *slave, const char *slavename, const char *nodename,
525  const char *filename, Bool_t isStart)
526 {
527  if (fDoTrace && fTrace != 0) {
528  TPerfEvent pe(&fTzero);
529 
530  pe.fType = kFile;
531  pe.fSlaveName = slavename;
532  pe.fNodeName = nodename;
533  pe.fFileName = filename;
534  pe.fSlave = slave;
535  pe.fIsStart = isStart;
536 
537  fPerfEvent = &pe;
538  fTrace->SetBranchAddress("PerfEvents",&fPerfEvent);
539  fTrace->Fill();
540  fPerfEvent = 0;
541  }
542 
543  if (fDoHist && fPacketsHist != 0) {
544  fNodeHist->Fill(nodename, isStart ? 1 : -1);
545  }
546 }
547 
548 ////////////////////////////////////////////////////////////////////////////////
549 /// Open file event.
550 
551 void TPerfStats::FileOpenEvent(TFile *file, const char *filename, Double_t start)
552 {
553  if (fDoTrace && fTrace != 0) {
554  TPerfEvent pe(&fTzero);
555 
556  pe.fType = kFileOpen;
557  pe.fFileName = filename;
558  pe.fFileClass = file != 0 ? file->ClassName() : "none";
559  pe.fProcTime = double(TTimeStamp())-start;
560  pe.fIsOk = (file != 0);
561 
562  fPerfEvent = &pe;
563  fTrace->SetBranchAddress("PerfEvents",&fPerfEvent);
564  fTrace->Fill();
565  fPerfEvent = 0;
566  }
567 }
568 
569 ////////////////////////////////////////////////////////////////////////////////
570 /// Read file event.
571 
572 void TPerfStats::FileReadEvent(TFile *file, Int_t len, Double_t start)
573 {
574  if (fDoTrace && fTrace != 0) {
575  TPerfEvent pe(&fTzero);
576 
577  pe.fType = kFileRead;
578  pe.fFileName = file->GetName();
579  pe.fFileClass = file->ClassName();
580  pe.fLen = len;
581  pe.fProcTime = double(TTimeStamp())-start;
582 
583  fPerfEvent = &pe;
584  fTrace->SetBranchAddress("PerfEvents",&fPerfEvent);
585  fTrace->Fill();
586  fPerfEvent = 0;
587  }
588 }
589 
590 ////////////////////////////////////////////////////////////////////////////////
591 /// Record TTree file unzip event.
592 /// start is the TimeStamp before unzip
593 /// pos is where in the file the compressed buffer came from
594 /// complen is the length of the compressed buffer
595 /// objlen is the length of the de-compressed buffer
596 
597 void TPerfStats::UnzipEvent(TObject * /* tree */, Long64_t /* pos */,
598  Double_t /* start */, Int_t /* complen */,
599  Int_t /* objlen */)
600 {
601  // Do nothing for now.
602 }
603 
604 ////////////////////////////////////////////////////////////////////////////////
605 /// Rate event.
606 
607 void TPerfStats::RateEvent(Double_t proctime, Double_t deltatime,
608  Long64_t eventsprocessed, Long64_t bytesRead)
609 {
610  if ((fDoTrace || fDoTraceRate) && fTrace != 0) {
611  TPerfEvent pe(&fTzero);
612 
613  pe.fType = kRate;
614  pe.fEventsProcessed = eventsprocessed;
615  pe.fBytesRead = bytesRead;
616  pe.fProcTime = proctime;
617  pe.fLatency = deltatime;
618 
619  fPerfEvent = &pe;
620  fTrace->SetBranchAddress("PerfEvents",&fPerfEvent);
621  fTrace->Fill();
622  fPerfEvent = 0;
623  }
624 }
625 
626 ////////////////////////////////////////////////////////////////////////////////
627 /// Set number of bytes read.
628 
629 void TPerfStats::SetBytesRead(Long64_t num)
630 {
631  fBytesRead = num;
632 }
633 
634 ////////////////////////////////////////////////////////////////////////////////
635 /// Get number of bytes read.
636 
637 Long64_t TPerfStats::GetBytesRead() const
638 {
639  return fBytesRead;
640 }
641 
642 ////////////////////////////////////////////////////////////////////////////////
643 /// Send to the connected monitoring servers information related to this query.
644 /// The information is of three types: 'summary', 'dataset' and 'files'.
645 /// Actual 'table' formatting is done by the relevant sender, implementation of
646 /// TProofMonSender, where the details are given.
647 
648 void TPerfStats::WriteQueryLog()
649 {
650  TTimeStamp stop;
651 
652  // Write to monitoring system
653  if (!fMonSenders.IsEmpty()) {
654  TQueryResult *qr = (gProofServ && gProofServ->GetProof()) ?
655  gProofServ->GetProof()->GetQueryResult() : 0;
656  if (!gProofServ || !gProofServ->GetSessionTag() || !gProofServ->GetProof() || !qr) {
657  Error("WriteQueryLog", "some required object are undefined (%p %p %p %p)",
658  gProofServ, (gProofServ ? gProofServ->GetSessionTag() : 0),
659  (gProofServ ? gProofServ->GetProof() : 0),
660  ((gProofServ && gProofServ->GetProof()) ? qr : 0));
661  return;
662  }
663 
664  TString identifier;
665  identifier.Form("%s-q%d", gProofServ->GetSessionTag(), qr->GetSeqNum());
666 
667  TList values;
668  values.SetOwner();
669  values.Add(new TParameter<int>("id", 0));
670  values.Add(new TNamed("user", gProofServ->GetUser()));
671  values.Add(new TNamed("proofgroup", gProofServ->GetGroup()));
672  values.Add(new TNamed("begin", fTzero.AsString("s")));
673  values.Add(new TNamed("end", stop.AsString("s")));
674  values.Add(new TParameter<int>("walltime", stop.GetSec()-fTzero.GetSec()));
675  values.Add(new TParameter<float>("cputime", fTotCpuTime));
676  values.Add(new TParameter<Long64_t>("bytesread", fTotBytesRead));
677  values.Add(new TParameter<Long64_t>("events", fTotEvents));
678  values.Add(new TParameter<Long64_t>("totevents", fTotEvents));
679  values.Add(new TParameter<int>("workers", fSlaves));
680  values.Add(new TNamed("querytag", identifier.Data()));
681 
682  TList *mfls = (fOutput) ? (TList *) fOutput->FindObject("MissingFiles") : 0;
683  // Memory usage on workers
684  TStatus *pst = (fOutput) ? (TStatus *) fOutput->FindObject("PROOF_Status") : 0;
685  Long64_t vmxw = (pst) ? (Long64_t) pst->GetVirtMemMax() : -1;
686  Long64_t rmxw = (pst) ? (Long64_t) pst->GetResMemMax() : -1;
687  values.Add(new TParameter<Long64_t>("vmemmxw", vmxw));
688  values.Add(new TParameter<Long64_t>("rmemmxw", rmxw));
689  // Memory usage on master
690  values.Add(new TParameter<Long64_t>("vmemmxm", (Long64_t) fgVirtMemMax));
691  values.Add(new TParameter<Long64_t>("rmemmxm", (Long64_t) fgResMemMax));
692  // Dataset information
693  values.Add(new TNamed("dataset", fDataSet.Data()));
694  values.Add(new TParameter<int>("numfiles", fDataSetSize));
695  // Missing files
696  Int_t nmiss = (mfls && mfls->GetSize() > 0) ? mfls->GetSize() : 0;
697  values.Add(new TParameter<int>("missfiles", nmiss));
698  // Query status
699  Int_t est = (pst) ? pst->GetExitStatus() : -1;
700  values.Add(new TParameter<int>("status", est));
701  // Root version
702  TString rver = TString::Format("%s|%s", gROOT->GetVersion(), gROOT->GetGitCommit());
703  values.Add(new TNamed("rootver", rver.Data()));
704 
705  for (Int_t i = 0; i < fMonSenders.GetEntries(); i++) {
706  TProofMonSender *m = (TProofMonSender *) fMonSenders[i];
707  if (m) {
708  // Send query summary
709  if (m->SendSummary(&values, identifier) != 0)
710  Error("WriteQueryLog", "sending of summary info failed (%s)", m->GetName());
711  // Send dataset information
712  if (m->SendDataSetInfo(fDSet, mfls, fTzero.AsString("s"), identifier) != 0)
713  Error("WriteQueryLog", "sending of dataset info failed (%s)", m->GetName());
714  // Send file information
715  if (m->SendFileInfo(fDSet, mfls, fTzero.AsString("s"), identifier) != 0)
716  Error("WriteQueryLog", "sending of files info failed (%s)", m->GetName());
717  } else {
718  Warning("WriteQueryLog", "undefined entry found in monitors array for id: %d", i);
719  }
720  }
721  }
722 }
723 
724 ////////////////////////////////////////////////////////////////////////////////
725 /// Setup the PROOF input list with requested statistics and tracing options.
726 
727 void TPerfStats::Setup(TList *input)
728 {
729  const Int_t ntags=3;
730  const char *tags[ntags] = {"StatsHist", "StatsTrace", "SlaveStatsTrace"};
731 
732  TString varname, parname;
733  for (Int_t i=0; i<ntags; i++) {
734  varname.Form("Proof.%s", tags[i]);
735  parname.Form("PROOF_%s", tags[i]);
736  if (!input->FindObject(parname))
737  if (gEnv->GetValue(varname, 0)) input->Add(new TNamed(parname.Data(),""));
738  }
739 }
740 
741 ////////////////////////////////////////////////////////////////////////////////
742 /// Initialize PROOF statistics run.
743 
744 void TPerfStats::Start(TList *input, TList *output)
745 {
746  if (gPerfStats)
747  delete gPerfStats;
748  fgVirtMemMax = -1;
749  fgResMemMax = -1;
750  TPerfStats::SetMemValues();
751 
752  gPerfStats = new TPerfStats(input, output);
753  if (gPerfStats && !gPerfStats->TestBit(TObject::kInvalidObject)) {
754  // This measures the time taken by the constructor: not negligeable ...
755  gPerfStats->SimpleEvent(TVirtualPerfStats::kStart);
756  } else {
757  SafeDelete(gPerfStats);
758  }
759 }
760 
761 ////////////////////////////////////////////////////////////////////////////////
762 /// Terminate the PROOF statistics run.
763 
764 void TPerfStats::Stop()
765 {
766  if (!gPerfStats) return;
767 
768  TPerfStats::SetMemValues();
769  gPerfStats->SimpleEvent(TVirtualPerfStats::kStop);
770 
771  delete gPerfStats;
772  gPerfStats = 0;
773 }
774 
775 ////////////////////////////////////////////////////////////////////////////////
776 /// Record memory usage
777 
778 void TPerfStats::SetMemValues()
779 {
780  ProcInfo_t pi;
781  if (!gSystem->GetProcInfo(&pi)){
782  if (pi.fMemVirtual > fgVirtMemMax) fgVirtMemMax = pi.fMemVirtual;
783  if (pi.fMemResident > fgResMemMax) fgResMemMax = pi.fMemResident;
784  }
785 }
786 
787 ////////////////////////////////////////////////////////////////////////////////
788 /// Get memory usage
789 
790 void TPerfStats::GetMemValues(Long_t &vmax, Long_t &rmax)
791 {
792  vmax = fgVirtMemMax;
793  rmax = fgResMemMax;
794 }