+ //
+ // removes local file
+ //
+
+ TString command(Form("rm -f %s", filename));
+
+ Int_t result = gSystem->Exec(command.Data());
+ if(result != 0)
+ {
+ Log("SHUTTLE", Form("RemoveFile - %s: Cannot remove file %s!",
+ fCurrentDetector.Data(), filename));
+ }
+}
+
+//______________________________________________________________________________________________
+AliShuttleStatus* AliShuttle::ReadShuttleStatus()
+{
+ //
+ // Reads the AliShuttleStatus from the CDB
+ //
+
+ if (fStatusEntry){
+ delete fStatusEntry;
+ fStatusEntry = 0;
+ }
+
+ fStatusEntry = AliCDBManager::Instance()->GetStorage(GetLocalCDB())
+ ->Get(Form("/SHUTTLE/STATUS/%s", fCurrentDetector.Data()), GetCurrentRun());
+
+ if (!fStatusEntry) return 0;
+ fStatusEntry->SetOwner(1);
+
+ AliShuttleStatus* status = dynamic_cast<AliShuttleStatus*> (fStatusEntry->GetObject());
+ if (!status) {
+ AliError("Invalid object stored to CDB!");
+ return 0;
+ }
+
+ return status;
+}
+
+//______________________________________________________________________________________________
+Bool_t AliShuttle::WriteShuttleStatus(AliShuttleStatus* status)
+{
+ //
+ // writes the status for one subdetector
+ //
+
+ if (fStatusEntry){
+ delete fStatusEntry;
+ fStatusEntry = 0;
+ }
+
+ Int_t run = GetCurrentRun();
+
+ AliCDBId id(AliCDBPath("SHUTTLE", "STATUS", fCurrentDetector), run, run);
+
+ fStatusEntry = new AliCDBEntry(status, id, new AliCDBMetaData);
+ fStatusEntry->SetOwner(1);
+
+ UInt_t result = AliCDBManager::Instance()->GetStorage(fgkLocalCDB)->Put(fStatusEntry);
+
+ if (!result) {
+ Log("SHUTTLE", Form("WriteShuttleStatus - Failed for %s, run %d",
+ fCurrentDetector.Data(), run));
+ return kFALSE;
+ }
+
+ SendMLInfo();
+
+ return kTRUE;
+}
+
+//______________________________________________________________________________________________
+void AliShuttle::UpdateShuttleStatus(AliShuttleStatus::Status newStatus, Bool_t increaseCount)
+{
+ //
+ // changes the AliShuttleStatus for the given detector and run to the given status
+ //
+
+ if (!fStatusEntry){
+ AliError("UNEXPECTED: fStatusEntry empty");
+ return;
+ }
+
+ AliShuttleStatus* status = dynamic_cast<AliShuttleStatus*> (fStatusEntry->GetObject());
+
+ if (!status){
+ Log("SHUTTLE", "UpdateShuttleStatus - UNEXPECTED: status could not be read from current CDB entry");
+ return;
+ }
+
+ TString actionStr = Form("UpdateShuttleStatus - %s: Changing state from %s to %s",
+ fCurrentDetector.Data(),
+ status->GetStatusName(),
+ status->GetStatusName(newStatus));
+ Log("SHUTTLE", actionStr);
+ SetLastAction(actionStr);
+
+ status->SetStatus(newStatus);
+ if (increaseCount) status->IncreaseCount();
+
+ AliCDBManager::Instance()->GetStorage(fgkLocalCDB)->Put(fStatusEntry);
+
+ SendMLInfo();
+}
+
+//______________________________________________________________________________________________
+void AliShuttle::SendMLInfo()
+{
+ //
+ // sends ML information about the current status of the current detector being processed
+ //
+
+ AliShuttleStatus* status = dynamic_cast<AliShuttleStatus*> (fStatusEntry->GetObject());
+
+ if (!status){
+ Log("SHUTTLE", "SendMLInfo - UNEXPECTED: status could not be read from current CDB entry");
+ return;
+ }
+
+ TMonaLisaText mlStatus(Form("%s_status", fCurrentDetector.Data()), status->GetStatusName());
+ TMonaLisaValue mlRetryCount(Form("%s_count", fCurrentDetector.Data()), status->GetCount());
+
+ TList mlList;
+ mlList.Add(&mlStatus);
+ mlList.Add(&mlRetryCount);
+
+ TString mlID;
+ mlID.Form("%d", GetCurrentRun());
+ fMonaLisa->SendParameters(&mlList, mlID);
+}
+
+//______________________________________________________________________________________________
+Bool_t AliShuttle::ContinueProcessing()
+{
+ // this function reads the AliShuttleStatus information from CDB and
+ // checks if the processing should be continued
+ // if yes it returns kTRUE and updates the AliShuttleStatus with nextStatus
+
+ if (!fConfig->HostProcessDetector(fCurrentDetector)) return kFALSE;
+
+ AliPreprocessor* aPreprocessor =
+ dynamic_cast<AliPreprocessor*> (fPreprocessorMap.GetValue(fCurrentDetector));
+ if (!aPreprocessor)
+ {
+ Log("SHUTTLE", Form("ContinueProcessing - %s: no preprocessor registered", fCurrentDetector.Data()));
+ return kFALSE;
+ }
+
+ AliShuttleLogbookEntry::Status entryStatus =
+ fLogbookEntry->GetDetectorStatus(fCurrentDetector);
+
+ if(entryStatus != AliShuttleLogbookEntry::kUnprocessed) {
+ Log("SHUTTLE", Form("ContinueProcessing - %s is %s",
+ fCurrentDetector.Data(),
+ fLogbookEntry->GetDetectorStatusName(entryStatus)));
+ return kFALSE;
+ }
+
+ // if we get here, according to Shuttle logbook subdetector is in UNPROCESSED state
+
+ // check if current run is first unprocessed run for current detector
+ if (fConfig->StrictRunOrder(fCurrentDetector) &&
+ !fFirstUnprocessed[GetDetPos(fCurrentDetector)])
+ {
+ if (fTestMode == kNone)
+ {
+ Log("SHUTTLE", Form("ContinueProcessing - %s requires strict run ordering"
+ " but this is not the first unprocessed run!"));
+ return kFALSE;
+ }
+ else
+ {
+ Log("SHUTTLE", Form("ContinueProcessing - In TESTMODE - "
+ "Although %s requires strict run ordering "
+ "and this is not the first unprocessed run, "
+ "the SHUTTLE continues"));
+ }
+ }
+
+ AliShuttleStatus* status = ReadShuttleStatus();
+ if (!status) {
+ // first time
+ Log("SHUTTLE", Form("ContinueProcessing - %s: Processing first time",
+ fCurrentDetector.Data()));
+ status = new AliShuttleStatus(AliShuttleStatus::kStarted);
+ return WriteShuttleStatus(status);
+ }
+
+ // The following two cases shouldn't happen if Shuttle Logbook was correctly updated.
+ // If it happens it may mean Logbook updating failed... let's do it now!
+ if (status->GetStatus() == AliShuttleStatus::kDone ||
+ status->GetStatus() == AliShuttleStatus::kFailed){
+ Log("SHUTTLE", Form("ContinueProcessing - %s is already %s. Updating Shuttle Logbook",
+ fCurrentDetector.Data(),
+ status->GetStatusName(status->GetStatus())));
+ UpdateShuttleLogbook(fCurrentDetector.Data(),
+ status->GetStatusName(status->GetStatus()));
+ return kFALSE;
+ }
+
+ if (status->GetStatus() == AliShuttleStatus::kStoreStarted || status->GetStatus() == AliShuttleStatus::kStoreError) {
+ Log("SHUTTLE",
+ Form("ContinueProcessing - %s: Grid storage of one or more "
+ "objects failed. Trying again now",
+ fCurrentDetector.Data()));
+ UpdateShuttleStatus(AliShuttleStatus::kStoreStarted);
+ if (StoreOCDB()){
+ Log("SHUTTLE", Form("ContinueProcessing - %s: all objects "
+ "successfully stored into main storage",
+ fCurrentDetector.Data()));
+ UpdateShuttleStatus(AliShuttleStatus::kDone);
+ UpdateShuttleLogbook(fCurrentDetector, "DONE");
+ } else {
+ Log("SHUTTLE",
+ Form("ContinueProcessing - %s: Grid storage failed again",
+ fCurrentDetector.Data()));
+ UpdateShuttleStatus(AliShuttleStatus::kStoreError);
+ }
+ return kFALSE;
+ }
+
+ // if we get here, there is a restart
+ Bool_t cont = kFALSE;
+
+ // abort conditions
+ if (status->GetCount() >= fConfig->GetMaxRetries()) {
+ Log("SHUTTLE", Form("ContinueProcessing - %s failed %d times in status %s - "
+ "Updating Shuttle Logbook", fCurrentDetector.Data(),
+ status->GetCount(), status->GetStatusName()));
+ UpdateShuttleLogbook(fCurrentDetector.Data(), "FAILED");
+ UpdateShuttleStatus(AliShuttleStatus::kFailed);
+
+ // there may still be objects in local OCDB and reference storage
+ // and FXS databases may be not updated: do it now!
+
+ // TODO Currently disabled, we want to keep files in case of failure!
+ // CleanLocalStorage(fgkLocalCDB);
+ // CleanLocalStorage(fgkLocalRefStorage);
+ // UpdateTableFailCase();
+
+ // Send mail to detector expert!
+ Log("SHUTTLE", Form("ContinueProcessing - Sending mail to %s expert...",
+ fCurrentDetector.Data()));
+ if (!SendMail())
+ Log("SHUTTLE", Form("ContinueProcessing - Could not send mail to %s expert",
+ fCurrentDetector.Data()));
+
+ } else {
+ Log("SHUTTLE", Form("ContinueProcessing - %s: restarting. "
+ "Aborted before with %s. Retry number %d.", fCurrentDetector.Data(),
+ status->GetStatusName(), status->GetCount()));
+ Bool_t increaseCount = kTRUE;
+ if (status->GetStatus() == AliShuttleStatus::kDCSError ||
+ status->GetStatus() == AliShuttleStatus::kDCSStarted)
+ increaseCount = kFALSE;
+
+ UpdateShuttleStatus(AliShuttleStatus::kStarted, increaseCount);
+ cont = kTRUE;
+ }
+
+ return cont;
+}
+
+//______________________________________________________________________________________________
+Bool_t AliShuttle::Process(AliShuttleLogbookEntry* entry)
+{
+ //
+ // Makes data retrieval for all detectors in the configuration.
+ // entry: Shuttle logbook entry, contains run paramenters and status of detectors
+ // (Unprocessed, Inactive, Failed or Done).
+ // Returns kFALSE in case of error occured and kTRUE otherwise
+ //
+
+ if (!entry) return kFALSE;
+
+ fLogbookEntry = entry;
+
+ Log("SHUTTLE", Form("\t\t\t^*^*^*^*^*^*^*^*^*^*^*^* run %d: START ^*^*^*^*^*^*^*^*^*^*^*^*",
+ GetCurrentRun()));
+
+ // Send the information to ML
+ TMonaLisaText mlStatus("SHUTTLE_status", "Processing");
+ TMonaLisaText mlRunType("SHUTTLE_runtype", Form("%s (%s)", entry->GetRunType(), entry->GetRunParameter("log")));
+
+ TList mlList;
+ mlList.Add(&mlStatus);
+ mlList.Add(&mlRunType);
+
+ TString mlID;
+ mlID.Form("%d", GetCurrentRun());
+ fMonaLisa->SendParameters(&mlList, mlID);
+
+ if (fLogbookEntry->IsDone())
+ {
+ Log("SHUTTLE","Process - Shuttle is already DONE. Updating logbook");
+ UpdateShuttleLogbook("shuttle_done");
+ fLogbookEntry = 0;
+ return kTRUE;
+ }
+
+ // read test mode if flag is set
+ if (fReadTestMode)
+ {
+ fTestMode = kNone;
+ TString logEntry(entry->GetRunParameter("log"));
+ //printf("log entry = %s\n", logEntry.Data());
+ TString searchStr("Testmode: ");
+ Int_t pos = logEntry.Index(searchStr.Data());
+ //printf("%d\n", pos);
+ if (pos >= 0)
+ {
+ TSubString subStr = logEntry(pos + searchStr.Length(), logEntry.Length());
+ //printf("%s\n", subStr.String().Data());
+ TString newStr(subStr.Data());
+ TObjArray* token = newStr.Tokenize(' ');
+ if (token)
+ {
+ //token->Print();
+ TObjString* tmpStr = dynamic_cast<TObjString*> (token->First());
+ if (tmpStr)
+ {
+ Int_t testMode = tmpStr->String().Atoi();
+ if (testMode > 0)
+ {
+ Log("SHUTTLE", Form("Process - Enabling test mode %d", testMode));
+ SetTestMode((TestMode) testMode);
+ }
+ }
+ delete token;
+ }
+ }
+ }
+
+ fLogbookEntry->Print("all");
+
+ // Initialization
+ Bool_t hasError = kFALSE;
+
+ // Set the CDB and Reference folders according to the year and LHC period
+ TString lhcPeriod(GetLHCPeriod());
+ if (lhcPeriod.Length() == 0)
+ {
+ Log("SHUTTLE","Process - LHCPeriod not found in logbook!");
+ return 0;
+ }
+
+ if (fgkMainCDB.Length() == 0)
+ fgkMainCDB = Form("alien://folder=/alice/data/%d/%s/OCDB?user=alidaq?cacheFold=/tmp/OCDBCache",
+ GetCurrentYear(), lhcPeriod.Data());
+
+ if (fgkMainRefStorage.Length() == 0)
+ fgkMainRefStorage = Form("alien://folder=/alice/data/%d/%s/Reference?user=alidaq?cacheFold=/tmp/OCDBCache",
+ GetCurrentYear(), lhcPeriod.Data());
+
+ // Loop on detectors in the configuration
+ TIter iter(fConfig->GetDetectors());
+ TObjString* aDetector = 0;
+
+ Bool_t first = kTRUE;
+
+ while ((aDetector = (TObjString*) iter.Next()))
+ {
+ fCurrentDetector = aDetector->String();
+
+ if (ContinueProcessing() == kFALSE) continue;
+
+ if (first)
+ {
+ // only read QueryCDB when needed and only once
+ AliCDBStorage *mainCDBSto = AliCDBManager::Instance()->GetStorage(fgkMainCDB);
+ if(mainCDBSto) mainCDBSto->QueryCDB(GetCurrentRun());
+ AliCDBStorage *mainRefSto = AliCDBManager::Instance()->GetStorage(fgkMainRefStorage);
+ if(mainRefSto) mainRefSto->QueryCDB(GetCurrentRun());
+ first = kFALSE;
+ }
+
+ Log("SHUTTLE", Form("\t\t\t****** run %d - %s: START ******",
+ GetCurrentRun(), aDetector->GetName()));
+
+ for(Int_t iSys=0;iSys<3;iSys++) fFXSCalled[iSys]=kFALSE;
+
+ Log(fCurrentDetector.Data(), "Process - Starting processing");
+
+ Int_t pid = fork();
+
+ if (pid < 0)
+ {
+ Log("SHUTTLE", "Process - ERROR: Forking failed");
+ }
+ else if (pid > 0)
+ {
+ // parent
+ Log("SHUTTLE", Form("Process - In parent process of %d - %s: Starting monitoring",
+ GetCurrentRun(), aDetector->GetName()));
+
+ Long_t begin = time(0);
+
+ int status; // to be used with waitpid, on purpose an int (not Int_t)!
+ while (waitpid(pid, &status, WNOHANG) == 0)
+ {
+ Long_t expiredTime = time(0) - begin;
+
+ if (expiredTime > fConfig->GetPPTimeOut())
+ {
+ TString tmp;
+ tmp.Form("Process - Process of %s time out. "
+ "Run time: %d seconds. Killing...",
+ fCurrentDetector.Data(), expiredTime);
+ Log("SHUTTLE", tmp);
+ Log(fCurrentDetector, tmp);
+
+ kill(pid, 9);
+
+ UpdateShuttleStatus(AliShuttleStatus::kPPTimeOut);
+ hasError = kTRUE;
+
+ gSystem->Sleep(1000);
+ }
+ else
+ {
+ gSystem->Sleep(1000);
+
+ TString checkStr;
+ checkStr.Form("ps -o vsize --pid %d | tail -n 1", pid);
+ FILE* pipe = gSystem->OpenPipe(checkStr, "r");
+ if (!pipe)
+ {
+ Log("SHUTTLE", Form("Process - Error: "
+ "Could not open pipe to %s", checkStr.Data()));
+ continue;
+ }
+
+ char buffer[100];
+ if (!fgets(buffer, 100, pipe))
+ {
+ Log("SHUTTLE", "Process - Error: ps did not return anything");
+ gSystem->ClosePipe(pipe);
+ continue;
+ }
+ gSystem->ClosePipe(pipe);
+
+ //Log("SHUTTLE", Form("ps returned %s", buffer));
+
+ Int_t mem = 0;
+ if ((sscanf(buffer, "%d\n", &mem) != 1) || !mem)
+ {
+ Log("SHUTTLE", "Process - Error: Could not parse output of ps");
+ continue;
+ }
+
+ if (expiredTime % 60 == 0)
+ {
+ Log("SHUTTLE", Form("Process - %s: Checking process. "
+ "Run time: %d seconds - Memory consumption: %d KB",
+ fCurrentDetector.Data(), expiredTime, mem));
+ SendAlive();
+ }
+
+ if (mem > fConfig->GetPPMaxMem())
+ {
+ TString tmp;
+ tmp.Form("Process - Process exceeds maximum allowed memory "
+ "(%d KB > %d KB). Killing...",
+ mem, fConfig->GetPPMaxMem());
+ Log("SHUTTLE", tmp);
+ Log(fCurrentDetector, tmp);
+
+ kill(pid, 9);
+
+ UpdateShuttleStatus(AliShuttleStatus::kPPOutOfMemory);
+ hasError = kTRUE;
+
+ gSystem->Sleep(1000);
+ }
+ }
+ }
+
+ Log("SHUTTLE", Form("Process - In parent process of %d - %s: Client has terminated.",
+ GetCurrentRun(), aDetector->GetName()));
+
+ if (WIFEXITED(status))
+ {
+ Int_t returnCode = WEXITSTATUS(status);
+
+ Log("SHUTTLE", Form("Process - %s: the return code is %d", fCurrentDetector.Data(),
+ returnCode));
+
+ if (returnCode == 0) hasError = kTRUE;
+ }
+ }
+ else if (pid == 0)
+ {
+ // client
+ Log("SHUTTLE", Form("Process - In client process of %d - %s", GetCurrentRun(),
+ aDetector->GetName()));
+
+ Log("SHUTTLE", Form("Process - Redirecting output to %s log",fCurrentDetector.Data()));
+
+ if ((freopen(GetLogFileName(fCurrentDetector), "a", stdout)) == 0)
+ {
+ Log("SHUTTLE", "Process - Could not freopen stdout");
+ }
+ else
+ {
+ fOutputRedirected = kTRUE;
+ if ((dup2(fileno(stdout), fileno(stderr))) < 0)
+ Log("SHUTTLE", "Process - Could not redirect stderr");
+
+ }
+
+ TString wd = gSystem->WorkingDirectory();
+ TString tmpDir = Form("%s/%s_%d_process", GetShuttleTempDir(),
+ fCurrentDetector.Data(), GetCurrentRun());
+
+ Int_t result = gSystem->GetPathInfo(tmpDir.Data(), 0, (Long64_t*) 0, 0, 0);
+ if (!result) // temp dir already exists!
+ {
+ Log(fCurrentDetector.Data(),
+ Form("Process - %s dir already exists! Removing...", tmpDir.Data()));
+ gSystem->Exec(Form("rm -rf %s",tmpDir.Data()));
+ }
+
+ if (gSystem->mkdir(tmpDir.Data(), 1))
+ {
+ Log(fCurrentDetector.Data(), "Process - could not make temp directory!!");
+ gSystem->Exit(1);
+ }
+
+ if (!gSystem->ChangeDirectory(tmpDir.Data()))
+ {
+ Log(fCurrentDetector.Data(), "Process - could not change directory!!");
+ gSystem->Exit(1);
+ }
+
+ Bool_t success = ProcessCurrentDetector();
+
+ gSystem->ChangeDirectory(wd.Data());
+
+ if (success) // Preprocessor finished successfully!
+ {
+ // remove temporary folder or DCS map
+ if (!fConfig->KeepTempFolder())
+ {
+ gSystem->Exec(Form("rm -rf %s",tmpDir.Data()));
+ } else if (!fConfig->KeepDCSMap())
+ {
+ gSystem->Exec(Form("rm -f %s/DCSMap.root",tmpDir.Data()));
+ }
+
+ // Update time_processed field in FXS DB
+ if (UpdateTable() == kFALSE)
+ Log("SHUTTLE", Form("Process - %s: Could not update FXS databases!",
+ fCurrentDetector.Data()));
+
+ // Transfer the data from local storage to main storage (Grid)
+ UpdateShuttleStatus(AliShuttleStatus::kStoreStarted);
+ if (StoreOCDB() == kFALSE)
+ {
+ Log("SHUTTLE",
+ Form("\t\t\t****** run %d - %s: STORAGE ERROR ******",
+ GetCurrentRun(), aDetector->GetName()));
+ UpdateShuttleStatus(AliShuttleStatus::kStoreError);
+ success = kFALSE;
+ } else {
+ Log("SHUTTLE",
+ Form("\t\t\t****** run %d - %s: DONE ******",
+ GetCurrentRun(), aDetector->GetName()));
+ UpdateShuttleStatus(AliShuttleStatus::kDone);
+ UpdateShuttleLogbook(fCurrentDetector, "DONE");
+ }
+ } else
+ {
+ Log("SHUTTLE",
+ Form("\t\t\t****** run %d - %s: PP ERROR ******",
+ GetCurrentRun(), aDetector->GetName()));
+ }
+
+ for (UInt_t iSys=0; iSys<3; iSys++)
+ {
+ if (fFXSCalled[iSys]) fFXSlist[iSys].Clear();
+ }
+
+ Log("SHUTTLE", Form("Process - Client process of %d - %s is exiting now with %d.",
+ GetCurrentRun(), aDetector->GetName(), success));
+
+ // the client exits here
+ gSystem->Exit(success);
+
+ AliError("We should never get here!!!");
+ }
+ }
+
+ Log("SHUTTLE", Form("\t\t\t^*^*^*^*^*^*^*^*^*^*^*^* run %d: FINISH ^*^*^*^*^*^*^*^*^*^*^*^*",
+ GetCurrentRun()));
+
+ //check if shuttle is done for this run, if so update logbook
+ TObjArray checkEntryArray;
+ checkEntryArray.SetOwner(1);
+ TString whereClause = Form("where run=%d", GetCurrentRun());
+ if (!QueryShuttleLogbook(whereClause.Data(), checkEntryArray) ||
+ checkEntryArray.GetEntries() == 0) {
+ Log("SHUTTLE", Form("Process - Warning: Cannot check status of run %d on Shuttle logbook!",
+ GetCurrentRun()));
+ return hasError == kFALSE;
+ }
+
+ AliShuttleLogbookEntry* checkEntry = dynamic_cast<AliShuttleLogbookEntry*>
+ (checkEntryArray.At(0));
+
+ if (checkEntry)
+ {
+ if (checkEntry->IsDone())
+ {
+ Log("SHUTTLE","Process - Shuttle is DONE. Updating logbook");
+ UpdateShuttleLogbook("shuttle_done");
+ }
+ else
+ {
+ for (UInt_t iDet=0; iDet<NDetectors(); iDet++)
+ {
+ if (checkEntry->GetDetectorStatus(iDet) == AliShuttleLogbookEntry::kUnprocessed)
+ {
+ AliDebug(2, Form("Run %d: setting %s as \"not first time unprocessed\"",
+ checkEntry->GetRun(), GetDetName(iDet)));
+ fFirstUnprocessed[iDet] = kFALSE;
+ }
+ }
+ }
+ }
+
+ fLogbookEntry = 0;
+
+ return hasError == kFALSE;
+}
+
+//______________________________________________________________________________________________
+Bool_t AliShuttle::ProcessCurrentDetector()
+{
+ //
+ // Makes data retrieval just for a specific detector (fCurrentDetector).
+ // Threre should be a configuration for this detector.
+
+ Log("SHUTTLE", Form("ProcessCurrentDetector - Retrieving values for %s, run %d",
+ fCurrentDetector.Data(), GetCurrentRun()));
+
+ TString wd = gSystem->WorkingDirectory();
+
+ if (!CleanReferenceStorage(fCurrentDetector.Data()))
+ return kFALSE;
+
+ gSystem->ChangeDirectory(wd.Data());
+
+ TMap* dcsMap = new TMap();
+
+ // call preprocessor
+ AliPreprocessor* aPreprocessor =
+ dynamic_cast<AliPreprocessor*> (fPreprocessorMap.GetValue(fCurrentDetector));
+
+ aPreprocessor->Initialize(GetCurrentRun(), GetCurrentStartTime(), GetCurrentEndTime());
+
+ Bool_t processDCS = aPreprocessor->ProcessDCS();
+
+ if (!processDCS)
+ {
+ Log(fCurrentDetector, "ProcessCurrentDetector -"
+ " The preprocessor requested to skip the retrieval of DCS values");
+ }
+ else if (fTestMode & kSkipDCS)
+ {
+ Log(fCurrentDetector, "ProcessCurrentDetector - In TESTMODE: Skipping DCS processing");
+ }
+ else if (fTestMode & kErrorDCS)
+ {
+ Log(fCurrentDetector, "ProcessCurrentDetector - In TESTMODE: Simulating DCS error");
+ UpdateShuttleStatus(AliShuttleStatus::kDCSStarted);
+ UpdateShuttleStatus(AliShuttleStatus::kDCSError);
+ delete dcsMap;
+ return kFALSE;
+ } else {
+
+ UpdateShuttleStatus(AliShuttleStatus::kDCSStarted);
+
+ // Query DCS archive
+ Int_t nServers = fConfig->GetNServers(fCurrentDetector);
+
+ for (int iServ=0; iServ<nServers; iServ++)
+ {
+
+ TString host(fConfig->GetDCSHost(fCurrentDetector, iServ));
+ Int_t port = fConfig->GetDCSPort(fCurrentDetector, iServ);
+ Int_t multiSplit = fConfig->GetMultiSplit(fCurrentDetector, iServ);
+
+ Log(fCurrentDetector, Form("ProcessCurrentDetector -"
+ " Querying DCS Amanda server %s:%d (%d of %d)",
+ host.Data(), port, iServ+1, nServers));
+
+ TMap* aliasMap = 0;
+ TMap* dpMap = 0;
+
+ if (fConfig->GetDCSAliases(fCurrentDetector, iServ)->GetEntries() > 0)
+ {
+ aliasMap = GetValueSet(host, port,
+ fConfig->GetDCSAliases(fCurrentDetector, iServ),
+ kAlias, multiSplit);
+ if (!aliasMap)
+ {
+ Log(fCurrentDetector,
+ Form("ProcessCurrentDetector -"
+ " Error retrieving DCS aliases from server %s."
+ " Sending mail to DCS experts!", host.Data()));
+ UpdateShuttleStatus(AliShuttleStatus::kDCSError);
+
+ if (!SendMailToDCS())
+ Log("SHUTTLE", Form("ProcessCurrentDetector - "
+ "Could not send mail to DCS experts!"));
+
+ delete dcsMap;
+ return kFALSE;
+ }
+ }
+
+ if (fConfig->GetDCSDataPoints(fCurrentDetector, iServ)->GetEntries() > 0)
+ {
+ dpMap = GetValueSet(host, port,
+ fConfig->GetDCSDataPoints(fCurrentDetector, iServ),
+ kDP, multiSplit);
+ if (!dpMap)
+ {
+ Log(fCurrentDetector,
+ Form("ProcessCurrentDetector -"
+ " Error retrieving DCS data points from server %s."
+ " Sending mail to DCS experts!", host.Data()));
+ UpdateShuttleStatus(AliShuttleStatus::kDCSError);
+
+ if (!SendMailToDCS())
+ Log("SHUTTLE", Form("ProcessCurrentDetector - "
+ "Could not send mail to DCS experts!"));
+
+ if (aliasMap) delete aliasMap;
+ delete dcsMap;
+ return kFALSE;
+ }
+ }
+
+ // merge aliasMap and dpMap into dcsMap
+ if(aliasMap) {
+ TIter iter(aliasMap);
+ TObjString* key = 0;
+ while ((key = (TObjString*) iter.Next()))
+ dcsMap->Add(key, aliasMap->GetValue(key->String()));
+
+ aliasMap->SetOwner(kFALSE);
+ delete aliasMap;
+ }
+
+ if(dpMap) {
+ TIter iter(dpMap);
+ TObjString* key = 0;
+ while ((key = (TObjString*) iter.Next()))
+ dcsMap->Add(key, dpMap->GetValue(key->String()));
+
+ dpMap->SetOwner(kFALSE);
+ delete dpMap;
+ }
+ }
+ }
+
+ // save map into file, to help debugging in case of preprocessor error
+ TFile* f = TFile::Open("DCSMap.root","recreate");
+ f->cd();
+ dcsMap->Write("DCSMap", TObject::kSingleKey);
+ f->Close();
+ delete f;
+
+ // DCS Archive DB processing successful. Call Preprocessor!
+ UpdateShuttleStatus(AliShuttleStatus::kPPStarted);
+
+ UInt_t returnValue = aPreprocessor->Process(dcsMap);
+
+ if (returnValue > 0) // Preprocessor error!
+ {
+ Log(fCurrentDetector, Form("ProcessCurrentDetector - "
+ "Preprocessor failed. Process returned %d.", returnValue));
+ UpdateShuttleStatus(AliShuttleStatus::kPPError);
+ dcsMap->DeleteAll();
+ delete dcsMap;
+ return kFALSE;
+ }
+
+ // preprocessor ok!
+ UpdateShuttleStatus(AliShuttleStatus::kPPDone);
+ Log(fCurrentDetector, Form("ProcessCurrentDetector - %s preprocessor returned success",
+ fCurrentDetector.Data()));
+
+ dcsMap->DeleteAll();
+ delete dcsMap;
+
+ return kTRUE;
+}
+
+//______________________________________________________________________________________________
+void AliShuttle::CountOpenRuns()
+{
+ // Query DAQ's Shuttle logbook and sends the number of open runs to ML
+
+ // check connection, in case connect
+ if (!Connect(3))
+ return;
+
+ TString sqlQuery;
+ sqlQuery = Form("select count(*) from %s where shuttle_done=0", fConfig->GetShuttlelbTable());
+
+ TSQLResult* aResult = fServer[3]->Query(sqlQuery);
+ if (!aResult) {
+ AliError(Form("Can't execute query <%s>!", sqlQuery.Data()));
+ return;
+ }
+
+ AliDebug(2,Form("Query = %s", sqlQuery.Data()));
+
+ if (aResult->GetRowCount() == 0) {
+ AliError(Form("No result for query %s received", sqlQuery.Data()));
+ return;
+ }
+
+ if (aResult->GetFieldCount() != 1) {
+ AliError(Form("Invalid field count for query %s received", sqlQuery.Data()));
+ return;
+ }
+
+ TSQLRow* aRow = aResult->Next();
+ if (!aRow) {
+ AliError(Form("Could not receive result of query %s", sqlQuery.Data()));
+ return;
+ }
+
+ TString result(aRow->GetField(0), aRow->GetFieldLength(0));
+ Int_t count = result.Atoi();
+
+ Log("SHUTTLE", Form("%d unprocessed runs", count));
+
+ delete aRow;
+ delete aResult;
+
+ TMonaLisaValue mlStatus("SHUTTLE_openruns", count);
+
+ TList mlList;
+ mlList.Add(&mlStatus);
+
+ fMonaLisa->SendParameters(&mlList, "__PROCESSINGINFO__");
+}
+
+//______________________________________________________________________________________________
+Bool_t AliShuttle::QueryShuttleLogbook(const char* whereClause,
+ TObjArray& entries)
+{
+ // Query DAQ's Shuttle logbook and fills detector status object.
+ // Call QueryRunParameters to query DAQ logbook for run parameters.
+ //
+
+ entries.SetOwner(1);
+
+ // check connection, in case connect
+ if (!Connect(3)) return kFALSE;
+
+ TString sqlQuery;
+ sqlQuery = Form("select * from %s %s order by run", fConfig->GetShuttlelbTable(), whereClause);
+
+ TSQLResult* aResult = fServer[3]->Query(sqlQuery);
+ if (!aResult) {
+ AliError(Form("Can't execute query <%s>!", sqlQuery.Data()));
+ return kFALSE;
+ }
+
+ AliDebug(2,Form("Query = %s", sqlQuery.Data()));
+
+ if(aResult->GetRowCount() == 0) {
+ Log("SHUTTLE", "No entries in Shuttle Logbook match request");
+ delete aResult;
+ return kTRUE;
+ }
+
+ // TODO Check field count!
+ const UInt_t nCols = 23;
+ if (aResult->GetFieldCount() != (Int_t) nCols) {
+ Log("SHUTTLE", "Invalid SQL result field number!");
+ delete aResult;
+ return kFALSE;
+ }
+
+ TSQLRow* aRow;
+ while ((aRow = aResult->Next())) {
+ TString runString(aRow->GetField(0), aRow->GetFieldLength(0));
+ Int_t run = runString.Atoi();
+
+ AliShuttleLogbookEntry *entry = QueryRunParameters(run);
+ if (!entry)
+ continue;
+
+ // loop on detectors
+ for(UInt_t ii = 0; ii < nCols; ii++)
+ entry->SetDetectorStatus(aResult->GetFieldName(ii), aRow->GetField(ii));
+
+ entries.AddLast(entry);
+ delete aRow;
+ }
+
+ delete aResult;
+ return kTRUE;
+}
+
+//______________________________________________________________________________________________
+AliShuttleLogbookEntry* AliShuttle::QueryRunParameters(Int_t run)
+{
+ //
+ // Retrieve run parameters written in the DAQ logbook and sets them into AliShuttleLogbookEntry object
+ //
+
+ // check connection, in case connect
+ if (!Connect(3))
+ return 0;
+
+ TString sqlQuery;
+ sqlQuery.Form("select * from %s where run=%d", fConfig->GetDAQlbTable(), run);
+
+ TSQLResult* aResult = fServer[3]->Query(sqlQuery);
+ if (!aResult) {
+ Log("SHUTTLE", Form("Can't execute query <%s>!", sqlQuery.Data()));
+ return 0;
+ }
+
+ if (aResult->GetRowCount() == 0) {
+ Log("SHUTTLE", Form("QueryRunParameters - No entry in DAQ Logbook for run %d. Skipping", run));
+ delete aResult;
+ return 0;
+ }
+
+ if (aResult->GetRowCount() > 1) {
+ Log("SHUTTLE", Form("QueryRunParameters - UNEXPECTED: "
+ "more than one entry in DAQ Logbook for run %d!", run));
+ delete aResult;
+ return 0;
+ }
+
+ TSQLRow* aRow = aResult->Next();
+ if (!aRow)
+ {
+ Log("SHUTTLE", Form("QueryRunParameters - Could not retrieve row for run %d. Skipping", run));
+ delete aResult;
+ return 0;
+ }
+
+ AliShuttleLogbookEntry* entry = new AliShuttleLogbookEntry(run);
+
+ for (Int_t ii = 0; ii < aResult->GetFieldCount(); ii++)
+ entry->SetRunParameter(aResult->GetFieldName(ii), aRow->GetField(ii));
+
+ UInt_t startTime = entry->GetStartTime();
+ UInt_t endTime = entry->GetEndTime();
+
+// if (!startTime || !endTime || startTime > endTime)
+// {
+// Log("SHUTTLE",
+// Form("QueryRunParameters - Invalid parameters for Run %d: startTime = %d, endTime = %d. Skipping!",
+// run, startTime, endTime));
+//
+// Log("SHUTTLE", Form("Marking SHUTTLE done for run %d", run));
+// fLogbookEntry = entry;
+// if (!UpdateShuttleLogbook("shuttle_done"))
+// {
+// AliError(Form("Could not update logbook for run %d !", run));
+// }
+// fLogbookEntry = 0;
+//
+// delete entry;
+// delete aRow;
+// delete aResult;
+// return 0;
+// }
+
+ if (!startTime)
+ {
+ Log("SHUTTLE",
+ Form("QueryRunParameters - Invalid parameters for Run %d: "
+ "startTime = %d, endTime = %d. Skipping!",
+ run, startTime, endTime));
+
+ Log("SHUTTLE", Form("Marking SHUTTLE done for run %d", run));
+ fLogbookEntry = entry;
+ if (!UpdateShuttleLogbook("shuttle_ignored"))
+ {
+ AliError(Form("Could not update logbook for run %d !", run));
+ }
+ fLogbookEntry = 0;
+
+ delete entry;
+ delete aRow;
+ delete aResult;
+ return 0;
+ }
+
+ if (startTime && !endTime)
+ {
+ // TODO Here we don't mark SHUTTLE done, because this may mean
+ //the run is still ongoing!!
+ Log("SHUTTLE",
+ Form("QueryRunParameters - Invalid parameters for Run %d: "
+ "startTime = %d, endTime = %d. Skipping (Shuttle won't be marked as DONE)!",
+ run, startTime, endTime));
+
+ //Log("SHUTTLE", Form("Marking SHUTTLE done for run %d", run));
+ //fLogbookEntry = entry;
+ //if (!UpdateShuttleLogbook("shuttle_done"))
+ //{
+ // AliError(Form("Could not update logbook for run %d !", run));
+ //}
+ //fLogbookEntry = 0;
+
+ delete entry;
+ delete aRow;
+ delete aResult;
+ return 0;
+ }
+
+ if (startTime && endTime && (startTime > endTime))
+ {
+ Log("SHUTTLE",
+ Form("QueryRunParameters - Invalid parameters for Run %d: "
+ "startTime = %d, endTime = %d. Skipping!",
+ run, startTime, endTime));
+
+ Log("SHUTTLE", Form("Marking SHUTTLE done for run %d", run));
+ fLogbookEntry = entry;
+ if (!UpdateShuttleLogbook("shuttle_ignored"))
+ {
+ AliError(Form("Could not update logbook for run %d !", run));
+ }
+ fLogbookEntry = 0;
+
+ delete entry;
+ delete aRow;
+ delete aResult;
+ return 0;
+ }
+
+ TString totEventsStr = entry->GetRunParameter("totalEvents");
+ Int_t totEvents = totEventsStr.Atoi();
+ if (totEvents < 1)
+ {
+ Log("SHUTTLE",
+ Form("QueryRunParameters - Run %d has 0 events - Skipping!", run));
+
+ Log("SHUTTLE", Form("Marking SHUTTLE done for run %d", run));
+ fLogbookEntry = entry;
+ if (!UpdateShuttleLogbook("shuttle_ignored"))
+ {
+ AliError(Form("Could not update logbook for run %d !", run));
+ }
+ fLogbookEntry = 0;
+
+ delete entry;
+ delete aRow;
+ delete aResult;
+ return 0;
+ }
+
+ delete aRow;
+ delete aResult;
+
+ return entry;
+}
+
+//______________________________________________________________________________________________
+TMap* AliShuttle::GetValueSet(const char* host, Int_t port, const TSeqCollection* entries,
+ DCSType type, Int_t multiSplit)
+{
+ // Retrieve all "entry" data points from the DCS server
+ // host, port: TSocket connection parameters
+ // entries: list of name of the alias or data point
+ // type: kAlias or kDP
+ // returns TMap of values, 0 when failure
+
+ AliDCSClient client(host, port, fTimeout, fRetries, multiSplit);
+
+ TMap* result = 0;
+ if (type == kAlias)
+ {
+ result = client.GetAliasValues(entries, GetCurrentStartTime(),
+ GetCurrentEndTime());
+ }
+ else if (type == kDP)
+ {
+ result = client.GetDPValues(entries, GetCurrentStartTime(),
+ GetCurrentEndTime());
+ }
+
+ if (result == 0)
+ {
+ Log(fCurrentDetector.Data(), Form("GetValueSet - Can't get entries! Reason: %s",
+ client.GetErrorString(client.GetResultErrorCode())));
+ if (client.GetResultErrorCode() == AliDCSClient::fgkServerError)
+ Log(fCurrentDetector.Data(), Form("GetValueSet - Server error code: %s",
+ client.GetServerError().Data()));
+
+ return 0;
+ }
+
+ return result;
+}
+
+//______________________________________________________________________________________________
+const char* AliShuttle::GetFile(Int_t system, const char* detector,
+ const char* id, const char* source)
+{
+ // Get calibration file from file exchange servers
+ // First queris the FXS database for the file name, using the run, detector, id and source info
+ // then calls RetrieveFile(filename) for actual copy to local disk
+ // run: current run being processed (given by Logbook entry fLogbookEntry)
+ // detector: the Preprocessor name
+ // id: provided as a parameter by the Preprocessor
+ // source: provided by the Preprocessor through GetFileSources function
+
+ // check if test mode should simulate a FXS error
+ if (fTestMode & kErrorFXSFiles)
+ {
+ Log(detector, Form("GetFile - In TESTMODE - Simulating error while connecting to %s FXS", GetSystemName(system)));
+ return 0;
+ }
+
+ // check connection, in case connect
+ if (!Connect(system))
+ {
+ Log(detector, Form("GetFile - Couldn't connect to %s FXS database", GetSystemName(system)));
+ return 0;
+ }
+
+ // Query preparation
+ TString sourceName(source);
+ Int_t nFields = 3;
+ TString sqlQueryStart = Form("select filePath,size,fileChecksum from %s where",
+ fConfig->GetFXSdbTable(system));
+ TString whereClause = Form("run=%d and detector=\"%s\" and fileId=\"%s\"",
+ GetCurrentRun(), detector, id);
+
+ if (system == kDAQ)
+ {
+ whereClause += Form(" and DAQsource=\"%s\"", source);
+ }
+ else if (system == kDCS)
+ {
+ sourceName="none";
+ }
+ else if (system == kHLT)
+ {
+ whereClause += Form(" and DDLnumbers=\"%s\"", source);
+ nFields = 3;
+ }
+
+ TString sqlQuery = Form("%s %s", sqlQueryStart.Data(), whereClause.Data());
+
+ AliDebug(2, Form("SQL query: \n%s",sqlQuery.Data()));
+
+ // Query execution
+ TSQLResult* aResult = 0;
+ aResult = dynamic_cast<TSQLResult*> (fServer[system]->Query(sqlQuery));
+ if (!aResult) {
+ Log(detector, Form("GetFileName - Can't execute SQL query to %s database for: id = %s, source = %s",
+ GetSystemName(system), id, sourceName.Data()));
+ return 0;
+ }
+
+ if(aResult->GetRowCount() == 0)
+ {
+ Log(detector,
+ Form("GetFileName - No entry in %s FXS db for: id = %s, source = %s",
+ GetSystemName(system), id, sourceName.Data()));
+ delete aResult;
+ return 0;
+ }
+
+ if (aResult->GetRowCount() > 1) {
+ Log(detector,
+ Form("GetFileName - More than one entry in %s FXS db for: id = %s, source = %s",
+ GetSystemName(system), id, sourceName.Data()));
+ delete aResult;
+ return 0;
+ }
+
+ if (aResult->GetFieldCount() != nFields) {
+ Log(detector,
+ Form("GetFileName - Wrong field count in %s FXS db for: id = %s, source = %s",
+ GetSystemName(system), id, sourceName.Data()));
+ delete aResult;
+ return 0;
+ }
+
+ TSQLRow* aRow = dynamic_cast<TSQLRow*> (aResult->Next());
+
+ if (!aRow){
+ Log(detector, Form("GetFileName - Empty set result in %s FXS db from query: id = %s, source = %s",
+ GetSystemName(system), id, sourceName.Data()));
+ delete aResult;
+ return 0;
+ }
+
+ TString filePath(aRow->GetField(0), aRow->GetFieldLength(0));
+ TString fileSize(aRow->GetField(1), aRow->GetFieldLength(1));
+ TString fileChecksum(aRow->GetField(2), aRow->GetFieldLength(2));
+
+ delete aResult;
+ delete aRow;
+
+ AliDebug(2, Form("filePath = %s; size = %s, fileChecksum = %s",
+ filePath.Data(), fileSize.Data(), fileChecksum.Data()));
+
+ // retrieved file is renamed to make it unique
+ TString localFileName = Form("%s/%s_%d_process/%s_%s_%d_%s_%s.shuttle",
+ GetShuttleTempDir(), detector, GetCurrentRun(),
+ GetSystemName(system), detector, GetCurrentRun(),
+ id, sourceName.Data());
+
+
+ // file retrieval from FXS
+ UInt_t nRetries = 0;
+ UInt_t maxRetries = 3;
+ Bool_t result = kFALSE;
+
+ // copy!! if successful TSystem::Exec returns 0
+ while(nRetries++ < maxRetries) {
+ AliDebug(2, Form("Trying to copy file. Retry # %d", nRetries));
+ result = RetrieveFile(system, filePath.Data(), localFileName.Data());
+ if(!result)
+ {
+ Log(detector, Form("GetFileName - Copy of file %s from %s FXS failed",
+ filePath.Data(), GetSystemName(system)));
+ continue;
+ }
+
+ if (fileChecksum.Length()>0)
+ {
+ // compare md5sum of local file with the one stored in the FXS DB
+ Int_t md5Comp = gSystem->Exec(Form("md5sum %s |grep %s 2>&1 > /dev/null",
+ localFileName.Data(), fileChecksum.Data()));
+
+ if (md5Comp != 0)
+ {
+ Log(detector, Form("GetFileName - md5sum of file %s does not match with local copy!",
+ filePath.Data()));
+ result = kFALSE;
+ continue;
+ }
+ } else {
+ Log(fCurrentDetector, Form("GetFile - md5sum of file %s not set in %s database, skipping comparison",
+ filePath.Data(), GetSystemName(system)));
+ }
+ if (result) break;
+ }
+
+ if(!result) return 0;
+
+ fFXSCalled[system]=kTRUE;
+ TObjString *fileParams = new TObjString(Form("%s#!?!#%s", id, sourceName.Data()));
+ fFXSlist[system].Add(fileParams);
+
+ static TString staticLocalFileName;
+ staticLocalFileName.Form("%s", localFileName.Data());
+
+ Log(fCurrentDetector, Form("GetFile - Retrieved file with id %s and "
+ "source %s from %s to %s", id, source,
+ GetSystemName(system), localFileName.Data()));
+
+ return staticLocalFileName.Data();
+}
+
+//______________________________________________________________________________________________
+Bool_t AliShuttle::RetrieveFile(UInt_t system, const char* fxsFileName, const char* localFileName)
+{
+ //
+ // Copies file from FXS to local Shuttle machine
+ //
+
+ // check temp directory: trying to cd to temp; if it does not exist, create it
+ AliDebug(2, Form("Copy file %s from %s FXS into %s",
+ GetSystemName(system), fxsFileName, localFileName));
+
+ TString tmpDir(localFileName);
+
+ tmpDir = tmpDir(0,tmpDir.Last('/'));
+
+ Int_t noDir = gSystem->GetPathInfo(tmpDir.Data(), 0, (Long64_t*) 0, 0, 0);
+ if (noDir) // temp dir does not exists!
+ {
+ if (gSystem->mkdir(tmpDir.Data(), 1))
+ {
+ Log(fCurrentDetector.Data(), "RetrieveFile - could not make temp directory!!");
+ return kFALSE;
+ }
+ }
+
+ TString baseFXSFolder;
+ if (system == kDAQ)
+ {
+ baseFXSFolder = "FES/";
+ }
+ else if (system == kDCS)
+ {
+ baseFXSFolder = "";
+ }
+ else if (system == kHLT)
+ {
+ baseFXSFolder = "/opt/FXS/";
+ }
+
+
+ TString command = Form("scp -oPort=%d -2 %s@%s:%s%s %s",
+ fConfig->GetFXSPort(system),
+ fConfig->GetFXSUser(system),
+ fConfig->GetFXSHost(system),
+ baseFXSFolder.Data(),
+ fxsFileName,
+ localFileName);
+
+ AliDebug(2, Form("%s",command.Data()));
+
+ Bool_t result = (gSystem->Exec(command.Data()) == 0);
+
+ return result;
+}
+
+//______________________________________________________________________________________________
+TList* AliShuttle::GetFileSources(Int_t system, const char* detector, const char* id)
+{
+ //
+ // Get sources producing the condition file Id from file exchange servers
+ // if id is NULL all sources are returned (distinct)
+ //
+
+ if (id)
+ {
+ Log(detector, Form("GetFileSources - Querying %s FXS for files with id %s produced by %s", GetSystemName(system), id, detector));
+ } else {
+ Log(detector, Form("GetFileSources - Querying %s FXS for files produced by %s", GetSystemName(system), detector));
+ }
+
+ // check if test mode should simulate a FXS error
+ if (fTestMode & kErrorFXSSources)
+ {
+ Log(detector, Form("GetFileSources - In TESTMODE - Simulating error while connecting to %s FXS", GetSystemName(system)));
+ return 0;
+ }
+
+ if (system == kDCS)
+ {
+ Log(detector, "GetFileSources - WARNING: DCS system has only one source of data!");
+ TList *list = new TList();
+ list->SetOwner(1);
+ list->Add(new TObjString(" "));
+ return list;
+ }
+
+ // check connection, in case connect
+ if (!Connect(system))
+ {
+ Log(detector, Form("GetFileSources - Couldn't connect to %s FXS database", GetSystemName(system)));
+ return NULL;
+ }
+
+ TString sourceName = 0;
+ if (system == kDAQ)
+ {
+ sourceName = "DAQsource";
+ } else if (system == kHLT)
+ {
+ sourceName = "DDLnumbers";
+ }
+
+ TString sqlQueryStart = Form("select distinct %s from %s where", sourceName.Data(), fConfig->GetFXSdbTable(system));
+ TString whereClause = Form("run=%d and detector=\"%s\"",
+ GetCurrentRun(), detector);
+ if (id)
+ whereClause += Form(" and fileId=\"%s\"", id);
+ TString sqlQuery = Form("%s %s", sqlQueryStart.Data(), whereClause.Data());
+
+ AliDebug(2, Form("SQL query: \n%s",sqlQuery.Data()));
+
+ // Query execution
+ TSQLResult* aResult;
+ aResult = fServer[system]->Query(sqlQuery);
+ if (!aResult) {
+ Log(detector, Form("GetFileSources - Can't execute SQL query to %s database for id: %s",
+ GetSystemName(system), id));
+ return 0;
+ }
+
+ TList *list = new TList();
+ list->SetOwner(1);
+
+ if (aResult->GetRowCount() == 0)
+ {
+ Log(detector,
+ Form("GetFileSources - No entry in %s FXS table for id: %s", GetSystemName(system), id));
+ delete aResult;
+ return list;
+ }
+
+ Log(detector, Form("GetFileSources - Found %d sources", aResult->GetRowCount()));
+
+ TSQLRow* aRow;
+ while ((aRow = aResult->Next()))
+ {
+
+ TString source(aRow->GetField(0), aRow->GetFieldLength(0));
+ AliDebug(2, Form("%s = %s", sourceName.Data(), source.Data()));
+ list->Add(new TObjString(source));
+ delete aRow;
+ }
+
+ delete aResult;
+
+ return list;
+}
+
+//______________________________________________________________________________________________
+TList* AliShuttle::GetFileIDs(Int_t system, const char* detector, const char* source)
+{
+ //
+ // Get all ids of condition files produced by a given source from file exchange servers
+ //
+
+ Log(detector, Form("GetFileIDs - Retrieving ids with source %s with %s", source, GetSystemName(system)));
+
+ // check if test mode should simulate a FXS error
+ if (fTestMode & kErrorFXSSources)
+ {
+ Log(detector, Form("GetFileIDs - In TESTMODE - Simulating error while connecting to %s FXS", GetSystemName(system)));
+ return 0;
+ }
+
+ // check connection, in case connect
+ if (!Connect(system))
+ {
+ Log(detector, Form("GetFileIDs - Couldn't connect to %s FXS database", GetSystemName(system)));
+ return NULL;
+ }
+
+ TString sourceName = 0;
+ if (system == kDAQ)
+ {
+ sourceName = "DAQsource";
+ } else if (system == kHLT)
+ {
+ sourceName = "DDLnumbers";
+ }
+
+ TString sqlQueryStart = Form("select fileId from %s where", fConfig->GetFXSdbTable(system));
+ TString whereClause = Form("run=%d and detector=\"%s\"",
+ GetCurrentRun(), detector);
+ if (sourceName.Length() > 0 && source)
+ whereClause += Form(" and %s=\"%s\"", sourceName.Data(), source);
+ TString sqlQuery = Form("%s %s", sqlQueryStart.Data(), whereClause.Data());
+
+ AliDebug(2, Form("SQL query: \n%s",sqlQuery.Data()));
+
+ // Query execution
+ TSQLResult* aResult;
+ aResult = fServer[system]->Query(sqlQuery);
+ if (!aResult) {
+ Log(detector, Form("GetFileIDs - Can't execute SQL query to %s database for source: %s",
+ GetSystemName(system), source));
+ return 0;
+ }
+
+ TList *list = new TList();
+ list->SetOwner(1);
+
+ if (aResult->GetRowCount() == 0)
+ {
+ Log(detector,
+ Form("GetFileIDs - No entry in %s FXS table for source: %s", GetSystemName(system), source));
+ delete aResult;
+ return list;
+ }
+
+ Log(detector, Form("GetFileIDs - Found %d ids", aResult->GetRowCount()));
+
+ TSQLRow* aRow;
+
+ while ((aRow = aResult->Next()))
+ {
+
+ TString id(aRow->GetField(0), aRow->GetFieldLength(0));
+ AliDebug(2, Form("fileId = %s", id.Data()));
+ list->Add(new TObjString(id));
+ delete aRow;
+ }
+
+ delete aResult;
+
+ return list;
+}
+
+//______________________________________________________________________________________________
+Bool_t AliShuttle::Connect(Int_t system)
+{
+ // Connect to MySQL Server of the system's FXS MySQL databases
+ // DAQ Logbook, Shuttle Logbook and DAQ FXS db are on the same host
+ //
+
+ // check connection: if already connected return
+ if(fServer[system] && fServer[system]->IsConnected()) return kTRUE;
+
+ TString dbHost, dbUser, dbPass, dbName;
+
+ if (system < 3) // FXS db servers
+ {
+ dbHost = Form("mysql://%s:%d", fConfig->GetFXSdbHost(system), fConfig->GetFXSdbPort(system));
+ dbUser = fConfig->GetFXSdbUser(system);
+ dbPass = fConfig->GetFXSdbPass(system);
+ dbName = fConfig->GetFXSdbName(system);
+ } else { // Run & Shuttle logbook servers
+ // TODO Will the Shuttle logbook server be the same as the Run logbook server ???
+ dbHost = Form("mysql://%s:%d", fConfig->GetDAQlbHost(), fConfig->GetDAQlbPort());
+ dbUser = fConfig->GetDAQlbUser();
+ dbPass = fConfig->GetDAQlbPass();
+ dbName = fConfig->GetDAQlbDB();
+ }
+
+ fServer[system] = TSQLServer::Connect(dbHost.Data(), dbUser.Data(), dbPass.Data());
+ if (!fServer[system] || !fServer[system]->IsConnected()) {
+ if(system < 3)
+ {
+ AliError(Form("Can't establish connection to FXS database for %s",
+ AliShuttleInterface::GetSystemName(system)));
+ } else {
+ AliError("Can't establish connection to Run logbook.");
+ }
+ if(fServer[system]) delete fServer[system];
+ return kFALSE;
+ }
+
+ // Get tables
+ TSQLResult* aResult=0;
+ switch(system){
+ case kDAQ:
+ aResult = fServer[kDAQ]->GetTables(dbName.Data());
+ break;
+ case kDCS:
+ aResult = fServer[kDCS]->GetTables(dbName.Data());
+ break;
+ case kHLT:
+ aResult = fServer[kHLT]->GetTables(dbName.Data());
+ break;
+ default:
+ aResult = fServer[3]->GetTables(dbName.Data());
+ break;
+ }
+
+ delete aResult;
+ return kTRUE;
+}
+
+//______________________________________________________________________________________________
+Bool_t AliShuttle::UpdateTable()
+{
+ //
+ // Update FXS table filling time_processed field in all rows corresponding to current run and detector
+ //
+
+ Bool_t result = kTRUE;
+
+ for (UInt_t system=0; system<3; system++)
+ {
+ if(!fFXSCalled[system]) continue;
+
+ // check connection, in case connect
+ if (!Connect(system))
+ {
+ Log(fCurrentDetector, Form("UpdateTable - Couldn't connect to %s FXS database", GetSystemName(system)));
+ result = kFALSE;
+ continue;
+ }
+
+ TTimeStamp now; // now
+
+ // Loop on FXS list entries
+ TIter iter(&fFXSlist[system]);
+ TObjString *aFXSentry=0;
+ while ((aFXSentry = dynamic_cast<TObjString*> (iter.Next())))
+ {
+ TString aFXSentrystr = aFXSentry->String();
+ TObjArray *aFXSarray = aFXSentrystr.Tokenize("#!?!#");
+ if (!aFXSarray || aFXSarray->GetEntries() != 2 )
+ {
+ Log(fCurrentDetector, Form("UpdateTable - error updating %s FXS entry. Check string: <%s>",
+ GetSystemName(system), aFXSentrystr.Data()));
+ if(aFXSarray) delete aFXSarray;
+ result = kFALSE;
+ continue;
+ }
+ const char* fileId = ((TObjString*) aFXSarray->At(0))->GetName();
+ const char* source = ((TObjString*) aFXSarray->At(1))->GetName();
+
+ TString whereClause;
+ if (system == kDAQ)
+ {
+ whereClause = Form("where run=%d and detector=\"%s\" and fileId=\"%s\" and DAQsource=\"%s\";",
+ GetCurrentRun(), fCurrentDetector.Data(), fileId, source);
+ }
+ else if (system == kDCS)
+ {
+ whereClause = Form("where run=%d and detector=\"%s\" and fileId=\"%s\";",
+ GetCurrentRun(), fCurrentDetector.Data(), fileId);
+ }
+ else if (system == kHLT)
+ {
+ whereClause = Form("where run=%d and detector=\"%s\" and fileId=\"%s\" and DDLnumbers=\"%s\";",
+ GetCurrentRun(), fCurrentDetector.Data(), fileId, source);
+ }
+
+ delete aFXSarray;
+
+ TString sqlQuery = Form("update %s set time_processed=%d %s", fConfig->GetFXSdbTable(system),
+ now.GetSec(), whereClause.Data());
+
+ AliDebug(2, Form("SQL query: \n%s",sqlQuery.Data()));
+
+ // Query execution
+ TSQLResult* aResult;
+ aResult = dynamic_cast<TSQLResult*> (fServer[system]->Query(sqlQuery));
+ if (!aResult)
+ {
+ Log(fCurrentDetector, Form("UpdateTable - %s db: can't execute SQL query <%s>",
+ GetSystemName(system), sqlQuery.Data()));
+ result = kFALSE;
+ continue;
+ }
+ delete aResult;
+ }
+ }
+
+ return result;
+}
+
+//______________________________________________________________________________________________
+Bool_t AliShuttle::UpdateTableFailCase()
+{
+ // Update FXS table filling time_processed field in all rows corresponding to current run and detector
+ // this is called in case the preprocessor is declared failed for the current run, because
+ // the fields are updated only in case of success
+
+ Bool_t result = kTRUE;
+
+ for (UInt_t system=0; system<3; system++)
+ {
+ // check connection, in case connect
+ if (!Connect(system))
+ {
+ Log(fCurrentDetector, Form("UpdateTableFailCase - Couldn't connect to %s FXS database",
+ GetSystemName(system)));
+ result = kFALSE;
+ continue;
+ }
+
+ TTimeStamp now; // now
+
+ // Loop on FXS list entries
+
+ TString whereClause = Form("where run=%d and detector=\"%s\";",
+ GetCurrentRun(), fCurrentDetector.Data());
+
+
+ TString sqlQuery = Form("update %s set time_processed=%d %s", fConfig->GetFXSdbTable(system),
+ now.GetSec(), whereClause.Data());
+
+ AliDebug(2, Form("SQL query: \n%s",sqlQuery.Data()));
+
+ // Query execution
+ TSQLResult* aResult;
+ aResult = dynamic_cast<TSQLResult*> (fServer[system]->Query(sqlQuery));
+ if (!aResult)
+ {
+ Log(fCurrentDetector, Form("UpdateTableFailCase - %s db: can't execute SQL query <%s>",
+ GetSystemName(system), sqlQuery.Data()));
+ result = kFALSE;
+ continue;
+ }
+ delete aResult;
+ }
+
+ return result;
+}
+
+//______________________________________________________________________________________________
+Bool_t AliShuttle::UpdateShuttleLogbook(const char* detector, const char* status)
+{
+ //
+ // Update Shuttle logbook filling detector or shuttle_done column
+ // ex. of usage: UpdateShuttleLogbook("PHOS", "DONE") or UpdateShuttleLogbook("shuttle_done")
+ //
+
+ // check connection, in case connect
+ if(!Connect(3)){
+ Log("SHUTTLE", "UpdateShuttleLogbook - Couldn't connect to DAQ Logbook.");
+ return kFALSE;
+ }
+
+ TString detName(detector);
+ TString setClause;
+ if (detName == "shuttle_done" || detName == "shuttle_ignored")
+ {
+ setClause = "set shuttle_done=1";
+
+ if (detName == "shuttle_done")
+ {
+ // Send the information to ML
+ TMonaLisaText mlStatus("SHUTTLE_status", "Done");
+
+ TList mlList;
+ mlList.Add(&mlStatus);
+
+ TString mlID;
+ mlID.Form("%d", GetCurrentRun());
+ fMonaLisa->SendParameters(&mlList, mlID);
+ }
+ } else {
+ TString statusStr(status);
+ if(statusStr.Contains("done", TString::kIgnoreCase) ||
+ statusStr.Contains("failed", TString::kIgnoreCase)){
+ setClause = Form("set %s=\"%s\"", detector, status);
+ } else {
+ Log("SHUTTLE",
+ Form("UpdateShuttleLogbook - Invalid status <%s> for detector %s",
+ status, detector));
+ return kFALSE;
+ }
+ }
+
+ TString whereClause = Form("where run=%d", GetCurrentRun());
+
+ TString sqlQuery = Form("update %s %s %s",
+ fConfig->GetShuttlelbTable(), setClause.Data(), whereClause.Data());
+
+ AliDebug(2, Form("SQL query: \n%s",sqlQuery.Data()));
+
+ // Query execution
+ TSQLResult* aResult;
+ aResult = dynamic_cast<TSQLResult*> (fServer[3]->Query(sqlQuery));
+ if (!aResult) {
+ Log("SHUTTLE", Form("UpdateShuttleLogbook - Can't execute query <%s>", sqlQuery.Data()));
+ return kFALSE;
+ }
+ delete aResult;
+
+ return kTRUE;
+}
+
+//______________________________________________________________________________________________
+Int_t AliShuttle::GetCurrentRun() const
+{
+ //
+ // Get current run from logbook entry
+ //
+
+ return fLogbookEntry ? fLogbookEntry->GetRun() : -1;
+}
+
+//______________________________________________________________________________________________
+UInt_t AliShuttle::GetCurrentStartTime() const
+{
+ //
+ // get current start time
+ //
+
+ return fLogbookEntry ? fLogbookEntry->GetStartTime() : 0;
+}
+
+//______________________________________________________________________________________________
+UInt_t AliShuttle::GetCurrentEndTime() const
+{
+ //
+ // get current end time from logbook entry
+ //
+
+ return fLogbookEntry ? fLogbookEntry->GetEndTime() : 0;
+}
+
+//______________________________________________________________________________________________
+UInt_t AliShuttle::GetCurrentYear() const
+{
+ //
+ // Get current year from logbook entry
+ //
+
+ if (!fLogbookEntry) return 0;
+
+ TTimeStamp startTime(GetCurrentStartTime());
+ TString year = Form("%d",startTime.GetDate());
+ year = year(0,4);
+
+ return year.Atoi();
+}
+
+//______________________________________________________________________________________________
+const char* AliShuttle::GetLHCPeriod() const
+{
+ //
+ // Get current LHC period from logbook entry
+ //
+
+ if (!fLogbookEntry) return 0;
+
+ return fLogbookEntry->GetRunParameter("LHCperiod");
+}
+
+//______________________________________________________________________________________________
+void AliShuttle::Log(const char* detector, const char* message)
+{
+ //
+ // Fill log string with a message
+ //
+
+ TString logRunDir = GetShuttleLogDir();
+ if (GetCurrentRun() >=0)
+ logRunDir += Form("/%d", GetCurrentRun());
+
+ void* dir = gSystem->OpenDirectory(logRunDir.Data());
+ if (dir == NULL) {
+ if (gSystem->mkdir(logRunDir.Data(), kTRUE)) {
+ AliError(Form("Can't open directory <%s>", GetShuttleLogDir()));
+ return;
+ }
+
+ } else {
+ gSystem->FreeDirectory(dir);
+ }
+
+ TString toLog = Form("%s (%d): %s - ", TTimeStamp(time(0)).AsString("s"), getpid(), detector);
+ if (GetCurrentRun() >= 0)
+ toLog += Form("run %d - ", GetCurrentRun());
+ toLog += Form("%s", message);
+
+ AliInfo(toLog.Data());
+
+ // if we redirect the log output already to the file, leave here
+ if (fOutputRedirected && strcmp(detector, "SHUTTLE") != 0)
+ return;
+
+ TString fileName = GetLogFileName(detector);
+
+ gSystem->ExpandPathName(fileName);
+
+ ofstream logFile;
+ logFile.open(fileName, ofstream::out | ofstream::app);
+
+ if (!logFile.is_open()) {
+ AliError(Form("Could not open file %s", fileName.Data()));
+ return;
+ }
+
+ logFile << toLog.Data() << "\n";
+
+ logFile.close();
+}
+
+//______________________________________________________________________________________________
+TString AliShuttle::GetLogFileName(const char* detector) const
+{
+ //
+ // returns the name of the log file for a given sub detector
+ //
+
+ TString fileName;
+
+ if (GetCurrentRun() >= 0)
+ {
+ fileName.Form("%s/%d/%s_%d.log", GetShuttleLogDir(), GetCurrentRun(),
+ detector, GetCurrentRun());
+ } else {
+ fileName.Form("%s/%s.log", GetShuttleLogDir(), detector);
+ }
+
+ return fileName;
+}
+
+//______________________________________________________________________________________________
+void AliShuttle::SendAlive()
+{
+ // sends alive message to ML
+
+ TMonaLisaText mlStatus("SHUTTLE_status", "Alive");
+
+ TList mlList;
+ mlList.Add(&mlStatus);
+
+ fMonaLisa->SendParameters(&mlList, "__PROCESSINGINFO__");
+}
+
+//______________________________________________________________________________________________
+Bool_t AliShuttle::Collect(Int_t run)
+{
+ //
+ // Collects conditions data for all UNPROCESSED run written to DAQ LogBook in case of run = -1 (default)
+ // If a dedicated run is given this run is processed
+ //
+ // In operational mode, this is the Shuttle function triggered by the EOR signal.
+ //
+
+ if (run == -1)
+ Log("SHUTTLE","Collect - Shuttle called. Collecting conditions data for unprocessed runs");
+ else
+ Log("SHUTTLE", Form("Collect - Shuttle called. Collecting conditions data for run %d", run));
+
+ SetLastAction("Starting");