#ifdef HAVE_CONFIG
#include "config.h"
#endif
#include "RConfig.h"
#include "Riostream.h"
#ifdef WIN32
#include <io.h>
typedef long off_t;
#endif
#include <errno.h>
#include <time.h>
#include <fcntl.h>
#include <sys/types.h>
#include <sys/stat.h>
#if (defined(__FreeBSD__) && (__FreeBSD__ < 4)) || \
(defined(__APPLE__) && (!defined(MAC_OS_X_VERSION_10_3) || \
(MAC_OS_X_VERSION_MAX_ALLOWED < MAC_OS_X_VERSION_10_3)))
#include <sys/file.h>
#define lockf(fd, op, sz) flock((fd), (op))
#ifndef F_LOCK
#define F_LOCK (LOCK_EX | LOCK_NB)
#endif
#ifndef F_ULOCK
#define F_ULOCK LOCK_UN
#endif
#endif
#include "TProofServ.h"
#include "TAuthenticate.h"
#include "TDSetProxy.h"
#include "TEnv.h"
#include "TError.h"
#include "TException.h"
#include "TFile.h"
#include "TInterpreter.h"
#include "TKey.h"
#include "TMessage.h"
#include "TPerfStats.h"
#include "TProofDebug.h"
#include "TProof.h"
#include "TProofLimitsFinder.h"
#include "TProofPlayer.h"
#include "TProofQueryResult.h"
#include "TRegexp.h"
#include "TROOT.h"
#include "TSelector.h"
#include "TSocket.h"
#include "TStopwatch.h"
#include "TSysEvtHandler.h"
#include "TSystem.h"
#include "TTimeStamp.h"
#include "TUrl.h"
#include "TTree.h"
#include "TPluginManager.h"
#include "TObjString.h"
#include "compiledata.h"
#include "TProofResourcesStatic.h"
#include "TProofNodeInfo.h"
#include "TFileInfo.h"
#include "TTimer.h"
#include "TMutex.h"
#ifndef R__WIN32
const char* const kCP = "/bin/cp -f";
const char* const kRM = "/bin/rm -rf";
const char* const kLS = "/bin/ls -l";
const char* const kUNTAR = "%s -c %s/%s | (cd %s; tar xf -)";
const char* const kGUNZIP = "gunzip";
#else
const char* const kCP = "copy";
const char* const kRM = "delete";
const char* const kLS = "dir";
const char* const kUNTAR = "...";
const char* const kGUNZIP = "gunzip";
#endif
TProofServ *gProofServ;
static volatile Int_t gProofServDebug = 1;
Int_t TProofServ::fgMaxQueries = -1;
class TProofServInterruptHandler : public TSignalHandler {
TProofServ *fServ;
public:
TProofServInterruptHandler(TProofServ *s)
: TSignalHandler(kSigUrgent, kFALSE) { fServ = s; }
Bool_t Notify();
};
Bool_t TProofServInterruptHandler::Notify()
{
fServ->HandleUrgentData();
if (TROOT::Initialized()) {
Throw(GetSignal());
}
return kTRUE;
}
class TProofServSigPipeHandler : public TSignalHandler {
TProofServ *fServ;
public:
TProofServSigPipeHandler(TProofServ *s) : TSignalHandler(kSigPipe, kFALSE)
{ fServ = s; }
Bool_t Notify();
};
Bool_t TProofServSigPipeHandler::Notify()
{
fServ->HandleSigPipe();
return kTRUE;
}
class TProofServInputHandler : public TFileHandler {
TProofServ *fServ;
public:
TProofServInputHandler(TProofServ *s, Int_t fd) : TFileHandler(fd, 1)
{ fServ = s; }
Bool_t Notify();
Bool_t ReadNotify() { return Notify(); }
};
Bool_t TProofServInputHandler::Notify()
{
fServ->HandleSocketInput();
return kTRUE;
}
ClassImp(TProofServ)
extern "C" {
TApplication *GetTProofServ(Int_t *argc, char **argv, FILE *flog)
{ return new TProofServ(argc, argv, flog); }
}
TProofServ::TProofServ(Int_t *argc, char **argv, FILE *flog)
: TApplication("proofserv", argc, argv, 0, -1)
{
Bool_t test = (*argc >= 4 && !strcmp(argv[3], "test")) ? kTRUE : kFALSE;
if ((gEnv->GetValue("Proof.GdbHook",0) == 3 && !test) ||
(gEnv->GetValue("Proof.GdbHook",0) == 4 && test)) {
while (gProofServDebug)
;
}
if (*argc >= 4)
if (!strcmp(argv[3], "test"))
fService = "prooftest";
if (*argc < 2) {
Error("TProofServ", "Must have at least 1 arguments (see proofd).");
exit(1);
}
gErrorAbortLevel = kSysError + 1;
SetErrorHandler(ErrorHandler);
fNcmd = 0;
fInterrupt = kFALSE;
fProtocol = 0;
fOrdinal = "-1";
if (gSystem->Getenv("ROOTPROOFORDINAL"))
fOrdinal = gSystem->Getenv("ROOTPROOFORDINAL");
fGroupId = -1;
fGroupSize = 0;
fRealTime = 0.0;
fCpuTime = 0.0;
fProof = 0;
fPlayer = 0;
fSocket = 0;
fEnabledPackages = new TList;
fEnabledPackages->SetOwner();
fLogFile = flog;
fLogFileDes = -1;
fArchivePath = "";
fPackageLock = 0;
fCacheLock = 0;
fQueryLock = 0;
fDataSetLock = 0;
fSeqNum = 0;
fDrawQueries = 0;
fKeptQueries = 0;
fQueries = new TList;
fWaitingQueries = new TList;
fPreviousQueries = 0;
fIdle = kTRUE;
fShutdownWhenIdle = kTRUE;
fShutdownTimer = 0;
fShutdownTimerMtx = 0;
if (gErrorIgnoreLevel == kUnset) {
gErrorIgnoreLevel = 0;
if (gEnv) {
TString level = gEnv->GetValue("Root.ErrorIgnoreLevel", "Info");
if (!level.CompareTo("Info",TString::kIgnoreCase))
gErrorIgnoreLevel = kInfo;
else if (!level.CompareTo("Warning",TString::kIgnoreCase))
gErrorIgnoreLevel = kWarning;
else if (!level.CompareTo("Error",TString::kIgnoreCase))
gErrorIgnoreLevel = kError;
else if (!level.CompareTo("Break",TString::kIgnoreCase))
gErrorIgnoreLevel = kBreak;
else if (!level.CompareTo("SysError",TString::kIgnoreCase))
gErrorIgnoreLevel = kSysError;
else if (!level.CompareTo("Fatal",TString::kIgnoreCase))
gErrorIgnoreLevel = kFatal;
}
}
if (gSystem->Getenv("ROOTPROOFLOGLEVEL"))
gProofDebugLevel = atoi(gSystem->Getenv("ROOTPROOFLOGLEVEL"));
else
gProofDebugLevel = gEnv->GetValue("Proof.DebugLevel",0);
fLogLevel = gProofDebugLevel;
gProofDebugMask = (TProofDebug::EProofDebugMask) gEnv->GetValue("Proof.DebugMask",~0);
if (gProofDebugLevel > 0)
Info("TProofServ", "DebugLevel %d Mask %u", gProofDebugLevel, gProofDebugMask);
GetOptions(argc, argv);
gProofServ = this;
}
void TProofServ::CreateServer()
{
if (!(gSystem->Getenv("ROOTOPENSOCK"))) {
Fatal("CreateServer", "Socket setup by proofd undefined");
exit(1);
}
Int_t sock = strtol(gSystem->Getenv("ROOTOPENSOCK"), (char **)0, 10);
if (sock <= 0) {
Fatal("CreateServer", "Invalid socket descriptor number (%d)", sock);
exit(1);
}
fSocket = new TSocket(sock);
if (IsMaster()) {
if (gEnv->GetValue("Proof.GdbHook",0) == 1) {
while (gProofServDebug)
;
}
} else {
if (gEnv->GetValue("Proof.GdbHook",0) == 2) {
while (gProofServDebug)
;
}
}
if (gProofDebugLevel > 0)
Info("CreateServer", "Service %s ConfDir %s IsMaster %d\n",
GetService(), GetConfDir(), (Int_t)fMasterServ);
Setup();
if (!fLogFile) {
RedirectOutput();
if (!fLogFile || (fLogFileDes = fileno(fLogFile)) < 0) {
SendLogFile(-98);
Terminate(0);
}
} else {
if ((fLogFileDes = fileno(fLogFile)) < 0) {
SendLogFile(-98);
Terminate(0);
}
}
if (IsMaster()) {
if (CatMotd() == -1) {
SendLogFile(-99);
Terminate(0);
}
} else {
THLimitsFinder::SetLimitsFinder(new TProofLimitsFinder);
}
ProcessLine("#include <iostream>", kTRUE);
ProcessLine("#include <_string>",kTRUE);
ProcessLine("#include <RtypesCint.h>", kTRUE);
ProcessLine("#define ROOT_Rtypes 0", kTRUE);
ProcessLine("#define ROOT_TError 0", kTRUE);
ProcessLine("#define ROOT_TGenericClassInfo 0", kTRUE);
gROOT->LoadClass("TMinuit", "Minuit");
gROOT->LoadClass("TPostScript", "Postscript");
const char *logon;
logon = gEnv->GetValue("Proof.Load", (char *)0);
if (logon) {
char *mac = gSystem->Which(TROOT::GetMacroPath(), logon, kReadPermission);
if (mac)
ProcessLine(Form(".L %s", logon), kTRUE);
delete [] mac;
}
logon = gEnv->GetValue("Proof.Logon", (char *)0);
if (logon && !NoLogOpt()) {
char *mac = gSystem->Which(TROOT::GetMacroPath(), logon, kReadPermission);
if (mac)
ProcessFile(logon);
delete [] mac;
}
gInterpreter->SaveContext();
gInterpreter->SaveGlobalsContext();
gSystem->AddSignalHandler(new TProofServInterruptHandler(this));
gSystem->AddFileHandler(new TProofServInputHandler(this, sock));
if (IsMaster()) {
TString master = "proof:;
TInetAddress a = gSystem->GetSockName(sock);
if (a.IsValid()) {
master += ":";
master += a.GetPort();
}
TPluginManager *pm = gROOT->GetPluginManager();
if (!pm) {
Error("CreateServer", "no plugin manager found");
SendLogFile(-99);
Terminate(0);
}
TPluginHandler *h = pm->FindHandler("TVirtualProof", fConfFile);
if (!h) {
Error("CreateServer", "no plugin found for TVirtualProof with a"
" config file of '%s'", fConfFile.Data());
SendLogFile(-99);
Terminate(0);
}
if (h->LoadPlugin() == -1) {
Error("CreateServer", "plugin for TVirtualProof could not be loaded");
SendLogFile(-99);
Terminate(0);
}
fProof = reinterpret_cast<TProof*>(h->ExecPlugin(4, master.Data(),
fConfFile.Data(),
GetConfDir(),
fLogLevel));
if (!fProof || !fProof->IsValid()) {
Error("CreateServer", "plugin for TVirtualProof could not be executed");
delete fProof;
fProof = 0;
SendLogFile(-99);
Terminate(0);
}
fEndMaster = fProof->IsEndMaster();
SendLogFile();
}
}
TProofServ::~TProofServ()
{
SafeDelete(fQueries);
SafeDelete(fPreviousQueries);
SafeDelete(fWaitingQueries);
SafeDelete(fEnabledPackages);
SafeDelete(fSocket);
SafeDelete(fPackageLock);
SafeDelete(fCacheLock);
SafeDelete(fQueryLock);
SafeDelete(fDataSetLock);
close(fLogFileDes);
}
Int_t TProofServ::CatMotd()
{
TString lastname;
FILE *motd;
Bool_t show = kFALSE;
TString motdname(GetConfDir());
motdname += "/etc/proof/noproof";
if ((motd = fopen(motdname, "r"))) {
Int_t c;
printf("\n");
while ((c = getc(motd)) != EOF)
putchar(c);
fclose(motd);
printf("\n");
return -1;
}
lastname = TString(GetWorkDir()) + "/.prooflast";
char *last = gSystem->ExpandPathName(lastname.Data());
Long64_t size;
Long_t id, flags, modtime, lasttime;
if (gSystem->GetPathInfo(last, &id, &size, &flags, &lasttime) == 1)
lasttime = 0;
if (time(0) - lasttime > (time_t)86400)
show = kTRUE;
motdname = GetConfDir();
motdname += "/etc/proof/motd";
if (gSystem->GetPathInfo(motdname, &id, &size, &flags, &modtime) == 0) {
if (modtime > lasttime || show) {
if ((motd = fopen(motdname, "r"))) {
Int_t c;
printf("\n");
while ((c = getc(motd)) != EOF)
putchar(c);
fclose(motd);
printf("\n");
}
}
}
Int_t fd = creat(last, 0600);
close(fd);
delete [] last;
return 0;
}
TObject *TProofServ::Get(const char *namecycle)
{
fSocket->Send(namecycle, kPROOF_GETOBJECT);
TMessage *mess;
if (fSocket->Recv(mess) < 0)
return 0;
TObject *idcur = 0;
if (mess->What() == kMESS_OBJECT)
idcur = mess->ReadObject(mess->GetClass());
delete mess;
return idcur;
}
TDSetElement *TProofServ::GetNextPacket(Long64_t totalEntries)
{
Long64_t bytesRead = 0;
if (gPerfStats != 0) bytesRead = gPerfStats->GetBytesRead();
if (fCompute.Counter() > 0)
fCompute.Stop();
TMessage req(kPROOF_GETPACKET);
req << fLatency.RealTime() << fCompute.RealTime()
<< fCompute.CpuTime() << bytesRead
<< totalEntries;
if (fPlayer)
req << fPlayer->GetEventsProcessed();
fLatency.Start();
Int_t rc = fSocket->Send(req);
if (rc <= 0) {
Error("GetNextPacket","Send() failed, returned %d", rc);
return 0;
}
TMessage *mess;
if ((rc = fSocket->Recv(mess)) <= 0) {
fLatency.Stop();
Error("GetNextPacket","Recv() failed, returned %d", rc);
return 0;
}
fLatency.Stop();
TDSetElement *e = 0;
TString file;
TString dir;
TString obj;
Int_t what = mess->What();
switch (what) {
case kPROOF_GETPACKET:
(*mess) >> e;
if (e != 0) {
fCompute.Start();
PDB(kLoop, 2) Info("GetNextPacket", "'%s' '%s' '%s' %lld %lld",
e->GetFileName(), e->GetDirectory(),
e->GetObjName(), e->GetFirst(),e->GetNum());
} else {
PDB(kLoop, 2) Info("GetNextPacket", "Done");
}
delete mess;
return e;
case kPROOF_STOPPROCESS:
PDB(kLoop, 2) Info("GetNextPacket:kPROOF_STOPPROCESS","received");
break;
default:
Error("GetNextPacket","unexpected answer message type: %d",what);
break;
}
delete mess;
return 0;
}
void TProofServ::GetOptions(Int_t *argc, char **argv)
{
if (*argc <= 1) {
Fatal("GetOptions", "Must be started from proofd with arguments");
exit(1);
}
if (!strcmp(argv[1], "proofserv")) {
fMasterServ = kTRUE;
fEndMaster = kTRUE;
} else if (!strcmp(argv[1], "proofslave")) {
fMasterServ = kFALSE;
fEndMaster = kFALSE;
} else {
Fatal("GetOptions", "Must be started as proofmaster or proofslave");
exit(1);
}
fService = argv[1];
if (!(gSystem->Getenv("ROOTCONFDIR"))) {
Fatal("GetOptions", "Must specify a config directory");
exit(1);
}
fConfDir = gSystem->Getenv("ROOTCONFDIR");
}
void TProofServ::HandleSocketInput()
{
static Int_t recursive = 0;
if (recursive > 0) {
HandleSocketInputDuringProcess();
return;
}
recursive++;
static TStopwatch timer;
TMessage *mess;
char str[2048];
Int_t what;
if (fSocket->Recv(mess) <= 0) {
Error("HandleSocketInput", "retrieving message from input socket");
Terminate(0);
}
what = mess->What();
timer.Start();
fNcmd++;
if (fProof) fProof->SetActive();
switch (what) {
case kMESS_CINT:
mess->ReadString(str, sizeof(str));
if (IsMaster() && IsParallel()) {
fProof->SendCommand(str);
} else {
PDB(kGlobal, 1) Info("HandleSocketInput:kMESS_CINT", "processing: %s...", str);
ProcessLine(str);
}
SendLogFile();
break;
case kMESS_STRING:
mess->ReadString(str, sizeof(str));
break;
case kMESS_OBJECT:
mess->ReadObject(mess->GetClass());
break;
case kPROOF_GROUPVIEW:
mess->ReadString(str, sizeof(str));
sscanf(str, "%d %d", &fGroupId, &fGroupSize);
break;
case kPROOF_LOGLEVEL:
{
UInt_t mask;
mess->ReadString(str, sizeof(str));
sscanf(str, "%d %u", &fLogLevel, &mask);
gProofDebugLevel = fLogLevel;
gProofDebugMask = (TProofDebug::EProofDebugMask) mask;
if (IsMaster())
fProof->SetLogLevel(fLogLevel, mask);
}
break;
case kPROOF_PING:
if (IsMaster())
fProof->Ping();
break;
case kPROOF_PRINT:
mess->ReadString(str, sizeof(str));
Print(str);
SendLogFile();
break;
case kPROOF_RESET:
mess->ReadString(str, sizeof(str));
Reset(str);
break;
case kPROOF_STATUS:
Warning("HandleSocketInput:kPROOF_STATUS",
"kPROOF_STATUS message is obsolete");
fSocket->Send(fProof->GetParallel(), kPROOF_STATUS);
break;
case kPROOF_GETSTATS:
SendStatistics();
break;
case kPROOF_GETPARALLEL:
SendParallel();
break;
case kPROOF_STOP:
Terminate(0);
break;
case kPROOF_STOPPROCESS:
PDB(kGlobal, 1) Info("HandleSocketInput:kPROOF_STOPPROCESS","enter");
break;
case kPROOF_PROCESS:
HandleProcess(mess);
SendLogFile();
break;
case kPROOF_QUERYLIST:
HandleQueryList(mess);
SendLogFile();
break;
case kPROOF_REMOVE:
HandleRemove(mess);
SendLogFile();
break;
case kPROOF_RETRIEVE:
HandleRetrieve(mess);
SendLogFile();
break;
case kPROOF_ARCHIVE:
HandleArchive(mess);
SendLogFile();
break;
case kPROOF_MAXQUERIES:
{
PDB(kGlobal, 1)
Info("HandleSocketInput:kPROOF_MAXQUERIES", "Enter");
TMessage m(kPROOF_MAXQUERIES);
m << fgMaxQueries;
fSocket->Send(m);
}
SendLogFile();
break;
case kPROOF_CLEANUPSESSION:
{
PDB(kGlobal, 1)
Info("HandleSocketInput:kPROOF_CLEANUPSESSION", "Enter");
TString stag;
(*mess) >> stag;
if (CleanupSession(stag) == 0) {
Printf("Session %s cleaned up", stag.Data());
} else {
Printf("Could not cleanup session %s", stag.Data());
}
}
SendLogFile();
break;
case kPROOF_GETENTRIES:
{
PDB(kGlobal, 1) Info("HandleSocketInput:kPROOF_GETENTRIES", "Enter");
Bool_t isTree;
TString filename;
TString dir;
TString objname;
Long64_t entries;
(*mess) >> isTree >> filename >> dir >> objname;
PDB(kGlobal, 2) Info("HandleSocketInput:kPROOF_GETENTRIES",
"Report size of object %s (%s) in dir %s in file %s",
objname.Data(), isTree ? "T" : "O",
dir.Data(), filename.Data());
entries = TDSet::GetEntries(isTree, filename, dir, objname);
PDB(kGlobal, 2) Info("HandleSocketInput:kPROOF_GETENTRIES",
"Found %lld %s", entries, isTree ? "entries" : "objects");
TMessage answ(kPROOF_GETENTRIES);
answ << entries;
SendLogFile();
fSocket->Send(answ);
PDB(kGlobal, 1) Info("HandleSocketInput:kPROOF_GETENTRIES", "Done");
}
break;
case kPROOF_CHECKFILE:
HandleCheckFile(mess);
break;
case kPROOF_SENDFILE:
mess->ReadString(str, sizeof(str));
{
Long_t size;
Int_t bin, fw = 1;
char name[1024];
if (fProtocol > 5)
sscanf(str, "%s %d %ld %d", name, &bin, &size, &fw);
else
sscanf(str, "%s %d %ld", name, &bin, &size);
ReceiveFile(name, bin ? kTRUE : kFALSE, size);
if (size > 0) {
fCacheLock->Lock();
gSystem->Exec(Form("%s %s %s", kCP, name, fCacheDir.Data()));
fCacheLock->Unlock();
}
if (IsMaster() && fw == 1)
fProof->SendFile(name, bin);
}
break;
case kPROOF_LOGFILE:
{
Int_t start, end;
(*mess) >> start >> end;
PDB(kGlobal, 1)
Info("HandleSocketInput:kPROOF_LOGFILE",
"Logfile request - byte range: %d - %d", start, end);
SendLogFile(0, start, end);
}
break;
case kPROOF_PARALLEL:
if (IsMaster()) {
Int_t nodes;
(*mess) >> nodes;
fProof->SetParallel(nodes);
SendLogFile();
}
break;
case kPROOF_CACHE:
{
Int_t status = HandleCache(mess);
SendLogFile(status);
}
break;
case kPROOF_WORKERLISTS:
{
if (IsMaster())
HandleWorkerLists(mess);
else
Warning("HandleSocketInput:kPROOF_WORKERLISTS",
"Action meaning-less on worker nodes: protocol error?");
SendLogFile();
}
break;
case kPROOF_GETSLAVEINFO:
{
PDB(kGlobal, 1) Info("HandleSocketInput:kPROOF_GETSLAVEINFO", "Enter");
if (IsMaster()) {
TList *info = fProof->GetSlaveInfo();
TMessage answ(kPROOF_GETSLAVEINFO);
answ << info;
fSocket->Send(answ);
} else {
TMessage answ(kPROOF_GETSLAVEINFO);
answ << (TList *)0;
fSocket->Send(answ);
}
PDB(kGlobal, 1) Info("HandleSocketInput:kPROOF_GETSLAVEINFO", "Done");
}
break;
case kPROOF_GETTREEHEADER:
{
PDB(kGlobal, 1) Info("HandleSocketInput:kPROOF_GETTREEHEADER", "Enter");
TMessage answ(kMESS_OBJECT);
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("HandleSocketInput:kPROOF_GETTREEHEADER",
"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);
PDB(kGlobal, 1) Info("HandleSocketInput:kPROOF_GETTREEHEADER", "Done");
}
break;
case kPROOF_GETOUTPUTLIST:
{
PDB(kGlobal, 1) Info("HandleSocketInput:kPROOF_GETOUTPUTLIST", "Enter");
TList* outputList = 0;
if (IsMaster()) {
outputList = fProof->GetOutputList();
if (!outputList)
outputList = new TList();
} else {
outputList = new TList();
if (fProof->GetPlayer()) {
TList *olist = fProof->GetPlayer()->GetOutputList();
TIter next(olist);
TObject *o;
while ( (o = next()) ) {
outputList->Add(new TNamed(o->GetName(), ""));
}
}
}
outputList->SetOwner();
TMessage answ(kPROOF_GETOUTPUTLIST);
answ << outputList;
fSocket->Send(answ);
delete outputList;
PDB(kGlobal, 1) Info("HandleSocketInput:kPROOF_GETOUTPUTLIST", "Done");
}
break;
case kPROOF_VALIDATE_DSET:
{
PDB(kGlobal, 1) Info("HandleSocketInput:kPROOF_VALIDATE_DSET", "Enter");
TDSet* dset = 0;
(*mess) >> dset;
if (IsMaster()) fProof->ValidateDSet(dset);
else dset->Validate();
TMessage answ(kPROOF_VALIDATE_DSET);
answ << dset;
fSocket->Send(answ);
delete dset;
PDB(kGlobal, 1) Info("HandleSocketInput:kPROOF_VALIDATE_DSET", "Done");
SendLogFile();
}
break;
case kPROOF_DATA_READY:
{
PDB(kGlobal, 1) Info("HandleSocketInput:kPROOF_DATA_READY", "Enter");
TMessage answ(kPROOF_DATA_READY);
if (IsMaster()) {
Long64_t totalbytes = 0, bytesready = 0;
Bool_t dataready = fProof->IsDataReady(totalbytes, bytesready);
answ << dataready << totalbytes << bytesready;
} else {
Error("HandleSocketInput:kPROOF_DATA_READY",
"This message should not be sent to slaves");
answ << kFALSE << Long64_t(0) << Long64_t(0);
}
fSocket->Send(answ);
PDB(kGlobal, 1) Info("HandleSocketInput:kPROOF_DATA_READY", "Done");
SendLogFile();
}
break;
case kPROOF_DATASETS:
{
HandleDataSets(mess);
SendLogFile();
}
break;
case kPROOF_LIB_INC_PATH:
HandleLibIncPath(mess);
SendLogFile();
break;
default:
Error("HandleSocketInput", "unknown command %d", what);
break;
}
recursive--;
if (fProof) fProof->SetActive(kFALSE);
fRealTime += (Float_t)timer.RealTime();
fCpuTime += (Float_t)timer.CpuTime();
SetShutdownTimer(kTRUE, -1);
delete mess;
}
void TProofServ::HandleSocketInputDuringProcess()
{
PDB(kGlobal,1) Info("HandleSocketInputDuringProcess", "enter");
TMessage *mess;
char str[2048];
Int_t what;
Bool_t aborted = kFALSE;
if (fSocket->Recv(mess) <= 0) {
Error("HandleSocketInputDuringProcess", "retrieving message from input socket");
Terminate(0);
}
what = mess->What();
switch (what) {
case kPROOF_PROCESS:
HandleProcess(mess);
SendLogFile();
break;
case kPROOF_GETSTATS:
SendStatistics();
break;
case kPROOF_LOGFILE:
{
Int_t start, end;
(*mess) >> start >> end;
PDB(kGlobal, 1)
Info("HandleSocketInputDuringProcess:kPROOF_LOGFILE",
"Logfile request - byte range: %d - %d", start, end);
SendLogFile(0, start, end);
}
break;
case kPROOF_QUERYLIST:
HandleQueryList(mess);
SendLogFile();
break;
case kPROOF_ARCHIVE:
HandleArchive(mess);
SendLogFile();
break;
case kPROOF_REMOVE:
HandleRemove(mess);
SendLogFile();
break;
case kPROOF_RETRIEVE:
HandleRetrieve(mess);
SendLogFile();
break;
case kPROOF_STOPPROCESS:
{ Long_t timeout = -1;
(*mess) >> aborted;
if (fProtocol > 9)
(*mess) >> timeout;
PDB(kGlobal, 1)
Info("HandleSocketInputDuringProcess:kPROOF_STOPPROCESS",
"enter %d, %d", aborted, timeout);
if (fProof)
fProof->StopProcess(aborted, timeout);
else
if(fPlayer)
fPlayer->StopProcess(aborted, timeout);
}
break;
case kPROOF_CACHE:
{
Int_t status = HandleCache(mess);
SendLogFile(status);
}
break;
case kPROOF_CHECKFILE:
HandleCheckFile(mess);
break;
case kPROOF_SENDFILE:
mess->ReadString(str, sizeof(str));
{
Long_t size;
Int_t bin, fw = 1;
char name[1024];
if (fProtocol > 5)
sscanf(str, "%s %d %ld %d", name, &bin, &size, &fw);
else
sscanf(str, "%s %d %ld", name, &bin, &size);
ReceiveFile(name, bin ? kTRUE : kFALSE, size);
if (size > 0) {
fCacheLock->Lock();
gSystem->Exec(Form("%s %s %s", kCP, name, fCacheDir.Data()));
fCacheLock->Unlock();
}
if (IsMaster() && fw == 1)
fProof->SendFile(name, bin);
}
break;
case kPROOF_DATASETS:
{
HandleDataSets(mess);
SendLogFile();
}
break;
default:
Error("HandleSocketInputDuringProcess", "unknown command %d", what);
break;
}
delete mess;
}
void TProofServ::HandleUrgentData()
{
char oob_byte;
Int_t n, nch, wasted = 0;
const Int_t kBufSize = 1024;
char waste[kBufSize];
PDB(kGlobal, 5)
Info("HandleUrgentData", "handling oob...");
while ((n = fSocket->RecvRaw(&oob_byte, 1, kOob)) < 0) {
if (n == -2) {
fSocket->GetOption(kBytesToRead, nch);
if (nch == 0) {
gSystem->Sleep(1000);
continue;
}
if (nch > kBufSize) nch = kBufSize;
n = fSocket->RecvRaw(waste, nch);
if (n <= 0) {
Error("HandleUrgentData", "error receiving waste");
break;
}
wasted = 1;
} else {
Error("HandleUrgentData", "error receiving OOB");
return;
}
}
PDB(kGlobal, 5)
Info("HandleUrgentData", "got OOB byte: %d\n", oob_byte);
if (fProof) fProof->SetActive();
switch (oob_byte) {
case TProof::kHardInterrupt:
Info("HandleUrgentData", "*** Hard Interrupt");
if (IsMaster())
fProof->Interrupt(TProof::kHardInterrupt);
while (1) {
Int_t atmark;
fSocket->GetOption(kAtMark, atmark);
if (atmark) {
n = fSocket->SendRaw(&oob_byte, 1, kOob);
if (n <= 0)
Error("HandleUrgentData", "error sending OOB");
break;
}
fSocket->GetOption(kBytesToRead, nch);
if (nch == 0) {
gSystem->Sleep(1000);
continue;
}
if (nch > kBufSize) nch = kBufSize;
n = fSocket->RecvRaw(waste, nch);
if (n <= 0) {
Error("HandleUrgentData", "error receiving waste (2)");
break;
}
}
break;
case TProof::kSoftInterrupt:
Info("HandleUrgentData", "Soft Interrupt");
if (IsMaster())
fProof->Interrupt(TProof::kSoftInterrupt);
if (wasted) {
Error("HandleUrgentData", "soft interrupt flushed stream");
break;
}
Interrupt();
break;
case TProof::kShutdownInterrupt:
Info("HandleUrgentData", "Shutdown Interrupt");
if (IsMaster())
fProof->Interrupt(TProof::kShutdownInterrupt);
Terminate(0);
break;
default:
Error("HandleUrgentData", "unexpected OOB byte");
break;
}
SendLogFile();
if (fProof) fProof->SetActive(kFALSE);
}
void TProofServ::HandleSigPipe()
{
if (IsMaster()) {
if (fSocket->Send(kPROOF_PING | kMESS_ACK) < 0) {
Info("HandleSigPipe", "keepAlive probe failed");
fProof->SetActive();
fProof->Interrupt(TProof::kShutdownInterrupt);
fProof->SetActive(kFALSE);
Terminate(0);
}
} else {
Info("HandleSigPipe", "keepAlive probe failed");
Terminate(0);
}
}
Bool_t TProofServ::IsParallel() const
{
if (IsMaster())
return fProof->IsParallel();
return kFALSE;
}
void TProofServ::Print(Option_t *option) const
{
if (IsMaster())
fProof->Print(option);
else
Printf("This is slave %s", gSystem->HostName());
}
void TProofServ::RedirectOutput()
{
char logfile[512];
if (IsMaster()) {
sprintf(logfile, "%s/master.log", fSessionDir.Data());
} else {
sprintf(logfile, "%s/slave-%s.log", fSessionDir.Data(), fOrdinal.Data());
}
if ((freopen(logfile, "w", stdout)) == 0)
SysError("RedirectOutput", "could not freopen stdout");
if ((dup2(fileno(stdout), fileno(stderr))) < 0)
SysError("RedirectOutput", "could not redirect stderr");
if ((fLogFile = fopen(logfile, "r")) == 0)
SysError("RedirectOutput", "could not open logfile");
if (fProtocol < 4 && fWorkDir != kPROOF_WorkDir) {
Warning("RedirectOutput", "no way to tell master (or client) where"
" to upload packages");
}
}
void TProofServ::Reset(const char *dir)
{
gDirectory->cd(dir);
gROOT->Reset();
if (gDirectory != gROOT) {
gDirectory->Delete();
}
if (IsMaster()) fProof->SendCurrentState();
}
Int_t TProofServ::ReceiveFile(const char *file, Bool_t bin, Long64_t size)
{
if (size <= 0) return 0;
Int_t fd = open(file, O_CREAT | O_TRUNC | O_WRONLY, 0600);
if (fd < 0) {
SysError("ReceiveFile", "error opening file %s", file);
return -1;
}
const Int_t kMAXBUF = 16384;
char buf[kMAXBUF], cpy[kMAXBUF];
Int_t left, r;
Long64_t filesize = 0;
while (filesize < size) {
left = Int_t(size - filesize);
if (left > kMAXBUF)
left = kMAXBUF;
r = fSocket->RecvRaw(&buf, left);
if (r > 0) {
char *p = buf;
filesize += r;
while (r) {
Int_t w;
if (!bin) {
Int_t k = 0, i = 0, j = 0;
char *q;
while (i < r) {
if (p[i] == '\r') {
i++;
k++;
}
cpy[j++] = buf[i++];
}
q = cpy;
r -= k;
w = write(fd, q, r);
} else {
w = write(fd, p, r);
}
if (w < 0) {
SysError("ReceiveFile", "error writing to file %s", file);
close(fd);
return -1;
}
r -= w;
p += w;
}
} else if (r < 0) {
Error("ReceiveFile", "error during receiving file %s", file);
close(fd);
return -1;
}
}
close(fd);
chmod(file, 0644);
return 0;
}
void TProofServ::Run(Bool_t retrn)
{
CreateServer();
TApplication::Run(retrn);
}
void TProofServ::SendLogFile(Int_t status, Int_t start, Int_t end)
{
fflush(stdout);
off_t ltot=0, lnow=0;
Int_t left = -1;
Bool_t adhoc = kFALSE;
if (fLogFileDes > -1) {
ltot = lseek(fileno(stdout), (off_t) 0, SEEK_END);
lnow = lseek(fLogFileDes, (off_t) 0, SEEK_CUR);
if (start > -1) {
lseek(fLogFileDes, (off_t) start, SEEK_SET);
if (end <= start || end > ltot)
end = ltot;
left = (Int_t)(end - start);
if (end < ltot)
left++;
adhoc = kTRUE;
} else {
left = (Int_t)(ltot - lnow);
}
}
if (left > 0) {
fSocket->Send(left, kPROOF_LOGFILE);
const Int_t kMAXBUF = 32768;
char buf[kMAXBUF];
Int_t wanted = (left > kMAXBUF) ? kMAXBUF : left;
Int_t len;
do {
while ((len = read(fLogFileDes, buf, wanted)) < 0 &&
TSystem::GetErrno() == EINTR)
TSystem::ResetErrno();
if (len < 0) {
SysError("SendLogFile", "error reading log file");
break;
}
if (end == ltot && len == wanted)
buf[len-1] = '\n';
if (fSocket->SendRaw(buf, len) < 0) {
SysError("SendLogFile", "error sending log file");
break;
}
left -= len;
wanted = (left > kMAXBUF) ? kMAXBUF : left;
} while (len > 0 && left > 0);
}
if (adhoc)
lseek(fLogFileDes, lnow, SEEK_SET);
TMessage mess(kPROOF_LOGDONE);
if (IsMaster())
mess << status << (fProof ? fProof->GetParallel() : 0);
else
mess << status << (Int_t) 1;
fSocket->Send(mess);
}
void TProofServ::SendStatistics()
{
Long64_t bytesread = 0;
if (IsMaster()) bytesread = fProof->GetBytesRead();
else bytesread = TFile::GetFileBytesRead();
TMessage mess(kPROOF_GETSTATS);
TString workdir = gSystem->WorkingDirectory();
mess << bytesread << fRealTime << fCpuTime << workdir;
if (fProtocol >= 4) mess << TString(gProofServ->GetWorkDir());
fSocket->Send(mess);
}
void TProofServ::SendParallel()
{
Int_t nparallel = 0;
if (IsMaster()) {
fProof->AskParallel();
nparallel = fProof->GetParallel();
} else {
nparallel = 1;
}
fSocket->Send(nparallel, kPROOF_GETPARALLEL);
}
Int_t TProofServ::UnloadPackage(const char *package)
{
TObjString *pack = (TObjString *) fEnabledPackages->FindObject(package);
if (pack) {
TString aclicincpath = gSystem->GetIncludePath();
TString cintincpath = gInterpreter->GetIncludePath();
aclicincpath.Remove(aclicincpath.Length() - cintincpath.Length() - 1);
aclicincpath.ReplaceAll(TString(" -I") + package, "");
gSystem->SetIncludePath(aclicincpath);
delete fEnabledPackages->Remove(pack);
PDB(kPackage, 1)
Info("UnloadPackage",
"package %s successfully unloaded", package);
}
if (!gSystem->AccessPathName(package))
if (gSystem->Unlink(package) != 0)
Warning("UnloadPackage", "unable to remove symlink to %s", package);
return 0;
}
Int_t TProofServ::UnloadPackages()
{
TIter nextpackage(fEnabledPackages);
while (TObjString* objstr = dynamic_cast<TObjString*>(nextpackage()))
if (UnloadPackage(objstr->String()) != 0)
return -1;
PDB(kPackage, 1)
Info("UnloadPackages",
"packages successfully unloaded");
return 0;
}
void TProofServ::Setup()
{
char str[512];
if (IsMaster()) {
sprintf(str, "**** Welcome to the PROOF server @ %s ****", gSystem->HostName());
} else {
sprintf(str, "**** PROOF slave server @ %s started ****", gSystem->HostName());
}
if (fSocket->Send(str) != 1+static_cast<Int_t>(strlen(str))) {
Error("Setup", "failed to send proof server startup message");
gSystem->Exit(1);
}
Int_t what;
if (fSocket->Recv(fProtocol, what) != 2*sizeof(Int_t)) {
Error("Setup", "failed to receive remote proof protocol");
gSystem->Exit(1);
}
if (fSocket->Send(kPROOF_Protocol, kROOTD_PROTOCOL) != 2*sizeof(Int_t)) {
Error("Setup", "failed to send local proof protocol");
gSystem->Exit(1);
}
if (fProtocol < 5) {
TString wconf;
if (OldAuthSetup(wconf) != 0) {
Error("Setup", "OldAuthSetup: failed to setup authentication");
gSystem->Exit(1);
}
if (IsMaster()) {
fConfFile = wconf;
fWorkDir = kPROOF_WorkDir;
} else {
if (fProtocol < 4) {
fWorkDir = kPROOF_WorkDir;
} else {
fWorkDir = wconf;
if (fWorkDir.IsNull()) fWorkDir = kPROOF_WorkDir;
}
}
} else {
TMessage *mess;
if ((fSocket->Recv(mess) <= 0) || !mess) {
Error("Setup", "failed to receive ordinal and config info");
gSystem->Exit(1);
}
if (IsMaster()) {
(*mess) >> fUser >> fOrdinal >> fConfFile;
if (gSystem->Getenv("ROOTPROOFWORKDIR"))
fWorkDir = gSystem->Getenv("ROOTPROOFWORKDIR");
else
fWorkDir = kPROOF_WorkDir;
} else {
(*mess) >> fUser >> fOrdinal >> fWorkDir;
if (fWorkDir.IsNull())
if (gSystem->Getenv("ROOTPROOFWORKDIR"))
fWorkDir = gSystem->Getenv("ROOTPROOFWORKDIR");
else
fWorkDir = kPROOF_WorkDir;
}
delete mess;
}
if (IsMaster()) {
TString conffile = fConfFile;
conffile.Remove(0, 1 + conffile.Index(":"));
TProofResourcesStatic resources(fConfDir, conffile);
if (resources.IsValid()) {
if (resources.GetMaster()) {
TString tmpWorkDir = resources.GetMaster()->GetWorkDir();
if (tmpWorkDir != "")
fWorkDir = tmpWorkDir;
}
} else {
Error("Setup", "reading config file %s",
resources.GetFileName().Data());
gSystem->Exit(1);
}
}
if (fWorkDir.BeginsWith("/") &&
!fWorkDir.BeginsWith(gSystem->HomeDirectory())) {
if (!fWorkDir.EndsWith("/"))
fWorkDir += "/";
UserGroup_t *u = gSystem->GetUserInfo();
if (u) {
fWorkDir += u->fUser;
delete u;
}
}
char *workdir = gSystem->ExpandPathName(fWorkDir.Data());
fWorkDir = workdir;
delete [] workdir;
gSystem->Umask(022);
gSystem->Setenv("HOME", gSystem->HomeDirectory());
#ifdef R__UNIX
TString bindir;
# ifdef ROOTBINDIR
bindir = ROOTBINDIR;
# else
bindir = gSystem->Getenv("ROOTSYS");
if (!bindir.IsNull()) bindir += "/bin";
# endif
# ifdef COMPILER
TString compiler = COMPILER;
compiler.Remove(0, compiler.Index("is ") + 3);
compiler = gSystem->DirName(compiler);
if (!bindir.IsNull()) bindir += ":";
bindir += compiler;
#endif
if (!bindir.IsNull()) bindir += ":";
bindir += "/bin:/usr/bin:/usr/local/bin";
gSystem->Setenv("PATH", bindir);
#endif
if (gSystem->AccessPathName(fWorkDir)) {
gSystem->mkdir(fWorkDir, kTRUE);
if (!gSystem->ChangeDirectory(fWorkDir)) {
SysError("Setup", "can not change to PROOF directory %s",
fWorkDir.Data());
}
} else {
if (!gSystem->ChangeDirectory(fWorkDir)) {
gSystem->Unlink(fWorkDir);
gSystem->mkdir(fWorkDir, kTRUE);
if (!gSystem->ChangeDirectory(fWorkDir)) {
SysError("Setup", "can not change to PROOF directory %s",
fWorkDir.Data());
}
}
}
fCacheDir = fWorkDir;
fCacheDir += TString("/") + kPROOF_CacheDir;
if (gSystem->AccessPathName(fCacheDir))
gSystem->MakeDirectory(fCacheDir);
fCacheLock =
new TProofLockPath(Form("%s%s",kPROOF_CacheLockFile,fUser.Data()));
fPackageDir = fWorkDir;
fPackageDir += TString("/") + kPROOF_PackDir;
if (gSystem->AccessPathName(fPackageDir))
gSystem->MakeDirectory(fPackageDir);
fPackageLock =
new TProofLockPath(Form("%s%s",kPROOF_PackageLockFile,fUser.Data()));
TString host = gSystem->HostName();
if (host.Index(".") != kNPOS)
host.Remove(host.Index("."));
fSessionTag = Form("%s-%s-%d-%d", fOrdinal.Data(), host.Data(),
TTimeStamp().GetSec(),gSystem->GetPid());
fSessionDir = fWorkDir;
if (IsMaster())
fSessionDir += "/master-";
else
fSessionDir += "/slave-";
fSessionDir += fSessionTag;
if (gSystem->AccessPathName(fSessionDir)) {
gSystem->MakeDirectory(fSessionDir);
if (!gSystem->ChangeDirectory(fSessionDir)) {
SysError("Setup", "can not change to working directory %s",
fSessionDir.Data());
} else {
gSystem->Setenv("PROOF_SANDBOX", fSessionDir);
}
}
if (IsMaster()) {
fQueryDir = fWorkDir;
fQueryDir += TString("/") + kPROOF_QueryDir;
if (gSystem->AccessPathName(fQueryDir))
gSystem->MakeDirectory(fQueryDir);
else
ScanPreviousQueries(fQueryDir);
fQueryDir += TString("/session-") + fSessionTag;
if (gSystem->AccessPathName(fQueryDir))
gSystem->MakeDirectory(fQueryDir);
fQueryLock = new TProofLockPath(Form("%s%s-%s",
kPROOF_QueryLockFile,fSessionTag.Data(),fUser.Data()));
fQueryLock->Lock();
fDataSetDir = fWorkDir;
fDataSetDir += TString("/") + kPROOF_DataSetDir;
if (gSystem->AccessPathName(fDataSetDir))
gSystem->MakeDirectory(fDataSetDir);
fDataSetLock =
new TProofLockPath(Form("%s%s", kPROOF_DataSetLockFile, fUser.Data()));
if (fProtocol > 6) {
TMessage m(kPROOF_SESSIONTAG);
m << fSessionTag;
fSocket->Send(m);
}
}
fSocket->SetOption(kProcessGroup, gSystem->GetPid());
fSocket->SetOption(kNoDelay, 1);
fSocket->SetOption(kKeepAlive, 1);
gSystem->AddSignalHandler(new TProofServSigPipeHandler(this));
}
void TProofServ::Terminate(Int_t status)
{
if (status == 0) {
gSystem->ChangeDirectory("/");
gSystem->MakeDirectory(fSessionDir+"/.delete");
gSystem->Exec(Form("%s %s", kRM, fSessionDir.Data()));
}
if (IsMaster()) {
if (!(fQueries->GetSize())) {
gSystem->ChangeDirectory("/");
gSystem->MakeDirectory(fQueryDir+"/.delete");
gSystem->Exec(Form("%s %s", kRM, fQueryDir.Data()));
if (fQueryLock)
gSystem->Unlink(fQueryLock->GetName());
}
if (fQueryLock)
fQueryLock->Unlock();
}
TIter next(gSystem->GetListOfFileHandlers());
TObject *fh = 0;
while ((fh = next())) {
TProofServInputHandler *ih = dynamic_cast<TProofServInputHandler *>(fh);
if (ih)
gSystem->RemoveFileHandler(ih);
}
gSystem->Exit(status);
}
Bool_t TProofServ::IsActive()
{
return gProofServ ? kTRUE : kFALSE;
}
TProofServ *TProofServ::This()
{
return gProofServ;
}
Int_t TProofServ::OldAuthSetup(TString &conf)
{
OldProofServAuthSetup_t oldAuthSetupHook = 0;
if (!oldAuthSetupHook) {
#ifdef ROOTLIBDIR
TString authlib = TString(ROOTLIBDIR) + "/libRootAuth";
#else
TString authlib = TString(gRootDir) + "/lib/libRootAuth";
#endif
char *p = 0;
if ((p = gSystem->DynamicPathName(authlib, kTRUE))) {
delete[] p;
if (gSystem->Load(authlib) == -1) {
Error("OldAuthSetup", "can't load %s",authlib.Data());
return kFALSE;
}
} else {
Error("OldAuthSetup", "can't locate %s",authlib.Data());
return -1;
}
Func_t f = gSystem->DynFindSymbol(authlib,"OldProofServAuthSetup");
if (f)
oldAuthSetupHook = (OldProofServAuthSetup_t)(f);
else {
Error("OldAuthSetup", "can't find OldProofServAuthSetup");
return -1;
}
}
if (oldAuthSetupHook) {
return (*oldAuthSetupHook)(fSocket, IsMaster(), fProtocol,
fUser, fOrdinal, conf);
} else {
Error("OldAuthSetup",
"hook to method OldProofServAuthSetup is undefined");
return -1;
}
}
TProofQueryResult *TProofServ::MakeQueryResult(Long64_t nent, const char *opt,
TList *inlist, Long64_t fst,
TDSet *dset, const char *selec,
TEventList *evl)
{
fSeqNum++;
TProofQueryResult *pqr =
new TProofQueryResult(fSeqNum, opt, inlist, nent, fst, dset, selec, evl);
pqr->SetTitle(gSystem->BaseName(fQueryDir));
return pqr;
}
void TProofServ::SetQueryRunning(TProofQueryResult *pq)
{
fflush(stdout);
Int_t startlog = lseek(fileno(stdout), (off_t) 0, SEEK_END);
Printf(" ");
Info("SetQueryRunning", "starting query: %d", pq->GetSeqNum());
TString parlist = "";
TIter nxp(fEnabledPackages);
TObjString *os= 0;
while ((os = (TObjString *)nxp())) {
if (parlist.Length() <= 0)
parlist = os->GetName();
else
parlist += Form(";%s",os->GetName());
}
pq->SetRunning(startlog, parlist);
pq->SetProcessInfo(pq->GetEntries(),
fProof->GetCpuTime(), fProof->GetBytesRead());
}
void TProofServ::AddLogFile(TProofQueryResult *pq)
{
if (!pq)
return;
fflush(stdout);
off_t lnow = lseek(fLogFileDes, (off_t) 0, SEEK_CUR);
Int_t start = pq->fStartLog;
if (start > -1)
lseek(fLogFileDes, (off_t) start, SEEK_SET);
const Int_t kMAXBUF = 4096;
char line[kMAXBUF];
while (fgets(line, sizeof(line), fLogFile)) {
if (line[strlen(line)-1] == '\n')
line[strlen(line)-1] = 0;
pq->AddLogLine((const char *)line);
}
lseek(fLogFileDes, lnow, SEEK_SET);
}
void TProofServ::FinalizeQuery(TProofPlayer *p, TProofQueryResult *pq)
{
if (!pq || !p) {
Warning("FinalizeQuery",
"bad inputs: query = %p, player = %p ", pq ? pq : 0, p ? p : 0);
return;
}
Int_t qn = pq->GetSeqNum();
Long64_t np = p->GetEventsProcessed();
TProofPlayer::EExitStatus est = p->GetExitStatus();
TList *out = p->GetOutputList();
fProof->AskStatistics();
Float_t cpu = fProof->GetCpuTime();
Long64_t bytes = fProof->GetBytesRead();
TQueryResult::EQueryStatus st = TQueryResult::kAborted;
PDB(kGlobal, 2) Info("FinalizeQuery","query #%d", qn);
PDB(kGlobal, 1)
Info("FinalizeQuery","%.1f %lld", cpu, bytes);
Bool_t save = kTRUE;
switch (est) {
case TProofPlayer::kAborted:
PDB(kGlobal, 1)
Info("FinalizeQuery", "query %d has been ABORTED <====", qn);
out = 0;
save = kFALSE;
break;
case TProofPlayer::kStopped:
PDB(kGlobal, 1)
Info("FinalizeQuery",
"query %d has been STOPPED: %d events processed", qn, np);
st = TQueryResult::kStopped;
break;
case TProofPlayer::kFinished:
PDB(kGlobal, 1)
Info("FinalizeQuery",
"query %d has been completed: %d events processed", qn, np);
st = TQueryResult::kCompleted;
break;
default:
Warning("FinalizeQuery",
"query %d: unknown exit status (%d)", qn, p->GetExitStatus());
}
pq->SetProcessInfo(np, cpu - pq->GetUsedCPU(), bytes - pq->GetBytes());
pq->RecordEnd(st, out);
AddLogFile(pq);
if (save) {
if (fgMaxQueries > -1) {
if (fQueries && fKeptQueries >= fgMaxQueries) {
TQueryResult *fcom = 0;
TQueryResult *farc = 0;
TIter nxq(fQueries);
TQueryResult *qr = 0;
while ((qr = (TQueryResult *) nxq())) {
if (qr->IsArchived()) {
if (qr->GetOutputList() && !farc)
farc = qr;
} else if (qr->GetStatus() > TQueryResult::kRunning && !fcom) {
fcom = qr;
}
if (farc && fcom)
break;
}
if (farc) {
RemoveQuery(farc, kTRUE);
fKeptQueries--;
} else if (fcom) {
RemoveQuery(fcom);
fKeptQueries--;
}
}
if (fKeptQueries < fgMaxQueries) {
SaveQuery(pq);
fKeptQueries++;
}
} else {
SaveQuery(pq);
fKeptQueries++;
}
}
}
void TProofServ::ScanPreviousQueries(const char *dir)
{
if (fPreviousQueries) {
fPreviousQueries->Delete();
SafeDelete(fPreviousQueries);
}
void *dirs = gSystem->OpenDirectory(dir);
char *sess = 0;
while ((sess = (char *) gSystem->GetDirEntry(dirs))) {
if (strlen(sess) < 7 || strncmp(sess,"session",7))
continue;
if (strstr(sess, fSessionTag))
continue;
void *dirq = gSystem->OpenDirectory(Form("%s/%s", dir, sess));
char *qry = 0;
while ((qry = (char *) gSystem->GetDirEntry(dirq))) {
if (qry[0] == '.')
continue;
TString fn = Form("%s/%s/%s/query-result.root", dir, sess, qry);
TFile *f = TFile::Open(fn);
if (f) {
f->ReadKeys();
TIter nxk(f->GetListOfKeys());
TKey *k = 0;
TProofQueryResult *pqr = 0;
while ((k = (TKey *)nxk())) {
if (!strcmp(k->GetClassName(), "TProofQueryResult")) {
pqr = (TProofQueryResult *) f->Get(k->GetName());
if (pqr) {
TQueryResult *qr = pqr->CloneInfo();
if (!fPreviousQueries)
fPreviousQueries = new TList;
if (qr->GetStatus() > TQueryResult::kRunning) {
fPreviousQueries->Add(qr);
} else {
TProofLockPath *lck = 0;
if (LockSession(qr->GetTitle(), &lck) == 0) {
RemoveQuery(qr);
SafeDelete(lck);
}
}
}
}
}
f->Close();
delete f;
}
}
gSystem->FreeDirectory(dirq);
}
gSystem->FreeDirectory(dirs);
}
Int_t TProofServ::LockSession(const char *sessiontag, TProofLockPath **lck)
{
if (strstr(sessiontag, fSessionTag))
return 0;
if (!lck) {
Info("LockSession","locker space undefined");
return -1;
}
*lck = 0;
TString stag = sessiontag;
TRegexp re("session-.*-.*-.*-.*");
Int_t i1 = stag.Index(re);
if (i1 == kNPOS) {
Info("LockSession","bad format: %s", sessiontag);
return -1;
}
stag.ReplaceAll("session-","");
Int_t i2 = stag.Index(":q");
if (i2 != kNPOS)
stag.Remove(i2);
TString parlog = fSessionDir;
parlog = parlog.Remove(parlog.Index("master-")+strlen("master-"));
parlog += stag;
if (!gSystem->AccessPathName(parlog)) {
Info("LockSession","parent still running: do nothing");
return -1;
}
TString qlock = fQueryLock->GetName();
qlock.ReplaceAll(fSessionTag, stag);
if (!gSystem->AccessPathName(qlock)) {
*lck = new TProofLockPath(qlock);
if (((*lck)->Lock()) < 0) {
Info("LockSession","problems locking query lock file");
SafeDelete(*lck);
return -1;
}
}
return 0;
}
Int_t TProofServ::CleanupSession(const char *sessiontag)
{
if (!sessiontag) {
Info("CleanupSession","session tag undefined");
return -1;
}
TString qdir = fQueryDir;
qdir.ReplaceAll(Form("session-%s", fSessionTag.Data()), sessiontag);
Int_t idx = qdir.Index(":q");
if (idx != kNPOS)
qdir.Remove(idx);
if (gSystem->AccessPathName(qdir)) {
Info("CleanupSession","query dir %s does not exist", qdir.Data());
return -1;
}
TProofLockPath *lck = 0;
if (LockSession(sessiontag, &lck) == 0) {
gSystem->Exec(Form("%s %s", kRM, qdir.Data()));
if (lck) {
gSystem->Unlink(lck->GetName());
SafeDelete(lck);
}
return 0;
}
Info("CleanupSession", "could not lock session %s", sessiontag);
return -1;
}
void TProofServ::SaveQuery(TQueryResult *qr, const char *fout)
{
if (!qr || qr->IsDraw())
return;
TString querydir = Form("%s/%d",fQueryDir.Data(), qr->GetSeqNum());
if (gSystem->AccessPathName(querydir))
gSystem->MakeDirectory(querydir);
TString ofn = fout ? fout : Form("%s/query-result.root", querydir.Data());
TFile *f = TFile::Open(ofn, "RECREATE");
if (f) {
f->cd();
if (!(qr->IsArchived()))
qr->fResultFile = ofn;
qr->Write();
f->Close();
delete f;
}
}
void TProofServ::RemoveQuery(const char *queryref)
{
PDB(kGlobal, 1)
Info("RemoveQuery", "Enter");
Int_t qry = -1;
TString qdir;
TProofQueryResult *pqr = LocateQuery(queryref, qry, qdir);
if (pqr) {
if (qry > -1) {
fQueries->Remove(pqr);
fWaitingQueries->Remove(pqr);
} else
fPreviousQueries->Remove(pqr);
delete pqr;
pqr = 0;
}
Info("RemoveQuery", "removing directory: %s", qdir.Data());
gSystem->Exec(Form("%s %s", kRM, qdir.Data()));
return;
}
void TProofServ::RemoveQuery(TQueryResult *qr, Bool_t soft)
{
PDB(kGlobal, 1)
Info("RemoveQuery", "Enter");
if (!qr)
return;
TString qdir = fQueryDir;
qdir = qdir.Remove(qdir.Index(kPROOF_QueryDir)+strlen(kPROOF_QueryDir));
qdir = Form("%s/%s/%d", qdir.Data(), qr->GetTitle(), qr->GetSeqNum());
PDB(kGlobal, 1)
Info("RemoveQuery", "removing directory: %s", qdir.Data());
gSystem->Exec(Form("%s %s", kRM, qdir.Data()));
if (soft) {
TQueryResult *qrn = qr->CloneInfo();
Int_t idx = fQueries->IndexOf(qr);
if (idx > -1)
fQueries->AddAt(qrn, idx);
else
SafeDelete(qrn);
}
fQueries->Remove(qr);
SafeDelete(qr);
return;
}
TProofQueryResult *TProofServ::LocateQuery(TString queryref,
Int_t &qry, TString &qdir)
{
TProofQueryResult *pqr = 0;
qry = -1;
if (queryref.IsDigit()) {
qry = queryref.Atoi();
} else if (queryref.Contains(fSessionTag)) {
Int_t i1 = queryref.Index(":q");
if (i1 != kNPOS) {
queryref.Remove(0,i1+2);
qry = queryref.Atoi();
}
}
qdir = "";
if (qry > -1) {
PDB(kGlobal, 1)
Info("LocateQuery", "local query: %d", qry);
if (fQueries) {
TIter nxq(fQueries);
while ((pqr = (TProofQueryResult *) nxq())) {
if (pqr->GetSeqNum() == qry) {
qdir = Form("%s/%d", fQueryDir.Data(), qry);
break;
}
}
}
} else {
PDB(kGlobal, 1)
Info("LocateQuery", "previously processed query: %s", queryref.Data());
if (fPreviousQueries) {
TIter nxq(fPreviousQueries);
while ((pqr = (TProofQueryResult *) nxq())) {
if (queryref.Contains(pqr->GetTitle()) &&
queryref.Contains(pqr->GetName()))
break;
}
}
queryref.ReplaceAll(":q","/");
qdir = fQueryDir;
qdir = qdir.Remove(qdir.Index(kPROOF_QueryDir)+strlen(kPROOF_QueryDir));
qdir = Form("%s/%s", qdir.Data(), queryref.Data());
}
return pqr;
}
void TProofServ::HandleArchive(TMessage *mess)
{
PDB(kGlobal, 1)
Info("HandleArchive", "Enter");
TString queryref;
TString path;
(*mess) >> queryref >> path;
if (queryref == "Default") {
fArchivePath = path;
Info("HandleArchive",
"default path set to %s", fArchivePath.Data());
return;
}
Int_t qry = -1;
TString qdir;
TProofQueryResult *pqr = LocateQuery(queryref, qry, qdir);
TProofQueryResult *pqm = pqr;
if (path.Length() <= 0) {
if (fArchivePath.Length() <= 0) {
Info("HandleArchive",
"archive paths are not defined - do nothing");
return;
}
if (qry > 0) {
path = Form("%s/session-%s-%d.root",
fArchivePath.Data(), fSessionTag.Data(), qry);
} else {
path = queryref;
path.ReplaceAll(":q","-");
path.Insert(0, Form("%s/",fArchivePath.Data()));
path += ".root";
}
}
if (!pqr || qry < 0) {
TString fout = qdir;
fout += "/query-result.root";
TFile *f = TFile::Open(fout,"READ");
pqr = 0;
if (f) {
f->ReadKeys();
TIter nxk(f->GetListOfKeys());
TKey *k = 0;
while ((k = (TKey *)nxk())) {
if (!strcmp(k->GetClassName(), "TProofQueryResult")) {
pqr = (TProofQueryResult *) f->Get(k->GetName());
if (pqr)
break;
}
}
f->Close();
delete f;
} else {
Info("HandleArchive",
"file cannot be open (%s)",fout.Data());
return;
}
}
if (pqr) {
PDB(kGlobal, 1) Info("HandleArchive",
"archive path for query #%d: %s",
qry, path.Data());
TFile *farc = 0;
if (gSystem->AccessPathName(path))
farc = TFile::Open(path,"NEW");
else
farc = TFile::Open(path,"UPDATE");
if (!farc || !(farc->IsOpen())) {
Info("HandleArchive",
"archive file cannot be open (%s)",path.Data());
return;
}
farc->cd();
pqr->SetArchived(path);
if (pqm)
pqm->SetArchived(path);
pqr->Write();
if (qry > -1)
SaveQuery(pqr);
Info("HandleArchive",
"results of query %s archived to file %s",
queryref.Data(), path.Data());
}
return;
}
void TProofServ::HandleProcess(TMessage *mess)
{
PDB(kGlobal, 1)
Info("HandleProcess", "Enter");
if (!IsTopMaster() && !fIdle)
return;
TDSet *dset;
TString filename, opt;
TList *input;
Long64_t nentries, first;
TEventList *evl;
Bool_t sync;
(*mess) >> dset >> filename >> input >> opt >> nentries >> first >> evl >> sync;
if (evl)
dset->SetEventList(evl);
TProofPlayer *p = 0;
if (IsTopMaster()) {
TProofQueryResult *pq = 0;
pq = MakeQueryResult(nentries, opt, input, first, dset, filename, evl);
if (!(pq->IsDraw())) {
fQueries->Add(pq);
SaveQuery(pq);
}
fWaitingQueries->Add(pq);
if (!sync) {
TMessage m(kPROOF_QUERYSUBMITTED);
m << pq->GetSeqNum();
fSocket->Send(m);
}
if (!fIdle) {
Info("HandleProcess",
"query \"%s:%s\" submitted", pq->GetTitle(), pq->GetName());
return;
}
while (fWaitingQueries->GetSize() > 0) {
fIdle = kFALSE;
Reset("");
pq = (TProofQueryResult *)(fWaitingQueries->First());
if (pq) {
dset = pq->GetDSet();
opt = pq->GetOptions();
input = pq->GetInputList();
nentries = pq->GetEntries();
first = pq->GetFirst();
evl = pq->GetEventList();
if (pq->GetSelecImp()) {
gSystem->Exec(Form("%s %s", kRM, pq->GetSelecImp()->GetName()));
pq->GetSelecImp()->SaveSource(pq->GetSelecImp()->GetName());
}
if (pq->GetSelecHdr() &&
!strstr(pq->GetSelecHdr()->GetName(), "TProofDrawHist")) {
gSystem->Exec(Form("%s %s", kRM, pq->GetSelecHdr()->GetName()));
pq->GetSelecHdr()->SaveSource(pq->GetSelecHdr()->GetName());
}
fWaitingQueries->Remove(pq);
} else {
Error("HandleProcess", "empty query in queue!");
continue;
}
SetQueryRunning(pq);
if (!(pq->IsDraw()))
SaveQuery(pq);
else
fDrawQueries++;
TMessage m(kPROOF_STARTPROCESS);
m << TString(pq->GetSelecImp()->GetName())
<< pq->GetDSet()->GetListOfElements()->GetSize()
<< pq->GetFirst() << pq->GetEntries();
fSocket->Send(m);
if (IsParallel()) {
p = fProof->MakePlayer();
} else {
p = new TProofPlayerSlave(fSocket);
fProof->SetPlayer(p);
}
p->AddQueryResult(pq);
fPlayer = p;
if (dset->IsA() == TDSetProxy::Class())
((TDSetProxy*)dset)->SetProofServ(this);
TIter next(input);
for (TObject *o; (o = next()); ) {
PDB(kGlobal, 2) Info("HandleProcess", "adding: %s", o->GetName());
p->AddInput(o);
}
p->Process(dset, filename, opt, nentries, first);
if (p->GetExitStatus() != TProofPlayer::kFinished) {
Bool_t abort =
(p->GetExitStatus() == TProofPlayer::kAborted) ? kTRUE : kFALSE;
TMessage m(kPROOF_STOPPROCESS);
if (fProtocol > 8) {
m << p->GetEventsProcessed() << abort;
} else {
m << p->GetEventsProcessed();
}
fSocket->Send(m);
}
FinalizeQuery(p, pq);
if (p->GetExitStatus() != TProofPlayer::kAborted) {
if (fProtocol > 6) {
PDB(kGlobal, 2) Info("HandleProcess","Sending results");
fSocket->SendObject(pq, kPROOF_OUTPUTLIST);
} else {
PDB(kGlobal, 2) Info("HandleProcess","Sending output list");
fSocket->SendObject(p->GetOutputList(), kPROOF_OUTPUTLIST);
}
} else {
fSocket->SendObject(0,kPROOF_OUTPUTLIST);
}
if (p->GetExitStatus() == TProofPlayer::kAborted)
RemoveQuery(pq);
if (fProof != 0)
fProof->SetPlayer(0);
SafeDelete(p);
}
fSocket->Send(kPROOF_SETIDLE);
} else {
fIdle = kFALSE;
if (IsMaster() && IsParallel()) {
p = fProof->MakePlayer();
} else {
p = new TProofPlayerSlave(fSocket);
if (IsMaster())
fProof->SetPlayer(p);
}
fPlayer = p;
if (dset->IsA() == TDSetProxy::Class())
((TDSetProxy*)dset)->SetProofServ(this);
TIter next(input);
for (TObject *o; (o = next()); ) {
PDB(kGlobal, 2) Info("HandleProcess", "adding: %s", o->GetName());
p->AddInput(o);
}
p->Process(dset, filename, opt, nentries, first);
TMessage m(kPROOF_STOPPROCESS);
m << p->GetEventsProcessed();
fSocket->Send(m);
if (p->GetExitStatus() != TProofPlayer::kAborted) {
PDB(kGlobal, 2) Info("HandleProcess","Sending output list");
fSocket->SendObject(p->GetOutputList(), kPROOF_OUTPUTLIST);
} else {
fSocket->SendObject(0,kPROOF_OUTPUTLIST);
}
SafeDelete(dset);
SafeDelete(input)
SafeDelete(p);
}
fPlayer = 0;
fIdle = kTRUE;
PDB(kGlobal, 1) Info("HandleProcess", "Done");
return;
}
void TProofServ::HandleQueryList(TMessage *mess)
{
PDB(kGlobal, 1)
Info("HandleQueryList", "Enter");
Bool_t all;
(*mess) >> all;
TList *ql = new TList;
Int_t ntot = 0;
if (all) {
TString qdir = fQueryDir;
Int_t idx = qdir.Index("session-");
if (idx != kNPOS)
qdir.Remove(idx);
ScanPreviousQueries(qdir);
if (fPreviousQueries) {
TIter nxq(fPreviousQueries);
TProofQueryResult *pqr = 0;
while ((pqr = (TProofQueryResult *)nxq())) {
ntot++;
pqr->fSeqNum = ntot;
ql->Add(pqr);
}
}
}
Int_t npre = ntot;
if (fQueries) {
TIter nxq(fQueries);
TProofQueryResult *pqr = 0;
TQueryResult *pqm = 0;
while ((pqr = (TProofQueryResult *)nxq())) {
ntot++;
pqm = pqr->CloneInfo();
pqm->fSeqNum = ntot;
ql->Add(pqm);
}
}
TMessage m(kPROOF_QUERYLIST);
m << npre << fDrawQueries << ql;
fSocket->Send(m);
delete ql;
return;
}
void TProofServ::HandleRemove(TMessage *mess)
{
PDB(kGlobal, 1)
Info("HandleRemove", "Enter");
TString queryref;
(*mess) >> queryref;
if (queryref == "cleanupqueue") {
Int_t pend = fWaitingQueries->GetSize();
fWaitingQueries->Delete();
Info("HandleRemove", "%d queries removed from the waiting list", pend);
return;
}
TProofLockPath *lck = 0;
if (LockSession(queryref, &lck) == 0) {
RemoveQuery(queryref);
if (lck) {
gSystem->Unlink(lck->GetName());
SafeDelete(lck);
}
return;
}
Info("HandleRemove",
"query %s could not be removed (unable to lock session)", queryref.Data());
return;
}
void TProofServ::HandleRetrieve(TMessage *mess)
{
PDB(kGlobal, 1)
Info("HandleRetrieve", "Enter");
TString queryref;
(*mess) >> queryref;
Int_t qry = -1;
TString qdir;
TProofQueryResult *pqr = LocateQuery(queryref, qry, qdir);
if (!pqr || qry < 0) {
TString fout = qdir;
fout += "/query-result.root";
TFile *f = TFile::Open(fout,"READ");
pqr = 0;
if (f) {
f->ReadKeys();
TIter nxk(f->GetListOfKeys());
TKey *k = 0;
while ((k = (TKey *)nxk())) {
if (!strcmp(k->GetClassName(), "TProofQueryResult")) {
pqr = (TProofQueryResult *) f->Get(k->GetName());
if (pqr) {
fSocket->SendObject(pqr, kPROOF_RETRIEVE);
} else {
Info("HandleRetrieve",
"query not found in file %s",fout.Data());
fSocket->SendObject(0, kPROOF_RETRIEVE);
}
break;
}
}
f->Close();
delete f;
} else {
Info("HandleRetrieve",
"file cannot be open (%s)",fout.Data());
fSocket->SendObject(0, kPROOF_RETRIEVE);
return;
}
} else {
fSocket->SendObject(pqr, kPROOF_RETRIEVE);
}
return;
}
void TProofServ::HandleLibIncPath(TMessage *mess)
{
TString type;
Bool_t add;
TString path;
(*mess) >> type >> add >> path;
if ((type != "lib") && (type != "inc")) {
Error("HandleLibIncPath","unknown action type: %s", type.Data());
return;
}
path.ReplaceAll(","," ");
TObjArray *op = 0;
if (path.Length() > 0 && path != "-") {
if (!(op = path.Tokenize(" "))) {
Error("HandleLibIncPath","decomposing path %s", path.Data());
return;
}
}
if (add) {
if (type == "lib") {
TIter nxl(op, kIterBackward);
TObjString *lib = 0;
while ((lib = (TObjString *) nxl())) {
TString xlib = lib->GetName();
gSystem->ExpandPathName(xlib);
if (!gSystem->AccessPathName(xlib, kReadPermission)) {
TString newlibpath = gSystem->GetDynamicPath();
Int_t pos = 0;
if (newlibpath.BeginsWith(".:"))
pos = 2;
if (newlibpath.Index(xlib) == kNPOS) {
newlibpath.Insert(pos,Form("%s:", xlib.Data()));
gSystem->SetDynamicPath(newlibpath);
}
} else {
Info("HandleLibIncPath",
"libpath %s does not exist or cannot be read - not added", xlib.Data());
}
}
if (IsMaster())
fProof->AddDynamicPath(path);
} else {
TIter nxi(op);
TObjString *inc = 0;
while ((inc = (TObjString *) nxi())) {
TString xinc = inc->GetName();
gSystem->ExpandPathName(xinc);
if (!gSystem->AccessPathName(xinc, kReadPermission)) {
TString curincpath = gSystem->GetIncludePath();
if (curincpath.Index(xinc) == kNPOS)
gSystem->AddIncludePath(Form("-I%s", xinc.Data()));
} else
Info("HandleLibIncPath",
"incpath %s does not exist or cannot be read - not added", xinc.Data());
}
if (IsMaster())
fProof->AddIncludePath(path);
}
} else {
if (type == "lib") {
TIter nxl(op);
TObjString *lib = 0;
while ((lib = (TObjString *) nxl())) {
TString xlib = lib->GetName();
gSystem->ExpandPathName(xlib);
TString newlibpath = gSystem->GetDynamicPath();
newlibpath.ReplaceAll(Form("%s:", xlib.Data()),"");
gSystem->SetDynamicPath(newlibpath);
}
if (IsMaster())
fProof->RemoveDynamicPath(path);
} else {
TIter nxi(op);
TObjString *inc = 0;
while ((inc = (TObjString *) nxi())) {
TString newincpath = gSystem->GetIncludePath();
newincpath.ReplaceAll(Form("-I%s", inc->GetName()),"");
newincpath.ReplaceAll(gInterpreter->GetIncludePath(),"");
gSystem->SetIncludePath(newincpath);
}
if (IsMaster())
fProof->RemoveIncludePath(path);
}
}
}
void TProofServ::HandleCheckFile(TMessage *mess)
{
TString filenam;
TMD5 md5;
UInt_t opt = TVirtualProof::kUntar;
(*mess) >> filenam >> md5;
if ((mess->BufferSize() > mess->Length()) && (fProtocol > 8))
(*mess) >> opt;
if (filenam.BeginsWith("-")) {
Int_t st = 0;
Bool_t err = kFALSE;
filenam = filenam.Strip(TString::kLeading, '-');
TString packnam = filenam;
packnam.Remove(packnam.Length() - 4);
TMD5 *md5local = TMD5::FileChecksum(fPackageDir + "/" + filenam);
if (md5local && md5 == (*md5local)) {
if ((opt & TVirtualProof::kRemoveOld)) {
st = gSystem->Exec(Form("%s %s/%s", kRM, fPackageDir.Data(),
packnam.Data()));
if (st)
Error("HandleCheckFile", "failure executing: %s %s/%s",
kRM, fPackageDir.Data(), packnam.Data());
}
char *gunzip = gSystem->Which(gSystem->Getenv("PATH"),kGUNZIP,
kExecutePermission);
if (gunzip) {
st = gSystem->Exec(Form(kUNTAR, gunzip, fPackageDir.Data(),
filenam.Data(), fPackageDir.Data()));
if (st)
Error("HandleCheckFile", "failure executing: %s",
Form(kUNTAR, gunzip, fPackageDir.Data(),
filenam.Data(), fPackageDir.Data()));
delete [] gunzip;
} else
Error("HandleCheckFile", "%s not found",
kGUNZIP);
if (gSystem->AccessPathName(fPackageDir + "/" + packnam, kWritePermission)) {
fSocket->Send(kPROOF_FATAL);
err = kTRUE;
PDB(kPackage, 1)
Info("HandleCheckFile",
"package %s did not unpack into %s", filenam.Data(),
packnam.Data());
} else {
TString md5f = fPackageDir + "/" + packnam + "/PROOF-INF/md5.txt";
TMD5::WriteChecksum(md5f, md5local);
fSocket->Send(kPROOF_CHECKFILE);
PDB(kPackage, 1)
Info("HandleCheckFile",
"package %s installed on node", filenam.Data());
}
} else {
fSocket->Send(kPROOF_FATAL);
err = kTRUE;
}
if (!IsMaster() || err) {
gSystem->Exec(Form("%s %s/%s", kRM, fPackageDir.Data(),
filenam.Data()));
fPackageLock->Unlock();
} else {
fPackageLock->Unlock();
fProof->UploadPackage(fPackageDir + "/" + filenam);
}
delete md5local;
} else if (filenam.BeginsWith("+")) {
filenam = filenam.Strip(TString::kLeading, '+');
TString packnam = filenam;
packnam.Remove(packnam.Length() - 4);
TString md5f = fPackageDir + "/" + packnam + "/PROOF-INF/md5.txt";
fPackageLock->Lock();
TMD5 *md5local = TMD5::ReadChecksum(md5f);
if (md5local && md5 == (*md5local)) {
fPackageLock->Unlock();
fSocket->Send(kPROOF_CHECKFILE);
PDB(kPackage, 1)
Info("HandleCheckFile",
"package %s already on node", filenam.Data());
if (IsMaster())
fProof->UploadPackage(fPackageDir + "/" + filenam);
} else {
fSocket->Send(kPROOF_FATAL);
PDB(kPackage, 1)
Info("HandleCheckFile",
"package %s not yet on node", filenam.Data());
}
delete md5local;
} else if (filenam.BeginsWith("=")) {
filenam = filenam.Strip(TString::kLeading, '=');
TString packnam = filenam;
packnam.Remove(packnam.Length() - 4);
TString md5f = fPackageDir + "/" + packnam + "/PROOF-INF/md5.txt";
fPackageLock->Lock();
TMD5 *md5local = TMD5::ReadChecksum(md5f);
fPackageLock->Unlock();
if (md5local && md5 == (*md5local)) {
fSocket->Send(kPROOF_CHECKFILE);
PDB(kPackage, 1)
Info("HandleCheckFile",
"package %s already on node", filenam.Data());
if (IsMaster())
fProof->UploadPackage(fPackageDir + "/" + filenam);
} else {
fSocket->Send(kPROOF_FATAL);
PDB(kPackage, 1)
Info("HandleCheckFile",
"package %s not yet on node", filenam.Data());
}
delete md5local;
} else {
TString cachef = fCacheDir + "/" + filenam;
fCacheLock->Lock();
TMD5 *md5local = TMD5::FileChecksum(cachef);
if (md5local && md5 == (*md5local)) {
gSystem->Exec(Form("%s %s .", kCP, cachef.Data()));
fSocket->Send(kPROOF_CHECKFILE);
PDB(kPackage, 1)
Info("HandleCheckFile", "file %s already on node", filenam.Data());
} else {
fSocket->Send(kPROOF_FATAL);
PDB(kPackage, 1)
Info("HandleCheckFile", "file %s not yet on node", filenam.Data());
}
delete md5local;
fCacheLock->Unlock();
}
}
Int_t TProofServ::HandleCache(TMessage *mess)
{
PDB(kGlobal, 1)
Info("HandleCache", "Enter");
Int_t status = 0;
Int_t type = 0;
Bool_t all = kFALSE;
TMessage msg;
TString package, pdir, ocwd;
(*mess) >> type;
switch (type) {
case TProof::kShowCache:
(*mess) >> all;
printf("*** File cache %s:%s ***\n", gSystem->HostName(),
fCacheDir.Data());
fflush(stdout);
gSystem->Exec(Form("%s %s", kLS, fCacheDir.Data()));
if (IsMaster() && all)
fProof->ShowCache(all);
break;
case TProof::kClearCache:
fCacheLock->Lock();
gSystem->Exec(Form("%s %s, kRM, fCacheDir.Data()));
fCacheLock->Unlock();
if (IsMaster())
fProof->ClearCache();
break;
case TProof::kShowPackages:
(*mess) >> all;
printf("*** Package cache %s:%s ***\n", gSystem->HostName(),
fPackageDir.Data());
fflush(stdout);
gSystem->Exec(Form("%s %s", kLS, fPackageDir.Data()));
if (IsMaster() && all)
fProof->ShowPackages(all);
break;
case TProof::kClearPackages:
status = UnloadPackages();
if (status == 0) {
fPackageLock->Lock();
gSystem->Exec(Form("%s %s, kRM, fPackageDir.Data()));
fPackageLock->Unlock();
if (IsMaster())
status = fProof->ClearPackages();
}
break;
case TProof::kClearPackage:
(*mess) >> package;
status = UnloadPackage(package);
if (status == 0) {
fPackageLock->Lock();
// remove package directory and par file
gSystem->Exec(Form("%s %s/%s", kRM, fPackageDir.Data(),
package.Data()));
if (IsMaster())
gSystem->Exec(Form("%s %s/%s.par", kRM, fPackageDir.Data(),
package.Data()));
fPackageLock->Unlock();
if (IsMaster())
status = fProof->ClearPackage(package);
}
break;
case TProof::kBuildPackage:
(*mess) >> package;
fPackageLock->Lock();
pdir = fPackageDir + "/" + package;
if (gSystem->AccessPathName(pdir)) {
Error("HandleCache", "package %s does not exist",
package.Data());
status = -1;
} else if (gSystem->AccessPathName(pdir + "/PROOF-INF")) {
Error("HandleCache", "package %s does not have a PROOF-INF directory",
package.Data());
status = -1;
}
if (!status) {
PDB(kPackage, 1)
Info("HandleCache",
"package %s exists and has PROOF-INF directory", package.Data());
ocwd = gSystem->WorkingDirectory();
gSystem->ChangeDirectory(pdir);
if (!gSystem->AccessPathName(pdir + "/PROOF-INF/BUILD.sh")) {
if (gSystem->Exec("PROOF-INF/BUILD.sh"))
status = -1;
}
gSystem->ChangeDirectory(ocwd);
}
fPackageLock->Unlock();
if (!status) {
if (IsMaster())
fProof->BuildPackage(package);
PDB(kPackage, 1)
Info("HandleCache",
"package %s successfully built", package.Data());
}
break;
case TProof::kLoadPackage:
(*mess) >> package;
pdir = fPackageDir + "/" + package;
ocwd = gSystem->WorkingDirectory();
gSystem->ChangeDirectory(pdir);
if (!gSystem->AccessPathName(pdir + "/PROOF-INF/SETUP.C")) {
gROOT->Macro("PROOF-INF/SETUP.C");
}
gSystem->ChangeDirectory(ocwd);
gSystem->Symlink(pdir, package);
gSystem->AddIncludePath(TString("-I") + package);
gROOT->ProcessLine(TString(".include ") + package);
if (!status) {
fEnabledPackages->Add(new TObjString(package));
if (IsMaster())
fProof->LoadPackage(package);
PDB(kPackage, 1)
Info("HandleCache",
"package %s successfully loaded", package.Data());
}
break;
case TProof::kShowEnabledPackages:
(*mess) >> all;
if (IsMaster()) {
if (all)
printf("*** Enabled packages on master %s on %s\n",
fOrdinal.Data(), gSystem->HostName());
else
printf("*** Enabled packages ***\n");
} else {
printf("*** Enabled packages on slave %s on %s\n",
fOrdinal.Data(), gSystem->HostName());
}
{
TIter next(fEnabledPackages);
while (TObjString *str = (TObjString*) next())
printf("%s\n", str->GetName());
}
if (IsMaster() && all)
fProof->ShowEnabledPackages(all);
break;
case TProof::kShowSubCache:
(*mess) >> all;
if (IsMaster() && all)
fProof->ShowCache(all);
break;
case TProof::kClearSubCache:
if (IsMaster())
fProof->ClearCache();
break;
case TProof::kShowSubPackages:
(*mess) >> all;
if (IsMaster() && all)
fProof->ShowPackages(all);
break;
case TProof::kDisableSubPackages:
if (IsMaster())
fProof->DisablePackages();
break;
case TProof::kDisableSubPackage:
(*mess) >> package;
if (IsMaster())
fProof->DisablePackage(package);
break;
case TProof::kBuildSubPackage:
(*mess) >> package;
if (IsMaster())
fProof->BuildPackage(package);
break;
case TProof::kUnloadPackage:
(*mess) >> package;
status = UnloadPackage(package);
if (IsMaster() && status == 0)
status = fProof->UnloadPackage(package);
break;
case TProof::kDisablePackage:
(*mess) >> package;
fPackageLock->Lock();
gSystem->Exec(Form("%s %s/%s", kRM, fPackageDir.Data(),
package.Data()));
gSystem->Exec(Form("%s %s/%s.par", kRM, fPackageDir.Data(),
package.Data()));
fPackageLock->Unlock();
if (IsMaster())
fProof->DisablePackage(package);
break;
case TProof::kUnloadPackages:
status = UnloadPackages();
if (IsMaster() && status == 0)
status = fProof->UnloadPackages();
break;
case TProof::kDisablePackages:
fPackageLock->Lock();
gSystem->Exec(Form("%s %s, kRM, fPackageDir.Data()));
fPackageLock->Unlock();
if (IsMaster())
fProof->DisablePackages();
break;
case TProof::kListEnabledPackages:
msg.Reset(kPROOF_PACKAGE_LIST);
msg << type << fEnabledPackages;
fSocket->Send(msg);
break;
case TProof::kListPackages:
{
TList *pack = new TList;
void *dir = gSystem->OpenDirectory(fPackageDir);
if (dir) {
TString pac(gSystem->GetDirEntry(dir));
while (pac.Length() > 0) {
if (pac.EndsWith(".par")) {
pac.ReplaceAll(".par","");
pack->Add(new TObjString(pac.Data()));
}
pac = gSystem->GetDirEntry(dir);
}
}
gSystem->FreeDirectory(dir);
msg.Reset(kPROOF_PACKAGE_LIST);
msg << type << pack;
fSocket->Send(msg);
}
break;
default:
Error("HandleCache", "unknown type %d", type);
break;
}
return status;
}
void TProofServ::HandleWorkerLists(TMessage *mess)
{
PDB(kGlobal, 1)
Info("HandleWorkerLists", "Enter");
Int_t type = 0;
TString ord;
(*mess) >> type;
switch (type) {
case TProof::kActivateWorker:
(*mess) >> ord;
if (fProof) {
fProof->ActivateWorker(ord);
if (ord == "*")
Info("HandleWorkerList","all workers (re-)activated");
else
Info("HandleWorkerList","worker %s (re-)activated", ord.Data());
} else {
Warning("HandleWorkerList","undefined PROOF session: protocol error?");
}
break;
case TProof::kDeactivateWorker:
(*mess) >> ord;
if (fProof) {
fProof->DeactivateWorker(ord);
if (ord == "*")
Info("HandleWorkerList","all workers deactivated");
else
Info("HandleWorkerList","worker %s deactivated", ord.Data());
} else {
Warning("HandleWorkerList","undefined PROOF session: protocol error?");
}
break;
default:
Warning("HandleWorkerList","unknown action type (%d)", type);
}
}
Int_t TProofServ::HandleDataSets(TMessage *mess)
{
PDB(kGlobal, 1)
Info("HandleDataSets", "Enter");
TList *previousDataSet = 0;
TString dataSetName;
Int_t type = 0;
(*mess) >> type;
switch (type) {
case TProof::kCheckDataSetName:
{
TString fileListName;
(*mess) >> fileListName;
char *fileListPath =
Form("%s/%s.root", fDataSetDir.Data(), fileListName.Data());
if (gSystem->AccessPathName(fileListPath, kFileExists) == kFALSE) {
fSocket->Send(kMESS_NOTOK);
} else {
fSocket->Send("", kMESS_OK);
}
}
break;
case TProof::kAppendDataSet:
{
(*mess) >> dataSetName;
previousDataSet = GetDataSet(dataSetName.Data());
}
case TProof::kCreateDataSet:
{
if (type == TProof::kCreateDataSet) {
(*mess) >> dataSetName;
}
char *fileListPath =
Form("%s/%s.root", fDataSetDir.Data(), dataSetName.Data());
TList *fileList =
(TList *) (mess->ReadObject(TList::Class()));
if (previousDataSet) {
TIter nextOldFile(previousDataSet);
while (TFileInfo *obj = (TFileInfo*)nextOldFile())
fileList->Add(obj);
delete previousDataSet;
}
if (fileList->GetSize() > 0) {
fileList->Sort();
TList *newFileList = new TList();
TIter nextFile(fileList);
TFileInfo *prevFile = (TFileInfo*)nextFile();
newFileList->Add(prevFile);
while (TFileInfo *obj = (TFileInfo*)nextFile())
if (prevFile->Compare(obj)) {
newFileList->Add(obj);
prevFile = obj;
}
if (gSystem->AccessPathName(gSystem->DirName(fileListPath)))
gSystem->mkdir(gSystem->DirName(fileListPath), kTRUE);
TFile *f = TFile::Open(fileListPath, "RECREATE");
if (f) {
f->cd();
newFileList->Write("fileList", TObject::kSingleKey);
f->Close();
fSocket->Send(kMESS_OK);
} else {
fSocket->Send(kMESS_NOTOK);
Error("HandleSocketInput",
"can't open dataset file for writing");
}
delete f;
delete newFileList;
fileList->SetOwner();
delete fileList;
} else {
fSocket->Send(kMESS_NOTOK);
Printf("Can not save an empty list.");
}
}
break;
case TProof::kGetDataSets:
{
TString dir;
(*mess) >> dir;
char *dataSetDirPath;
void *dataSetDir;
if (dir.Length())
if (strstr(dir, "public") == dir)
dataSetDirPath =
gSystem->ExpandPathName(Form("%s/%s/public/",
kPROOF_WorkDir,
kPROOF_DataSetDir));
else {
char *userName = (char *)malloc(strlen(dir));
strcpy(userName, dir.Data() + 1);
strtok(userName, "/");
dataSetDirPath =
gSystem->ExpandPathName(Form("~%s/proof/%s/public/",
userName,
kPROOF_DataSetDir));
}
else
dataSetDirPath = Form("%s/%s",
gSystem->ExpandPathName(kPROOF_WorkDir),
kPROOF_DataSetDir);
if ((dataSetDir = gSystem->OpenDirectory(dataSetDirPath))) {
TRegexp rg(".*.root");
TList *fileList = new TList();
const char *ent;
while ((ent = gSystem->GetDirEntry(dataSetDir))) {
if (TString(ent).Index(rg) != kNPOS)
fileList->Add(new TObjString(TString(ent, strlen(ent) - 5)));
}
fileList->Sort();
fSocket->SendObject(fileList, kMESS_OBJECT);
fileList->SetOwner();
delete fileList;
} else {
Printf("Can not open the dataset directory.");
fSocket->Send(kMESS_NOTOK);
}
}
break;
case TProof::kGetDataSet:
{
TString name;
(*mess) >> name;
if (TList *fileList = GetDataSet(name.Data())) {
fSocket->SendObject(fileList, kMESS_OK);
delete fileList;
} else
fSocket->Send(kMESS_NOTOK);
}
break;
case TProof::kRemoveDataSet:
{
TString name;
(*mess) >> name;
const char *fileListPath = Form("%s/%s.root",fDataSetDir.Data(),name.Data());
TProofLockPathGuard dslguard(fDataSetLock);
if (gSystem->AccessPathName(fileListPath, kFileExists) == kFALSE) {
if (gSystem->Unlink(fileListPath)) {
Printf("Error removing dataset %s", name.Data());
fSocket->Send(kMESS_NOTOK);
} else
fSocket->Send(kMESS_OK);
} else {
Printf("The dataset does not exist");
fSocket->Send(kMESS_NOTOK);
}
}
break;
case TProof::kVerifyDataSet:
{
TString name;
(*mess) >> name;
if (TList *fileList = GetDataSet(name.Data())) {
TList *missingFileList = new TList();
TIter next(fileList);
TFileInfo *fileInfo;
while ((fileInfo = (TFileInfo *)next())) {
if (gSystem->AccessPathName(fileInfo->GetFirstUrl()->GetUrl(),
kFileExists) != kFALSE)
missingFileList->Add(fileInfo);
}
fSocket->SendObject(missingFileList, kMESS_OK);
} else
fSocket->Send(kMESS_NOTOK);
}
break;
default:
Error("HandleDataSets", "unknown type %d", type);
break;
}
return 0;
}
TProofServ::EQueryAction TProofServ::GetWorkers(TList *workers,
Int_t & )
{
if (!workers) {
Error("GetWorkers", "output list undefined");
return kQueryStop;
}
TProofResourcesStatic *resources =
new TProofResourcesStatic(fConfDir, fConfFile);
fConfFile = resources->GetFileName();
PDB(kGlobal,1)
Info("GetWorkers", "using PROOF config file: %s", fConfFile.Data());
TProofNodeInfo *master = resources->GetMaster();
if (master)
fImage = master->GetImage();
if (!master || (fImage.Length() == 0)) {
Error("GetWorkers",
"no appropriate master line found in %s", fConfFile.Data());
return kQueryStop;
}
if (resources->GetWorkers()) {
TProofNodeInfo *ni = 0;
TIter nw(resources->GetWorkers());
while ((ni = (TProofNodeInfo *) nw()))
workers->Add(new TProofNodeInfo(*ni));
}
return kQueryOK;
}
TList *TProofServ::GetDataSet(const char *name)
{
const char *fileListPath;
if (strchr(name, '~') == name) {
char *nameCopy = new char[strlen(name)];
strcpy(nameCopy, name + 1);
char *userName = strtok(nameCopy, "/");
if (strcmp(strtok(0, "/"), "public"))
return 0;
fileListPath =
gSystem->ExpandPathName(Form("~%s/proof/%s/public/%s.root",
userName,
kPROOF_DataSetDir,
strtok(0, "/")));
delete[] nameCopy;
} else if (strchr(name, '/') && strstr(name, "public") != name) {
Printf("Dataset name should be of form [[~user/]public/]dataset");
return 0;
} else
fileListPath = Form("%s/%s/%s.root",
gSystem->ExpandPathName(kPROOF_WorkDir),
kPROOF_DataSetDir,
name);
TList *fileList = 0;
if (gSystem->AccessPathName(fileListPath, kFileExists) == kFALSE) {
TFile *f = TFile::Open(fileListPath);
f->cd();
fileList = (TList *) f->Get("fileList");
f->Close();
delete f;
if (strchr(name, '~') == name)
delete[] fileListPath;
}
return fileList;
}
void TProofServ::ErrorHandler(Int_t level, Bool_t abort, const char *location,
const char *msg)
{
if (level < gErrorIgnoreLevel)
return;
static TString syslogService;
if (syslogService.IsNull()) {
syslogService = gProofServ != 0 ? gProofServ->GetService() : "proof";
gSystem->Openlog(syslogService, kLogPid | kLogCons, kLogLocal5);
} else if (gProofServ != 0 && syslogService != gProofServ->GetService()) {
syslogService = gProofServ->GetService();
gSystem->Openlog(syslogService, kLogPid | kLogCons, kLogLocal5);
}
const char *type = 0;
ELogLevel loglevel = kLogInfo;
if (level >= kInfo) {
loglevel = kLogInfo;
type = "Info";
}
if (level >= kWarning) {
loglevel = kLogWarning;
type = "Warning";
}
if (level >= kError) {
loglevel = kLogErr;
type = "Error";
}
if (level >= kBreak) {
loglevel = kLogErr;
type = "*** Break ***";
}
if (level >= kSysError) {
loglevel = kLogErr;
type = "SysError";
}
if (level >= kFatal) {
loglevel = kLogErr;
type = "Fatal";
}
TString node = "proof";
TString user = "unknown";
if (gProofServ) {
node = gProofServ->IsMaster() ? "master" : "slave";
node += gProofServ->GetOrdinal();
user = gProofServ->GetUser();
}
TString buf;
if (!location || strlen(location) == 0 ||
(level >= kBreak && level < kSysError)) {
fprintf(stderr, "%s on %s: %s\n", type, node.Data(), msg);
buf.Form("%s:%s:%s:%s", user.Data(), node.Data(), type, msg);
} else {
fprintf(stderr, "%s in <%s> on %s: %s\n", type, location, node.Data(), msg);
buf.Form("%s:%s:%s:<%s>:%s", user.Data(), node.Data(), type, location, msg);
}
fflush(stderr);
gSystem->Syslog(loglevel, buf);
if (abort) {
static Bool_t recursive = kFALSE;
if (gProofServ != 0 && !recursive) {
recursive = kTRUE;
gProofServ->GetSocket()->Send(kPROOF_FATAL);
recursive = kFALSE;
}
fprintf(stderr, "aborting\n");
fflush(stderr);
gSystem->StackTrace();
gSystem->Abort();
}
}
Int_t TProofLockPath::Lock()
{
const char *pname = GetName();
if (gSystem->AccessPathName(pname))
fLockId = open(pname, O_CREAT|O_RDWR, 0644);
else
fLockId = open(pname, O_RDWR);
if (fLockId == -1) {
SysError("Lock", "cannot open lock file %s", pname);
return -1;
}
#if !defined(R__WIN32) && !defined(R__WINGCC)
if (lockf(fLockId, F_LOCK, (off_t) 1) == -1) {
SysError("Lock", "error locking %s", pname);
close(fLockId);
fLockId = -1;
return -1;
}
#endif
PDB(kPackage, 2)
Info("Lock", "file %s locked", pname);
return 0;
}
Int_t TProofLockPath::Unlock()
{
if (!IsLocked())
return 0;
lseek(fLockId, 0, SEEK_SET);
#if !defined(R__WIN32) && !defined(R__WINGCC)
if (lockf(fLockId, F_ULOCK, (off_t)1) == -1) {
SysError("Unlock", "error unlocking %s", GetName());
close(fLockId);
fLockId = -1;
return -1;
}
#endif
PDB(kPackage, 2)
Info("Unlock", "file %s unlocked", GetName());
close(fLockId);
fLockId = -1;
return 0;
}
ROOT page - Class index - Class Hierarchy - Top of the page
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.