From 028b5231b2267da30cb7d2b9c8cc3d7b45470894 Mon Sep 17 00:00:00 2001
From: Eric Cano <Eric.Cano@cern.ch>
Date: Tue, 5 Mar 2019 09:02:23 +0100
Subject: [PATCH] Expanded retrieve requests for repack
The retrieve requests now carry all the information necessary to create the archive jobs.
Retrive requests carry an isRepack boolean. The distinction between repack and non-repack behaviour
is pushed as low as possible in the api (in the object store) so that client code can be more
generic. For the same reason, the
The expand repack mode (re-creation of missing tape files) is not supported yet.
Changed the represenation of copy numbers to uint32 (as this is the smalled number supported but protocol buffers).
Added ArchiveRoute::StorageClassMap and ArchiveRoute::FullMap to store archive routing in memory.
Added repack buffer base URL support, and mechanism to compute temporary copy location in disk cache.
Fixed bug in ObjectOps<>::asyncInsert();
Inplemented the path generation for repack disk storage (<BaseURL>/<VID>/<FSeq (09%d)>).
---
common/dataStructures/ArchiveFile.cpp | 6 +-
common/dataStructures/ArchiveFile.hpp | 4 +-
common/dataStructures/ArchiveRoute.hpp | 3 +
common/dataStructures/RepackInfo.hpp | 1 +
common/dataStructures/RetrieveJob.hpp | 2 +-
common/dataStructures/RetrieveRequest.cpp | 8 +-
common/dataStructures/RetrieveRequest.hpp | 3 +-
common/dataStructures/utils.cpp | 4 +-
common/dataStructures/utils.hpp | 4 +-
objectstore/Helpers.cpp | 2 +
objectstore/ObjectOps.hpp | 1 +
objectstore/RepackRequest.cpp | 56 ++-
objectstore/RepackRequest.hpp | 17 +-
objectstore/RetrieveQueueAlgorithms.hpp | 5 +-
.../RetrieveQueueToTransferAlgorithms.cpp | 3 +-
objectstore/RetrieveRequest.cpp | 97 ++--
objectstore/RetrieveRequest.hpp | 37 +-
objectstore/RootEntry.cpp | 3 +-
objectstore/cta.proto | 19 +-
scheduler/OStoreDB/OStoreDB.cpp | 424 +++++++++++++++---
scheduler/OStoreDB/OStoreDB.hpp | 34 +-
scheduler/RepackRequest.cpp | 31 +-
scheduler/RepackRequest.hpp | 3 +-
scheduler/RetrieveJob.cpp | 11 +-
scheduler/RetrieveJob.hpp | 9 +-
scheduler/RetrieveMount.cpp | 14 +-
scheduler/RetrieveMount.hpp | 2 +-
scheduler/Scheduler.cpp | 99 ++--
scheduler/Scheduler.hpp | 1 -
scheduler/SchedulerDatabase.hpp | 32 +-
scheduler/SchedulerTest.cpp | 51 ++-
scheduler/testingMocks/MockRetrieveJob.hpp | 3 +-
scheduler/testingMocks/MockRetrieveMount.hpp | 2 +-
.../tapeserver/daemon/DiskWriteTaskTest.cpp | 3 +-
.../daemon/DiskWriteThreadPoolTest.cpp | 3 +-
.../tapeserver/daemon/RecallReportPacker.cpp | 6 +-
.../daemon/RecallReportPackerTest.cpp | 4 +-
.../daemon/RecallTaskInjectorTest.cpp | 4 +-
38 files changed, 748 insertions(+), 263 deletions(-)
diff --git a/common/dataStructures/ArchiveFile.cpp b/common/dataStructures/ArchiveFile.cpp
index b302c57b72..c8ee856db2 100644
--- a/common/dataStructures/ArchiveFile.cpp
+++ b/common/dataStructures/ArchiveFile.cpp
@@ -59,7 +59,7 @@ bool ArchiveFile::operator!=(const ArchiveFile &rhs) const {
//------------------------------------------------------------------------------
// operator<<
//------------------------------------------------------------------------------
-std::ostream &operator<<(std::ostream &os, ArchiveFile &obj) {
+std::ostream &operator<<(std::ostream &os, const ArchiveFile &obj) {
os <<
"{"
"archiveFileID=" << obj.archiveFileID << ","
@@ -69,8 +69,8 @@ std::ostream &operator<<(std::ostream &os, ArchiveFile &obj) {
"checksumType=" << obj.checksumType << ","
"checksumValue=" << obj.checksumValue << ","
"storageClass=" << obj.storageClass << ","
- "diskFileInfo=" << obj.diskFileInfo << ","
- "tapeFiles=" << obj.tapeFiles << ","
+ "diskFileInfo=" << obj.diskFileInfo << ","
+ "tapeFiles=" << obj.tapeFiles << ","
"creationTime=" << obj.creationTime << ","
"reconciliationTime=" << obj.reconciliationTime <<
"}";
diff --git a/common/dataStructures/ArchiveFile.hpp b/common/dataStructures/ArchiveFile.hpp
index 725b83aa7c..f2875077e5 100644
--- a/common/dataStructures/ArchiveFile.hpp
+++ b/common/dataStructures/ArchiveFile.hpp
@@ -68,8 +68,8 @@ struct ArchiveFile {
* to be listed by the operator. For example, if the listing requested is
* for a single tape, the map will contain only one element.
*/
- std::map<uint64_t,TapeFile> tapeFiles;
- time_t creationTime;
+ std::map<uint32_t,TapeFile> tapeFiles;
+ time_t creationTime;
time_t reconciliationTime;
}; // struct ArchiveFile
diff --git a/common/dataStructures/ArchiveRoute.hpp b/common/dataStructures/ArchiveRoute.hpp
index 6f668cede5..454af64f40 100644
--- a/common/dataStructures/ArchiveRoute.hpp
+++ b/common/dataStructures/ArchiveRoute.hpp
@@ -61,6 +61,9 @@ struct ArchiveRoute {
EntryLog creationLog;
EntryLog lastModificationLog;
std::string comment;
+
+ typedef std::map<uint32_t, ArchiveRoute> StorageClassMap;
+ typedef std::map<std::tuple<std::string/*disk instance*/, std::string /*storage class*/>, StorageClassMap> FullMap;
}; // struct ArchiveRoute
diff --git a/common/dataStructures/RepackInfo.hpp b/common/dataStructures/RepackInfo.hpp
index 4d2f47c55b..aeb5ea92fd 100644
--- a/common/dataStructures/RepackInfo.hpp
+++ b/common/dataStructures/RepackInfo.hpp
@@ -30,6 +30,7 @@ namespace dataStructures {
struct RepackInfo {
std::string vid;
+ std::string repackBufferBaseURL;
enum class Type {
ExpandAndRepack,
ExpandOnly,
diff --git a/common/dataStructures/RetrieveJob.hpp b/common/dataStructures/RetrieveJob.hpp
index 188ddda69d..c1958bc656 100644
--- a/common/dataStructures/RetrieveJob.hpp
+++ b/common/dataStructures/RetrieveJob.hpp
@@ -43,7 +43,7 @@ struct RetrieveJob {
RetrieveRequest request;
uint64_t fileSize;
- std::map<std::string,std::pair<uint64_t,TapeFile>> tapeCopies;
+ std::map<std::string,std::pair<uint32_t,TapeFile>> tapeCopies;
std::list<std::string> failurelogs;
}; // struct RetrieveJob
diff --git a/common/dataStructures/RetrieveRequest.cpp b/common/dataStructures/RetrieveRequest.cpp
index 35154d638f..43d170d5de 100644
--- a/common/dataStructures/RetrieveRequest.cpp
+++ b/common/dataStructures/RetrieveRequest.cpp
@@ -27,7 +27,7 @@ namespace dataStructures {
//------------------------------------------------------------------------------
// constructor
//------------------------------------------------------------------------------
-RetrieveRequest::RetrieveRequest(): archiveFileID(0),isRepack(false) {}
+RetrieveRequest::RetrieveRequest(): archiveFileID(0) {}
//------------------------------------------------------------------------------
// operator==
@@ -37,8 +37,7 @@ bool RetrieveRequest::operator==(const RetrieveRequest &rhs) const {
&& archiveFileID==rhs.archiveFileID
&& dstURL==rhs.dstURL
&& diskFileInfo==rhs.diskFileInfo
- && creationLog==rhs.creationLog
- && isRepack == rhs.isRepack;
+ && creationLog==rhs.creationLog;
}
//------------------------------------------------------------------------------
@@ -56,8 +55,7 @@ std::ostream &operator<<(std::ostream &os, const RetrieveRequest &obj) {
<< " archiveFileID=" << obj.archiveFileID
<< " dstURL=" << obj.dstURL
<< " diskFileInfo=" << obj.diskFileInfo
- << " creationLog=" << obj.creationLog
- << " isRepack=" << obj.isRepack<<")";
+ << " creationLog=" << obj.creationLog <<")";
return os;
}
diff --git a/common/dataStructures/RetrieveRequest.hpp b/common/dataStructures/RetrieveRequest.hpp
index a452545286..f319ec08a0 100644
--- a/common/dataStructures/RetrieveRequest.hpp
+++ b/common/dataStructures/RetrieveRequest.hpp
@@ -26,6 +26,7 @@
#include "common/dataStructures/DiskFileInfo.hpp"
#include "common/dataStructures/EntryLog.hpp"
#include "common/dataStructures/UserIdentity.hpp"
+#include "common/dataStructures/ArchiveRoute.hpp"
namespace cta {
namespace common {
@@ -48,8 +49,6 @@ struct RetrieveRequest {
std::string errorReportURL;
DiskFileInfo diskFileInfo;
EntryLog creationLog;
- bool isRepack;
- std::string tapePool;
}; // struct RetrieveRequest
std::ostream &operator<<(std::ostream &os, const RetrieveRequest &obj);
diff --git a/common/dataStructures/utils.cpp b/common/dataStructures/utils.cpp
index 181052e751..689026cd7c 100644
--- a/common/dataStructures/utils.cpp
+++ b/common/dataStructures/utils.cpp
@@ -22,7 +22,7 @@ namespace cta {
namespace common {
namespace dataStructures {
-std::ostream &operator<<(std::ostream &os, const std::map<uint64_t,TapeFile> &map) {
+std::ostream &operator<<(std::ostream &os, const std::map<uint32_t,TapeFile> &map) {
os << "(";
for(auto it = map.begin(); it != map.end(); it++) {
os << " key=" << it->first << " value=" << it->second << " ";
@@ -54,7 +54,7 @@ std::ostream &operator<<(std::ostream &os, const std::map<uint64_t,std::pair<std
return os;
}
-std::ostream &operator<<(std::ostream &os, const std::map<std::string,std::pair<uint64_t,TapeFile>> &map) {
+std::ostream &operator<<(std::ostream &os, const std::map<std::string,std::pair<uint32_t,TapeFile>> &map) {
os << "(";
for(auto it = map.begin(); it != map.end(); it++) {
os << " key=" << it->first << " value.first=" << it->second.first << " value.second=" << it->second.second;
diff --git a/common/dataStructures/utils.hpp b/common/dataStructures/utils.hpp
index b7fb7ea76e..c3e989a42d 100644
--- a/common/dataStructures/utils.hpp
+++ b/common/dataStructures/utils.hpp
@@ -27,11 +27,11 @@ namespace cta {
namespace common {
namespace dataStructures {
-std::ostream &operator<<(std::ostream &os, const std::map<uint64_t,TapeFile> &map);
+std::ostream &operator<<(std::ostream &os, const std::map<uint32_t,TapeFile> &map);
std::ostream &operator<<(std::ostream &os, const std::map<uint64_t,std::string> &map);
std::ostream &operator<<(std::ostream &os, const std::pair<std::string,std::string> &pair);
std::ostream &operator<<(std::ostream &os, const std::map<uint64_t,std::pair<std::string,std::string>> &map);
-std::ostream &operator<<(std::ostream &os, const std::map<std::string,std::pair<uint64_t,TapeFile>> &map);
+std::ostream &operator<<(std::ostream &os, const std::map<std::string,std::pair<uint32_t,TapeFile>> &map);
std::ostream &operator<<(std::ostream &os, const std::map<uint64_t,std::pair<std::string,std::string>> &map);
} // namespace dataStructures
diff --git a/objectstore/Helpers.cpp b/objectstore/Helpers.cpp
index 4e3328961f..cef85a9fb6 100644
--- a/objectstore/Helpers.cpp
+++ b/objectstore/Helpers.cpp
@@ -197,6 +197,8 @@ void Helpers::getLockedAndFetchedJobQueue<RetrieveQueue>(RetrieveQueue& retrieve
queueFetchTime = t.secs(utils::Timer::resetCounter);
log::ScopedParamContainer params(lc);
params.add("attemptNb", i+1)
+ .add("queueName", vid.value())
+ .add("queueType", toString(queueType))
.add("queueObject", retrieveQueue.getAddressIfSet())
.add("rootFetchNoLockTime", rootFetchNoLockTime)
.add("rootRelockExclusiveTime", rootRelockExclusiveTime)
diff --git a/objectstore/ObjectOps.hpp b/objectstore/ObjectOps.hpp
index 5fe5868fde..aceb9d6230 100644
--- a/objectstore/ObjectOps.hpp
+++ b/objectstore/ObjectOps.hpp
@@ -496,6 +496,7 @@ public:
AsyncInserter * asyncInsert() {
std::unique_ptr<AsyncInserter> ret;
+ ret.reset(new AsyncInserter(*this));
// Current simplification: the parsing of the header/payload is synchronous.
// This could be delegated to the backend.
// Check that we are not dealing with an existing object
diff --git a/objectstore/RepackRequest.cpp b/objectstore/RepackRequest.cpp
index 979891c5f0..bf6e210984 100644
--- a/objectstore/RepackRequest.cpp
+++ b/objectstore/RepackRequest.cpp
@@ -59,7 +59,7 @@ void RepackRequest::initialize() {
m_payload.set_retrievedbytes(0);
m_payload.set_archivedfiles(0);
m_payload.set_archivedbytes(0);
- m_payload.set_failedtoretievefiles(0);
+ m_payload.set_failedtoretrievefiles(0);
m_payload.set_failedtoretrievebytes(0);
m_payload.set_failedtoarchivefiles(0);
m_payload.set_failedtoarchivebytes(0);
@@ -98,6 +98,16 @@ void RepackRequest::setType(common::dataStructures::RepackInfo::Type repackType)
}
}
+//------------------------------------------------------------------------------
+// RepackRequest::setStatus()
+//------------------------------------------------------------------------------
+void RepackRequest::setStatus(common::dataStructures::RepackInfo::Status repackStatus) {
+ checkPayloadWritable();
+ // common::dataStructures::RepackInfo::Status and serializers::RepackRequestStatus are defined using the same values,
+ // hence the cast.
+ m_payload.set_status((serializers::RepackRequestStatus)repackStatus);
+}
+
//------------------------------------------------------------------------------
// RepackRequest::getInfo()
//------------------------------------------------------------------------------
@@ -107,6 +117,7 @@ common::dataStructures::RepackInfo RepackRequest::getInfo() {
RepackInfo ret;
ret.vid = m_payload.vid();
ret.status = (RepackInfo::Status) m_payload.status();
+ ret.repackBufferBaseURL = m_payload.buffer_url();
if (m_payload.repackmode()) {
if (m_payload.expandmode()) {
ret.type = RepackInfo::Type::ExpandAndRepack;
@@ -121,6 +132,14 @@ common::dataStructures::RepackInfo RepackRequest::getInfo() {
return ret;
}
+//------------------------------------------------------------------------------
+// RepackRequest::setBufferURL()
+//------------------------------------------------------------------------------
+void RepackRequest::setBufferURL(const std::string& bufferURL) {
+ checkPayloadWritable();
+ m_payload.set_buffer_url(bufferURL);
+}
+
//------------------------------------------------------------------------------
// RepackRequest::RepackSubRequestPointer::serialize()
//------------------------------------------------------------------------------
@@ -148,7 +167,7 @@ void RepackRequest::RepackSubRequestPointer::deserialize(const serializers::Repa
//------------------------------------------------------------------------------
// RepackRequest::getOrPrepareSubrequestInfo()
//------------------------------------------------------------------------------
-auto RepackRequest::getOrPrepareSubrequestInfo(std::set<uint32_t> fSeqs, AgentReference& agentRef)
+auto RepackRequest::getOrPrepareSubrequestInfo(std::set<uint64_t> fSeqs, AgentReference& agentRef)
-> SubrequestInfo::set {
checkPayloadWritable();
RepackSubRequestPointer::Map pointerMap;
@@ -165,7 +184,7 @@ auto RepackRequest::getOrPrepareSubrequestInfo(std::set<uint32_t> fSeqs, AgentRe
retInfo.fSeq = srp.fSeq;
retInfo.subrequestDeleted = srp.subrequestDeleted;
} catch (std::out_of_range &) {
- retInfo.address = agentRef.nextId("repackSubRequest");
+ retInfo.address = agentRef.nextId("RepackSubRequest");
retInfo.fSeq = fs;
retInfo.subrequestDeleted = false;
auto & p = pointerMap[fs];
@@ -179,7 +198,7 @@ auto RepackRequest::getOrPrepareSubrequestInfo(std::set<uint32_t> fSeqs, AgentRe
// Record changes, if any.
if (newElementCreated) {
m_payload.mutable_subrequests()->Clear();
- for (auto & p: pointerMap) p.second.deserialize(*m_payload.mutable_subrequests()->Add());
+ for (auto & p: pointerMap) p.second.serialize(*m_payload.mutable_subrequests()->Add());
}
return ret;
}
@@ -188,7 +207,7 @@ auto RepackRequest::getOrPrepareSubrequestInfo(std::set<uint32_t> fSeqs, AgentRe
// RepackRequest::setLastExpandedFSeq()
//------------------------------------------------------------------------------
void RepackRequest::setLastExpandedFSeq(uint64_t lastExpandedFSeq) {
- checkWritable();
+ checkPayloadWritable();
m_payload.set_lastexpandedfseq(lastExpandedFSeq);
}
@@ -243,7 +262,7 @@ void RepackRequest::reportRetriveFailures(SubrequestStatistics::List& retrieveFa
if (!p.failureAccounted) {
p.failureAccounted = true;
m_payload.set_failedtoretrievebytes(m_payload.failedtoretrievebytes() + rs.bytes);
- m_payload.set_failedtoretievefiles(m_payload.failedtoretievefiles() + rs.files);
+ m_payload.set_failedtoretrievefiles(m_payload.failedtoretrievefiles() + rs.files);
didUpdate = true;
}
} catch (std::out_of_range &) {
@@ -338,6 +357,30 @@ void RepackRequest::reportSubRequestsForDeletion(std::list<uint64_t>& fSeqs) {
}
}
+//------------------------------------------------------------------------------
+// RepackRequest::reportSubRequestsForDeletion()
+//------------------------------------------------------------------------------
+auto RepackRequest::getStats() -> std::map<StatsType, StatsValues> {
+ checkPayloadReadable();
+ std::map<StatsType, StatsValues> ret;
+ ret[StatsType::ArchiveTotal].files = m_payload.totalfilestoarchive();
+ ret[StatsType::ArchiveTotal].bytes = m_payload.totalbytestoarchive();
+ ret[StatsType::RetrieveTotal].files = m_payload.totalfilestoretrieve();
+ ret[StatsType::RetrieveTotal].bytes = m_payload.totalbytestoretrieve();
+ ret[StatsType::UserProvided].files = m_payload.userprovidedfiles();
+ ret[StatsType::UserProvided].bytes = m_payload.userprovidedbytes();
+ ret[StatsType::RetrieveFailure].files = m_payload.failedtoretrievefiles();
+ ret[StatsType::RetrieveFailure].bytes = m_payload.failedtoretrievebytes();
+ ret[StatsType::RetrieveSuccess].files = m_payload.retrievedfiles();
+ ret[StatsType::RetrieveSuccess].bytes = m_payload.retrievedbytes();
+ ret[StatsType::ArchiveFailure].files = m_payload.failedtoarchivefiles();
+ ret[StatsType::ArchiveFailure].bytes = m_payload.failedtoarchivebytes();
+ ret[StatsType::ArchiveSuccess].files = m_payload.archivedfiles();
+ ret[StatsType::ArchiveSuccess].bytes = m_payload.archivedbytes();
+ return ret;
+}
+
+
//------------------------------------------------------------------------------
// RepackRequest::garbageCollect()
//------------------------------------------------------------------------------
@@ -387,6 +430,7 @@ RepackRequest::AsyncOwnerAndStatusUpdater* RepackRequest::asyncUpdateOwnerAndSta
typedef common::dataStructures::RepackInfo RepackInfo;
retRef.m_repackInfo.status = (RepackInfo::Status) payload.status();
retRef.m_repackInfo.vid = payload.vid();
+ retRef.m_repackInfo.repackBufferBaseURL = payload.buffer_url();
if (payload.repackmode()) {
if (payload.expandmode()) {
retRef.m_repackInfo.type = RepackInfo::Type::ExpandAndRepack;
diff --git a/objectstore/RepackRequest.hpp b/objectstore/RepackRequest.hpp
index 1cf2a9a3b0..8513caa2af 100644
--- a/objectstore/RepackRequest.hpp
+++ b/objectstore/RepackRequest.hpp
@@ -41,6 +41,7 @@ public:
void setType(common::dataStructures::RepackInfo::Type repackType);
void setStatus(common::dataStructures::RepackInfo::Status repackStatus);
common::dataStructures::RepackInfo getInfo();
+ void setBufferURL(const std::string & bufferURL);
// Sub request management
struct SubrequestInfo {
@@ -59,7 +60,7 @@ public:
* yet not update the object to reflect the last fSeq created.
* This function implicitly records the information it generates (commit up t the caller);
*/
- SubrequestInfo::set getOrPrepareSubrequestInfo (std::set<uint32_t> fSeqs, AgentReference & agentRef);
+ SubrequestInfo::set getOrPrepareSubrequestInfo (std::set<uint64_t> fSeqs, AgentReference & agentRef);
private:
struct RepackSubRequestPointer {
std::string address;
@@ -90,6 +91,20 @@ public:
void reportArchiveSuccesses (SubrequestStatistics::List & archiveSuccesses);
void reportArchiveFailures (SubrequestStatistics::List & archiveFailures);
void reportSubRequestsForDeletion (std::list<uint64_t>& fSeqs);
+ enum class StatsType: uint8_t {
+ UserProvided,
+ RetrieveSuccess,
+ RetrieveFailure,
+ RetrieveTotal,
+ ArchiveSuccess,
+ ArchiveFailure,
+ ArchiveTotal,
+ };
+ struct StatsValues {
+ uint64_t files = 0;
+ uint64_t bytes = 0;
+ };
+ std::map<StatsType, StatsValues> getStats();
void garbageCollect(const std::string &presumedOwner, AgentReference & agentReference, log::LogContext & lc,
cta::catalogue::Catalogue & catalogue) override;
diff --git a/objectstore/RetrieveQueueAlgorithms.hpp b/objectstore/RetrieveQueueAlgorithms.hpp
index 02532eb0a2..1cb0627f45 100644
--- a/objectstore/RetrieveQueueAlgorithms.hpp
+++ b/objectstore/RetrieveQueueAlgorithms.hpp
@@ -55,6 +55,7 @@ struct ContainerTraits<RetrieveQueue,C>
common::dataStructures::RetrieveRequest rr;
std::string errorReportURL;
SchedulerDatabase::RetrieveJob::ReportType reportType;
+ RetrieveRequest::RepackInfo repackInfo;
};
struct PoppedElementsSummary;
struct PopCriteria {
@@ -382,6 +383,7 @@ switchElementsOwnership(PoppedElementsBatch &poppedElementBatch, const Container
u.get()->wait();
e.archiveFile = u.get()->getArchiveFile();
e.rr = u.get()->getRetrieveRequest();
+ e.repackInfo = u.get()->getRepackInfo();
switch(u.get()->getJobStatus()) {
case serializers::RetrieveJobStatus::RJS_ToReportToUserForFailure:
e.reportType = SchedulerDatabase::RetrieveJob::ReportType::FailureReport;
@@ -486,7 +488,8 @@ getPoppingElementsCandidates(Container &cont, PopCriteria &unfulfilledCriteria,
cjfq.size,
common::dataStructures::ArchiveFile(),
common::dataStructures::RetrieveRequest(),
- "", SchedulerDatabase::RetrieveJob::ReportType::NoReportRequired
+ "", SchedulerDatabase::RetrieveJob::ReportType::NoReportRequired,
+ RetrieveRequest::RepackInfo()
});
ret.summary.files++;
}
diff --git a/objectstore/RetrieveQueueToTransferAlgorithms.cpp b/objectstore/RetrieveQueueToTransferAlgorithms.cpp
index 2367b0920e..2d80b46b33 100644
--- a/objectstore/RetrieveQueueToTransferAlgorithms.cpp
+++ b/objectstore/RetrieveQueueToTransferAlgorithms.cpp
@@ -55,7 +55,8 @@ getPoppingElementsCandidates(Container &cont, PopCriteria &unfulfilledCriteria,
common::dataStructures::ArchiveFile(),
common::dataStructures::RetrieveRequest(),
"",
- SchedulerDatabase::RetrieveJob::ReportType::NoReportRequired
+ SchedulerDatabase::RetrieveJob::ReportType::NoReportRequired,
+ RetrieveRequest::RepackInfo()
});
ret.summary.bytes += cjfq.size;
ret.summary.files++;
diff --git a/objectstore/RetrieveRequest.cpp b/objectstore/RetrieveRequest.cpp
index 5bedfaa473..a46aa7358d 100644
--- a/objectstore/RetrieveRequest.cpp
+++ b/objectstore/RetrieveRequest.cpp
@@ -58,7 +58,6 @@ void RetrieveRequest::initialize() {
m_payload.set_failurereportlog("");
m_payload.set_failurereporturl("");
m_payload.set_isrepack(false);
- m_payload.set_tapepool("");
// This object is good to go (to storage)
m_payloadInterpreted = true;
}
@@ -372,7 +371,6 @@ std::string RetrieveRequest::getLastActiveVid() {
return m_payload.archivefile().tapefiles(0).vid();
}
-
//------------------------------------------------------------------------------
// RetrieveRequest::setSchedulerRequest()
//------------------------------------------------------------------------------
@@ -402,8 +400,6 @@ cta::common::dataStructures::RetrieveRequest RetrieveRequest::getSchedulerReques
objectstore::EntryLogSerDeser el(ret.creationLog);
el.deserialize(m_payload.schedulerrequest().entrylog());
ret.dstURL = m_payload.schedulerrequest().dsturl();
- ret.isRepack = m_payload.isrepack();
- ret.tapePool = m_payload.tapepool();
ret.errorReportURL = m_payload.schedulerrequest().retrieveerrorreporturl();
objectstore::DiskFileInfoSerDeser dfisd;
dfisd.deserialize(m_payload.schedulerrequest().diskfileinfo());
@@ -516,6 +512,37 @@ bool RetrieveRequest::addJobFailure(uint32_t copyNumber, uint64_t mountId,
throw NoSuchJob ("In RetrieveRequest::addJobFailure(): could not find job");
}
+//------------------------------------------------------------------------------
+// RetrieveRequest::setRepackInfo()
+//------------------------------------------------------------------------------
+void RetrieveRequest::setRepackInfo(const RepackInfo& repackInfo) {
+ checkPayloadWritable();
+ m_payload.set_isrepack(repackInfo.isRepack);
+ if (repackInfo.isRepack) {
+ for (auto & ar: repackInfo.archiveRouteMap) {
+ auto * plar=m_payload.mutable_repack_info()->mutable_archive_routes()->Add();
+ plar->set_copynb(ar.first);
+ plar->set_tapepool(ar.second);
+ }
+ for (auto cntr: repackInfo.copyNbsToRearchive) {
+ m_payload.mutable_repack_info()->mutable_copy_nbs_to_rearchive()->Add(cntr);
+ }
+ m_payload.mutable_repack_info()->set_file_buffer_url(repackInfo.fileBufferURL);
+ m_payload.mutable_repack_info()->set_repack_request_address(repackInfo.repackRequestAddress);
+ }
+}
+
+//------------------------------------------------------------------------------
+// RetrieveRequest::getRepackInfo()
+//------------------------------------------------------------------------------
+RetrieveRequest::RepackInfo RetrieveRequest::getRepackInfo() {
+ checkPayloadReadable();
+ RepackInfoSerDeser ret;
+ if (m_payload.isrepack())
+ ret.deserialize(m_payload.repack_info());
+ return ret;
+}
+
//------------------------------------------------------------------------------
// RetrieveRequest::getRetryStatus()
//------------------------------------------------------------------------------
@@ -744,8 +771,6 @@ auto RetrieveRequest::asyncUpdateJobOwner(uint32_t copyNumber, const std::string
dfi.deserialize(payload.schedulerrequest().diskfileinfo());
retRef.m_retrieveRequest.diskFileInfo = dfi;
retRef.m_retrieveRequest.dstURL = payload.schedulerrequest().dsturl();
- retRef.m_retrieveRequest.isRepack = payload.isrepack();
- retRef.m_retrieveRequest.tapePool = payload.tapepool();
retRef.m_retrieveRequest.errorReportURL = payload.schedulerrequest().retrieveerrorreporturl();
retRef.m_retrieveRequest.requester.name = payload.schedulerrequest().requester().name();
retRef.m_retrieveRequest.requester.group = payload.schedulerrequest().requester().group();
@@ -753,6 +778,18 @@ auto RetrieveRequest::asyncUpdateJobOwner(uint32_t copyNumber, const std::string
af.deserialize(payload.archivefile());
retRef.m_archiveFile = af;
retRef.m_jobStatus = j.status();
+ if (payload.isrepack()) {
+ RetrieveRequest::RepackInfo & ri = retRef.m_repackInfo;
+ for (auto &ar: payload.repack_info().archive_routes()) {
+ ri.archiveRouteMap[ar.copynb()] = ar.tapepool();
+ }
+ for (auto cntr: payload.repack_info().copy_nbs_to_rearchive()) {
+ ri.copyNbsToRearchive.insert(cntr);
+ }
+ ri.fileBufferURL = payload.repack_info().file_buffer_url();
+ ri.isRepack = true;
+ ri.repackRequestAddress = payload.repack_info().repack_request_address();
+ }
// TODO serialization of payload maybe not necessary
oh.set_payload(payload.SerializePartialAsString());
return oh.SerializeAsString();
@@ -779,6 +816,13 @@ const common::dataStructures::ArchiveFile& RetrieveRequest::AsyncJobOwnerUpdater
return m_archiveFile;
}
+//------------------------------------------------------------------------------
+// RetrieveRequest::AsyncJobOwnerUpdater::getRepackInfo()
+//------------------------------------------------------------------------------
+const RetrieveRequest::RepackInfo& RetrieveRequest::AsyncJobOwnerUpdater::getRepackInfo() {
+ return m_repackInfo;
+}
+
//------------------------------------------------------------------------------
// RetrieveRequest::AsyncJobOwnerUpdater::getRetrieveRequest()
//------------------------------------------------------------------------------
@@ -970,24 +1014,29 @@ RetrieveRequest::AsyncRetrieveToArchiveTransformer * RetrieveRequest::asyncTrans
//TODO : Should creation log just be initialized or should it be copied from the retrieveRequest ?
cta::objectstore::serializers::EntryLog *archiveRequestCL = archiveRequestPayload.mutable_creationlog();
archiveRequestCL->CopyFrom(retrieveRequestMP.creationlog());
- //Add the jobs of the old RetrieveRequest to the new ArchiveRequest
- for(auto retrieveJob: retrieveRequestPayload.jobs()){
+ //Create archive jobs for each copyNb ro rearchive
+ RetrieveRequest::RepackInfoSerDeser repackInfoSerDeser;
+ repackInfoSerDeser.deserialize(retrieveRequestPayload.repack_info());
+ // TODO: for the moment we just clone the retrieve request's policy.
+ auto maxRetriesWithinMount = retrieveRequestPayload.jobs(0).maxretrieswithinmount();
+ auto maxTotalRetries = retrieveRequestPayload.jobs(0).maxtotalretries();
+ auto maxReportRetries = retrieveRequestPayload.jobs(0).maxreportretries();
+ for(auto cntr: repackInfoSerDeser.copyNbsToRearchive) {
auto *archiveJob = archiveRequestPayload.add_jobs();
archiveJob->set_status(cta::objectstore::serializers::ArchiveJobStatus::AJS_ToTransferForUser);
- archiveJob->set_copynb(retrieveJob.copynb());
+ archiveJob->set_copynb(cntr);
archiveJob->set_archivequeueaddress("");
archiveJob->set_totalreportretries(0);
archiveJob->set_lastmountwithfailure(0);
archiveJob->set_totalretries(0);
archiveJob->set_retrieswithinmount(0);
- archiveJob->set_maxretrieswithinmount(retrieveJob.maxretrieswithinmount()); //TODO : should we put the same value as the retrieveJob ?
+ archiveJob->set_maxretrieswithinmount(maxRetriesWithinMount); //TODO : should we put the same value as the retrieveJob ?
archiveJob->set_totalreportretries(0);
- archiveJob->set_maxtotalretries(retrieveJob.maxtotalretries()); //TODO : should we put the same value as the retrieveJob ?
- archiveJob->set_maxreportretries(retrieveJob.maxreportretries()); //TODO : should we put the same value as the retrieveJob ?
- archiveJob->set_tapepool(retrieveRequestPayload.tapepool());
+ archiveJob->set_maxtotalretries(maxTotalRetries); //TODO : should we put the same value as the retrieveJob ?
+ archiveJob->set_maxreportretries(maxReportRetries); //TODO : should we put the same value as the retrieveJob ?
+ archiveJob->set_tapepool(repackInfoSerDeser.archiveRouteMap[cntr]);
archiveJob->set_owner(processAgentAddress);
}
-
//Serialize the new ArchiveRequest so that it replaces the RetrieveRequest
oh.set_payload(archiveRequestPayload.SerializeAsString());
//Change the type of the RetrieveRequest to ArchiveRequest
@@ -1044,24 +1093,4 @@ void RetrieveRequest::setJobStatus(uint32_t copyNumber, const serializers::Retri
throw exception::Exception("In RetrieveRequest::setJobStatus(): job not found.");
}
-bool RetrieveRequest::isRepack(){
- checkPayloadReadable();
- return m_payload.isrepack();
-}
-
-void RetrieveRequest::setIsRepack(bool isRepack){
- checkPayloadWritable();
- m_payload.set_isrepack(isRepack);
-}
-
-std::string RetrieveRequest::getTapePool(){
- checkPayloadReadable();
- return m_payload.tapepool();
-}
-
-void RetrieveRequest::setTapePool(const std::string tapePool)
-{
- checkPayloadWritable();
- m_payload.set_tapepool(tapePool);
-}
}} // namespace cta::objectstore
diff --git a/objectstore/RetrieveRequest.hpp b/objectstore/RetrieveRequest.hpp
index 3a55e8706e..330642d080 100644
--- a/objectstore/RetrieveRequest.hpp
+++ b/objectstore/RetrieveRequest.hpp
@@ -143,11 +143,38 @@ public:
//! The copy number to enqueue. It could be different from the updated one in mixed success/failure scenario.
serializers::RetrieveJobStatus nextStatus;
};
- bool isRepack();
- void setIsRepack(bool isRepack);
- std::string getTapePool();
- void setTapePool(const std::string tapePool);
+ struct RepackInfo {
+ bool isRepack = false;
+ std::map<uint32_t, std::string> archiveRouteMap;
+ std::set<uint32_t> copyNbsToRearchive;
+ std::string repackRequestAddress;
+ std::string fileBufferURL;
+ };
+ void setRepackInfo(const RepackInfo & repackInfo);
+ RepackInfo getRepackInfo();
+ struct RepackInfoSerDeser: public RepackInfo {
+ operator RepackInfo() { return RepackInfo(*this); }
+ void serialize(cta::objectstore::serializers::RetrieveRequestRepackInfo & rrri) {
+ if (!isRepack) throw exception::Exception("In RetrieveRequest::RepackInfoSerDeser::serialize(): isRepack is false.");
+ for (auto &route: archiveRouteMap) {
+ auto * ar = rrri.mutable_archive_routes()->Add();
+ ar->set_copynb(route.first);
+ ar->set_tapepool(route.second);
+ }
+ for (auto cntr: copyNbsToRearchive) rrri.mutable_copy_nbs_to_rearchive()->Add(cntr);
+ rrri.set_file_buffer_url(fileBufferURL);
+ rrri.set_repack_request_address(repackRequestAddress);
+ }
+
+ void deserialize(const cta::objectstore::serializers::RetrieveRequestRepackInfo & rrri) {
+ isRepack = true;
+ for(auto &route: rrri.archive_routes()) { archiveRouteMap[route.copynb()] = route.tapepool(); }
+ for(auto &cntr: rrri.copy_nbs_to_rearchive()) { copyNbsToRearchive.insert(cntr); }
+ fileBufferURL = rrri.file_buffer_url();
+ repackRequestAddress = rrri.repack_request_address();
+ }
+ };
private:
/*!
* Determine and set the new status of the job.
@@ -185,11 +212,13 @@ public:
serializers::RetrieveJobStatus getJobStatus() { return m_jobStatus; }
const common::dataStructures::RetrieveRequest &getRetrieveRequest();
const common::dataStructures::ArchiveFile &getArchiveFile();
+ const RepackInfo &getRepackInfo();
private:
std::function<std::string(const std::string &)> m_updaterCallback;
std::unique_ptr<Backend::AsyncUpdater> m_backendUpdater;
common::dataStructures::RetrieveRequest m_retrieveRequest;
common::dataStructures::ArchiveFile m_archiveFile;
+ RepackInfo m_repackInfo;
serializers::RetrieveJobStatus m_jobStatus;
};
// An owner updater factory. The owner MUST be previousOwner for the update to be executed.
diff --git a/objectstore/RootEntry.cpp b/objectstore/RootEntry.cpp
index 43cce9b125..7f84020940 100644
--- a/objectstore/RootEntry.cpp
+++ b/objectstore/RootEntry.cpp
@@ -427,7 +427,8 @@ std::string RootEntry::getRetrieveQueueAddress(const std::string& vid, JobQueueT
auto & rqp = serializers::findElement(retrieveQueuePointers(queueType), vid);
return rqp.address();
} catch (serializers::NotFound &) {
- throw NoSuchRetrieveQueue("In RootEntry::getRetreveQueueAddress: retrieve queue not allocated");
+ throw NoSuchRetrieveQueue(std::string("In RootEntry::getRetreveQueueAddress: retrieve queue not allocated ")+
+ vid+"/"+toString(queueType));
}
}
diff --git a/objectstore/cta.proto b/objectstore/cta.proto
index a3b8897795..a8e17d816d 100644
--- a/objectstore/cta.proto
+++ b/objectstore/cta.proto
@@ -367,6 +367,18 @@ message RetrieveJob {
repeated string reportfailurelogs = 9210;
}
+message RetrieveRequestArchiveRoute {
+ required uint32 copynb = 9400;
+ required string tapepool = 9410;
+}
+
+message RetrieveRequestRepackInfo {
+ repeated RetrieveRequestArchiveRoute archive_routes = 9500;
+ repeated uint32 copy_nbs_to_rearchive = 9510;
+ required string repack_request_address = 9520;
+ required string file_buffer_url = 9530;
+}
+
message RetrieveRequest {
required SchedulerRetrieveRequest schedulerrequest = 9150;
required MountPolicy mountpolicy = 9151;
@@ -375,8 +387,8 @@ message RetrieveRequest {
repeated RetrieveJob jobs = 9154;
required string failurereporturl = 9155;
required string failurereportlog = 9156;
- required bool isrepack = 9157; //In protobuf, default values for bool is false
- optional string tapepool = 9158;
+ required bool isrepack = 9157;
+ optional RetrieveRequestRepackInfo repack_info = 9158;
}
message ValueCountPair {
@@ -492,6 +504,7 @@ message RepackSubRequestPointer {
message RepackRequest {
required string vid = 11000;
+ required string buffer_url = 11005;
required RepackRequestStatus status = 11010;
required bool expandmode = 11400;
required bool repackmode = 11410;
@@ -507,7 +520,7 @@ message RepackRequest {
required uint64 retrievedbytes = 11490;
required uint64 archivedfiles = 11500;
required uint64 archivedbytes = 11510;
- required uint64 failedtoretievefiles = 11520;
+ required uint64 failedtoretrievefiles = 11520;
required uint64 failedtoretrievebytes = 11530;
required uint64 failedtoarchivefiles = 11540;
required uint64 failedtoarchivebytes = 11550;
diff --git a/scheduler/OStoreDB/OStoreDB.cpp b/scheduler/OStoreDB/OStoreDB.cpp
index 33d893edef..db6b9abebc 100644
--- a/scheduler/OStoreDB/OStoreDB.cpp
+++ b/scheduler/OStoreDB/OStoreDB.cpp
@@ -28,6 +28,7 @@
#include "objectstore/RepackRequest.hpp"
#include "objectstore/RepackIndex.hpp"
#include "objectstore/RepackQueue.hpp"
+#include "objectstore/Sorter.hpp"
#include "objectstore/Helpers.hpp"
#include "common/exception/Exception.hpp"
#include "common/utils/utils.hpp"
@@ -1062,7 +1063,6 @@ std::string OStoreDB::queueRetrieve(const cta::common::dataStructures::RetrieveR
rReq->initialize();
rReq->setSchedulerRequest(rqst);
rReq->setRetrieveFileQueueCriteria(criteria);
- rReq->setTapePool(rqst.tapePool);
// Find the job corresponding to the vid (and check we indeed have one).
auto jobs = rReq->getJobs();
objectstore::RetrieveRequest::JobDump job;
@@ -1088,8 +1088,6 @@ std::string OStoreDB::queueRetrieve(const cta::common::dataStructures::RetrieveR
rReq->setOwner(m_agentReference->getAgentAddress());
// "Select" an arbitrary copy number. This is needed to serialize the object.
rReq->setActiveCopyNumber(criteria.archiveFile.tapeFiles.begin()->second.copyNb);
- rReq->setIsRepack(rqst.isRepack);
- rReq->setTapePool(rqst.tapePool);
rReq->insert();
double insertionTime = timer.secs(cta::utils::Timer::reset_t::resetCounter);
m_taskQueueSize++;
@@ -1294,6 +1292,7 @@ void OStoreDB::queueRepack(const std::string& vid, const std::string& bufferURL,
rr->setOwner(m_agentReference->getAgentAddress());
rr->setVid(vid);
rr->setType(repackType);
+ rr->setBufferURL(bufferURL);
// Try to reference the object in the index (will fail if there is already a request with this VID.
try {
Helpers::registerRepackRequestToIndex(vid, rr->getAddressIfSet(), *m_agentReference, m_objectStore, lc);
@@ -1511,14 +1510,270 @@ std::unique_ptr<SchedulerDatabase::RepackRequest> OStoreDB::getNextRepackJobToEx
auto repackInfo = jobs.elements.front().repackInfo;
//build the repackRequest with the repack infos
std::unique_ptr<OStoreDB::RepackRequest> ret;
- ret.reset(new OStoreDB::RepackRequest(repackRequest->getAddressIfSet(),*this));
+ ret.reset(new OStoreDB::RepackRequest(repackRequest->getAddressIfSet(), *this));
ret->repackInfo.vid = repackInfo.vid;
ret->repackInfo.type = repackInfo.type;
ret->repackInfo.status = repackInfo.status;
+ ret->repackInfo.repackBufferBaseURL = repackInfo.repackBufferBaseURL;
return std::move(ret);
}
}
+//------------------------------------------------------------------------------
+// OStoreDB::RepackRequest::getLastExpandedFSeq()
+//------------------------------------------------------------------------------
+uint64_t OStoreDB::RepackRequest::getLastExpandedFSeq() {
+ // We own the repack request, so we are only users of it.
+ m_repackRequest.fetchNoLock();
+ return m_repackRequest.getLastExpandedFSeq();
+}
+
+//------------------------------------------------------------------------------
+// OStoreDB::RepackRequest::addSubrequests()
+//------------------------------------------------------------------------------
+void OStoreDB::RepackRequest::addSubrequests(std::list<Subrequest>& repackSubrequests, cta::common::dataStructures::ArchiveRoute::FullMap& archiveRoutesMap, uint64_t maxFSeqLowBound, log::LogContext& lc) {
+ // We need to prepare retrieve requests names and reference them, create them, enqueue them.
+ objectstore::ScopedExclusiveLock rrl (m_repackRequest);
+ m_repackRequest.fetch();
+ std::set<uint64_t> fSeqs;
+ for (auto rsr: repackSubrequests) fSeqs.insert(rsr.fSeq);
+ auto subrequestsNames = m_repackRequest.getOrPrepareSubrequestInfo(fSeqs, *m_oStoreDB.m_agentReference);
+ // We make sure the references to subrequests exist persistently before creating them.
+ m_repackRequest.commit();
+ // We keep holding the repack request lock: we need to ensure de deleted boolean of each subrequest does
+ // not change while we attempt creating them (or we would face double creation).
+ // Sort the naming results in a fSeq->requestName map for easier access.
+ std::map<uint64_t, objectstore::RepackRequest::SubrequestInfo> subReqInfoMap;
+ for (auto &rn: subrequestsNames) { subReqInfoMap[rn.fSeq] = rn; }
+ // Try to create the retrieve subrequests (owned by this process, to be queued in a second step)
+ // subrequests can already fail at that point if we cannot find a copy on a valid tape.
+ std::list<uint64_t> failedFSeqs;
+ uint64_t failedFiles = 0;
+ uint64_t failedBytes = 0;
+ // First loop: we will issue the async insertions of the subrequests.
+ struct AsyncInsertionInfo {
+ Subrequest & rsr;
+ std::shared_ptr<RetrieveRequest> request;
+ std::shared_ptr<RetrieveRequest::AsyncInserter> inserter;
+ std::string bestVid;
+ uint32_t activeCopyNb;
+ };
+ std::list<AsyncInsertionInfo> asyncInsertionInfoList;
+ for (auto &rsr: repackSubrequests) {
+ // Requests marked as deleted are guaranteed to have already been created => we will not re-attempt.
+ if (!subReqInfoMap.at(rsr.fSeq).subrequestDeleted) {
+ // We need to try and create the subrequest.
+ // Create the sub request (it's a retrieve request now).
+ auto rr=std::make_shared<objectstore::RetrieveRequest>(subReqInfoMap.at(rsr.fSeq).address, m_oStoreDB.m_objectStore);
+ rr->initialize();
+ // Set the file info
+ common::dataStructures::RetrieveRequest schedReq;
+ schedReq.archiveFileID = rsr.archiveFile.archiveFileID;
+ schedReq.dstURL = rsr.fileBufferURL;
+ schedReq.diskFileInfo = rsr.archiveFile.diskFileInfo;
+ // dsrr.errorReportURL: We leave this bank as the reporting will be done to the repack request,
+ // stored in the repack info.
+ rr->setSchedulerRequest(schedReq);
+ // Set the repack info.
+ RetrieveRequest::RepackInfo rRRepackInfo;
+ try {
+ for (auto & ar: archiveRoutesMap.at(std::make_tuple(rsr.archiveFile.diskInstance, rsr.archiveFile.storageClass))) {
+ rRRepackInfo.archiveRouteMap[ar.second.copyNb] = ar.second.tapePoolName;
+ }
+ } catch (std::out_of_range &) {
+ failedFSeqs.emplace_back(rsr.fSeq);
+ failedFiles++;
+ failedBytes += rsr.archiveFile.fileSize;
+ log::ScopedParamContainer params(lc);
+ params.add("fileID", rsr.archiveFile.archiveFileID)
+ .add("diskInstance", rsr.archiveFile.diskInstance)
+ .add("storageClass", rsr.archiveFile.storageClass);
+ std::stringstream storageClassList;
+ bool first=true;
+ for (auto & sc: archiveRoutesMap) {
+ std::string diskInstance, storageClass;
+ std::tie(diskInstance, storageClass) = sc.first;
+ storageClassList << (first?"":" ") << "di=" << diskInstance << " sc=" << storageClass << " rc=" << sc.second.size();
+ }
+ params.add("storageClassList", storageClassList.str());
+ lc.log(log::ERR, "In OStoreDB::RepackRequest::addSubrequests(): not such archive route.");
+ continue;
+ }
+ rRRepackInfo.copyNbsToRearchive = rsr.copyNbsToRearchive;
+ rRRepackInfo.fileBufferURL = rsr.fileBufferURL;
+ rRRepackInfo.isRepack = true;
+ rRRepackInfo.repackRequestAddress = m_repackRequest.getAddressIfSet();
+ rr->setRepackInfo(rRRepackInfo);
+ // Set the queueing parameters
+ common::dataStructures::RetrieveFileQueueCriteria fileQueueCriteria;
+ fileQueueCriteria.archiveFile = rsr.archiveFile;
+ fileQueueCriteria.mountPolicy = common::dataStructures::MountPolicy::s_defaultMountPolicyForRepack;
+ rr->setRetrieveFileQueueCriteria(fileQueueCriteria);
+ // Decide of which vid we are going to retrieve from. Here, if we can retrieve from the repack VID, we
+ // will set the initial recall on it. Retries will we requeue to best VID as usual if needed.
+ std::string bestVid;
+ uint32_t activeCopyNumber = 0;
+ // Make sure we have a copy on the vid we intend to repack.
+ for (auto & tc: rsr.archiveFile.tapeFiles) {
+ if (tc.second.vid == repackInfo.vid) {
+ try {
+ // Try to select the repack VID from a one-vid list.
+ Helpers::selectBestRetrieveQueue({repackInfo.vid}, m_oStoreDB.m_catalogue, m_oStoreDB.m_objectStore);
+ bestVid = repackInfo.vid;
+ activeCopyNumber = tc.second.copyNb;
+ } catch (Helpers::NoTapeAvailableForRetrieve &) {}
+ break;
+ }
+ }
+ // The repack vid was not appropriate, let's try all candidates.
+ if (bestVid.empty()) {
+ std::set<std::string> candidateVids;
+ for (auto & tc: rsr.archiveFile.tapeFiles) candidateVids.insert(tc.second.vid);
+ try {
+ bestVid = Helpers::selectBestRetrieveQueue(candidateVids, m_oStoreDB.m_catalogue, m_oStoreDB.m_objectStore);
+ } catch (Helpers::NoTapeAvailableForRetrieve &) {
+ // Count the failure for this subrequest.
+ failedFSeqs.emplace_back(rsr.fSeq);
+ failedFiles++;
+ failedBytes += rsr.archiveFile.fileSize;
+ log::ScopedParamContainer params(lc);
+ params.add("fileId", rsr.archiveFile.archiveFileID)
+ .add("repackVid", repackInfo.vid);
+ lc.log(log::ERR,
+ "In OStoreDB::RepackRequest::addSubrequests(): could not queue a retrieve subrequest. Subrequest failed.");
+ continue;
+ }
+ }
+ for (auto &tc: rsr.archiveFile.tapeFiles)
+ if (tc.second.vid == bestVid) {
+ activeCopyNumber = tc.second.copyNb;
+ goto copyNbFound;
+ }
+ {
+ // Count the failure for this subrequest.
+ failedFSeqs.emplace_back(rsr.fSeq);
+ failedFiles++;
+ failedBytes += rsr.archiveFile.fileSize;
+ log::ScopedParamContainer params(lc);
+ params.add("fileId", rsr.archiveFile.archiveFileID)
+ .add("repackVid", repackInfo.vid)
+ .add("bestVid", bestVid);
+ lc.log(log::ERR,
+ "In OStoreDB::RepackRequest::addSubrequests(): could not find the copyNb for the chosen VID. Subrequest failed.");
+ continue;
+ }
+ copyNbFound:;
+ // We have the best VID. The request is ready to be created after comleting its information.
+ rr->setOwner(m_oStoreDB.m_agentReference->getAgentAddress());
+ rr->setActiveCopyNumber(activeCopyNumber);
+ // We can now try to insert the request. It could alredy have been created (in which case it must exist).
+ // We hold the lock to the repack request, no better not waste time, so we async create.
+ try {
+ std::shared_ptr<objectstore::RetrieveRequest::AsyncInserter> rrai(rr->asyncInsert());
+ asyncInsertionInfoList.emplace_back(AsyncInsertionInfo{rsr, rr, rrai, bestVid, activeCopyNumber});
+ } catch (exception::Exception & ex) {
+ // We can fail to serialize here...
+ // Count the failure for this subrequest.
+ failedFSeqs.emplace_back(rsr.fSeq);
+ failedFiles++;
+ failedBytes += rsr.archiveFile.fileSize;
+ failedFSeqs.emplace_back(rsr.fSeq);
+ log::ScopedParamContainer params(lc);
+ params.add("fileId", rsr.archiveFile)
+ .add("repackVid", repackInfo.vid)
+ .add("bestVid", bestVid)
+ .add("ExceptionMessage", ex.getMessageValue());
+ lc.log(log::ERR,
+ "In OStoreDB::RepackRequest::addSubrequests(): could not asyncInsert the subrequest.");
+ }
+ }
+ }
+ // We can now check the subrequests creations succeeded, and prepare their queueing.
+ struct AsyncInsertedSubrequestInfo {
+ Subrequest & rsr;
+ std::string bestVid;
+ uint32_t activeCopyNb;
+ std::shared_ptr<RetrieveRequest> request;
+ };
+ std::list <AsyncInsertedSubrequestInfo> asyncInsertedSubrequestInfoList;
+ for (auto & aii: asyncInsertionInfoList) {
+ // Check the insertion succeeded.
+ try {
+ aii.inserter->wait();
+ log::ScopedParamContainer params(lc);
+ params.add("fileID", aii.rsr.archiveFile.archiveFileID);
+ std::stringstream copyNbList;
+ bool first = true;
+ for (auto cn: aii.rsr.copyNbsToRearchive) { copyNbList << (first?"":" ") << cn; first = true; }
+ params.add("copyNbsToRearchive", copyNbList.str())
+ .add("subrequestObject", aii.request->getAddressIfSet())
+ .add("fileBufferURL", aii.rsr.fileBufferURL);
+ lc.log(log::INFO, "In OStoreDB::RepackRequest::addSubrequests(): subrequest created.");
+ asyncInsertedSubrequestInfoList.emplace_back(AsyncInsertedSubrequestInfo{aii.rsr, aii.bestVid, aii.activeCopyNb, aii.request});
+ } catch (exception::Exception & ex) {
+ // Count the failure for this subrequest.
+ failedFSeqs.emplace_back(aii.rsr.fSeq);
+ failedFiles++;
+ failedBytes += aii.rsr.archiveFile.fileSize;
+ log::ScopedParamContainer params(lc);
+ params.add("fileId", aii.rsr.archiveFile)
+ .add("repackVid", repackInfo.vid)
+ .add("bestVid", aii.bestVid)
+ .add("bestCopyNb", aii.activeCopyNb)
+ .add("ExceptionMessage", ex.getMessageValue());
+ lc.log(log::ERR,
+ "In OStoreDB::RepackRequest::addSubrequests(): could not asyncInsert the subrequest.");
+ }
+ }
+ // We now have created the subrequests. Time to enqueue.
+ {
+ objectstore::Sorter sorter(*m_oStoreDB.m_agentReference, m_oStoreDB.m_objectStore, m_oStoreDB.m_catalogue);
+ std::list<std::unique_ptr<objectstore::ScopedExclusiveLock>> locks;
+ for (auto &is: asyncInsertedSubrequestInfoList) {
+ locks.push_back(cta::make_unique<objectstore::ScopedExclusiveLock>(*is.request));
+ is.request->fetch();
+ sorter.insertRetrieveRequest(is.request, *m_oStoreDB.m_agentReference, is.activeCopyNb, lc);
+ }
+ locks.clear();
+ sorter.flushAll(lc);
+ }
+
+}
+
+//------------------------------------------------------------------------------
+// OStoreDB::RepackRequest::expandDone()
+//------------------------------------------------------------------------------
+void OStoreDB::RepackRequest::expandDone() {
+ // We are now done with the repack request. We can set its status.
+ ScopedSharedLock rrl(m_repackRequest);
+ m_repackRequest.fetch();
+ // After expansion, 2 statuses are possible: starting (nothing reported as done) or running (anything reported as done).
+ // We can find that out from the statistics...
+ typedef objectstore::RepackRequest::StatsType StatsType;
+ bool running=false;
+ auto stats=m_repackRequest.getStats();
+ for (auto t: {StatsType::ArchiveFailure, StatsType::ArchiveSuccess, StatsType::RetrieveSuccess, StatsType::RetrieveFailure}) {
+ if (stats.at(t).files) {
+ running=true;
+ break;
+ }
+ }
+ typedef common::dataStructures::RepackInfo::Status Status;
+ m_repackRequest.setStatus(running? Status::Running: Status::Starting);
+ m_repackRequest.commit();
+}
+
+//------------------------------------------------------------------------------
+// OStoreDB::RepackRequest::fail()
+//------------------------------------------------------------------------------
+void OStoreDB::RepackRequest::fail() {
+ ScopedExclusiveLock rrl(m_repackRequest);
+ m_repackRequest.fetch();
+ m_repackRequest.setStatus(common::dataStructures::RepackInfo::Status::Failed);
+ m_repackRequest.commit();
+}
+
+
//------------------------------------------------------------------------------
// OStoreDB::cancelRepack()
//------------------------------------------------------------------------------
@@ -1601,6 +1856,7 @@ getNextRetrieveJobsToReportBatch(uint64_t filesRequested, log::LogContext &logCo
rj->selectedCopyNb = j.copyNb;
rj->errorReportURL = j.errorReportURL;
rj->reportType = j.reportType;
+ rj->m_repackInfo = j.repackInfo;
rj->setJobOwned();
ret.emplace_back(std::move(rj));
}
@@ -2490,6 +2746,8 @@ getNextJobBatch(uint64_t filesRequested, uint64_t bytesRequested, log::LogContex
rj->archiveFile = j.archiveFile;
rj->retrieveRequest = j.rr;
rj->selectedCopyNb = j.copyNb;
+ rj->isRepack = j.repackInfo.isRepack;
+ rj->m_repackInfo = j.repackInfo;
rj->m_jobOwned = true;
rj->m_mountId = mountInfo.mountId;
ret.emplace_back(std::move(rj));
@@ -2580,46 +2838,101 @@ OStoreDB::RetrieveJob * OStoreDB::castFromSchedDBJob(SchedulerDatabase::Retrieve
}
//------------------------------------------------------------------------------
-// OStoreDB::RetrieveMount::waitAndFinishSettingJobsBatchSuccessful()
+// OStoreDB::RetrieveMount::flushAsyncSuccessReports()
//------------------------------------------------------------------------------
-std::set<cta::SchedulerDatabase::RetrieveJob*> OStoreDB::RetrieveMount::finishSettingJobsBatchSuccessful(
- std::list<cta::SchedulerDatabase::RetrieveJob*>& jobsBatch, log::LogContext& lc) {
- std::set<cta::SchedulerDatabase::RetrieveJob*> ret;
+void OStoreDB::RetrieveMount::flushAsyncSuccessReports(std::list<cta::SchedulerDatabase::RetrieveJob*>& jobsBatch,
+ log::LogContext& lc) {
std::list<std::string> rjToUnown;
- // We will wait on the asynchronously started reports of jobs and remove them from
- // ownership.
+ std::map<std::string, std::list<OStoreDB::RetrieveJob*>> jobsToRequeueForRepackMap;
+ // We will wait on the asynchronously started reports of jobs, queue the retrieve jobs
+ // for report and remove them from ownership.
+ // 1) Check the async update result.
for (auto & sDBJob: jobsBatch) {
auto osdbJob = castFromSchedDBJob(sDBJob);
- rjToUnown.push_back(osdbJob->m_retrieveRequest.getAddressIfSet());
- ret.insert(sDBJob);
+ if (osdbJob->isRepack) {
+ try {
+ osdbJob->m_jobSucceedForRepackReporter->wait();
+ jobsToRequeueForRepackMap[osdbJob->m_repackInfo.repackRequestAddress].emplace_back(osdbJob);
+ } catch (cta::exception::Exception & ex) {
+ log::ScopedParamContainer params(lc);
+ params.add("fileId", osdbJob->archiveFile.archiveFileID)
+ .add("requestObject", osdbJob->m_retrieveRequest.getAddressIfSet())
+ .add("exceptionMessage", ex.getMessageValue());
+ lc.log(log::ERR,
+ "In OStoreDB::RetrieveMount::flushAsyncSuccessReports(): async status update failed. "
+ "Will leave job to garbage collection.");
+ }
+ } else {
+ try {
+ osdbJob->m_jobDelete->wait();
+ rjToUnown.push_back(osdbJob->m_retrieveRequest.getAddressIfSet());
+ } catch (cta::exception::Exception & ex) {
+ log::ScopedParamContainer params(lc);
+ params.add("fileId", osdbJob->archiveFile.archiveFileID)
+ .add("requestObject", osdbJob->m_retrieveRequest.getAddressIfSet())
+ .add("exceptionMessage", ex.getMessageValue());
+ lc.log(log::ERR,
+ "In OStoreDB::RetrieveMount::flushAsyncSuccessReports(): async deletion failed. "
+ "Will leave job to garbage collection.");
+ }
+ }
+ }
+ // 2) Queue the repack requests for repack.
+ for (auto & repackRequestQueue: jobsToRequeueForRepackMap) {
+ typedef objectstore::ContainerAlgorithms<RetrieveQueue,RetrieveQueueToReportToRepackForSuccess> RQTRTRFSAlgo;
+ RQTRTRFSAlgo::InsertedElement::list insertedRequests;
+ // Keep a map of objectstore request -> sDBJob to handle errors.
+ std::map<objectstore::RetrieveRequest *, OStoreDB::RetrieveJob *> requestToJobMap;
+ for (auto & req: repackRequestQueue.second) {
+ insertedRequests.push_back(RQTRTRFSAlgo::InsertedElement{&req->m_retrieveRequest, req->selectedCopyNb,
+ req->archiveFile.tapeFiles[req->selectedCopyNb].fSeq, req->archiveFile.fileSize,
+ cta::common::dataStructures::MountPolicy::s_defaultMountPolicyForRepack,
+ serializers::RetrieveJobStatus::RJS_ToReportToRepackForSuccess});
+ requestToJobMap[&req->m_retrieveRequest] = req;
+ }
+ RQTRTRFSAlgo rQTRTRFSAlgo(m_oStoreDB.m_objectStore, *m_oStoreDB.m_agentReference);
+ try {
+ rQTRTRFSAlgo.referenceAndSwitchOwnership(repackRequestQueue.first, insertedRequests, lc);
+ // In case all goes well, we can remove ownership of all requests.
+ for (auto & req: repackRequestQueue.second) { rjToUnown.push_back(req->m_retrieveRequest.getAddressIfSet()); }
+ } catch (RQTRTRFSAlgo::OwnershipSwitchFailure & failure) {
+ // Some requests did not make it to the queue. Log and leave them for GC to sort out (leave them in ownership).
+ std::set<std::string> failedElements;
+ for (auto & fe: failure.failedElements) {
+ // Log error.
+ log::ScopedParamContainer params(lc);
+ params.add("fileId", requestToJobMap.at(fe.element->retrieveRequest)->archiveFile.archiveFileID)
+ .add("copyNb", fe.element->copyNb)
+ .add("requestObject", fe.element->retrieveRequest->getAddressIfSet());
+ try {
+ std::rethrow_exception(fe.failure);
+ } catch (cta::exception::Exception & ex) {
+ params.add("exeptionMessage", ex.getMessageValue());
+ } catch (std::exception & ex) {
+ params.add("exceptionWhat", ex.what())
+ .add("exceptionTypeName", typeid(ex).name());
+ }
+ lc.log(log::ERR, "In OStoreDB::RetrieveMount::flushAsyncSuccessReports(): failed to queue request to report for repack."
+ "Leaving request to be garbage collected.");
+ // Add the failed request to the set.
+ failedElements.insert(fe.element->retrieveRequest->getAddressIfSet());
+ }
+ for (auto & req: repackRequestQueue.second) {
+ if (!failedElements.count(req->m_retrieveRequest.getAddressIfSet())) {
+ rjToUnown.push_back(req->m_retrieveRequest.getAddressIfSet());
+ }
+ }
+ } catch (exception::Exception & ex) {
+ // Something else happened. We just log the error and let the garbage collector go through all the requests.
+ log::ScopedParamContainer params(lc);
+ params.add("exceptionMessage", ex.getMessageValue());
+ lc.log(log::ERR, "In OStoreDB::RetrieveMount::flushAsyncSuccessReports(): failed to queue a batch of requests.");
+ }
}
+ // 3) Remove requests from ownership.
m_oStoreDB.m_agentReference->removeBatchFromOwnership(rjToUnown, m_oStoreDB.m_objectStore);
- return ret;
}
-//------------------------------------------------------------------------------
-// OStoreDB::RetrieveMount::batchSucceedRetrieveForRepack()
-//------------------------------------------------------------------------------
-std::set<cta::SchedulerDatabase::RetrieveJob *> OStoreDB::RetrieveMount::batchSucceedRetrieveForRepack(
- std::list<cta::SchedulerDatabase::RetrieveJob *> & jobsBatch, cta::log::LogContext & lc)
-{
- std::set<cta::SchedulerDatabase::RetrieveJob *> ret;
- typedef objectstore::ContainerAlgorithms<objectstore::RetrieveQueue,objectstore::RetrieveQueueToReportToRepackForSuccess> AqtrtrfsCa;
- AqtrtrfsCa aqtrtrfsCa(m_oStoreDB.m_objectStore, *m_oStoreDB.m_agentReference);
- AqtrtrfsCa::InsertedElement::list insertedElementsLists;
- std::string vid;
- for(auto & retrieveJob : jobsBatch){
- auto osdbJob = castFromSchedDBJob(retrieveJob);
- ret.insert(retrieveJob);
- osdbJob->asyncReportSucceedForRepack();
- osdbJob->checkReportSucceedForRepack();
- auto & tapeFile = osdbJob->archiveFile.tapeFiles[osdbJob->selectedCopyNb];
- vid = osdbJob->m_retrieveMount->mountInfo.vid;
- insertedElementsLists.push_back(AqtrtrfsCa::InsertedElement{&osdbJob->m_retrieveRequest, (uint16_t)osdbJob->selectedCopyNb, tapeFile.fSeq,osdbJob->archiveFile.fileSize,cta::common::dataStructures::MountPolicy::s_defaultMountPolicyForRepack,serializers::RetrieveJobStatus::RJS_ToReportToRepackForSuccess});
- }
- aqtrtrfsCa.referenceAndSwitchOwnership(vid,insertedElementsLists,lc);
- return ret;
-}
//------------------------------------------------------------------------------
// OStoreDB::ArchiveMount::setDriveStatus()
//------------------------------------------------------------------------------
@@ -3276,39 +3589,22 @@ OStoreDB::RetrieveJob::~RetrieveJob() {
}
//------------------------------------------------------------------------------
-// OStoreDB::RetrieveJob::asyncSucceed()
-//------------------------------------------------------------------------------
-void OStoreDB::RetrieveJob::asyncSucceed() {
- // set the request as successful (delete it).
- m_jobDelete.reset(m_retrieveRequest.asyncDeleteJob());
-}
-
-//------------------------------------------------------------------------------
-// OStoreDB::RetrieveJob::checkSucceed()
+// OStoreDB::RetrieveJob::asyncSetSuccessful()
//------------------------------------------------------------------------------
-void OStoreDB::RetrieveJob::checkSucceed() {
- m_jobDelete->wait();
- m_retrieveRequest.resetValues();
- // We no more own the job (which could be gone)
- m_jobOwned = false;
- // Ownership will be removed from agent by caller through retrieve mount object.
-}
-
-//------------------------------------------------------------------------------
-// OStoreDB::RetrieveJob::asyncReportSucceedForRepack()
-//------------------------------------------------------------------------------
-void OStoreDB::RetrieveJob::asyncReportSucceedForRepack()
-{
- m_jobSucceedForRepackReporter.reset(m_retrieveRequest.asyncReportSucceedForRepack(this->selectedCopyNb));
+void OStoreDB::RetrieveJob::asyncSetSuccessful() {
+ if (isRepack) {
+ // If the job is from a repack subrequest, we change its status (to report
+ // for repack success). Queueing will be done in batch in
+ m_jobSucceedForRepackReporter.reset(m_retrieveRequest.asyncReportSucceedForRepack(this->selectedCopyNb));
+ } else {
+ // set the user transfer request as successful (delete it).
+ m_jobDelete.reset(m_retrieveRequest.asyncDeleteJob());
+ }
}
//------------------------------------------------------------------------------
-// OStoreDB::RetrieveJob::checkReportSucceedForRepack()
+// OStoreDB::getNextSucceededRetrieveRequestForRepackBatch()
//------------------------------------------------------------------------------
-void OStoreDB::RetrieveJob::checkReportSucceedForRepack(){
- m_jobSucceedForRepackReporter->wait();
-}
-
std::list<std::unique_ptr<cta::objectstore::RetrieveRequest>> OStoreDB::getNextSucceededRetrieveRequestForRepackBatch(uint64_t filesRequested, log::LogContext& lc)
{
std::list<std::unique_ptr<cta::objectstore::RetrieveRequest>> ret;
diff --git a/scheduler/OStoreDB/OStoreDB.hpp b/scheduler/OStoreDB/OStoreDB.hpp
index dbe000f714..d9b364fe0d 100644
--- a/scheduler/OStoreDB/OStoreDB.hpp
+++ b/scheduler/OStoreDB/OStoreDB.hpp
@@ -216,8 +216,7 @@ public:
void setDriveStatus(cta::common::dataStructures::DriveStatus status, time_t completionTime) override;
void setTapeSessionStats(const castor::tape::tapeserver::daemon::TapeSessionStats &stats) override;
public:
- std::set<cta::SchedulerDatabase::RetrieveJob*> finishSettingJobsBatchSuccessful(std::list<cta::SchedulerDatabase::RetrieveJob*>& jobsBatch, log::LogContext& lc) override;
- std::set<cta::SchedulerDatabase::RetrieveJob*> batchSucceedRetrieveForRepack(std::list<cta::SchedulerDatabase::RetrieveJob*>& jobsBatch, cta::log::LogContext& lc) override;
+ void flushAsyncSuccessReports(std::list<cta::SchedulerDatabase::RetrieveJob*>& jobsBatch, log::LogContext& lc) override;
};
friend class RetrieveMount;
@@ -228,15 +227,10 @@ public:
public:
CTA_GENERATE_EXCEPTION_CLASS(JobNotOwned);
CTA_GENERATE_EXCEPTION_CLASS(NoSuchJob);
- virtual void asyncSucceed() override;
- virtual void checkSucceed() override;
- /**
- * Allows to asynchronously report this RetrieveJob as success
- * for repack. It will call the retrieveRequest.asyncReportSucceedForRepack(this->selectedCopyNb) method
- * that will set the status of the Job as RJS_Succeeded
+ /** Async set job successful. Either delete (user transfer) or change status (repack)
+ * Wait will happen in RetrieveMount::flushAsyncSuccessReports().
*/
- virtual void asyncReportSucceedForRepack() override;
- virtual void checkReportSucceedForRepack() override;
+ virtual void asyncSetSuccessful() override;
void failTransfer(const std::string& failureReason, log::LogContext& lc) override;
void failReport(const std::string& failureReason, log::LogContext& lc) override;
virtual ~RetrieveJob() override;
@@ -256,6 +250,7 @@ public:
OStoreDB::RetrieveMount *m_retrieveMount;
std::unique_ptr<objectstore::RetrieveRequest::AsyncJobDeleter> m_jobDelete;
std::unique_ptr<objectstore::RetrieveRequest::AsyncJobSucceedForRepackReporter> m_jobSucceedForRepackReporter;
+ objectstore::RetrieveRequest::RepackInfo m_repackInfo;
};
static RetrieveJob * castFromSchedDBJob(SchedulerDatabase::RetrieveJob * job);
@@ -347,18 +342,19 @@ public:
class RepackRequest: public SchedulerDatabase::RepackRequest
{
friend class OStoreDB;
- public:
- RepackRequest(const std::string &jobAddress, OStoreDB &oStoreDB) :
- m_jobOwned(false), m_oStoreDB(oStoreDB),
- m_repackRequest(jobAddress, m_oStoreDB.m_objectStore){}
- void setJobOwned(bool b = true) { m_jobOwned = b; }
-
+ public:
+ RepackRequest(const std::string &jobAddress, OStoreDB &oStoreDB) :
+ m_oStoreDB(oStoreDB), m_repackRequest(jobAddress, m_oStoreDB.m_objectStore){}
+ void addSubrequests(std::list<Subrequest>& repackSubrequests, cta::common::dataStructures::ArchiveRoute::FullMap& archiveRoutesMap,
+ uint64_t maxFSeqLowBound, log::LogContext& lc) override;
+ void expandDone() override;
+ void fail() override;
+ uint64_t getLastExpandedFSeq() override;
private:
- bool m_jobOwned;
- uint64_t m_mountId;
OStoreDB & m_oStoreDB;
objectstore::RepackRequest m_repackRequest;
};
+ friend class RepackRequest;
/**
* A class holding a lock on the pending repack request queue. This is the first
@@ -378,7 +374,7 @@ public:
objectstore::ScopedExclusiveLock m_lockOnPendingRepackRequestsQueue;
};
- class RepackRequestPromotionStatisticsNoLock: public SchedulerDatabase::RepackRequestStatistics {
+ class RepackRequestPromotionStatisticsNoLock: public SchedulerDatabase::RepackRequestStatistics {\
friend class OStoreDB;
public:
PromotionToToExpandResult promotePendingRequestsForExpansion(size_t requestCount, log::LogContext &lc) override {
diff --git a/scheduler/RepackRequest.cpp b/scheduler/RepackRequest.cpp
index 4fd5d440fc..49c73ec07e 100644
--- a/scheduler/RepackRequest.cpp
+++ b/scheduler/RepackRequest.cpp
@@ -1,8 +1,37 @@
+/*
+ * The CERN Tape Archive(CTA) project
+ * Copyright(C) 2015 CERN
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ *(at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY {
+
+} without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program. If not, see <http://www.gnu.org/licenses/>.
+ */
+
+
#include "RepackRequest.hpp"
-cta::RepackRequest::RepackRequest(){}
+namespace cta {
const cta::common::dataStructures::RepackInfo cta::RepackRequest::getRepackInfo() const
{
return m_dbReq->repackInfo;
}
+
+void RepackRequest::fail() {
+ m_dbReq->fail();
+}
+
+
+} // namespace cta
+
diff --git a/scheduler/RepackRequest.hpp b/scheduler/RepackRequest.hpp
index 836069cfc6..8a87735deb 100644
--- a/scheduler/RepackRequest.hpp
+++ b/scheduler/RepackRequest.hpp
@@ -30,10 +30,9 @@ namespace cta {
class RepackRequest {
friend class Scheduler;
public:
- RepackRequest();
void expand();
const cta::common::dataStructures::RepackInfo getRepackInfo() const;
-
+ void fail();
protected:
std::unique_ptr<cta::SchedulerDatabase::RepackRequest> m_dbReq;
}; // class RepackRequest
diff --git a/scheduler/RetrieveJob.cpp b/scheduler/RetrieveJob.cpp
index a50443aa3c..ccf0d7e4f5 100644
--- a/scheduler/RetrieveJob.cpp
+++ b/scheduler/RetrieveJob.cpp
@@ -47,15 +47,8 @@ cta::RetrieveJob::RetrieveJob(RetrieveMount *mount,
//------------------------------------------------------------------------------
// asyncComplete
//------------------------------------------------------------------------------
-void cta::RetrieveJob::asyncComplete() {
- m_dbJob->asyncSucceed();
-}
-
-//------------------------------------------------------------------------------
-// checkComplete
-//------------------------------------------------------------------------------
-void cta::RetrieveJob::checkComplete() {
- m_dbJob->checkSucceed();
+void cta::RetrieveJob::asyncSetSuccessful() {
+ m_dbJob->asyncSetSuccessful();
}
//------------------------------------------------------------------------------
diff --git a/scheduler/RetrieveJob.hpp b/scheduler/RetrieveJob.hpp
index f0cf494d17..8db9586fb0 100644
--- a/scheduler/RetrieveJob.hpp
+++ b/scheduler/RetrieveJob.hpp
@@ -79,14 +79,9 @@ public:
* The checksum and the size of the transfer should already stored in the
* object beforehand. Result setting and calling complete are done in 2
* different threads (disk write and reporter thread, respectively).
+ * Completion will be checked implicitly in RetrieveMount::flushAsyncSuccessReports()
*/
- virtual void asyncComplete();
-
- /**
- * Check that asynchronous complete is finished and cleanup the job structures
- *
- */
- virtual void checkComplete();
+ virtual void asyncSetSuccessful();
/**
* Triggers a scheduler update following the failure of the job. Retry policy will
diff --git a/scheduler/RetrieveMount.cpp b/scheduler/RetrieveMount.cpp
index 24478acf21..754d8cfb19 100644
--- a/scheduler/RetrieveMount.cpp
+++ b/scheduler/RetrieveMount.cpp
@@ -144,10 +144,9 @@ std::list<std::unique_ptr<cta::RetrieveJob> > cta::RetrieveMount::getNextJobBatc
//------------------------------------------------------------------------------
// waitAndFinishSettingJobsBatchRetrieved()
//------------------------------------------------------------------------------
-void cta::RetrieveMount::waitAndFinishSettingJobsBatchRetrieved(std::queue<std::unique_ptr<cta::RetrieveJob> >& successfulRetrieveJobs, cta::log::LogContext& logContext) {
+void cta::RetrieveMount::flushAsyncSuccessReports(std::queue<std::unique_ptr<cta::RetrieveJob> >& successfulRetrieveJobs, cta::log::LogContext& logContext) {
std::list<std::unique_ptr<cta::RetrieveJob> > validatedSuccessfulRetrieveJobs; //List to ensure the destruction of the retrieve jobs at the end of this method
std::list<cta::SchedulerDatabase::RetrieveJob *> validatedSuccessfulDBRetrieveJobs;
- std::list<cta::SchedulerDatabase::RetrieveJob *> retrieveRepackJobs;
std::unique_ptr<cta::RetrieveJob> job;
double waitUpdateCompletionTime=0;
double jobBatchFinishingTime=0;
@@ -163,21 +162,14 @@ void cta::RetrieveMount::waitAndFinishSettingJobsBatchRetrieved(std::queue<std::
if (!job.get()) continue;
files++;
bytes+=job->archiveFile.fileSize;
- bool isRepack = job->m_dbJob->retrieveRequest.isRepack;
- if(!isRepack){
- job->checkComplete();
- validatedSuccessfulDBRetrieveJobs.emplace_back(job->m_dbJob.get());
- } else {
- retrieveRepackJobs.emplace_back(job->m_dbJob.get());
- }
+ validatedSuccessfulDBRetrieveJobs.emplace_back(job->m_dbJob.get());
validatedSuccessfulRetrieveJobs.emplace_back(std::move(job));
job.reset();
}
waitUpdateCompletionTime=t.secs(utils::Timer::resetCounter);
tl.insertAndReset("waitUpdateCompletionTime",t);
// Complete the cleaning up of the jobs in the mount
- m_dbMount->finishSettingJobsBatchSuccessful(validatedSuccessfulDBRetrieveJobs, logContext);
- m_dbMount->batchSucceedRetrieveForRepack(retrieveRepackJobs,logContext);
+ m_dbMount->flushAsyncSuccessReports(validatedSuccessfulDBRetrieveJobs, logContext);
jobBatchFinishingTime=t.secs();
tl.insertOrIncrement("jobBatchFinishingTime",jobBatchFinishingTime);
schedulerDbTime=jobBatchFinishingTime + waitUpdateCompletionTime;
diff --git a/scheduler/RetrieveMount.hpp b/scheduler/RetrieveMount.hpp
index f2c5e54bbb..753056d894 100644
--- a/scheduler/RetrieveMount.hpp
+++ b/scheduler/RetrieveMount.hpp
@@ -173,7 +173,7 @@ namespace cta {
* @param successfulRetrieveJobs the jobs to report
* @param logContext
*/
- virtual void waitAndFinishSettingJobsBatchRetrieved(std::queue<std::unique_ptr<cta::RetrieveJob> > & successfulRetrieveJobs, cta::log::LogContext &logContext);
+ virtual void flushAsyncSuccessReports(std::queue<std::unique_ptr<cta::RetrieveJob> > & successfulRetrieveJobs, cta::log::LogContext &logContext);
/**
diff --git a/scheduler/Scheduler.cpp b/scheduler/Scheduler.cpp
index 7057a6b309..7d2e6e45e4 100644
--- a/scheduler/Scheduler.cpp
+++ b/scheduler/Scheduler.cpp
@@ -33,6 +33,7 @@
#include <iostream>
#include <sstream>
+#include <iomanip>
#include <sys/types.h>
#include <sys/stat.h>
#include <unistd.h>
@@ -208,13 +209,7 @@ void Scheduler::queueRetrieve(
utils::Timer t;
// Get the queue criteria
common::dataStructures::RetrieveFileQueueCriteria queueCriteria;
- if(!request.isRepack){
- queueCriteria = m_catalogue.prepareToRetrieveFile(instanceName, request.archiveFileID, request.requester, lc);
- } else {
- //Repack does not need policy
- queueCriteria.archiveFile = m_catalogue.getArchiveFileById(request.archiveFileID);
- queueCriteria.mountPolicy = common::dataStructures::MountPolicy::s_defaultMountPolicyForRepack;
- }
+ queueCriteria = m_catalogue.prepareToRetrieveFile(instanceName, request.archiveFileID, request.requester, lc);
auto catalogueTime = t.secs(cta::utils::Timer::resetCounter);
std::string selectedVid = m_db.queueRetrieve(request, queueCriteria, lc);
auto schedulerDbTime = t.secs();
@@ -251,13 +246,11 @@ void Scheduler::queueRetrieve(
}
spc.add("selectedVid", selectedVid)
.add("catalogueTime", catalogueTime)
- .add("schedulerDbTime", schedulerDbTime);
- if(!request.isRepack){
- spc.add("policyName", queueCriteria.mountPolicy.name)
+ .add("schedulerDbTime", schedulerDbTime)
+ .add("policyName", queueCriteria.mountPolicy.name)
.add("policyMaxDrives", queueCriteria.mountPolicy.maxDrivesAllowed)
.add("policyMinAge", queueCriteria.mountPolicy.retrieveMinRequestAge)
.add("policyPriority", queueCriteria.mountPolicy.retrievePriority);
- }
lc.log(log::INFO, "Queued retrieve request");
}
@@ -339,7 +332,8 @@ void Scheduler::queueRepack(const common::dataStructures::SecurityIdentity &cliI
tl.insertAndReset("schedulerDbTime", t);
log::ScopedParamContainer params(lc);
params.add("tapeVid", vid)
- .add("repackType", toString(repackType));
+ .add("repackType", toString(repackType))
+ .add("bufferURL", bufferURL);
tl.addToLog(params);
lc.log(log::INFO, "In Scheduler::queueRepack(): success.");
}
@@ -431,35 +425,80 @@ const std::string Scheduler::generateRetrieveDstURL(const cta::common::dataStruc
// expandRepackRequest
//------------------------------------------------------------------------------
void Scheduler::expandRepackRequest(std::unique_ptr<RepackRequest>& repackRequest, log::TimingList&, utils::Timer&, log::LogContext& lc) {
- uint64_t fseq = c_defaultFseqForRepack;
std::list<common::dataStructures::ArchiveFile> files;
- auto vid = repackRequest->getRepackInfo().vid;
+ auto repackInfo = repackRequest->getRepackInfo();
+ typedef cta::common::dataStructures::RepackInfo::Type RepackType;
+ if (repackInfo.type != RepackType::RepackOnly) {
+ log::ScopedParamContainer params(lc);
+ params.add("tapeVid", repackInfo.vid);
+ lc.log(log::ERR, "In Scheduler::expandRepackRequest(): failing repack request with unsupported (yet) type.");
+ repackRequest->fail();
+ return;
+ }
//We need to get the ArchiveRoutes to allow the retrieval of the tapePool in which the
//tape where the file is is located
std::list<common::dataStructures::ArchiveRoute> routes = m_catalogue.getArchiveRoutes();
//To identify the routes, we need to have both the dist instance name and the storage class name
//thus, the key of the map is a pair of string
- std::map<std::pair<std::string, std::string>,common::dataStructures::ArchiveRoute> mapRoutes;
+ cta::common::dataStructures::ArchiveRoute::FullMap archiveRoutesMap;
for(auto route: routes){
//insert the route into the map to allow a quick retrieval
- mapRoutes[std::make_pair(route.storageClassName,route.diskInstanceName)] = route;
+ archiveRoutesMap[std::make_pair(route.diskInstanceName,route.storageClassName)][route.copyNb] = route;
}
- while(true) {
- files = m_catalogue.getFilesForRepack(vid,fseq,c_defaultMaxNbFilesForRepack);
- for(auto &archiveFile : files)
+ uint64_t fSeq = repackRequest->m_dbReq->getLastExpandedFSeq() + 1;
+ cta::catalogue::ArchiveFileItor archiveFilesForCatalogue = m_catalogue.getArchiveFilesForRepackItor(repackInfo.vid, fSeq);
+ while(archiveFilesForCatalogue.hasMore()) {
+ size_t filesCount = 0;
+ uint64_t maxAddedFSeq = 0;
+ std::list<SchedulerDatabase::RepackRequest::Subrequest> retrieveSubrequests;
+ while(filesCount < c_defaultMaxNbFilesForRepack && archiveFilesForCatalogue.hasMore())
{
- cta::common::dataStructures::RetrieveRequest retrieveRequest;
- retrieveRequest.archiveFileID = archiveFile.archiveFileID;
- retrieveRequest.diskFileInfo = archiveFile.diskFileInfo;
- retrieveRequest.dstURL = generateRetrieveDstURL(archiveFile.diskFileInfo);
- retrieveRequest.isRepack = true;
- retrieveRequest.tapePool = mapRoutes[std::make_pair(archiveFile.storageClass,archiveFile.diskInstance)].tapePoolName;
- queueRetrieve(archiveFile.diskInstance,retrieveRequest,lc);
+ auto archiveFile = archiveFilesForCatalogue.next();
+ filesCount++;
+ fSeq++;
+ retrieveSubrequests.push_back(cta::SchedulerDatabase::RepackRequest::Subrequest());
+ auto & rsr = retrieveSubrequests.back();
+ rsr.archiveFile = archiveFile;
+ rsr.fSeq = std::numeric_limits<decltype(rsr.fSeq)>::max();
+ // We have to determine which copynbs we want to rearchive, and under which fSeq we record this file.
+ if (repackInfo.type == RepackType::ExpandAndRepack || repackInfo.type == RepackType::RepackOnly) {
+ // determine which fSeq(s) (normally only one) lives on this tape.
+ for (auto & tc: archiveFile.tapeFiles) if (tc.second.vid == repackInfo.vid) {
+ rsr.copyNbsToRearchive.insert(tc.second.copyNb);
+ // We make the (reasonable) assumption that the archive file only has one copy on this tape.
+ // If not, we will ensure the subrequest is filed under the lowest fSeq existing on this tape.
+ // This will prevent double subrequest creation (we already have such a mechanism in case of crash and
+ // restart of expansion.
+ rsr.fSeq = std::min(tc.second.fSeq, rsr.fSeq);
+ maxAddedFSeq = std::max(maxAddedFSeq, rsr.fSeq);
+ }
+ }
+ if (repackInfo.type == RepackType::ExpandAndRepack || repackInfo.type == RepackType::ExpandOnly) {
+ // We should not get here are the type is filtered at the beginning of the function.
+ // TODO: add support for expand.
+ throw cta::exception::Exception("In Scheduler::expandRepackRequest(): expand not yet supported.");
+ }
+ if ((rsr.fSeq == std::numeric_limits<decltype(rsr.fSeq)>::max()) || rsr.copyNbsToRearchive.empty()) {
+ log::ScopedParamContainer params(lc);
+ params.add("fileId", rsr.archiveFile.archiveFileID)
+ .add("repackVid", repackInfo.vid);
+ lc.log(log::ERR, "In Scheduler::expandRepackRequest(): no fSeq found for this file on this tape.");
+ retrieveSubrequests.pop_back();
+ } else {
+ // We found some copies to rearchive. We still have to decide which file path we are going to use.
+ // File path will be base URL + /<VID>/<fSeq>
+ std::stringstream fileBufferURL;
+ fileBufferURL << repackInfo.repackBufferBaseURL << "/" << repackInfo.vid << "/"
+ << std::setw(9) << std::setfill('0') << rsr.fSeq;
+ rsr.fileBufferURL = fileBufferURL.str();
+ }
}
- if (files.size()) {
- auto & tf=files.back().tapeFiles;
- fseq = std::find_if(tf.cbegin(), tf.cend(), [vid](decltype(*(tf.cbegin())) &f){ return f.second.vid == vid; })->second.fSeq + 1;
- } else break;
+ // Note: the highest fSeq will be recorded internally in the following call.
+ // We know that the fSeq processed on the tape are >= initial fSeq + filesCount - 1 (or fSeq - 1 as we counted).
+ // We pass this information to the db for recording in the repack request. This will allow restarting from the right
+ // value in case of crash.
+ repackRequest->m_dbReq->addSubrequests(retrieveSubrequests, archiveRoutesMap, fSeq - 1, lc);
+ fSeq = std::max(fSeq, maxAddedFSeq + 1);
}
}
diff --git a/scheduler/Scheduler.hpp b/scheduler/Scheduler.hpp
index 606e72b560..9bbbc751ee 100644
--- a/scheduler/Scheduler.hpp
+++ b/scheduler/Scheduler.hpp
@@ -290,7 +290,6 @@ public:
/*============== Actual mount scheduling and queue status reporting ========*/
private:
- const uint64_t c_defaultFseqForRepack = 1;
const size_t c_defaultMaxNbFilesForRepack = 500;
typedef std::pair<std::string, common::dataStructures::MountType> tpType;
diff --git a/scheduler/SchedulerDatabase.hpp b/scheduler/SchedulerDatabase.hpp
index d6c9457cf9..b437b1290a 100644
--- a/scheduler/SchedulerDatabase.hpp
+++ b/scheduler/SchedulerDatabase.hpp
@@ -25,6 +25,7 @@
#include "common/dataStructures/ArchiveFile.hpp"
#include "common/dataStructures/ArchiveRequest.hpp"
#include "common/dataStructures/ArchiveFileQueueCriteriaAndFileId.hpp"
+#include "common/dataStructures/ArchiveRoute.hpp"
#include "common/dataStructures/DriveInfo.hpp"
#include "common/dataStructures/MountType.hpp"
#include "common/dataStructures/MountPolicy.hpp"
@@ -357,10 +358,7 @@ public:
virtual void complete(time_t completionTime) = 0;
virtual void setDriveStatus(common::dataStructures::DriveStatus status, time_t completionTime) = 0;
virtual void setTapeSessionStats(const castor::tape::tapeserver::daemon::TapeSessionStats &stats) = 0;
- virtual std::set<cta::SchedulerDatabase::RetrieveJob *> finishSettingJobsBatchSuccessful(
- std::list<cta::SchedulerDatabase::RetrieveJob *> & jobsBatch, log::LogContext & lc) = 0;
- virtual std::set<cta::SchedulerDatabase::RetrieveJob *> batchSucceedRetrieveForRepack(
- std::list<cta::SchedulerDatabase::RetrieveJob *> & jobsBatch, cta::log::LogContext & lc) = 0;
+ virtual void flushAsyncSuccessReports(std::list<cta::SchedulerDatabase::RetrieveJob *> & jobsBatch, log::LogContext & lc) = 0;
virtual ~RetrieveMount() {}
uint32_t nbFilesCurrentlyOnTape;
};
@@ -377,14 +375,14 @@ public:
} reportType;
cta::common::dataStructures::ArchiveFile archiveFile;
cta::common::dataStructures::RetrieveRequest retrieveRequest;
- uint64_t selectedCopyNb;
- virtual void asyncSucceed() = 0;
- virtual void checkSucceed() = 0;
- virtual void asyncReportSucceedForRepack() = 0;
- virtual void checkReportSucceedForRepack() = 0;
+ uint32_t selectedCopyNb;
+ bool isRepack = false;
+ /** Set the job successful (async). Wait() and end of report happen in RetrieveMount::flushAsyncSuccessReports() */
+ virtual void asyncSetSuccessful() = 0;
virtual void failTransfer(const std::string &failureReason, log::LogContext &lc) = 0;
virtual void failReport(const std::string &failureReason, log::LogContext &lc) = 0;
virtual ~RetrieveJob() {}
+ private:
};
/*============ Repack management: user side ================================*/
@@ -429,13 +427,23 @@ public:
/**
* A class providing the per repack request interface. It is also used to create the per file
- * requests in the object store.
+ * subrequests in the object store.
*/
class RepackRequest {
public:
cta::common::dataStructures::RepackInfo repackInfo;
- uint64_t getLastExpandedFseq();
- void addFileRequestsBatch();
+ virtual uint64_t getLastExpandedFSeq() = 0;
+ struct Subrequest {
+ uint64_t fSeq;
+ cta::common::dataStructures::ArchiveFile archiveFile;
+ std::set<uint32_t> copyNbsToRearchive;
+ std::string fileBufferURL;
+ };
+ virtual void addSubrequests(std::list<Subrequest>& repackSubrequests,
+ cta::common::dataStructures::ArchiveRoute::FullMap & archiveRoutesMap, uint64_t maxFSeqLowBound, log::LogContext & lc) = 0;
+ virtual void expandDone() = 0;
+ virtual void fail() = 0;
+ virtual ~RepackRequest() {}
};
/***/
diff --git a/scheduler/SchedulerTest.cpp b/scheduler/SchedulerTest.cpp
index 03c9ecde35..efb7a06f79 100644
--- a/scheduler/SchedulerTest.cpp
+++ b/scheduler/SchedulerTest.cpp
@@ -36,6 +36,7 @@
#include "objectstore/BackendRadosTestSwitch.hpp"
#include "objectstore/RootEntry.hpp"
#include "objectstore/JobQueueType.hpp"
+#include "objectstore/RepackIndex.hpp"
#include "tests/TestsCompileTimeSwitches.hpp"
#include "common/Timer.hpp"
#include "tapeserver/castor/tape/tapeserver/daemon/RecallReportPacker.hpp"
@@ -565,8 +566,10 @@ TEST_P(SchedulerTest, archive_report_and_retrieve_new_file) {
ASSERT_EQ(1, jobBatch.size());
retrieveJob.reset(jobBatch.front().release());
ASSERT_NE(nullptr, retrieveJob.get());
- retrieveJob->asyncComplete();
- retrieveJob->checkComplete();
+ retrieveJob->asyncSetSuccessful();
+ std::queue<std::unique_ptr<cta::RetrieveJob> > jobQueue;
+ jobQueue.push(std::move(retrieveJob));
+ retrieveMount->flushAsyncSuccessReports(jobQueue, lc);
jobBatch = retrieveMount->getNextJobBatch(1,1,lc);
ASSERT_EQ(0, jobBatch.size());
}
@@ -1347,13 +1350,16 @@ TEST_P(SchedulerTest, expandRepackRequest) {
setupDefaultCatalogue();
- //cta::log::StdoutLogger dummyLogger("dummy","dummy");
- cta::log::DummyLogger dummyLogger("dummy","dummy");
- log::LogContext lc(dummyLogger);
+#ifdef STDOUT_LOGGING
+ log::StdoutLogger dl("dummy", "unitTest");
+#else
+ log::DummyLogger dl("", "");
+#endif
+ log::LogContext lc(dl);
//Create an agent to represent this test process
std::string agentReferenceName = "expandRepackRequestTest";
- std::unique_ptr<objectstore::AgentReference> agentReference(new objectstore::AgentReference(agentReferenceName, dummyLogger));
+ std::unique_ptr<objectstore::AgentReference> agentReference(new objectstore::AgentReference(agentReferenceName, dl));
const uint64_t capacityInBytes = (uint64_t)10 * 1000 * 1000 * 1000 * 1000;
@@ -1440,7 +1446,7 @@ TEST_P(SchedulerTest, expandRepackRequest) {
scheduler.waitSchedulerDbSubthreadsComplete();
{
for(uint64_t i = 0; i < nbTapesToRepack ; ++i) {
- scheduler.queueRepack(admin,allVid.at(i),"bufferURL",common::dataStructures::RepackInfo::Type::ExpandOnly,lc);
+ scheduler.queueRepack(admin,allVid.at(i),"root://repackData/buffer",common::dataStructures::RepackInfo::Type::RepackOnly,lc);
}
scheduler.waitSchedulerDbSubthreadsComplete();
//scheduler.waitSchedulerDbSubthreadsComplete();
@@ -1473,11 +1479,11 @@ TEST_P(SchedulerTest, expandRepackRequest) {
int j = 1;
for(auto retrieveJob : retrieveJobs){
//Test that the informations are correct for each file
- ASSERT_EQ(retrieveJob.request.tapePool,s_tapePoolName);
+ //ASSERT_EQ(retrieveJob.request.tapePool,s_tapePoolName);
ASSERT_EQ(retrieveJob.request.archiveFileID,archiveFileId++);
ASSERT_EQ(retrieveJob.fileSize,compressedFileSize);
std::stringstream ss;
- ss<<"repack://public_dir/public_file_"<<i<<"_"<<j;
+ ss<<"root://repackData/buffer/"<<allVid.at(i-1)<<"/"<<std::setw(9)<<std::setfill('0')<<j;
ASSERT_EQ(retrieveJob.request.dstURL, ss.str());
ASSERT_EQ(retrieveJob.tapeCopies[vid].second.copyNb,1);
ASSERT_EQ(retrieveJob.tapeCopies[vid].second.checksumType,checksumType);
@@ -1545,7 +1551,10 @@ TEST_P(SchedulerTest, expandRepackRequest) {
re.fetch();
//Get the retrieveQueueToReportToRepackForSuccess
- std::string retrieveQueueToReportToRepackForSuccessAddress = re.getRetrieveQueueAddress(allVid.at(i-1),cta::objectstore::JobQueueType::JobsToReportToRepackForSuccess);
+ // The queue is named after the repack request: we need to query the repack index
+ objectstore::RepackIndex ri(re.getRepackIndexAddress(), schedulerDB.getBackend());
+ ri.fetchNoLock();
+ std::string retrieveQueueToReportToRepackForSuccessAddress = re.getRetrieveQueueAddress(ri.getRepackRequestAddress(allVid.at(i-1)),cta::objectstore::JobQueueType::JobsToReportToRepackForSuccess);
cta::objectstore::RetrieveQueue rq(retrieveQueueToReportToRepackForSuccessAddress,schedulerDB.getBackend());
//Fetch the queue so that we can get the retrieveRequests from it
@@ -1575,16 +1584,16 @@ TEST_P(SchedulerTest, expandRepackRequest) {
//Testing scheduler retrieve request
ASSERT_EQ(schedulerRetrieveRequest.archiveFileID,archiveFileId++);
std::stringstream ss;
- ss<<"repack://public_dir/public_file_"<<i<<"_"<<j;
+ ss<<"root://repackData/buffer/"<<allVid.at(i-1)<<"/"<<std::setw(9)<<std::setfill('0')<<j;
ASSERT_EQ(schedulerRetrieveRequest.dstURL,ss.str());
- ASSERT_EQ(schedulerRetrieveRequest.isRepack,true);
- ASSERT_EQ(schedulerRetrieveRequest.tapePool,s_tapePoolName);
+ // TODO ASSERT_EQ(schedulerRetrieveRequest.isRepack,true);
+ // TODO ASSERT_EQ(schedulerRetrieveRequest.tapePool,s_tapePoolName);
std::ostringstream diskFilePath;
diskFilePath << "/public_dir/public_file_"<<i<<"_"<<j;
ASSERT_EQ(schedulerRetrieveRequest.diskFileInfo.path,diskFilePath.str());
//Testing the retrieve request
- ASSERT_EQ(retrieveRequest.isRepack(),true);
- ASSERT_EQ(retrieveRequest.getTapePool(),s_tapePoolName);
+ ASSERT_EQ(retrieveRequest.getRepackInfo().isRepack,true);
+ // TODO ASSERT_EQ(retrieveRequest.getTapePool(),s_tapePoolName);
ASSERT_EQ(retrieveRequest.getQueueType(),cta::objectstore::JobQueueType::JobsToReportToRepackForSuccess);
ASSERT_EQ(retrieveRequest.getRetrieveFileQueueCriteria().mountPolicy,cta::common::dataStructures::MountPolicy::s_defaultMountPolicyForRepack);
ASSERT_EQ(retrieveRequest.getActiveCopyNumber(),1);
@@ -1622,7 +1631,7 @@ TEST_P(SchedulerTest, expandRepackRequest) {
//Testing RetrieveRequest
common::dataStructures::RetrieveRequest schedulerRetrieveRequest = retrieveRequest->getSchedulerRequest();
- ASSERT_EQ(retrieveRequest->getTapePool(),s_tapePoolName);
+ // TODO ASSERT_EQ(retrieveRequest->getTapePool(),s_tapePoolName);
ASSERT_EQ(retrieveRequest->getJobs().size(),1);
ASSERT_EQ(retrieveRequest->getLastActiveVid(),currentVid);
ASSERT_EQ(retrieveRequest->getQueueType(),cta::objectstore::JobQueueType::JobsToReportToRepackForSuccess);
@@ -1639,15 +1648,15 @@ TEST_P(SchedulerTest, expandRepackRequest) {
//Testing scheduler retrieve request
ASSERT_EQ(schedulerRetrieveRequest.archiveFileID,archiveFileId++);
std::stringstream ss;
- ss<<"repack://public_dir/public_file_"<<i<<"_"<<j;
+ ss<<"root://repackData/buffer/"<<allVid.at(i-1)<<"/"<<std::setw(9)<<std::setfill('0')<<j;
ASSERT_EQ(schedulerRetrieveRequest.dstURL,ss.str());
- ASSERT_EQ(schedulerRetrieveRequest.isRepack,true);
- ASSERT_EQ(schedulerRetrieveRequest.tapePool,s_tapePoolName);
+ // TODO ASSERT_EQ(schedulerRetrieveRequest.isRepack,true);
+ // TODO ASSERT_EQ(schedulerRetrieveRequest.tapePool,s_tapePoolName);
std::ostringstream diskFilePath;
diskFilePath << "/public_dir/public_file_"<<i<<"_"<<j;
ASSERT_EQ(schedulerRetrieveRequest.diskFileInfo.path,diskFilePath.str());
//Testing the retrieve request
- ASSERT_EQ(schedulerRetrieveRequest.isRepack,true);
+ // TODO ASSERT_EQ(schedulerRetrieveRequest.isRepack,true);
//Testing the archive file associated to the retrieve request
ASSERT_EQ(archiveFile.storageClass,storageClass.name);
@@ -1702,7 +1711,7 @@ TEST_P(SchedulerTest, expandRepackRequest) {
ASSERT_EQ(archiveFile.fileSize,archiveFileSize);
ASSERT_EQ(archiveFile.storageClass,s_storageClassName);
std::stringstream ss;
- ss<<"repack://public_dir/public_file_"<<i<<"_"<<j;
+ ss<<"root://repackData/buffer/"<<allVid.at(i-1)<<"/"<<std::setw(9)<<std::setfill('0')<<j;
ASSERT_EQ(archiveRequest.getSrcURL(),ss.str());
for(auto archiveJob : archiveRequest.dumpJobs()){
ASSERT_EQ(archiveJob.status,cta::objectstore::serializers::ArchiveJobStatus::AJS_ToTransferForUser);
diff --git a/scheduler/testingMocks/MockRetrieveJob.hpp b/scheduler/testingMocks/MockRetrieveJob.hpp
index 94a7b71bb2..fb8128aadd 100644
--- a/scheduler/testingMocks/MockRetrieveJob.hpp
+++ b/scheduler/testingMocks/MockRetrieveJob.hpp
@@ -33,8 +33,7 @@ namespace cta {
cta::PositioningMethod::ByBlock), completes(0), failures(0) {
archiveFile.tapeFiles[1];
}
- virtual void asyncComplete() override { completes++; }
- virtual void checkComplete() override {}
+ virtual void asyncSetSuccessful() override { completes++; }
void transferFailed(const std::string &failureReason, cta::log::LogContext&) override { failures++; };
~MockRetrieveJob() throw() {}
diff --git a/scheduler/testingMocks/MockRetrieveMount.hpp b/scheduler/testingMocks/MockRetrieveMount.hpp
index 9625ad650b..4753487bc7 100644
--- a/scheduler/testingMocks/MockRetrieveMount.hpp
+++ b/scheduler/testingMocks/MockRetrieveMount.hpp
@@ -68,7 +68,7 @@ namespace cta {
void setTapeSessionStats(const castor::tape::tapeserver::daemon::TapeSessionStats &stats) override {};
- void waitAndFinishSettingJobsBatchRetrieved(std::queue<std::unique_ptr<cta::RetrieveJob> >& successfulRetrieveJobs, cta::log::LogContext& logContext) override {};
+ void flushAsyncSuccessReports(std::queue<std::unique_ptr<cta::RetrieveJob> >& successfulRetrieveJobs, cta::log::LogContext& logContext) override {};
private:
diff --git a/tapeserver/castor/tape/tapeserver/daemon/DiskWriteTaskTest.cpp b/tapeserver/castor/tape/tapeserver/daemon/DiskWriteTaskTest.cpp
index 348daddd53..9165dbac7e 100644
--- a/tapeserver/castor/tape/tapeserver/daemon/DiskWriteTaskTest.cpp
+++ b/tapeserver/castor/tape/tapeserver/daemon/DiskWriteTaskTest.cpp
@@ -46,8 +46,7 @@ namespace unitTests{
void complete(time_t completionTime) override { throw std::runtime_error("Not implemented"); }
void setDriveStatus(cta::common::dataStructures::DriveStatus status, time_t completionTime) override { throw std::runtime_error("Not implemented"); }
void setTapeSessionStats(const castor::tape::tapeserver::daemon::TapeSessionStats &stats) override { throw std::runtime_error("Not implemented"); }
- std::set<cta::SchedulerDatabase::RetrieveJob*> finishSettingJobsBatchSuccessful(std::list<cta::SchedulerDatabase::RetrieveJob*>& jobsBatch, cta::log::LogContext& lc) override { throw std::runtime_error("Not implemented"); }
- std::set<cta::SchedulerDatabase::RetrieveJob*> batchSucceedRetrieveForRepack(std::list<cta::SchedulerDatabase::RetrieveJob*>& jobsBatch, cta::log::LogContext& lc) override { throw std::runtime_error("Not implemented"); }
+ void flushAsyncSuccessReports(std::list<cta::SchedulerDatabase::RetrieveJob*>& jobsBatch, cta::log::LogContext& lc) override { throw std::runtime_error("Not implemented"); }
};
class TestingRetrieveMount: public cta::RetrieveMount {
diff --git a/tapeserver/castor/tape/tapeserver/daemon/DiskWriteThreadPoolTest.cpp b/tapeserver/castor/tape/tapeserver/daemon/DiskWriteThreadPoolTest.cpp
index 7866aceb81..92b9206f06 100644
--- a/tapeserver/castor/tape/tapeserver/daemon/DiskWriteThreadPoolTest.cpp
+++ b/tapeserver/castor/tape/tapeserver/daemon/DiskWriteThreadPoolTest.cpp
@@ -40,8 +40,7 @@ namespace unitTests{
void complete(time_t completionTime) override { throw std::runtime_error("Not implemented"); }
void setDriveStatus(cta::common::dataStructures::DriveStatus status, time_t completionTime) override { throw std::runtime_error("Not implemented"); }
void setTapeSessionStats(const castor::tape::tapeserver::daemon::TapeSessionStats &stats) override { throw std::runtime_error("Not implemented"); }
- std::set<cta::SchedulerDatabase::RetrieveJob*> finishSettingJobsBatchSuccessful(std::list<cta::SchedulerDatabase::RetrieveJob*>& jobsBatch, cta::log::LogContext& lc) override { throw std::runtime_error("Not implemented"); }
- std::set<cta::SchedulerDatabase::RetrieveJob*> batchSucceedRetrieveForRepack(std::list<cta::SchedulerDatabase::RetrieveJob*>& jobsBatch, cta::log::LogContext& lc) override { throw std::runtime_error("Not implemented"); }
+ void flushAsyncSuccessReports(std::list<cta::SchedulerDatabase::RetrieveJob*>& jobsBatch, cta::log::LogContext& lc) override { throw std::runtime_error("Not implemented"); }
};
class TestingRetrieveMount: public cta::RetrieveMount {
diff --git a/tapeserver/castor/tape/tapeserver/daemon/RecallReportPacker.cpp b/tapeserver/castor/tape/tapeserver/daemon/RecallReportPacker.cpp
index e6f218c0dd..39d6285b26 100644
--- a/tapeserver/castor/tape/tapeserver/daemon/RecallReportPacker.cpp
+++ b/tapeserver/castor/tape/tapeserver/daemon/RecallReportPacker.cpp
@@ -115,9 +115,7 @@ void RecallReportPacker::reportTestGoingToEnd(){
//ReportSuccessful::execute
//------------------------------------------------------------------------------
void RecallReportPacker::ReportSuccessful::execute(RecallReportPacker& parent){
- if(!m_successfulRetrieveJob->retrieveRequest.isRepack){
- m_successfulRetrieveJob->asyncComplete();
- }
+ m_successfulRetrieveJob->asyncSetSuccessful();
parent.m_successfulRetrieveJobs.push(std::move(m_successfulRetrieveJob));
}
@@ -366,7 +364,7 @@ bool RecallReportPacker::errorHappened() {
//fullCheckAndFinishAsyncExecute()
//------------------------------------------------------------------------------
void RecallReportPacker::fullCheckAndFinishAsyncExecute() {
- m_retrieveMount->waitAndFinishSettingJobsBatchRetrieved(m_successfulRetrieveJobs, m_lc);
+ m_retrieveMount->flushAsyncSuccessReports(m_successfulRetrieveJobs, m_lc);
}
//------------------------------------------------------------------------------
diff --git a/tapeserver/castor/tape/tapeserver/daemon/RecallReportPackerTest.cpp b/tapeserver/castor/tape/tapeserver/daemon/RecallReportPackerTest.cpp
index ea231fa232..152b2f4831 100644
--- a/tapeserver/castor/tape/tapeserver/daemon/RecallReportPackerTest.cpp
+++ b/tapeserver/castor/tape/tapeserver/daemon/RecallReportPackerTest.cpp
@@ -52,11 +52,9 @@ protected:
MockRetrieveJobExternalStats(cta::RetrieveMount & rm, int & completes, int &failures):
MockRetrieveJob(rm), completesRef(completes), failuresRef(failures) {}
- void asyncComplete() override {
+ void asyncSetSuccessful() override {
completesRef++;
}
-
- void checkComplete() override {}
void transferFailed(const std::string &failureReason, cta::log::LogContext&) override {
failuresRef++;
diff --git a/tapeserver/castor/tape/tapeserver/daemon/RecallTaskInjectorTest.cpp b/tapeserver/castor/tape/tapeserver/daemon/RecallTaskInjectorTest.cpp
index 68065bd4c4..4a331e51d2 100644
--- a/tapeserver/castor/tape/tapeserver/daemon/RecallTaskInjectorTest.cpp
+++ b/tapeserver/castor/tape/tapeserver/daemon/RecallTaskInjectorTest.cpp
@@ -135,9 +135,7 @@ namespace unitTests
void complete(time_t completionTime) override { throw std::runtime_error("Not implemented"); }
void setDriveStatus(cta::common::dataStructures::DriveStatus status, time_t completionTime) override { throw std::runtime_error("Not implemented"); }
void setTapeSessionStats(const castor::tape::tapeserver::daemon::TapeSessionStats &stats) override { throw std::runtime_error("Not implemented"); }
- std::set<cta::SchedulerDatabase::RetrieveJob*> finishSettingJobsBatchSuccessful(std::list<cta::SchedulerDatabase::RetrieveJob*>& jobsBatch, cta::log::LogContext& lc) override { throw std::runtime_error("Not implemented"); }
- std::set<cta::SchedulerDatabase::RetrieveJob*> batchSucceedRetrieveForRepack(std::list<cta::SchedulerDatabase::RetrieveJob*>& jobsBatch, cta::log::LogContext& lc) override { throw std::runtime_error("Not implemented"); }
-
+ void flushAsyncSuccessReports(std::list<cta::SchedulerDatabase::RetrieveJob*>& jobsBatch, cta::log::LogContext& lc) override { throw std::runtime_error("Not implemented"); }
};
TEST_F(castor_tape_tapeserver_daemonTest, RecallTaskInjectorNominal) {
--
GitLab