##// END OF EJS Templates
HG_REPOSITORIES » LPP » INSTRUMENTATION » USERS » JEANDET » SciQLOPCore
RSS
Moved all maps in VC2 to a dedicated struct to ease thread safety more...
jeandet -
r19:32323038b7f8
parent child
Show More
Show file before | Show file after | Hide comments
@@ -64,6 +64,11 public:
64 return this->_variables;
64 return this->_variables;
65 }
65 }
66
66
67 inline bool isEmpty()
68 {
69 return _variables.size()==0;
70 }
71
67 inline QUuid ID(){return _ID;}
72 inline QUuid ID(){return _ID;}
68
73
69 operator QUuid() {return _ID;}
74 operator QUuid() {return _ID;}
Show file before | Show file after | Hide comments
@@ -14,13 +14,27
14
14
15 struct VCTransaction
15 struct VCTransaction
16 {
16 {
17 VCTransaction(QUuid refVar, DateTimeRange range, int ready)
17 VCTransaction(QUuid refVar, DateTimeRange range, int varCount)
18 :refVar{refVar},range{range},ready{ready}
18 :refVar{refVar},range{range},_remainingVars{varCount}
19 {}
19 {}
20
20 QUuid refVar;
21 QUuid refVar;
21 DateTimeRange range;
22 DateTimeRange range;
22 int ready;
23 bool ready()
23 QReadWriteLock lock;
24 {
25 QReadLocker lock{&_lock};
26 return _remainingVars == 0;
27 }
28
29 bool done()
30 {
31 QWriteLocker lock{&_lock};
32 _remainingVars-=1;
33 return _remainingVars == 0;
34 }
35 private:
36 QReadWriteLock _lock;
37 int _remainingVars;
24 };
38 };
25
39
26 class TransactionExe:public QObject,public QRunnable
40 class TransactionExe:public QObject,public QRunnable
Show file before | Show file after | Hide comments
@@ -73,39 +73,99 public:
73 }
73 }
74 };
74 };
75
75
76
76 class VariableController2::VariableController2Private
77 class VariableController2::VariableController2Private
77 {
78 {
79 struct threadSafeVaraiblesMaps
80 {
81 inline void addVariable(const std::shared_ptr<Variable>& variable, const std::shared_ptr<IDataProvider>& provider, const std::shared_ptr<VariableSynchronizationGroup2>& synchronizationGroup)
82 {
83 QWriteLocker lock{&_lock};
84 _variables[*variable] = variable;
85 _providers[*variable] = provider;
86 _synchronizationGroups[*variable] = synchronizationGroup;
87 }
88
89 inline void removeVariable(const std::shared_ptr<Variable>& variable)
90 {
91 QWriteLocker lock{&_lock};
92 _variables.remove(*variable);
93 _providers.remove(*variable);
94 _synchronizationGroups.remove(*variable);
95 }
96
97 inline void synchronize(const std::shared_ptr<Variable>& variable, const std::optional<std::shared_ptr<Variable>>& with)
98 {
99 QWriteLocker lock{&_lock};
100 if(with.has_value())
101 {
102 auto newGroup = _synchronizationGroups[*with.value()];
103 newGroup->addVariable(*variable);
104 _synchronizationGroups[*variable] = newGroup;
105 }
106 else
107 {
108 _synchronizationGroups[*variable] = std::make_shared<VariableSynchronizationGroup2>(*variable);
109 }
110 }
111
112 inline std::shared_ptr<Variable> variable(QUuid variable)
113 {
114 QReadLocker lock{&_lock};
115 [[unlikely]]
116 if(!_variables.contains(variable))
117 SCIQLOP_ERROR(threadSafeVaraiblesMaps,"Unknown Variable");
118 return _variables[variable];
119 }
120
121 inline const std::set<std::shared_ptr<Variable>> variables()
122 {
123 std::set<std::shared_ptr<Variable>> vars;
124 QReadLocker lock{&_lock};
125 for(const auto &var:_variables)
126 {
127 vars.insert(var);
128 }
129 return vars;
130 }
131
132 inline std::shared_ptr<IDataProvider> provider(QUuid variable)
133 {
134 QReadLocker lock{&_lock};
135 [[unlikely]]
136 if(!_providers.contains(variable))
137 SCIQLOP_ERROR(threadSafeVaraiblesMaps,"Unknown Variable");
138 return _providers[variable];
139 }
140
141 inline std::shared_ptr<VariableSynchronizationGroup2> group(QUuid variable)
142 {
143 QReadLocker lock{&_lock};
144 [[unlikely]]
145 if(!_synchronizationGroups.contains(variable))
146 SCIQLOP_ERROR(threadSafeVaraiblesMaps,"Unknown Variable");
147 return _synchronizationGroups[variable];
148 }
149
150 inline bool has(const std::shared_ptr<Variable>& variable)
151 {
152 QReadLocker lock{&_lock};
153 return _variables.contains(*variable);
154 }
155
156 private:
157 QMap<QUuid,std::shared_ptr<Variable>> _variables;
158 QMap<QUuid,std::shared_ptr<IDataProvider>> _providers;
159 QMap<QUuid,std::shared_ptr<VariableSynchronizationGroup2>> _synchronizationGroups;
160 QReadWriteLock _lock{QReadWriteLock::Recursive};
161 }_maps;
78 QThreadPool _ThreadPool;
162 QThreadPool _ThreadPool;
79 QMap<QUuid,std::shared_ptr<Variable>> _variables;
80 QMap<QUuid,std::shared_ptr<IDataProvider>> _providers;
81 QMap<QUuid,std::shared_ptr<VariableSynchronizationGroup2>> _synchronizationGroups;
82 Transactions _transactions;
163 Transactions _transactions;
83 QReadWriteLock _lock{QReadWriteLock::Recursive};
84 std::unique_ptr<VariableCacheStrategy> _cacheStrategy;
164 std::unique_ptr<VariableCacheStrategy> _cacheStrategy;
85
165
86 bool p_contains(const std::shared_ptr<Variable>& variable)
87 {
88 QReadLocker lock{&_lock};
89 return _providers.contains(*variable);
90 }
91 bool v_contains(const std::shared_ptr<Variable>& variable)
92 {
93 QReadLocker lock{&_lock};
94 return SciQLop::containers::contains(this->_variables, variable);
95 }
96 bool sg_contains(const std::shared_ptr<Variable>& variable)
97 {
98 QReadLocker lock{&_lock};
99 return _synchronizationGroups.contains(*variable);
100 }
101
102 void _transactionComplete(QUuid group, std::shared_ptr<VCTransaction> transaction)
166 void _transactionComplete(QUuid group, std::shared_ptr<VCTransaction> transaction)
103 {
167 {
104 {
168 if(transaction->done())
105 QWriteLocker lock{&transaction->lock};
106 transaction->ready -=1;
107 }
108 if(transaction->ready==0)
109 {
169 {
110 _transactions.complete(group);
170 _transactions.complete(group);
111 }
171 }
@@ -113,7 +173,6 class VariableController2::VariableController2Private
113 }
173 }
114 void _processTransactions()
174 void _processTransactions()
115 {
175 {
116 QWriteLocker lock{&_lock};
117 auto nextTransactions = _transactions.nextTransactions();
176 auto nextTransactions = _transactions.nextTransactions();
118 auto pendingTransactions = _transactions.pendingTransactions();
177 auto pendingTransactions = _transactions.pendingTransactions();
119 for( auto [groupID, newTransaction] : nextTransactions)
178 for( auto [groupID, newTransaction] : nextTransactions)
@@ -121,14 +180,14 class VariableController2::VariableController2Private
121 if(newTransaction.has_value() && !pendingTransactions[groupID].has_value())
180 if(newTransaction.has_value() && !pendingTransactions[groupID].has_value())
122 {
181 {
123 _transactions.start(groupID);
182 _transactions.start(groupID);
124 auto refVar = _variables[newTransaction.value()->refVar];
183 auto refVar = _maps.variable(newTransaction.value()->refVar);
125 auto ranges = _computeAllRangesInGroup(refVar,newTransaction.value()->range);
184 auto ranges = _computeAllRangesInGroup(refVar,newTransaction.value()->range);
126 for( auto const& [ID, range] : ranges)
185 for( auto const& [ID, range] : ranges)
127 {
186 {
128 auto provider = _providers[ID];
187 auto provider = _maps.provider(ID);
129 auto variable = _variables[ID];
188 auto variable = _maps.variable(ID);
130 auto [missingRanges, newCacheRange] = _computeMissingRanges(variable,range);
189 auto [missingRanges, newCacheRange] = _computeMissingRanges(variable,range);
131 auto exe = new TransactionExe(_variables[ID], provider, missingRanges, range, newCacheRange);
190 auto exe = new TransactionExe(variable, provider, missingRanges, range, newCacheRange);
132 QObject::connect(exe,
191 QObject::connect(exe,
133 &TransactionExe::transactionComplete,
192 &TransactionExe::transactionComplete,
134 [groupID=groupID,transaction=newTransaction.value(),this]()
193 [groupID=groupID,transaction=newTransaction.value(),this]()
@@ -147,13 +206,13 class VariableController2::VariableController2Private
147 std::map<QUuid,DateTimeRange> ranges;
206 std::map<QUuid,DateTimeRange> ranges;
148 if(!DateTimeRangeHelper::hasnan(r))
207 if(!DateTimeRangeHelper::hasnan(r))
149 {
208 {
150 auto group = _synchronizationGroups[*refVar];
209 auto group = _maps.group(*refVar);
151 if(auto transformation = DateTimeRangeHelper::computeTransformation(refVar->range(),r);
210 if(auto transformation = DateTimeRangeHelper::computeTransformation(refVar->range(),r);
152 transformation.has_value())
211 transformation.has_value())
153 {
212 {
154 for(auto varId:group->variables())
213 for(auto varId:group->variables())
155 {
214 {
156 auto var = _variables[varId];
215 auto var = _maps.variable(varId);
157 auto newRange = var->range().transform(transformation.value());
216 auto newRange = var->range().transform(transformation.value());
158 ranges[varId] = newRange;
217 ranges[varId] = newRange;
159 }
218 }
@@ -163,7 +222,7 class VariableController2::VariableController2Private
163 {
222 {
164 for(auto varId:group->variables())
223 for(auto varId:group->variables())
165 {
224 {
166 auto var = _variables[varId];
225 auto var = _maps.variable(varId);
167 ranges[varId] = r;
226 ranges[varId] = r;
168 }
227 }
169 }
228 }
@@ -194,14 +253,11 class VariableController2::VariableController2Private
194
253
195 void _changeRange(QUuid id, DateTimeRange r)
254 void _changeRange(QUuid id, DateTimeRange r)
196 {
255 {
197 _lock.lockForRead();
256 _changeRange(_maps.variable(id) ,r);
198 auto var = _variables[id];
199 _lock.unlock();
200 _changeRange(var,r);
201 }
257 }
202 void _changeRange(const std::shared_ptr<Variable>& var, DateTimeRange r)
258 void _changeRange(const std::shared_ptr<Variable>& var, DateTimeRange r)
203 {
259 {
204 auto provider = _providers[*var];
260 auto provider = _maps.provider(*var);
205 auto [missingRanges, newCacheRange] = _computeMissingRanges(var,r);
261 auto [missingRanges, newCacheRange] = _computeMissingRanges(var,r);
206 std::vector<IDataSeries*> data;
262 std::vector<IDataSeries*> data;
207 for(auto range:missingRanges)
263 for(auto range:missingRanges)
@@ -218,6 +274,11 public:
218 this->_ThreadPool.setMaxThreadCount(32);
274 this->_ThreadPool.setMaxThreadCount(32);
219 }
275 }
220
276
277 /*
278 * This dtor has to like this even if this is ugly, because default dtor would rely on
279 * declaration order to destruct members and that would always lead to regressions when
280 * modifying with class members
281 */
221 ~VariableController2Private()
282 ~VariableController2Private()
222 {
283 {
223 while (this->_ThreadPool.activeThreadCount())
284 while (this->_ThreadPool.activeThreadCount())
@@ -228,21 +289,16 public:
228
289
229 std::shared_ptr<Variable> createVariable(const QString &name, const QVariantHash &metadata, std::shared_ptr<IDataProvider> provider)
290 std::shared_ptr<Variable> createVariable(const QString &name, const QVariantHash &metadata, std::shared_ptr<IDataProvider> provider)
230 {
291 {
231 QWriteLocker lock{&_lock};
232 auto newVar = std::make_shared<Variable>(name,metadata);
292 auto newVar = std::make_shared<Variable>(name,metadata);
233 this->_variables[*newVar] = newVar;
234 this->_providers[*newVar] = std::move(provider);
235 auto group = std::make_shared<VariableSynchronizationGroup2>(newVar->ID());
293 auto group = std::make_shared<VariableSynchronizationGroup2>(newVar->ID());
236 this->_synchronizationGroups[*newVar] = group;
294 _maps.addVariable(newVar,std::move(provider),group);
237 this->_transactions.addEntry(*group);
295 this->_transactions.addEntry(*group);
238 return newVar;
296 return newVar;
239 }
297 }
240
298
241 bool hasPendingTransactions(const std::shared_ptr<Variable>& variable)
299 bool hasPendingTransactions(const std::shared_ptr<Variable>& variable)
242 {
300 {
243 QReadLocker lock{&_lock};
301 return _transactions.active(*_maps.group(*variable));
244 auto group = _synchronizationGroups[*variable];
245 return _transactions.active(*group);
246 }
302 }
247
303
248 void deleteVariable(const std::shared_ptr<Variable>& variable)
304 void deleteVariable(const std::shared_ptr<Variable>& variable)
@@ -251,42 +307,23 public:
251 * Removing twice a var is ok but a var without provider has to be a hard error
307 * Removing twice a var is ok but a var without provider has to be a hard error
252 * this means we got the var controller in an inconsistent state
308 * this means we got the var controller in an inconsistent state
253 */
309 */
254 if(v_contains(variable))
310 _maps.removeVariable(variable);
255 {
256 QWriteLocker lock{&_lock};
257 this->_variables.remove(*variable);
258 }
259 if(p_contains(variable))
260 {
261 QWriteLocker lock{&_lock};
262 this->_providers.remove(*variable);
263 }
264 else
265 SCIQLOP_ERROR(VariableController2Private, "No provider found for given variable");
266 }
311 }
267
312
268 void asyncChangeRange(const std::shared_ptr<Variable>& variable, const DateTimeRange& r)
313 void asyncChangeRange(const std::shared_ptr<Variable>& variable, const DateTimeRange& r)
269 {
314 {
270 if(p_contains(variable))
315 if(!DateTimeRangeHelper::hasnan(r))
271 {
316 {
272 if(!DateTimeRangeHelper::hasnan(r))
317 auto group = _maps.group(*variable);
318 // Just overwrite next transaction
273 {
319 {
274 auto group = _synchronizationGroups[*variable];
320 _transactions.enqueue(*group,std::make_shared<VCTransaction>(variable->ID(), r, static_cast<int>(group->variables().size())));
275 // Just overwrite next transaction
276 {
277 QWriteLocker lock{&_lock};
278 _transactions.enqueue(*group,std::make_shared<VCTransaction>(variable->ID(), r, static_cast<int>(group->variables().size())));
279 }
280 _processTransactions();
281 }
282 else
283 {
284 SCIQLOP_ERROR(VariableController2Private, "Invalid range containing NaN");
285 }
321 }
322 _processTransactions();
286 }
323 }
287 else
324 else
288 {
325 {
289 SCIQLOP_ERROR(VariableController2Private, "No provider found for given variable");
326 SCIQLOP_ERROR(VariableController2Private, "Invalid range containing NaN");
290 }
327 }
291 }
328 }
292
329
@@ -299,37 +336,14 public:
299 }
336 }
300 }
337 }
301
338
302 void synchronize(const std::shared_ptr<Variable>& var, const std::shared_ptr<Variable>& with)
339 inline void synchronize(const std::shared_ptr<Variable>& var, const std::shared_ptr<Variable>& with)
303 {
340 {
304 if(v_contains(var) && v_contains(with))
341 _maps.synchronize(var, with);
305 {
306 if(sg_contains(var) && sg_contains(with))
307 {
308 QWriteLocker lock{&_lock};
309 auto dest_group = this->_synchronizationGroups[with->ID()];
310 this->_synchronizationGroups[*var] = dest_group;
311 dest_group->addVariable(*var);
312 }
313 else
314 {
315 SCIQLOP_ERROR(VariableController2Private, "At least one of the given variables isn't in a sync group");
316 }
317 }
318 else
319 {
320 SCIQLOP_ERROR(VariableController2Private, "At least one of the given variables is not found");
321 }
322 }
342 }
323
343
324 const std::set<std::shared_ptr<Variable>> variables()
344 inline const std::set<std::shared_ptr<Variable>> variables()
325 {
345 {
326 std::set<std::shared_ptr<Variable>> vars;
346 return _maps.variables();
327 QReadLocker lock{&_lock};
328 for(const auto &var:_variables)
329 {
330 vars.insert(var);
331 }
332 return vars;
333 }
347 }
334
348
335 };
349 };
General Comments 0
You need to be logged in to leave comments. Login now