ALPAKA_ACCELERATOR_NAMESPACE::EDMetadata Class Reference

#include <EDMetadata.h>

Public Member Functions
Device	device () const
void	discardEvent ()
	EDMetadata (std::shared_ptr< Queue > queue, std::shared_ptr< Event > event)
void	enqueueCallback (edm::WaitingTaskWithArenaHolder holder)
Queue &	queue () const
void	recordEvent ()
void	synchronize (EDMetadata &consumer, bool tryReuseQueue) const
	~EDMetadata ()

Private Member Functions
std::shared_ptr< Queue >	tryReuseQueue_ () const

Private Attributes
std::shared_ptr< Event >	event_
std::atomic< bool >	eventComplete_ = false
std::atomic< bool >	mayReuseQueue_ = true
std::shared_ptr< Queue >	queue_

Detailed Description

The EDMetadata class provides the exact synchronization mechanisms for Event data products for backends with asynchronous Queue. These include

• adding a notification for edm::WaitingTaskWithArenaHolder
• recording an Event
• synchronizing an Event data product and a consuming EDModule

For synchronous backends the EDMetadata acts as an owner of the Queue object, as no further synchronization is needed.

EDMetadata is used as the Metadata class for edm::DeviceProduct<T>, and is an implementation detail (not visible to user code).

TODO: What to do with device-synchronous backends? The data product needs to be wrapped into the edm::DeviceProduct, but the EDMetadata class used there does not need anything except "dummy" implementation of synchronize(). The question is clearly solvable, so maybe leave it to the time we would actually need one?

Definition at line 61 of file EDMetadata.h.

Constructor & Destructor Documentation

EDMetadata()

ALPAKA_ACCELERATOR_NAMESPACE::EDMetadata::EDMetadata ( std::shared_ptr< Queue >  queue, std::shared_ptr< Event >  event  )

inline

Definition at line 63 of file EDMetadata.h.

        : queue_(std::move(queue)), event_(std::move(event)) {}

~EDMetadata()

ALPAKA_ACCELERATOR_NAMESPACE::EDMetadata::~EDMetadata

(

)

Definition at line 10 of file EDMetadata.cc.

References CMS_SA_ALLOW, event_, eventComplete_, mayReuseQueue_, and SequenceTypes::wait().

                          {
    // Make sure that the production of the product in the GPU is
    // complete before destructing the product. This is to make sure
    // that the EDM stream does not move to the next event before all
    // asynchronous processing of the current is complete.

    // TODO: a callback notifying a WaitingTaskHolder (or similar)
    // would avoid blocking the CPU, but would also require more work.

    // If event_ is null, the EDMetadata was either
    // default-constructed, or fully synchronized before leaving the
    // produce() call, so no synchronization is needed.
    // If the queue was re-used, then some other EDMetadata object in
    // the same edm::Event records the event_ (in the same queue) and
    // calls alpaka::wait(), and therefore this wait() call can be
    // skipped).
    if (event_ and not eventComplete_ and mayReuseQueue_) {
      // Must not throw in a destructor, and if there were an
      // exception could not really propagate it anyway.
      CMS_SA_ALLOW try { alpaka::wait(*event_); } catch (...) {
      }
    }
  }

Member Function Documentation

device()

Device ALPAKA_ACCELERATOR_NAMESPACE::EDMetadata::device ( ) const

inline

Definition at line 67 of file EDMetadata.h.

References queue_.

{ return alpaka::getDev(*queue_); }

discardEvent()

void ALPAKA_ACCELERATOR_NAMESPACE::EDMetadata::discardEvent ( )

inline

Definition at line 76 of file EDMetadata.h.

References event_.

{ event_.reset(); }

enqueueCallback()

void ALPAKA_ACCELERATOR_NAMESPACE::EDMetadata::enqueueCallback

(

edm::WaitingTaskWithArenaHolder

holder

)

Definition at line 34 of file EDMetadata.cc.

References EDM_STRINGIZE, event_, eostools::move(), recordEvent(), edm::Async::runAsync(), and SequenceTypes::wait().

                                                                       {
    edm::Service<edm::Async> async;
    recordEvent();
    async->runAsync(
        std::move(holder),
        [event = event_]() mutable { alpaka::wait(*event); },
        []() { return "Enqueued via " EDM_STRINGIZE(ALPAKA_ACCELERATOR_NAMESPACE) "::EDMetadata::enqueueCallback()"; });
  }

queue()

Queue& ALPAKA_ACCELERATOR_NAMESPACE::EDMetadata::queue ( ) const

inline

Definition at line 71 of file EDMetadata.h.

References queue_.

Referenced by synchronize().

{ return *queue_; }

recordEvent()

void ALPAKA_ACCELERATOR_NAMESPACE::EDMetadata::recordEvent ( )

inline

Definition at line 75 of file EDMetadata.h.

References event_, and queue_.

Referenced by enqueueCallback().

{ alpaka::enqueue(*queue_, *event_); }

synchronize()

void ALPAKA_ACCELERATOR_NAMESPACE::EDMetadata::synchronize	(	EDMetadata &	consumer,
		bool	tryReuseQueue
	)		const

Synchronizes 'consumer' metadata wrt. 'this' in the event product

Definition at line 43 of file EDMetadata.cc.

References OfflineOutput_cfi::consumer, event_, eventComplete_, Exception, edm::errors::LogicError, queue(), queue_, tryReuseQueue_(), and SequenceTypes::wait().

                                                                             {
    if (*queue_ == *consumer.queue_) {
      return;
    }

    if (tryReuseQueue) {
      if (auto queue = tryReuseQueue_()) {
        consumer.queue_ = queue_;
        return;
      }
    }

    if (eventComplete_) {
      return;
    }

    // TODO: how necessary this check is?
    if (alpaka::getDev(*queue_) != alpaka::getDev(*consumer.queue_)) {
      throw edm::Exception(edm::errors::LogicError) << "Handling data from multiple devices is not yet supported";
    }

    // If the event has been discarded, the produce() function that
    // constructed this EDMetadata object did not launch any
    // asynchronous work.
    if (not event_) {
      return;
    }

    if (alpaka::isComplete(*event_)) {
      eventComplete_ = true;
    } else {
      // Event not yet occurred, so need to add synchronization
      // here. Sychronization is done by making the queue to wait
      // for an event, so all subsequent work in the queue will run
      // only after the event has "occurred" (i.e. data product
      // became available).
      alpaka::wait(*consumer.queue_, *event_);
    }
  }

tryReuseQueue_()

std::shared_ptr< Queue > ALPAKA_ACCELERATOR_NAMESPACE::EDMetadata::tryReuseQueue_ ( ) const

private

Returns a shared_ptr to the Queue if it can be reused, or a null shared_ptr if not

Definition at line 83 of file EDMetadata.cc.

References mayReuseQueue_, and queue_.

Referenced by synchronize().

                                                        {
    bool expected = true;
    if (mayReuseQueue_.compare_exchange_strong(expected, false)) {
      // If the current thread is the one flipping the flag, it may
      // reuse the queue.
      return queue_;
    }
    return nullptr;
  }

Member Data Documentation

event_

std::shared_ptr<Event> ALPAKA_ACCELERATOR_NAMESPACE::EDMetadata::event_

private

Definition at line 91 of file EDMetadata.h.

Referenced by discardEvent(), enqueueCallback(), recordEvent(), synchronize(), and ~EDMetadata().

eventComplete_

std::atomic<bool> ALPAKA_ACCELERATOR_NAMESPACE::EDMetadata::eventComplete_ = false

mutableprivate

Definition at line 97 of file EDMetadata.h.

Referenced by synchronize(), and ~EDMetadata().

mayReuseQueue_

std::atomic<bool> ALPAKA_ACCELERATOR_NAMESPACE::EDMetadata::mayReuseQueue_ = true

mutableprivate

Definition at line 95 of file EDMetadata.h.

Referenced by tryReuseQueue_(), and ~EDMetadata().

queue_

std::shared_ptr<Queue> ALPAKA_ACCELERATOR_NAMESPACE::EDMetadata::queue_

private

Definition at line 90 of file EDMetadata.h.

Referenced by device(), queue(), recordEvent(), synchronize(), and tryReuseQueue_().