Skip to content

Commit

Permalink
Added transformAsync ability to modules
Browse files Browse the repository at this point in the history 
This is similar to transform except two actions are specified similar to acquire/produce.
  • Loading branch information
@Dr15Jones
Dr15Jones committed Sep 30, 2022
1 parent 6aa1459 commit 62340cc
Show file tree
Hide file tree
Showing 33 changed files with 656 additions and 191 deletions.
15 changes: 12 additions & 3 deletions FWCore/Framework/interface/TransformerBase.h
View file
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,26 @@ 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 const&)>;
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;
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_;
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
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
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
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)));
},
[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 @@ -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
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
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
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
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
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

0 comments on commit 62340cc

Please sign in to comment.