Skip to content

Commit 5b284ab

Browse files
schlangsterschlangster
committed
Using specialized versions of TurnBase and ReactiveInput for sequential and parallel operation to avoid concurrent_vectors when they are not required.
1 parent 4dbe8e2 commit 5b284ab

7 files changed

Lines changed: 197 additions & 43 deletions

File tree

‎include/react/detail/EngineBase.h‎

Lines changed: 82 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -10,6 +10,7 @@
1010

1111
#include <functional>
1212
#include <memory>
13+
#include <mutex>
1314
#include <utility>
1415
#include <vector>
1516

@@ -29,19 +30,90 @@
2930
///////////////////////////////////////////////////////////////////////////////////////////////////
3031
/// TurnBase
3132
///////////////////////////////////////////////////////////////////////////////////////////////////
33+
template <bool is_thread_safe>
3234
class TurnBase
3335
{
3436
public:
35-
inline TurnBase(TurnIdT id, TurnFlagsT flags) :
37+
TurnBase(TurnIdT id, TurnFlagsT flags);
38+
39+
TurnIdT Id() const;
40+
41+
void QueueForDetach(IObserver& obs);
42+
43+
template <typename D>
44+
friend class InputManager;
45+
46+
template <typename D>
47+
friend class ContinuationHolder;
48+
49+
private:
50+
using ContinuationT = ContinuationInput<is_thread_safe>;
51+
52+
template <typename D>
53+
void detachObservers();
54+
};
55+
56+
// Not thread-safe
57+
template <>
58+
class TurnBase<false>
59+
{
60+
public:
61+
TurnBase(TurnIdT id, TurnFlagsT flags) :
62+
id_{ id }
63+
{}
64+
65+
TurnIdT Id() const { return id_; }
66+
67+
void QueueForDetach(IObserver& obs)
68+
{
69+
detachedObservers_.push_back(&obs);
70+
}
71+
72+
template <typename D>
73+
friend class InputManager;
74+
75+
template <typename D>
76+
friend class ContinuationHolder;
77+
78+
private:
79+
using ObsVectT = std::vector<IObserver*>;
80+
using ContinuationT = ContinuationInput<false>;
81+
82+
TurnIdT id_;
83+
84+
template <typename D>
85+
void detachObservers()
86+
{
87+
88+
auto& registry = DomainSpecificObserverRegistry<D>::Instance();
89+
90+
for (auto* o : detachedObservers_)
91+
registry.Unregister(o);
92+
93+
detachedObservers_.clear();
94+
}
95+
96+
ObsVectT detachedObservers_;
97+
ContinuationT continuation_;
98+
};
99+
100+
// Thread-safe
101+
template <>
102+
class TurnBase<true>
103+
{
104+
public:
105+
TurnBase(TurnIdT id, TurnFlagsT flags) :
36106
id_{ id }
37107
{}
38108

39-
inline TurnIdT Id() const { return id_; }
109+
TurnIdT Id() const { return id_; }
40110

41-
inline void QueueForDetach(IObserver& obs)
111+
void QueueForDetach(IObserver& obs)
42112
{
43-
if (detachedObserversPtr_ == nullptr)
44-
detachedObserversPtr_ = std::unique_ptr<ObsVectT>(new ObsVectT());
113+
// Allocation of concurrent vector is not cheap -> create on demand
114+
std::call_once(detachedObserversInit_, [this] {
115+
detachedObserversPtr_.reset(new ObsVectT());
116+
});
45117

46118
detachedObserversPtr_->push_back(&obs);
47119
}
@@ -54,6 +126,7 @@ class TurnBase
54126

55127
private:
56128
using ObsVectT = tbb::concurrent_vector<IObserver*>;
129+
using ContinuationT = ContinuationInput<true>;
57130

58131
TurnIdT id_;
59132

@@ -66,11 +139,14 @@ class TurnBase
66139

67140
for (auto* o : *detachedObserversPtr_)
68141
registry.Unregister(o);
142+
143+
detachedObserversPtr_->clear();
69144
}
70145
}
71146

147+
std::once_flag detachedObserversInit_;
72148
std::unique_ptr<ObsVectT> detachedObserversPtr_;
73-
ContinuationInput continuation_;
149+
ContinuationT continuation_;
74150
};
75151

76152
///////////////////////////////////////////////////////////////////////////////////////////////////

‎include/react/detail/ReactiveInput.h‎

Lines changed: 77 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -27,34 +27,100 @@ using TurnFlagsT = uint;
2727
///////////////////////////////////////////////////////////////////////////////////////////////////
2828
/// ContinuationInput
2929
///////////////////////////////////////////////////////////////////////////////////////////////////
30+
template <bool is_thread_safe>
3031
class ContinuationInput
3132
{
33+
public:
34+
ContinuationInput() = default;
35+
ContinuationInput(const ContinuationInput&) = delete;
36+
ContinuationInput(ContinuationInput&&);
37+
38+
bool IsEmpty() const;
39+
40+
template <typename F>
41+
void Add(F&& input);
42+
43+
void Execute();
44+
};
45+
46+
// Not thread-safe
47+
template <>
48+
class ContinuationInput<false>
49+
{
50+
public:
51+
using InputClosureT = std::function<void()>;
52+
using InputVectT = std::vector<InputClosureT>;
53+
54+
ContinuationInput() = default;
55+
ContinuationInput(const ContinuationInput&) = delete;
56+
57+
ContinuationInput(ContinuationInput&& other) :
58+
bufferedInputs_{ std::move(other.bufferedInputs_) }
59+
{}
60+
61+
ContinuationInput& operator=(ContinuationInput&& other)
62+
{
63+
bufferedInputs_ = std::move(other.bufferedInputs_);
64+
return *this;
65+
}
66+
67+
bool IsEmpty() const { return bufferedInputs_.empty(); }
68+
69+
template <typename F>
70+
void Add(F&& input)
71+
{
72+
bufferedInputs_.push_back(std::forward<F>(input));
73+
}
74+
75+
void Execute()
76+
{
77+
for (auto& f : bufferedInputs_)
78+
f();
79+
bufferedInputs_.clear();
80+
}
81+
82+
private:
83+
InputVectT bufferedInputs_;
84+
};
85+
86+
// Thread-safe
87+
template <>
88+
class ContinuationInput<true>
89+
{
3290
public:
3391
using InputClosureT = std::function<void()>;
3492
using InputVectT = tbb::concurrent_vector<InputClosureT>;
93+
94+
ContinuationInput() = default;
95+
ContinuationInput(const ContinuationInput&) = delete;
96+
97+
ContinuationInput(ContinuationInput&& other) :
98+
bufferedInputsPtr_{ std::move(other.bufferedInputsPtr_) }
99+
{}
35100

36-
inline ContinuationInput& operator=(ContinuationInput&& other)
101+
ContinuationInput& operator=(ContinuationInput&& other)
37102
{
38103
bufferedInputsPtr_ = std::move(other.bufferedInputsPtr_);
39104
return *this;
40105
}
41106

42-
inline bool IsEmpty() const { return bufferedInputsPtr_ == nullptr; }
107+
bool IsEmpty() const { return bufferedInputsPtr_ == nullptr; }
43108

44109
template <typename F>
45110
void Add(F&& input)
46111
{
112+
// Allocation of concurrent vector is not cheap -> create on demand
47113
std::call_once(bufferedInputsInit_, [this] {
48114
bufferedInputsPtr_.reset(new InputVectT());
49115
});
50116
bufferedInputsPtr_->push_back(std::forward<F>(input));
51117
}
52118

53-
inline void Execute()
119+
void Execute()
54120
{
55121
if (bufferedInputsPtr_ != nullptr)
56122
{
57-
for (auto f : *bufferedInputsPtr_)
123+
for (auto& f : *bufferedInputsPtr_)
58124
f();
59125
bufferedInputsPtr_->clear();
60126
}
@@ -73,19 +139,20 @@ class ContinuationHolder
73139
{
74140
public:
75141
using TurnT = typename D::TurnT;
142+
using ContinuationT = ContinuationInput<D::uses_parallel_updating>;
76143

77144
ContinuationHolder() = delete;
78145

79-
static void SetTurn(TurnT& turn) { ptr_ = &turn.continuation_; }
80-
static void Clear() { ptr_ = nullptr; }
81-
static ContinuationInput* Get() { return ptr_; }
146+
static void SetTurn(TurnT& turn) { ptr_ = &turn.continuation_; }
147+
static void Clear() { ptr_ = nullptr; }
148+
static ContinuationT* Get() { return ptr_; }
82149

83150
private:
84-
static REACT_TLS ContinuationInput* ptr_;
151+
static REACT_TLS ContinuationT* ptr_;
85152
};
86153

87154
template <typename D>
88-
ContinuationInput* ContinuationHolder<D>::ptr_(nullptr);
155+
typename ContinuationHolder<D>::ContinuationT* ContinuationHolder<D>::ptr_(nullptr);
89156

90157
///////////////////////////////////////////////////////////////////////////////////////////////////
91158
/// InputManager
@@ -219,7 +286,7 @@ class InputManager
219286
processContinuations(std::move(turn.continuation_), 0);
220287
}
221288

222-
static void processContinuations(ContinuationInput&& cont, TurnFlagsT flags)
289+
static void processContinuations(typename TurnT::ContinuationT&& cont, TurnFlagsT flags)
223290
{
224291
// No merging for continuations
225292
flags &= ~enable_input_merging;

‎include/react/engine/PulseCountEngine.h‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -34,7 +34,7 @@ using tbb::task_list;
3434
///////////////////////////////////////////////////////////////////////////////////////////////////
3535
/// Turn
3636
///////////////////////////////////////////////////////////////////////////////////////////////////
37-
struct Turn : public TurnBase
37+
struct Turn : public TurnBase<true>
3838
{
3939
public:
4040
Turn(TurnIdT id, TurnFlagsT flags);

‎include/react/engine/SubtreeEngine.h‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -34,7 +34,7 @@ using tbb::spin_rw_mutex;
3434
///////////////////////////////////////////////////////////////////////////////////////////////////
3535
/// Turn
3636
///////////////////////////////////////////////////////////////////////////////////////////////////
37-
class Turn : public TurnBase
37+
class Turn : public TurnBase<true>
3838
{
3939
public:
4040
Turn(TurnIdT id, TurnFlagsT flags);

‎include/react/engine/ToposortEngine.h‎

Lines changed: 17 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -30,6 +30,7 @@
3030

3131
/***************************************/ REACT_IMPL_BEGIN /**************************************/
3232

33+
template <bool is_thread_safe>
3334
class TurnBase;
3435

3536
namespace toposort {
@@ -92,13 +93,21 @@ struct DynRequestData
9293
};
9394

9495
///////////////////////////////////////////////////////////////////////////////////////////////////
95-
/// ExclusiveTurn
96+
/// ExclusiveSeqTurn
9697
///////////////////////////////////////////////////////////////////////////////////////////////////
98+
class ExclusiveSeqTurn : public REACT_IMPL::TurnBase<false>
99+
{
100+
public:
101+
ExclusiveSeqTurn(TurnIdT id, TurnFlagsT flags);
102+
};
97103

98-
class ExclusiveTurn : public REACT_IMPL::TurnBase
104+
///////////////////////////////////////////////////////////////////////////////////////////////////
105+
/// ExclusiveParTurn
106+
///////////////////////////////////////////////////////////////////////////////////////////////////
107+
class ExclusiveParTurn : public REACT_IMPL::TurnBase<true>
99108
{
100109
public:
101-
ExclusiveTurn(TurnIdT id, TurnFlagsT flags);
110+
ExclusiveParTurn(TurnIdT id, TurnFlagsT flags);
102111
};
103112

104113
///////////////////////////////////////////////////////////////////////////////////////////////////
@@ -191,16 +200,16 @@ class ParEngineBase : public EngineBase<ParNode,TTurn>
191200
///////////////////////////////////////////////////////////////////////////////////////////////////
192201
/// Concrete engines
193202
///////////////////////////////////////////////////////////////////////////////////////////////////
194-
class BasicSeqEngine : public SeqEngineBase<ExclusiveTurn> {};
195-
class QueuingSeqEngine : public DefaultQueuingEngine<SeqEngineBase,ExclusiveTurn> {};
203+
class BasicSeqEngine : public SeqEngineBase<ExclusiveSeqTurn> {};
204+
class QueuingSeqEngine : public DefaultQueuingEngine<SeqEngineBase,ExclusiveSeqTurn> {};
196205

197-
class BasicParEngine : public ParEngineBase<ExclusiveTurn> {};
198-
class QueuingParEngine : public DefaultQueuingEngine<ParEngineBase,ExclusiveTurn> {};
206+
class BasicParEngine : public ParEngineBase<ExclusiveParTurn> {};
207+
class QueuingParEngine : public DefaultQueuingEngine<ParEngineBase,ExclusiveParTurn> {};
199208

200209
///////////////////////////////////////////////////////////////////////////////////////////////////
201210
/// PipeliningTurn
202211
///////////////////////////////////////////////////////////////////////////////////////////////////
203-
class PipeliningTurn : public TurnBase
212+
class PipeliningTurn : public TurnBase<true>
204213
{
205214
public:
206215
using ConcNodeVectT = concurrent_vector<ParNode*>;

‎src/engine/SubtreeEngine.cpp‎

Lines changed: 1 addition & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -26,8 +26,7 @@ static const uint heavy_weight = 1000;
2626
///////////////////////////////////////////////////////////////////////////////////////////////////
2727
Turn::Turn(TurnIdT id, TurnFlagsT flags) :
2828
TurnBase(id, flags)
29-
{
30-
}
29+
{}
3130

3231
///////////////////////////////////////////////////////////////////////////////////////////////////
3332
/// PulsecountEngine
@@ -212,8 +211,6 @@ void EngineBase<TTurn>::OnNodeIdlePulse(Node& node, TTurn& turn)
212211
{
213212
if (isInPhase2_)
214213
node.ClearChangedFlag();
215-
else
216-
processChildren(node, turn);
217214
}
218215

219216
template <typename TTurn>
@@ -270,7 +267,6 @@ void EngineBase<TTurn>::applyAsyncDynamicAttach(Node& node, Node& parent, TTurn&
270267
node.SetShouldUpdate(true);
271268

272269
node.WaitCount = 1;
273-
274270
}
275271
}// ~parent.ShiftMutex (write)
276272

0 commit comments

Comments
 (0)