FazBrowse GitHub Viewer | Trending |
URL:
| Home
Tools: [Download Repo ZIP]   [Original HTTPS Page]

Refactored continuatons. Having a single active turn means it's no lo… · jchjava/cpp.react@55d76c1 · GitHub

/ cpp.react Public
forked from snakster/cpp.react

Commit 55d76c1

Browse files
schlangster
committed
Refactored continuatons. Having a single active turn means it's no longer necessary to store continuation input and detached observers per turn, but rather per input manager.
1 parent 3433c2c commit 55d76c1

6 files changed

Lines changed: 270 additions & 284 deletions

File tree

‎include/react/Observer.h‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -248,7 +248,7 @@ void DetachAllObservers(const TSubject& subject)
248248
///////////////////////////////////////////////////////////////////////////////////////////////////
249249
inline void DetachThisObserver()
250250
{
251-
REACT_IMPL::GlobalObserverState<>::ShouldDetach = true;
251+
REACT_IMPL::ThreadLocalObserverState<>::ShouldDetach = true;
252252
}
253253

254254
/******************************************/ REACT_END /******************************************/

‎include/react/detail/EngineBase.h‎

Lines changed: 48 additions & 126 deletions
Original file line numberDiff line numberDiff line change
@@ -17,7 +17,6 @@
1717
#include <utility>
1818
#include <vector>
1919

20-
#include "tbb/concurrent_vector.h"
2120
#include "tbb/queuing_mutex.h"
2221

2322
#include "react/common/Concurrency.h"
@@ -26,7 +25,6 @@
2625
#include "react/detail/IReactiveNode.h"
2726
#include "react/detail/IReactiveEngine.h"
2827
#include "react/detail/ObserverBase.h"
29-
#include "react/detail/Options.h"
3028

3129
/***************************************/ REACT_IMPL_BEGIN /**************************************/
3230

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

11242
TurnIdT Id() const { return id_; }
11343

114-
void QueueForDetach(IObserver& obs)
115-
{
116-
// Allocation of concurrent vector is not cheap -> create on demand
117-
std::call_once(detachedObserversInit_, [this] {
118-
detachedObserversPtr_.reset(new ObsVectT());
119-
});
120-
121-
detachedObserversPtr_->push_back(&obs);
122-
}
123-
124-
template <typename D>
125-
friend class InputManager;
126-
127-
template <typename D>
128-
friend class ContinuationHolder;
129-
13044
private:
131-
using ObsVectT = tbb::concurrent_vector<IObserver*>;
132-
using ContinuationT = ContinuationInput<true>;
133-
13445
TurnIdT id_;
135-
136-
template <typename D>
137-
void detachObservers()
138-
{
139-
if (detachedObserversPtr_ != nullptr)
140-
{
141-
auto& registry = DomainSpecificObserverRegistry<D>::Instance();
142-
143-
for (auto* o : *detachedObserversPtr_)
144-
registry.Unregister(o);
145-
146-
detachedObserversPtr_->clear();
147-
}
148-
}
149-
150-
std::once_flag detachedObserversInit_;
151-
std::unique_ptr<ObsVectT> detachedObserversPtr_;
152-
ContinuationT continuation_;
15346
};
15447

15548
///////////////////////////////////////////////////////////////////////////////////////////////////
@@ -179,21 +72,33 @@ class TurnQueueManager
17972
inline void RunMergedInputs() const
18073
{
18174
for (const auto& e : merged_)
182-
e.first();
75+
e.InputFunc();
18376
}
18477

18578
inline void UnblockSuccessors()
18679
{
80+
// Release merged
18781
for (const auto& e : merged_)
188-
if (e.second != nullptr)
189-
e.second->Unblock();
82+
{
83+
// Note: Since a merged input is either sync or async,
84+
// either cond or status will be null
19085

86+
// Sync
87+
if (e.Cond != nullptr)
88+
e.Cond->Unblock();
89+
90+
// Async
91+
else if (e.Status != nullptr)
92+
TransactionStatusInterface::DecrementWaitCount(*e.Status);
93+
}
94+
95+
// Release next thread in queue
19196
if (successor_)
19297
successor_->blockCondition_.Unblock();
19398
}
19499

195100
template <typename F>
196-
inline bool TryMerge(F&& inputFunc, BlockingCondition* caller)
101+
inline bool TryMerge(F&& inputFunc, BlockingCondition* caller, TransactionStatus* status)
197102
{
198103
if (!isMergeable_)
199104
return false;
@@ -202,18 +107,32 @@ class TurnQueueManager
202107
bool merged = blockCondition_.RunIfBlocked([&] {
203108
if (caller)
204109
caller->Block();
205-
merged_.emplace_back(std::make_pair(std::forward<F>(inputFunc), caller));
110+
merged_.emplace_back(std::forward<F>(inputFunc), caller, status);
206111
});
207112

208113
return merged;
209114
}
210115

211116
private:
212-
using MergedDataVectT =
213-
std::vector<
214-
std::pair<
215-
std::function<void()>,
216-
BlockingCondition*>>;
117+
struct MergedData
118+
{
119+
template <typename F>
120+
MergedData(F&& func, BlockingCondition* cond, TransactionStatus* status) :
121+
InputFunc( std::forward<F>(func) ),
122+
Cond( cond ),
123+
Status( status )
124+
{}
125+
126+
std::function<void()> InputFunc;
127+
128+
// Blocking condition variable for sync merged
129+
BlockingCondition* Cond;
130+
131+
// Status for async merged
132+
TransactionStatus* Status;
133+
};
134+
135+
using MergedDataVectT = std::vector<MergedData>;
217136

218137
bool isMergeable_;
219138
QueueEntry* successor_ = nullptr;
@@ -222,7 +141,7 @@ class TurnQueueManager
222141
};
223142

224143
template <typename F>
225-
inline bool TryMerge(F&& inputFunc)
144+
inline bool TryMergeSync(F&& inputFunc)
226145
{
227146
bool merged = false;
228147

@@ -232,7 +151,7 @@ class TurnQueueManager
232151
SeqMutexT::scoped_lock lock(seqMutex_);
233152

234153
if (tail_)
235-
merged = tail_->TryMerge(std::forward<F>(inputFunc), &caller);
154+
merged = tail_->TryMerge(std::forward<F>(inputFunc), &caller, nullptr);
236155
}// ~seqMutex_
237156

238157
if (merged)
@@ -242,15 +161,15 @@ class TurnQueueManager
242161
}
243162

244163
template <typename F>
245-
inline bool TryMergeAsync(F&& inputFunc)
164+
inline bool TryMergeAsync(F&& inputFunc, TransactionStatus* status)
246165
{
247166
bool merged = false;
248167

249168
{// seqMutex_
250169
SeqMutexT::scoped_lock lock(seqMutex_);
251170

252171
if (tail_)
253-
merged = tail_->TryMerge(std::forward<F>(inputFunc), nullptr);
172+
merged = tail_->TryMerge(std::forward<F>(inputFunc), nullptr, status);
254173
}// ~seqMutex_
255174

256175
return merged;
@@ -316,12 +235,15 @@ class DefaultQueuingEngine : public TTEngineBase<DefaultQueueableTurn<TTurnBase>
316235
using TurnT = DefaultQueueableTurn<TTurnBase>;
317236

318237
template <typename F>
319-
bool TryMergeInput(F&& f, bool isBlocking)
238+
bool TryMergeSync(F&& f)
239+
{
240+
return queueManager_.TryMergeSync(std::forward<F>(f));
241+
}
242+
243+
template <typename F>
244+
bool TryMergeAsync(F&& f, TransactionStatus* statusPtr)
320245
{
321-
if (isBlocking)
322-
return queueManager_.TryMerge(std::forward<F>(f));
323-
else
324-
return queueManager_.TryMergeAsync(std::forward<F>(f));
246+
return queueManager_.TryMergeAsync(std::forward<F>(f), statusPtr);
325247
}
326248

327249
void ApplyMergedInputs(TurnT& turn)

‎include/react/detail/IReactiveEngine.h‎

Lines changed: 17 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -17,6 +17,11 @@
1717

1818
/***************************************/ REACT_IMPL_BEGIN /**************************************/
1919

20+
///////////////////////////////////////////////////////////////////////////////////////////////////
21+
/// Forward declarations
22+
///////////////////////////////////////////////////////////////////////////////////////////////////
23+
class TransactionStatus;
24+
2025
///////////////////////////////////////////////////////////////////////////////////////////////////
2126
/// IReactiveEngine
2227
///////////////////////////////////////////////////////////////////////////////////////////////////
@@ -31,7 +36,10 @@ struct IReactiveEngine
3136
using TurnT = TTurn;
3237

3338
template <typename F>
34-
bool TryMergeInput(F&& f, bool isBlocking) { return false; }
39+
bool TryMergeSync(F&& f) { return false; }
40+
41+
template <typename F>
42+
bool TryMergeAsync(F&& f, TransactionStatus* status) { return false; }
3543

3644
void ApplyMergedInputs(TurnT& turn) {}
3745

@@ -78,9 +86,15 @@ struct EngineInterface
7886
}
7987

8088
template <typename F>
81-
static bool TryMergeInput(F&& f, bool isBlocking)
89+
static bool TryMergeSync(F&& f)
90+
{
91+
return Instance().TryMergeSync(std::forward<F>(f));
92+
}
93+
94+
template <typename F>
95+
static bool TryMergeAsync(F&& f, TransactionStatus* status)
8296
{
83-
return Instance().TryMergeInput(std::forward<F>(f), isBlocking);
97+
return Instance().TryMergeAsync(std::forward<F>(f), status);
8498
}
8599

86100
static void ApplyMergedInputs(TurnT& turn)

‎include/react/detail/ObserverBase.h‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -40,13 +40,13 @@ class IObserver
4040

4141
// tbb tasks are non-preemptible, thread local flag for each worker
4242
template <typename = void>
43-
struct GlobalObserverState
43+
struct ThreadLocalObserverState
4444
{
4545
static REACT_TLS bool ShouldDetach;
4646
};
4747

4848
template <typename T>
49-
REACT_TLS bool GlobalObserverState<T>::ShouldDetach(false);
49+
REACT_TLS bool ThreadLocalObserverState<T>::ShouldDetach(false);
5050

5151
///////////////////////////////////////////////////////////////////////////////////////////////////
5252
/// ObserverRegistry

0 commit comments

Comments
 (0)
Back | FazBrowse Home | New Git URL