Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Added transformAsync ability to modules #39557

Merged
merged 2 commits into from
Oct 17, 2022
Merged

Added transformAsync ability to modules #39557

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
1cf9461
Added transformAsync ability to modules
Dr15Jones Sep 30, 2022
8924cf4
Changes based on PR comments
Dr15Jones Oct 6, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
18 changes: 14 additions & 4 deletions FWCore/Framework/interface/TransformerBase.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,7 @@
#include <string>
#include <functional>
#include <memory>
#include <any>

namespace edm {
class ProducerBase;
Expand All @@ -25,6 +26,8 @@ namespace edm {
class BranchDescription;
class ProductResolverIndexHelper;
class ModuleDescription;
class WaitingTaskWithArenaHolder;
class WaitingTaskHolder;

class TransformerBase {
public:
Expand All @@ -34,20 +37,27 @@ namespace edm {
protected:
//The function takes the WrapperBase corresponding to the data product from the EDPutToken
// and returns the WrapperBase associated to the id and instanceName
using TransformFunction = std::function<std::unique_ptr<edm::WrapperBase>(edm::WrapperBase const&)>;
using TransformFunction = std::function<std::unique_ptr<edm::WrapperBase>(std::any)>;
using PreTransformFunction = std::function<std::any(edm::WrapperBase const&, edm::WaitingTaskWithArenaHolder)>;

void registerTransformImp(ProducerBase&, EDPutToken, const TypeID& id, std::string instanceName, TransformFunction);
void registerTransformAsyncImp(
ProducerBase&, EDPutToken, const TypeID& id, std::string instanceName, PreTransformFunction, TransformFunction);

std::size_t findMatchingIndex(ProducerBase const& iBase, edm::BranchDescription const&) const;
ProductResolverIndex prefetchImp(std::size_t iIndex) const { return transformInfo_.get<0>(iIndex); }
void transformImp(std::size_t iIndex, ProducerBase const& iBase, edm::EventForTransformer&) const;
ProductResolverIndex prefetchImp(std::size_t iIndex) const { return transformInfo_.get<kResolverIndex>(iIndex); }
void transformImpAsync(WaitingTaskHolder iTask,
std::size_t iIndex,
ProducerBase const& iBase,
edm::EventForTransformer&) const;

void extendUpdateLookup(ProducerBase const&,
ModuleDescription const& iModuleDesc,
ProductResolverIndexHelper const& iHelper);

private:
SoATuple<ProductResolverIndex, TypeID, EDPutToken, TransformFunction> transformInfo_;
enum InfoColumns { kResolverIndex, kType, kToken, kPreTransform, kTransform };
SoATuple<ProductResolverIndex, TypeID, EDPutToken, PreTransformFunction, TransformFunction> transformInfo_;
};
} // namespace edm

Expand Down
16 changes: 11 additions & 5 deletions FWCore/Framework/interface/global/EDFilterBase.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -39,6 +39,7 @@ namespace edm {
class ThinnedAssociationsHelper;
class WaitingTaskWithArenaHolder;
class EventForTransformer;
class ServiceWeakToken;

namespace maker {
template <typename T>
Expand Down Expand Up @@ -78,10 +79,12 @@ namespace edm {
ActivityRegistry*,
ModuleCallingContext const*,
WaitingTaskWithArenaHolder&);
void doTransform(size_t iTransformIndex,
EventPrincipal const& iEvent,
ActivityRegistry*,
ModuleCallingContext const*);
void doTransformAsync(WaitingTaskHolder iTask,
size_t iTransformIndex,
EventPrincipal const& iEvent,
ActivityRegistry*,
ModuleCallingContext const*,
ServiceWeakToken const&);
//For now this is a placeholder
/*virtual*/ void preActionBeforeRunEventAsync(WaitingTaskHolder iTask,
ModuleCallingContext const& iModuleCallingContext,
Expand Down Expand Up @@ -154,7 +157,10 @@ namespace edm {

virtual size_t transformIndex_(edm::BranchDescription const& iBranch) const;
virtual ProductResolverIndex transformPrefetch_(std::size_t iIndex) const;
virtual void transform_(std::size_t iIndex, edm::EventForTransformer& iEvent) const;
virtual void transformAsync_(WaitingTaskHolder iTask,
std::size_t iIndex,
edm::EventForTransformer& iEvent,
ServiceWeakToken const& iToken) const;

virtual void clearInputProcessBlockCaches();
virtual bool hasAcquire() const { return false; }
Expand Down
16 changes: 11 additions & 5 deletions FWCore/Framework/interface/global/EDProducerBase.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -40,6 +40,7 @@ namespace edm {
class ThinnedAssociationsHelper;
class WaitingTaskWithArenaHolder;
class EventForTransformer;
class ServiceWeakToken;

namespace maker {
template <typename T>
Expand Down Expand Up @@ -81,10 +82,12 @@ namespace edm {
ActivityRegistry*,
ModuleCallingContext const*,
WaitingTaskWithArenaHolder&);
void doTransform(size_t iTransformIndex,
EventPrincipal const& iEvent,
ActivityRegistry*,
ModuleCallingContext const*);
void doTransformAsync(WaitingTaskHolder iTask,
size_t iTransformIndex,
EventPrincipal const& iEvent,
ActivityRegistry*,
ModuleCallingContext const*,
ServiceWeakToken const&);
void doPreallocate(PreallocationConfiguration const&);
void doBeginJob();
void doEndJob();
Expand Down Expand Up @@ -157,7 +160,10 @@ namespace edm {

virtual size_t transformIndex_(edm::BranchDescription const& iBranch) const;
virtual ProductResolverIndex transformPrefetch_(std::size_t iIndex) const;
virtual void transform_(std::size_t iIndex, edm::EventForTransformer& iEvent) const;
virtual void transformAsync_(WaitingTaskHolder iTask,
std::size_t iIndex,
edm::EventForTransformer& iEvent,
ServiceWeakToken const& iToken) const;

virtual void clearInputProcessBlockCaches();
virtual bool hasAccumulator() const { return false; }
Expand Down
51 changes: 40 additions & 11 deletions FWCore/Framework/interface/global/implementors.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -29,6 +29,8 @@

// user include files
#include "FWCore/Common/interface/FWCoreCommonFwd.h"
#include "FWCore/Concurrency/interface/WaitingTaskHolder.h"
#include "FWCore/Concurrency/interface/WaitingTaskWithArenaHolder.h"
#include "FWCore/Framework/interface/CacheHandle.h"
#include "FWCore/Framework/interface/Frameworkfwd.h"
#include "FWCore/Framework/interface/InputProcessBlockCacheImpl.h"
Expand All @@ -47,6 +49,7 @@
namespace edm {

class WaitingTaskWithArenaHolder;
class ServiceWeakToken;

namespace global {
namespace impl {
Expand Down Expand Up @@ -462,15 +465,38 @@ namespace edm {
void registerTransform(edm::EDPutTokenT<G> iToken, F iF, std::string productInstance = std::string()) {
using ReturnTypeT = decltype(iF(std::declval<G>()));
TypeID returnType(typeid(ReturnTypeT));
TransformerBase::registerTransformImp(*this,
EDPutToken(iToken),
returnType,
std::move(productInstance),
[f = std::move(iF)](edm::WrapperBase const& iGotProduct) {
return std::make_unique<edm::Wrapper<ReturnTypeT>>(
WrapperBase::Emplace{},
f(*static_cast<edm::Wrapper<G> const&>(iGotProduct).product()));
});
TransformerBase::registerTransformImp(
*this,
EDPutToken(iToken),
returnType,
std::move(productInstance),
[f = std::move(iF)](std::any const& iGotProduct) {
auto pGotProduct = std::any_cast<edm::WrapperBase const*>(iGotProduct);
return std::make_unique<edm::Wrapper<ReturnTypeT>>(
WrapperBase::Emplace{}, f(*static_cast<edm::Wrapper<G> const*>(pGotProduct)->product()));
});
}

template <typename G, typename P, typename F>
void registerTransformAsync(edm::EDPutTokenT<G> iToken,
P iPre,
F iF,
std::string productInstance = std::string()) {
using CacheTypeT = decltype(iPre(std::declval<G>(), WaitingTaskWithArenaHolder()));
using ReturnTypeT = decltype(iF(std::declval<CacheTypeT>()));
TypeID returnType(typeid(ReturnTypeT));
TransformerBase::registerTransformAsyncImp(
*this,
EDPutToken(iToken),
returnType,
std::move(productInstance),
[p = std::move(iPre)](edm::WrapperBase const& iGotProduct, WaitingTaskWithArenaHolder iHolder) {
return std::any(p(*static_cast<edm::Wrapper<G> const&>(iGotProduct).product(), std::move(iHolder)));
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

IIUC, wrapping the temporary object (CacheTypeT) to std::any has the consequence that the CacheTypeT must be copyable, and e.g. std::unique_ptr would not work. This is probably ok for the foreseen use case (Alpaka buffers behave like shared_ptr), but might surprise in some other cases.

},
[f = std::move(iF)](std::any const& iCache) {
return std::make_unique<edm::Wrapper<ReturnTypeT>>(WrapperBase::Emplace{},
f(std::any_cast<CacheTypeT>(iCache)));
});
}

private:
Expand All @@ -480,8 +506,11 @@ namespace edm {
ProductResolverIndex transformPrefetch_(std::size_t iIndex) const final {
return TransformerBase::prefetchImp(iIndex);
}
void transform_(std::size_t iIndex, edm::EventForTransformer& iEvent) const final {
return TransformerBase::transformImp(iIndex, *this, iEvent);
void transformAsync_(WaitingTaskHolder iTask,
std::size_t iIndex,
edm::EventForTransformer& iEvent,
ServiceWeakToken const& iToken) const final {
return TransformerBase::transformImpAsync(std::move(iTask), iIndex, *this, iEvent);
}
void extendUpdateLookup(BranchType iBranchType, ProductResolverIndexHelper const& iHelper) override {
if (iBranchType == InEvent) {
Expand Down
16 changes: 11 additions & 5 deletions FWCore/Framework/interface/limited/EDFilterBase.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -39,6 +39,7 @@ namespace edm {
class ActivityRegistry;
class ThinnedAssociationsHelper;
class EventForTransformer;
class ServiceWeakToken;

namespace maker {
template <typename T>
Expand Down Expand Up @@ -78,10 +79,12 @@ namespace edm {

private:
bool doEvent(EventTransitionInfo const&, ActivityRegistry*, ModuleCallingContext const*);
void doTransform(size_t iTransformIndex,
EventPrincipal const& iEvent,
ActivityRegistry*,
ModuleCallingContext const*);
void doTransformAsync(WaitingTaskHolder iTask,
size_t iTransformIndex,
EventPrincipal const& iEvent,
ActivityRegistry*,
ModuleCallingContext const*,
ServiceWeakToken const&);

//For now this is a placeholder
/*virtual*/ void preActionBeforeRunEventAsync(WaitingTaskHolder,
Expand Down Expand Up @@ -155,7 +158,10 @@ namespace edm {

virtual size_t transformIndex_(edm::BranchDescription const& iBranch) const;
virtual ProductResolverIndex transformPrefetch_(std::size_t iIndex) const;
virtual void transform_(std::size_t iIndex, edm::EventForTransformer& iEvent) const;
virtual void transformAsync_(WaitingTaskHolder iTask,
std::size_t iIndex,
edm::EventForTransformer& iEvent,
ServiceWeakToken const& iToken) const;

virtual void clearInputProcessBlockCaches();

Expand Down
16 changes: 11 additions & 5 deletions FWCore/Framework/interface/limited/EDProducerBase.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -40,6 +40,7 @@ namespace edm {
class ActivityRegistry;
class ThinnedAssociationsHelper;
class EventForTransformer;
class ServiceWeakToken;

namespace maker {
template <typename T>
Expand Down Expand Up @@ -81,10 +82,12 @@ namespace edm {

private:
bool doEvent(EventTransitionInfo const&, ActivityRegistry*, ModuleCallingContext const*);
void doTransform(size_t iTransformIndex,
EventPrincipal const& iEvent,
ActivityRegistry*,
ModuleCallingContext const*);
void doTransformAsync(WaitingTaskHolder iTask,
size_t iTransformIndex,
EventPrincipal const& iEvent,
ActivityRegistry*,
ModuleCallingContext const*,
ServiceWeakToken const&);
void doPreallocate(PreallocationConfiguration const&);
void doBeginJob();
void doEndJob();
Expand Down Expand Up @@ -158,7 +161,10 @@ namespace edm {

virtual size_t transformIndex_(edm::BranchDescription const& iBranch) const;
virtual ProductResolverIndex transformPrefetch_(std::size_t iIndex) const;
virtual void transform_(std::size_t iIndex, edm::EventForTransformer& iEvent) const;
virtual void transformAsync_(WaitingTaskHolder iTask,
std::size_t iIndex,
edm::EventForTransformer& iEvent,
ServiceWeakToken const& iToken) const;

virtual void clearInputProcessBlockCaches();
virtual bool hasAccumulator() const { return false; }
Expand Down
51 changes: 40 additions & 11 deletions FWCore/Framework/interface/limited/implementors.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,6 +28,8 @@

// user include files
#include "FWCore/Common/interface/FWCoreCommonFwd.h"
#include "FWCore/Concurrency/interface/WaitingTaskHolder.h"
#include "FWCore/Concurrency/interface/WaitingTaskWithArenaHolder.h"
#include "FWCore/Framework/interface/CacheHandle.h"
#include "FWCore/Framework/interface/Frameworkfwd.h"
#include "FWCore/Framework/interface/InputProcessBlockCacheImpl.h"
Expand All @@ -44,6 +46,7 @@

// forward declarations
namespace edm {
class ServiceWeakToken;

namespace limited {
namespace impl {
Expand Down Expand Up @@ -449,15 +452,38 @@ namespace edm {
void registerTransform(edm::EDPutTokenT<G> iToken, F iF, std::string productInstance = std::string()) {
using ReturnTypeT = decltype(iF(std::declval<G>()));
TypeID returnType(typeid(ReturnTypeT));
TransformerBase::registerTransformImp(*this,
EDPutToken(iToken),
returnType,
std::move(productInstance),
[f = std::move(iF)](edm::WrapperBase const& iGotProduct) {
return std::make_unique<edm::Wrapper<ReturnTypeT>>(
WrapperBase::Emplace{},
f(*static_cast<edm::Wrapper<G> const&>(iGotProduct).product()));
});
TransformerBase::registerTransformImp(
*this,
EDPutToken(iToken),
returnType,
std::move(productInstance),
[f = std::move(iF)](std::any const& iGotProduct) {
auto pGotProduct = std::any_cast<edm::WrapperBase const*>(iGotProduct);
return std::make_unique<edm::Wrapper<ReturnTypeT>>(
WrapperBase::Emplace{}, f(*static_cast<edm::Wrapper<G> const*>(pGotProduct)->product()));
});
}

template <typename G, typename P, typename F>
void registerTransformAsync(edm::EDPutTokenT<G> iToken,
P iPre,
F iF,
std::string productInstance = std::string()) {
using CacheTypeT = decltype(iPre(std::declval<G>(), WaitingTaskWithArenaHolder()));
using ReturnTypeT = decltype(iF(std::declval<CacheTypeT>()));
TypeID returnType(typeid(ReturnTypeT));
TransformerBase::registerTransformAsyncImp(
*this,
EDPutToken(iToken),
returnType,
std::move(productInstance),
[p = std::move(iPre)](edm::WrapperBase const& iGotProduct, WaitingTaskWithArenaHolder iHolder) {
return std::any(p(*static_cast<edm::Wrapper<G> const&>(iGotProduct).product(), std::move(iHolder)));
},
[f = std::move(iF)](std::any const& iCache) {
auto cache = std::any_cast<CacheTypeT>(iCache);
return std::make_unique<edm::Wrapper<ReturnTypeT>>(WrapperBase::Emplace{}, f(cache));
});
}

private:
Expand All @@ -467,8 +493,11 @@ namespace edm {
ProductResolverIndex transformPrefetch_(std::size_t iIndex) const final {
return TransformerBase::prefetchImp(iIndex);
}
void transform_(std::size_t iIndex, edm::EventForTransformer& iEvent) const final {
return TransformerBase::transformImp(iIndex, *this, iEvent);
void transformAsync_(WaitingTaskHolder iTask,
std::size_t iIndex,
edm::EventForTransformer& iEvent,
ServiceWeakToken const& iToken) const final {
return TransformerBase::transformImpAsync(std::move(iTask), iIndex, *this, iEvent);
}
void extendUpdateLookup(BranchType iBranchType, ProductResolverIndexHelper const& iHelper) override {
if (iBranchType == InEvent) {
Expand Down
6 changes: 5 additions & 1 deletion FWCore/Framework/interface/maker/Worker.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -269,7 +269,11 @@ namespace edm {
ModuleCallingContext const*,
WaitingTaskWithArenaHolder&) = 0;

virtual void implDoTransform(size_t iTransformIndex, EventPrincipal const&, ParentContext const&) = 0;
virtual void implDoTransformAsync(WaitingTaskHolder,
size_t iTransformIndex,
EventPrincipal const&,
ParentContext const&,
ServiceWeakToken const&) = 0;
virtual ProductResolverIndex itemToGetForTransform(size_t iTransformIndex) const = 0;

virtual bool implDoPrePrefetchSelection(StreamID, EventPrincipal const&, ModuleCallingContext const*) = 0;
Expand Down
6 changes: 5 additions & 1 deletion FWCore/Framework/interface/maker/WorkerT.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -93,7 +93,11 @@ namespace edm {
void implDoAcquire(EventTransitionInfo const&, ModuleCallingContext const*, WaitingTaskWithArenaHolder&) final;

size_t transformIndex(edm::BranchDescription const&) const final;
void implDoTransform(size_t iTransformIndex, EventPrincipal const&, ParentContext const&) final;
void implDoTransformAsync(WaitingTaskHolder,
size_t iTransformIndex,
EventPrincipal const&,
ParentContext const&,
ServiceWeakToken const&) final;
ProductResolverIndex itemToGetForTransform(size_t iTransformIndex) const final;

bool implDoPrePrefetchSelection(StreamID, EventPrincipal const&, ModuleCallingContext const*) override;
Expand Down
Loading