diff --git a/objectstore/Algorithms.hpp b/objectstore/Algorithms.hpp index 8a4768f08ea20e6297863e32404dc2c1d046588b..bfd4df16f34e5a7d163a379c4cdbe22d12a1aaa3 100644 --- a/objectstore/Algorithms.hpp +++ b/objectstore/Algorithms.hpp @@ -29,117 +29,30 @@ namespace cta { namespace objectstore { -class EmptyClass {}; -class ArchiveQueue; -class ArchiveJob; - /** * Container traits definition. To be specialized class by class. * This is mostly a model. */ -template <class Container> +template <class C> class ContainerTraits{ public: + typedef C Container; typedef std::string ContainerAddress; typedef std::string ElementAddress; typedef std::string ContainerIdentifyer; - typedef cta::objectstore::EmptyClass Element; + class Element {}; typedef std::list<std::unique_ptr<Element>> ElementMemoryContainer; typedef std::list <Element *> ElementPointerContainer; - typedef cta::objectstore::EmptyClass ElementDescriptor; + class ElementDescriptor {}; typedef std::list<ElementDescriptor> ElementDescriptorContainer; + static ElementAddress getElementAddress(const Element & e); static void getLockedAndFetched(Container & cont, ScopedExclusiveLock & contLock, AgentReference & agRef, const ContainerIdentifyer & cId, log::LogContext & lc); static void addReferencesAndCommit(Container & cont, ElementMemoryContainer & elemMemCont); static ElementPointerContainer switchElementsOwnership(Container & cont, ElementMemoryContainer & elements, log::LogContext & lc); }; -template <> -class ContainerTraits<ArchiveQueue> { -public: - typedef ArchiveQueue Container; - typedef std::string ContainerAddress; - typedef std::string ElementAddress; - typedef std::string ContainerIdentifyer; - struct Element { - std::unique_ptr<ArchiveRequest> archiveRequest; - uint16_t copyNb; - cta::common::dataStructures::ArchiveFile archiveFile; - cta::common::dataStructures::MountPolicy mountPolicy; - }; - typedef std::list<Element> ElementMemoryContainer; - struct ElementOpFailure { - Element * element; - std::exception_ptr failure; - }; - typedef std::list<ElementOpFailure> ElementOpFailureContainer; - typedef ArchiveRequest::JobDump ElementDescriptor; - typedef std::list<ElementDescriptor> ElementDescriptorContainer; - - static ElementAddress getElementAddress(const Element & e) { return e.archiveRequest->getAddressIfSet(); } - - static void getLockedAndFetched(Container & cont, ScopedExclusiveLock & aqL, AgentReference & agRef, const ContainerIdentifyer & tapePool, - log::LogContext & lc) { - Helpers::getLockedAndFetchedQueue<ArchiveQueue>(cont, aqL, agRef, tapePool, QueueType::LiveJobs, lc); - } - - static void addReferencesAndCommit(Container & cont, ElementMemoryContainer & elemMemCont, - AgentReference & agentRef, log::LogContext & lc) { - std::list<ArchiveQueue::JobToAdd> jobsToAdd; - for (auto & e: elemMemCont) { - ArchiveRequest::JobDump jd; - jd.copyNb = e.copyNb; - jd.tapePool = cont.getTapePool(); - jd.owner = cont.getAddressIfSet(); - ArchiveRequest & ar = *e.archiveRequest; - jobsToAdd.push_back({jd, ar.getAddressIfSet(), e.archiveFile.archiveFileID, e.archiveFile.fileSize, - e.mountPolicy, time(NULL)}); - } - cont.addJobsAndCommit(jobsToAdd, agentRef, lc); - } - - static void removeReferencesAndCommit(Container & cont, ElementOpFailureContainer & elementsOpFailures) { - std::list<std::string> elementsToRemove; - for (auto & eof: elementsOpFailures) { - elementsToRemove.emplace_back(eof.element->archiveRequest->getAddressIfSet()); - } - cont.removeJobsAndCommit(elementsToRemove); - } - - static ElementOpFailureContainer switchElementsOwnership(ElementMemoryContainer & elemMemCont, Container & cont, - const ContainerAddress & previousOwnerAddress, log::LogContext & lc) { - std::list<std::unique_ptr<ArchiveRequest::AsyncJobOwnerUpdater>> updaters; - for (auto & e: elemMemCont) { - ArchiveRequest & ar = *e.archiveRequest; - auto copyNb = e.copyNb; - updaters.emplace_back(ar.asyncUpdateJobOwner(copyNb, cont.getAddressIfSet(), previousOwnerAddress)); - } - auto u = updaters.begin(); - auto e = elemMemCont.begin(); - ElementOpFailureContainer ret; - while (e != elemMemCont.end()) { - try { - u->get()->wait(); - } catch (...) { - ret.push_back(ElementOpFailure()); - ret.back().element = &(*e); - ret.back().failure = std::current_exception(); - } - u++; - e++; - } - return ret; - } - - - class OwnershipSwitchFailure: public cta::exception::Exception { - public: - OwnershipSwitchFailure(const std::string & message): cta::exception::Exception(message) {}; - ElementOpFailureContainer failedElements; - }; -}; - template <class C> class ContainerAlgorithms { public: diff --git a/objectstore/AlgorithmsTest.cpp b/objectstore/AlgorithmsTest.cpp index 7c8db1d3d3e25724c9f9feb359ee6adb2c7e2e74..d2b4b97c4a8a23d225828e296ae7dfda29c60e96 100644 --- a/objectstore/AlgorithmsTest.cpp +++ b/objectstore/AlgorithmsTest.cpp @@ -20,8 +20,7 @@ #include "RootEntry.hpp" #include "AgentReference.hpp" #include "Agent.hpp" -#include "Algorithms.hpp" -#include "ArchiveRequest.hpp" +#include "ArchiveQueueAlgorithms.hpp" #include "common/log/DummyLogger.hpp" #include "tests/TestsCompileTimeSwitches.hpp" #include "catalogue/DummyCatalogue.hpp" diff --git a/objectstore/ArchiveQueueAlgorithms.hpp b/objectstore/ArchiveQueueAlgorithms.hpp new file mode 100644 index 0000000000000000000000000000000000000000..8a5aa0db1c899c033f098a8f6f59f352df2232cd --- /dev/null +++ b/objectstore/ArchiveQueueAlgorithms.hpp @@ -0,0 +1,108 @@ +/* + * 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 "Algorithms.hpp" +#include "ArchiveQueue.hpp" + +namespace cta { namespace objectstore { + +template <> +class ContainerTraits<ArchiveQueue> { +public: + typedef ArchiveQueue Container; + typedef std::string ContainerAddress; + typedef std::string ElementAddress; + typedef std::string ContainerIdentifyer; + struct Element { + std::unique_ptr<ArchiveRequest> archiveRequest; + uint16_t copyNb; + cta::common::dataStructures::ArchiveFile archiveFile; + cta::common::dataStructures::MountPolicy mountPolicy; + }; + typedef std::list<Element> ElementMemoryContainer; + struct ElementOpFailure { + Element * element; + std::exception_ptr failure; + }; + typedef std::list<ElementOpFailure> ElementOpFailureContainer; + typedef ArchiveRequest::JobDump ElementDescriptor; + typedef std::list<ElementDescriptor> ElementDescriptorContainer; + + static ElementAddress getElementAddress(const Element & e) { return e.archiveRequest->getAddressIfSet(); } + + static void getLockedAndFetched(Container & cont, ScopedExclusiveLock & aqL, AgentReference & agRef, const ContainerIdentifyer & tapePool, + log::LogContext & lc) { + Helpers::getLockedAndFetchedQueue<ArchiveQueue>(cont, aqL, agRef, tapePool, QueueType::LiveJobs, lc); + } + + static void addReferencesAndCommit(Container & cont, ElementMemoryContainer & elemMemCont, + AgentReference & agentRef, log::LogContext & lc) { + std::list<ArchiveQueue::JobToAdd> jobsToAdd; + for (auto & e: elemMemCont) { + ArchiveRequest::JobDump jd; + jd.copyNb = e.copyNb; + jd.tapePool = cont.getTapePool(); + jd.owner = cont.getAddressIfSet(); + ArchiveRequest & ar = *e.archiveRequest; + jobsToAdd.push_back({jd, ar.getAddressIfSet(), e.archiveFile.archiveFileID, e.archiveFile.fileSize, + e.mountPolicy, time(NULL)}); + } + cont.addJobsAndCommit(jobsToAdd, agentRef, lc); + } + + static void removeReferencesAndCommit(Container & cont, ElementOpFailureContainer & elementsOpFailures) { + std::list<std::string> elementsToRemove; + for (auto & eof: elementsOpFailures) { + elementsToRemove.emplace_back(eof.element->archiveRequest->getAddressIfSet()); + } + cont.removeJobsAndCommit(elementsToRemove); + } + + static ElementOpFailureContainer switchElementsOwnership(ElementMemoryContainer & elemMemCont, Container & cont, + const ContainerAddress & previousOwnerAddress, log::LogContext & lc) { + std::list<std::unique_ptr<ArchiveRequest::AsyncJobOwnerUpdater>> updaters; + for (auto & e: elemMemCont) { + ArchiveRequest & ar = *e.archiveRequest; + auto copyNb = e.copyNb; + updaters.emplace_back(ar.asyncUpdateJobOwner(copyNb, cont.getAddressIfSet(), previousOwnerAddress)); + } + auto u = updaters.begin(); + auto e = elemMemCont.begin(); + ElementOpFailureContainer ret; + while (e != elemMemCont.end()) { + try { + u->get()->wait(); + } catch (...) { + ret.push_back(ElementOpFailure()); + ret.back().element = &(*e); + ret.back().failure = std::current_exception(); + } + u++; + e++; + } + return ret; + } + + class OwnershipSwitchFailure: public cta::exception::Exception { + public: + OwnershipSwitchFailure(const std::string & message): cta::exception::Exception(message) {}; + ElementOpFailureContainer failedElements; + }; +}; + +}} // namespace cta::objectstore \ No newline at end of file