Portable/interface/ProductBase.h

0001 #ifndef DataFormats_Portable_interface_ProductBase_h
0002 #define DataFormats_Portable_interface_ProductBase_h
0003
0004 #include <atomic>
0005 #include <memory>
0006 #include <utility>
0007
0008 #include <alpaka/alpaka.hpp>
0009
0010 #include "HeterogeneousCore/AlpakaInterface/interface/ScopedContextFwd.h"
0011 #include "HeterogeneousCore/AlpakaInterface/interface/traits.h"
0012
0013 namespace cms::alpakatools {
0014
0015   /**
0016    * Base class for all instantiations of Product<TQueue, T> to hold the
0017    * non-T-dependent members.
0018    */
0019   template <typename TQueue, typename = std::enable_if_t<cms::alpakatools::is_queue_v<TQueue>>>
0020   class ProductBase {
0021   public:
0022     using Queue = TQueue;
0023     using Event = alpaka::Event<Queue>;
0024     using Device = alpaka::Dev<Queue>;
0025
0026     ProductBase() = default;  // Needed only for ROOT dictionary generation
0027
0028     ~ProductBase() {
0029       // Make sure that the production of the product in the GPU is
0030       // complete before destructing the product. This is to make sure
0031       // that the EDM stream does not move to the next event before all
0032       // asynchronous processing of the current is complete.
0033
0034       // TODO: a callback notifying a WaitingTaskHolder (or similar)
0035       // would avoid blocking the CPU, but would also require more work.
0036
0037       // FIXME: this may throw an execption if the underlaying call fails.
0038       if (event_) {
0039         alpaka::wait(*event_);
0040       }
0041     }
0042
0043     ProductBase(const ProductBase&) = delete;
0044     ProductBase& operator=(const ProductBase&) = delete;
0045     ProductBase(ProductBase&& other)
0046         : queue_{std::move(other.queue_)}, event_{std::move(other.event_)}, mayReuseQueue_{other.mayReuseQueue_.load()} {}
0047     ProductBase& operator=(ProductBase&& other) {
0048       queue_ = std::move(other.queue_);
0049       event_ = std::move(other.event_);
0050       mayReuseQueue_ = other.mayReuseQueue_.load();
0051       return *this;
0052     }
0053
0054     bool isValid() const { return queue_.get() != nullptr; }
0055
0056     bool isAvailable() const {
0057       // if default-constructed, the product is not available
0058       if (not event_) {
0059         return false;
0060       }
0061       return alpaka::isComplete(*event_);
0062     }
0063
0064     // returning a const& requires changes in alpaka's getDev() implementations
0065     Device device() const { return alpaka::getDev(queue()); }
0066
0067     Queue const& queue() const { return *queue_; }
0068
0069     Event const& event() const { return *event_; }
0070
0071   protected:
0072     explicit ProductBase(std::shared_ptr<Queue> queue, std::shared_ptr<Event> event)
0073         : queue_{std::move(queue)}, event_{std::move(event)} {}
0074
0075   private:
0076     friend class impl::ScopedContextBase<Queue>;
0077     friend class ScopedContextProduce<Queue>;
0078
0079     // The following function is intended to be used only from ScopedContext
0080     const std::shared_ptr<Queue>& queuePtr() const { return queue_; }
0081
0082     bool mayReuseQueue() const {
0083       bool expected = true;
0084       bool changed = mayReuseQueue_.compare_exchange_strong(expected, false);
0085       // If the current thread is the one flipping the flag, it may
0086       // reuse the queue.
0087       return changed;
0088     }
0089
0090     // shared_ptr because of caching in QueueCache, and sharing across edm::Event products
0091     std::shared_ptr<Queue> queue_;  //!
0092     // shared_ptr because of caching in EventCache
0093     std::shared_ptr<Event> event_;  //!
0094
0095     // This flag tells whether the queue may be reused by a consumer or not.
0096     // The goal is to have a "chain" of modules to enqueue their work to the same queue.
0097     mutable std::atomic<bool> mayReuseQueue_ = true;  //!
0098   };
0099
0100 }  // namespace cms::alpakatools
0101
0102 #endif  // DataFormats_Portable_interface_ProductBase_h