#include "TPerfStats.h"
#include "Riostream.h"
#include "TCollection.h"
#include "TEnv.h"
#include "TError.h"
#include "TFile.h"
#include "TH1.h"
#include "TH2.h"
#include "TProofDebug.h"
#include "TProof.h"
#include "TProofServ.h"
#include "TSlave.h"
#include "TTree.h"
#include "TSQLServer.h"
#include "TSQLResult.h"
ClassImp(TPerfEvent)
ClassImp(TPerfStats)
TPerfEvent::TPerfEvent(TTimeStamp *offset)
: fEvtNode("-3"), fType(TVirtualPerfStats::kUnDefined), fSlave(),
fEventsProcessed(0), fBytesRead(0), fLen(0), fLatency(0.0), fProcTime(0.0), fCpuTime(0.0),
fIsStart(kFALSE), fIsOk(kFALSE)
{
if (gProofServ != 0) {
fEvtNode = gProofServ->GetOrdinal();
} else {
fEvtNode = "-2";
}
if (offset != 0) {
fTimeStamp = TTimeStamp(fTimeStamp.GetSec() - offset->GetSec(),
fTimeStamp.GetNanoSec() - offset->GetNanoSec());
}
}
Int_t TPerfEvent::Compare(const TObject *obj) const
{
const TPerfEvent *pe = dynamic_cast<const TPerfEvent*>(obj);
R__ASSERT(pe != 0);
if (fTimeStamp < pe->fTimeStamp) {
return -1;
} else if (fTimeStamp == pe->fTimeStamp) {
return 0;
} else {
return 1;
}
}
void TPerfEvent::Print(Option_t *) const
{
cout << "TPerfEvent: ";
if ( fEvtNode == -2 ) {
cout << "StandAlone ";
} else if ( fEvtNode == -1 ) {
cout << "Master ";
} else {
cout << "Slave " << fEvtNode << " ";
}
cout << TVirtualPerfStats::EventType(fType) << " "
<< double(fTimeStamp)
<< endl;
}
TPerfStats::TPerfStats(TList *input, TList *output)
: fTrace(0), fPerfEvent(0), fPacketsHist(0), fEventsHist(0), fLatencyHist(0),
fProcTimeHist(0), fCpuTimeHist(0), fBytesRead(0),
fTotCpuTime(0.), fTotBytesRead(0), fTotEvents(0), fSlaves(0), fDoHist(kFALSE),
fDoTrace(kFALSE), fDoTraceRate(kFALSE), fDoSlaveTrace(kFALSE), fDoQuota(kFALSE)
{
TProof *proof = gProofServ->GetProof();
TList *l = proof ? proof->GetListOfSlaveInfos() : 0 ;
TIter nextslaveinfo(l);
while (TSlaveInfo *si = dynamic_cast<TSlaveInfo*>(nextslaveinfo()))
if (si->fStatus == TSlaveInfo::kActive) fSlaves++;
PDB(kGlobal,1) Info("TPerfStats", "Statistics for %d slave(s)", fSlaves);
fDoHist = (input->FindObject("PROOF_StatsHist") != 0);
fDoTrace = (input->FindObject("PROOF_StatsTrace") != 0);
fDoTraceRate = (input->FindObject("PROOF_RateTrace") != 0);
fDoSlaveTrace = (input->FindObject("PROOF_SlaveStatsTrace") != 0);
if ((gProofServ->IsMaster() && (fDoTrace || fDoTraceRate)) ||
(!gProofServ->IsMaster() && fDoSlaveTrace)) {
fTrace = new TTree("PROOF_PerfStats", "PROOF Statistics");
fTrace->SetDirectory(0);
fTrace->Bronch("PerfEvents", "TPerfEvent", &fPerfEvent, 64000, 0);
output->Add(fTrace);
}
if (fDoHist && gProofServ->IsMaster()) {
Double_t time_per_bin = 1e-3;
Double_t min_time = 0;
Int_t ntime_bins = 1000;
fPacketsHist = new TH1D("PROOF_PacketsHist", "Packets processed per Slave",
fSlaves, 0, fSlaves);
fPacketsHist->SetDirectory(0);
fPacketsHist->SetMinimum(0);
output->Add(fPacketsHist);
fEventsHist = new TH1D("PROOF_EventsHist", "Events processed per Slave",
fSlaves, 0, fSlaves);
fEventsHist->SetDirectory(0);
fEventsHist->SetMinimum(0);
output->Add(fEventsHist);
fNodeHist = new TH1D("PROOF_NodeHist", "Slaves per Fileserving Node",
fSlaves, 0, fSlaves);
fNodeHist->SetDirectory(0);
fNodeHist->SetMinimum(0);
fNodeHist->SetBit(TH1::kCanRebin);
output->Add(fNodeHist);
fLatencyHist = new TH2D("PROOF_LatencyHist", "GetPacket Latency per Slave",
fSlaves, 0, fSlaves,
ntime_bins, min_time, time_per_bin);
fLatencyHist->SetDirectory(0);
fLatencyHist->SetMarkerStyle(4);
fLatencyHist->SetBit(TH1::kCanRebin);
output->Add(fLatencyHist);
fProcTimeHist = new TH2D("PROOF_ProcTimeHist", "Packet Processing Time per Slave",
fSlaves, 0, fSlaves,
ntime_bins, min_time, time_per_bin);
fProcTimeHist->SetDirectory(0);
fProcTimeHist->SetMarkerStyle(4);
fProcTimeHist->SetBit(TH1::kCanRebin);
output->Add(fProcTimeHist);
fCpuTimeHist = new TH2D("PROOF_CpuTimeHist", "Packet CPU Time per Slave",
fSlaves, 0, fSlaves,
ntime_bins, min_time, time_per_bin);
fCpuTimeHist->SetDirectory(0);
fCpuTimeHist->SetMarkerStyle(4);
fCpuTimeHist->SetBit(TH1::kCanRebin);
output->Add(fCpuTimeHist);
nextslaveinfo.Reset();
Int_t slavebin=1;
while (TSlaveInfo *si = dynamic_cast<TSlaveInfo*>(nextslaveinfo())) {
if (si->fStatus == TSlaveInfo::kActive) {
fPacketsHist->GetXaxis()->SetBinLabel(slavebin, si->GetOrdinal());
fEventsHist->GetXaxis()->SetBinLabel(slavebin, si->GetOrdinal());
fLatencyHist->GetXaxis()->SetBinLabel(slavebin, si->GetOrdinal());
fProcTimeHist->GetXaxis()->SetBinLabel(slavebin, si->GetOrdinal());
fCpuTimeHist->GetXaxis()->SetBinLabel(slavebin, si->GetOrdinal());
slavebin++;
}
}
}
if (gProofServ->IsMaster()) {
TString sqlserv = gEnv->GetValue("ProofServ.QueryLogDB","");
if (sqlserv != "")
fDoQuota = kTRUE;
}
}
void TPerfStats::SimpleEvent(EEventType type)
{
if (type == kStop && fPacketsHist != 0) {
fNodeHist->LabelsDeflate("X");
fNodeHist->LabelsOption("auv","X");
}
if (type == kStop && fDoQuota)
WriteQueryLog();
if (fTrace == 0) return;
TPerfEvent pe(&fTzero);
pe.fType = type;
fPerfEvent = &pe;
fTrace->SetBranchAddress("PerfEvents",&fPerfEvent);
fTrace->Fill();
fPerfEvent = 0;
}
void TPerfStats::PacketEvent(const char *slave, const char* slavename, const char* filename,
Long64_t eventsprocessed, Double_t latency, Double_t proctime,
Double_t cputime, Long64_t bytesRead)
{
if (fDoTrace && fTrace != 0) {
TPerfEvent pe(&fTzero);
pe.fType = kPacket;
pe.fSlaveName = slavename;
pe.fFileName = filename;
pe.fSlave = slave;
pe.fEventsProcessed = eventsprocessed;
pe.fBytesRead = bytesRead;
pe.fLatency = latency;
pe.fProcTime = proctime;
pe.fCpuTime = cputime;
fPerfEvent = &pe;
fTrace->SetBranchAddress("PerfEvents",&fPerfEvent);
fTrace->Fill();
fPerfEvent = 0;
}
if (fDoHist && fPacketsHist != 0) {
fPacketsHist->Fill(slave, 1);
fEventsHist->Fill(slave, eventsprocessed);
fLatencyHist->Fill(slave, latency, 1);
fProcTimeHist->Fill(slave, proctime, 1);
fCpuTimeHist->Fill(slave, cputime, 1);
}
if (fDoQuota) {
fTotCpuTime += cputime;
fTotBytesRead += bytesRead;
fTotEvents += eventsprocessed;
}
}
void TPerfStats::FileEvent(const char *slave, const char *slavename, const char *nodename,
const char *filename, Bool_t isStart)
{
if (fDoTrace && fTrace != 0) {
TPerfEvent pe(&fTzero);
pe.fType = kFile;
pe.fSlaveName = slavename;
pe.fNodeName = nodename;
pe.fFileName = filename;
pe.fSlave = slave;
pe.fIsStart = isStart;
fPerfEvent = &pe;
fTrace->SetBranchAddress("PerfEvents",&fPerfEvent);
fTrace->Fill();
fPerfEvent = 0;
}
if (fDoHist && fPacketsHist != 0) {
fNodeHist->Fill(nodename, isStart ? 1 : -1);
}
}
void TPerfStats::FileOpenEvent(TFile *file, const char *filename, Double_t proctime)
{
if (fDoTrace && fTrace != 0) {
TPerfEvent pe(&fTzero);
pe.fType = kFileOpen;
pe.fFileName = filename;
pe.fFileClass = file != 0 ? file->ClassName() : "none";
pe.fProcTime = proctime;
pe.fIsOk = (file != 0);
fPerfEvent = &pe;
fTrace->SetBranchAddress("PerfEvents",&fPerfEvent);
fTrace->Fill();
fPerfEvent = 0;
}
}
void TPerfStats::FileReadEvent(TFile *file, Int_t len, Double_t proctime)
{
if (fDoTrace && fTrace != 0) {
TPerfEvent pe(&fTzero);
pe.fType = kFileRead;
pe.fFileName = file->GetName();
pe.fFileClass = file->ClassName();
pe.fLen = len;
pe.fProcTime = proctime;
fPerfEvent = &pe;
fTrace->SetBranchAddress("PerfEvents",&fPerfEvent);
fTrace->Fill();
fPerfEvent = 0;
}
}
void TPerfStats::RateEvent(Double_t proctime, Double_t deltatime,
Long64_t eventsprocessed, Long64_t bytesRead)
{
if ((fDoTrace || fDoTraceRate) && fTrace != 0) {
TPerfEvent pe(&fTzero);
pe.fType = kRate;
pe.fEventsProcessed = eventsprocessed;
pe.fBytesRead = bytesRead;
pe.fProcTime = proctime;
pe.fLatency = deltatime;
fPerfEvent = &pe;
fTrace->SetBranchAddress("PerfEvents",&fPerfEvent);
fTrace->Fill();
fPerfEvent = 0;
}
}
void TPerfStats::SetBytesRead(Long64_t num)
{
fBytesRead = num;
}
Long64_t TPerfStats::GetBytesRead() const
{
return fBytesRead;
}
void TPerfStats::WriteQueryLog()
{
TTimeStamp stop;
TString sql;
sql.Form("INSERT INTO proofquerylog VALUES (0, '%s', '%s', "
"'%s', '%s', %d, %.2f, %lld, %lld, %d)",
gProofServ->GetUser(), gProofServ->GetGroup(),
fTzero.AsString("s"), stop.AsString("s"),
stop.GetSec()-fTzero.GetSec(), fTotCpuTime,
fTotBytesRead, fTotEvents, fSlaves);
TString sqlserv = gEnv->GetValue("ProofServ.QueryLogDB","");
TString sqluser = gEnv->GetValue("ProofServ.QueryLogUser","");
TString sqlpass = gEnv->GetValue("ProofServ.QueryLogPasswd","");
TSQLServer *db = TSQLServer::Connect(sqlserv, sqluser, sqlpass);
if (!db || db->IsZombie()) {
Error("WriteQueryLog", "failed to connect to SQL server %s as %s %s",
sqlserv.Data(), sqluser.Data(), sqlpass.Data());
printf("%s\n", sql.Data());
} else {
TSQLResult *res = db->Query(sql);
if (!res) {
Error("WriteQueryLog", "insert into proofquerylog failed");
printf("%s\n", sql.Data());
}
delete res;
}
delete db;
}
void TPerfStats::Setup(TList *input)
{
const Int_t ntags=3;
const Char_t *tags[ntags] = {"StatsHist",
"StatsTrace",
"SlaveStatsTrace"};
for (Int_t i=0; i<ntags; i++) {
TString envvar = "Proof.";
envvar += tags[i];
TString inputname = "PROOF_";
inputname += tags[i];
TObject* obj = input->FindObject(inputname.Data());
if (gEnv->GetValue(envvar.Data(), 0)) {
if (!obj)
input->Add(new TNamed(inputname.Data(),""));
} else {
if (obj) {
input->Remove(obj);
delete obj;
}
}
}
}
void TPerfStats::Start(TList *input, TList *output)
{
if (gPerfStats != 0) {
delete gPerfStats;
}
gPerfStats = new TPerfStats(input, output);
gPerfStats->SimpleEvent(TVirtualPerfStats::kStart);
}
void TPerfStats::Stop()
{
if (gPerfStats == 0) return;
gPerfStats->SimpleEvent(TVirtualPerfStats::kStop);
delete gPerfStats;
gPerfStats = 0;
}
This page has been automatically generated. If you have any comments or suggestions about the page layout send a mail to ROOT support, or contact the developers with any questions or problems regarding ROOT.