Skip to content

Commit 12d72a5

Browse files
schlangsterschlangster
committed
Added a new range-based signature option for event callbacks.
Instead of passing a function `func(E)` that is called for every event on the stream, EventRange<E> allows to iterate them manually. This allows client code to add node-level parallelization with parallel_for or parallel_reduce. It's supported by existing Observe and MakeContinuation, and also by a new operation `Process`, which is a more generic form of Transform/Filter. The signature of process callbacks is `void(EventRange<int> range, EventInserter<float> out)`. out is a back_insert_iterator.
1 parent 7b6009e commit 12d72a5

7 files changed

Lines changed: 415 additions & 97 deletions

File tree

‎include/react/Domain.h‎

Lines changed: 24 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -235,10 +235,21 @@ auto MakeContinuation(TransactionFlagsT flags, const Events<D,E>& trigger, FIn&&
235235
"MakeContinuation requires support for concurrent input to target domain.");
236236

237237
using REACT_IMPL::EventContinuationNode;
238+
using REACT_IMPL::AddContinuationRangeWrapper;
239+
using REACT_IMPL::IsCallableWith;
240+
using REACT_IMPL::EventRange;
241+
238242
using F = typename std::decay<FIn>::type;
239243

244+
using WrapperT =
245+
typename std::conditional<
246+
IsCallableWith<F,void, EventRange<E>>::value,
247+
F,
248+
AddContinuationRangeWrapper<E, F>
249+
>::type;
250+
240251
return Continuation<D,DOut>(
241-
std::make_shared<EventContinuationNode<D,DOut,E,F>>(
252+
std::make_shared<EventContinuationNode<D,DOut,E,WrapperT>>(
242253
flags, GetNodePtr(trigger), std::forward<FIn>(func)));
243254
}
244255

@@ -274,8 +285,19 @@ auto MakeContinuation(TransactionFlagsT flags, const Events<D,E>& trigger,
274285
"MakeContinuation requires support for concurrent input to target domain.");
275286

276287
using REACT_IMPL::SyncedContinuationNode;
288+
using REACT_IMPL::AddContinuationRangeWrapper;
289+
using REACT_IMPL::IsCallableWith;
290+
using REACT_IMPL::EventRange;
291+
277292
using F = typename std::decay<FIn>::type;
278293

294+
using WrapperT =
295+
typename std::conditional<
296+
IsCallableWith<F, void, EventRange<E>, TDepValues ...>::value,
297+
F,
298+
AddContinuationRangeWrapper<E, F, TDepValues ...>
299+
>::type;
300+
279301
struct NodeBuilder_
280302
{
281303
NodeBuilder_(TransactionFlagsT flags, const Events<D,E>& trigger, FIn&& func) :
@@ -288,7 +310,7 @@ auto MakeContinuation(TransactionFlagsT flags, const Events<D,E>& trigger,
288310
-> Continuation<D,DOut>
289311
{
290312
return Continuation<D,DOut>(
291-
std::make_shared<SyncedContinuationNode<D,DOut,E,F,TDepValues...>>(
313+
std::make_shared<SyncedContinuationNode<D,DOut,E,WrapperT,TDepValues...>>(
292314
MyFlags,
293315
GetNodePtr(MyTrigger),
294316
std::forward<FIn>(MyFunc), GetNodePtr(deps) ...));

‎include/react/Event.h‎

Lines changed: 66 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -345,6 +345,72 @@ auto Transform(const Events<D,TIn>& source, const SignalPack<D,TDepValues...>& d
345345
depPack.Data);
346346
}
347347

348+
///////////////////////////////////////////////////////////////////////////////////////////////////
349+
/// Process
350+
///////////////////////////////////////////////////////////////////////////////////////////////////
351+
using REACT_IMPL::EventRange;
352+
using REACT_IMPL::EventInserter;
353+
354+
template
355+
<
356+
typename TOut,
357+
typename D,
358+
typename TIn,
359+
typename FIn,
360+
typename F = typename std::decay<FIn>::type
361+
>
362+
auto Process(const Events<D,TIn>& src, FIn&& func)
363+
-> Events<D,TOut>
364+
{
365+
using REACT_IMPL::EventProcessingNode;
366+
367+
return Events<D,TOut>(
368+
std::make_shared<EventProcessingNode<D,TIn,TOut,F>>(
369+
GetNodePtr(src), std::forward<FIn>(func)));
370+
}
371+
372+
///////////////////////////////////////////////////////////////////////////////////////////////////
373+
/// Process - Synced
374+
///////////////////////////////////////////////////////////////////////////////////////////////////
375+
template
376+
<
377+
typename TOut,
378+
typename D,
379+
typename TIn,
380+
typename FIn,
381+
typename ... TDepValues
382+
>
383+
auto Process(const Events<D,TIn>& source, const SignalPack<D,TDepValues...>& depPack, FIn&& func)
384+
-> Events<D,TOut>
385+
{
386+
using REACT_IMPL::SyncedEventProcessingNode;
387+
388+
using F = typename std::decay<FIn>::type;
389+
390+
struct NodeBuilder_
391+
{
392+
NodeBuilder_(const Events<D,TIn>& source, FIn&& func) :
393+
MySource( source ),
394+
MyFunc( std::forward<FIn>(func) )
395+
{}
396+
397+
auto operator()(const Signal<D,TDepValues>& ... deps)
398+
-> Events<D,TOut>
399+
{
400+
return Events<D,TOut>(
401+
std::make_shared<SyncedEventProcessingNode<D,TIn,TOut,F,TDepValues ...>>(
402+
GetNodePtr(MySource), std::forward<FIn>(MyFunc), GetNodePtr(deps) ...));
403+
}
404+
405+
const Events<D,TIn>& MySource;
406+
FIn MyFunc;
407+
};
408+
409+
return REACT_IMPL::apply(
410+
NodeBuilder_( source, std::forward<FIn>(func) ),
411+
depPack.Data);
412+
}
413+
348414
///////////////////////////////////////////////////////////////////////////////////////////////////
349415
/// Flatten
350416
///////////////////////////////////////////////////////////////////////////////////////////////////

‎include/react/Observer.h‎

Lines changed: 41 additions & 19 deletions
Original file line numberDiff line numberDiff line change
@@ -185,7 +185,6 @@ auto Observe(const Signal<D,S>& subject, FIn&& func)
185185
SignalObserverNode<D,S,F>
186186
>::type;
187187

188-
189188
const auto& subjectPtr = GetNodePtr(subject);
190189

191190
std::unique_ptr<ObserverNode<D>> nodePtr( new TNode(subjectPtr, std::forward<FIn>(func)) );
@@ -212,18 +211,29 @@ auto Observe(const Events<D,E>& subject, FIn&& func)
212211
using REACT_IMPL::ObserverNode;
213212
using REACT_IMPL::EventObserverNode;
214213
using REACT_IMPL::AddDefaultReturnValueWrapper;
214+
using REACT_IMPL::AddObserverRangeWrapper;
215+
using REACT_IMPL::IsCallableWith;
216+
using REACT_IMPL::EventRange;
215217

216218
using F = typename std::decay<FIn>::type;
217-
using R = typename std::result_of<FIn(E)>::type;
218-
using WrapperT = AddDefaultReturnValueWrapper<F,ObserverAction,ObserverAction::next>;
219219

220-
// If return value of passed function is void, add ObserverAction::next as
221-
// default return value.
222-
using TNode = typename std::conditional<
223-
std::is_same<void,R>::value,
224-
EventObserverNode<D,E,WrapperT>,
225-
EventObserverNode<D,E,F>
226-
>::type;
220+
using WrapperT =
221+
typename std::conditional<
222+
IsCallableWith<F, ObserverAction, EventRange<E>>::value,
223+
F,
224+
typename std::conditional<
225+
IsCallableWith<F, ObserverAction, E>::value,
226+
AddObserverRangeWrapper<E, F>,
227+
typename std::conditional<
228+
IsCallableWith<F, void, EventRange<E>>::value,
229+
AddDefaultReturnValueWrapper<F,ObserverAction,ObserverAction::next>,
230+
AddObserverRangeWrapper<E,
231+
AddDefaultReturnValueWrapper<F,ObserverAction,ObserverAction::next>>
232+
>::type
233+
>::type
234+
>::type;
235+
236+
using TNode = EventObserverNode<D,E,WrapperT>;
227237

228238
const auto& subjectPtr = GetNodePtr(subject);
229239

@@ -253,18 +263,30 @@ auto Observe(const Events<D,E>& subject,
253263
using REACT_IMPL::ObserverNode;
254264
using REACT_IMPL::SyncedObserverNode;
255265
using REACT_IMPL::AddDefaultReturnValueWrapper;
266+
using REACT_IMPL::AddObserverRangeWrapper;
267+
using REACT_IMPL::IsCallableWith;
268+
using REACT_IMPL::EventRange;
256269

257270
using F = typename std::decay<FIn>::type;
258-
using R = typename std::result_of<FIn(E,TDepValues...)>::type;
259-
using WrapperT = AddDefaultReturnValueWrapper<F,ObserverAction,ObserverAction::next>;
260271

261-
// If return value of passed function is void, add ObserverAction::next as
262-
// default return value.
263-
using TNode = typename std::conditional<
264-
std::is_same<void,R>::value,
265-
SyncedObserverNode<D,E,WrapperT,TDepValues ...>,
266-
SyncedObserverNode<D,E,F,TDepValues ...>
267-
>::type;
272+
using WrapperT =
273+
typename std::conditional<
274+
IsCallableWith<F, ObserverAction, EventRange<E>, TDepValues ...>::value,
275+
F,
276+
typename std::conditional<
277+
IsCallableWith<F, ObserverAction, E, TDepValues ...>::value,
278+
AddObserverRangeWrapper<E, F, TDepValues ...>,
279+
typename std::conditional<
280+
IsCallableWith<F, void, EventRange<E>, TDepValues ...>::value,
281+
AddDefaultReturnValueWrapper<F, ObserverAction ,ObserverAction::next>,
282+
AddObserverRangeWrapper<E,
283+
AddDefaultReturnValueWrapper<F,ObserverAction,ObserverAction::next>,
284+
TDepValues...>
285+
>::type
286+
>::type
287+
>::type;
288+
289+
using TNode = SyncedObserverNode<D,E,WrapperT,TDepValues ...>;
268290

269291
struct NodeBuilder_
270292
{

‎include/react/detail/graph/ContinuationNodes.h‎

Lines changed: 37 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -30,6 +30,36 @@ class SignalNode;
3030
template <typename D, typename E>
3131
class EventStreamNode;
3232

33+
///////////////////////////////////////////////////////////////////////////////////////////////////
34+
/// AddContinuationRangeWrapper
35+
///////////////////////////////////////////////////////////////////////////////////////////////////
36+
template <typename E, typename F, typename ... TArgs>
37+
struct AddContinuationRangeWrapper
38+
{
39+
AddContinuationRangeWrapper(const AddContinuationRangeWrapper& other) = default;
40+
41+
AddContinuationRangeWrapper(AddContinuationRangeWrapper&& other) :
42+
MyFunc( std::move(other.MyFunc) )
43+
{}
44+
45+
template
46+
<
47+
typename FIn,
48+
class = typename DisableIfSame<FIn,AddContinuationRangeWrapper>::type
49+
>
50+
explicit AddContinuationRangeWrapper(FIn&& func) :
51+
MyFunc( std::forward<FIn>(func) )
52+
{}
53+
54+
void operator()(EventRange<E> range, const TArgs& ... args)
55+
{
56+
for (const auto& e : range)
57+
MyFunc(e, args ...);
58+
}
59+
60+
F MyFunc;
61+
};
62+
3363
///////////////////////////////////////////////////////////////////////////////////////////////////
3464
/// ContinuationNode
3565
///////////////////////////////////////////////////////////////////////////////////////////////////
@@ -171,13 +201,10 @@ class EventContinuationNode : public ContinuationNode<D>
171201
// Copy events and func
172202
[storedFunc,storedEvents] () mutable
173203
{
174-
for (const auto& e : storedEvents)
175-
storedFunc(e);
204+
storedFunc(EventRange<E>( storedEvents ));
176205
}
177206
);
178207

179-
TransactionFuncT cont2 = [] { return; };
180-
181208
DomainSpecificInputManager<D>::Instance()
182209
.StoreContinuation(
183210
DomainSpecificInputManager<DOut>::Instance(),
@@ -291,10 +318,12 @@ class SyncedContinuationNode : public ContinuationNode<D>
291318
// Copy events, func, value tuple (note: 2x copy)
292319
[storedFunc,storedEvents,storedValues] () mutable
293320
{
294-
for (const auto& e : storedEvents)
295-
{
296-
apply(EvalFunctor_( e, storedFunc ), storedValues);
297-
}
321+
apply(
322+
[&storedFunc,&storedEvents] (const TDepValues& ... vals)
323+
{
324+
storedFunc(EventRange<E>( storedEvents ), vals ...);
325+
},
326+
storedValues);
298327
}
299328
};
300329

0 commit comments

Comments
 (0)