Mirror/FairMQ
1
0
Fork 0
mirror of https://github.com/FairRootGroup/FairMQ.git synced 2025-10-13 16:46:47 +00:00
Code Issues Packages Projects Releases Wiki Activity

SDK: optimize TryChangeStateCompletion

Browse Source
This commit is contained in:
Alexey Rybalchenko 2019-09-16 12:58:20 +02:00 committed by Dennis Klein
parent 924320a0ac
commit 44bfbe02ed
1 changed files with 21 additions and 6 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

27
fairmq/sdk/Topology.h
View File

@ -319,6 +319,7 @@ class BasicTopology : public AsioBase<Executor, Allocator>
AsioBase<Executor, Allocator>::GetAllocator(), AsioBase<Executor, Allocator>::GetAllocator(),
std::move(handler)); std::move(handler));
fChangeStateTarget = expectedState.at(transition); fChangeStateTarget = expectedState.at(transition);
ResetTransitionedCount(fChangeStateTarget);
fDDSSession.SendCommand(GetTransitionName(transition)); fDDSSession.SendCommand(GetTransitionName(transition));
if (timeout > std::chrono::milliseconds(0)) { if (timeout > std::chrono::milliseconds(0)) {
fChangeStateOpTimer.expires_after(timeout); fChangeStateOpTimer.expires_after(timeout);
@ -394,6 +395,10 @@ class BasicTopology : public AsioBase<Executor, Allocator>
auto StateEqualsTo(DeviceState state) const -> bool { return sdk::StateEqualsTo(GetCurrentState(), state); } auto StateEqualsTo(DeviceState state) const -> bool { return sdk::StateEqualsTo(GetCurrentState(), state); }
private: private:
using TransitionedCount = unsigned int;
// using TransitionCounts = std::map<DeviceState, TransitionedCount>;
DDSSession fDDSSession; DDSSession fDDSSession;
DDSTopology fDDSTopo; DDSTopology fDDSTopo;
TopologyStateByTask fState; TopologyStateByTask fState;
@ -403,6 +408,7 @@ class BasicTopology : public AsioBase<Executor, Allocator>
ChangeStateOp fChangeStateOp; ChangeStateOp fChangeStateOp;
asio::steady_timer fChangeStateOpTimer; asio::steady_timer fChangeStateOpTimer;
DeviceState fChangeStateTarget; DeviceState fChangeStateTarget;
TransitionedCount fTransitionedCount;
static auto makeTopologyState(const DDSTopo& topo) -> TopologyStateByTask static auto makeTopologyState(const DDSTopo& topo) -> TopologyStateByTask
{ {
@ -422,6 +428,9 @@ class BasicTopology : public AsioBase<Executor, Allocator>
DeviceStatus& task = fState.at(taskId); DeviceStatus& task = fState.at(taskId);
task.initialized = true; task.initialized = true;
task.state = fair::mq::GetState(endState); task.state = fair::mq::GetState(endState);
if (task.state == fChangeStateTarget) {
++fTransitionedCount;
}
LOG(debug) << "Updated state entry: taskId=" << taskId << ",state=" << state; LOG(debug) << "Updated state entry: taskId=" << taskId << ",state=" << state;
TryChangeStateCompletion(); TryChangeStateCompletion();
} catch (const std::exception& e) { } catch (const std::exception& e) {
@ -432,17 +441,23 @@ class BasicTopology : public AsioBase<Executor, Allocator>
/// call only under locked fMtx! /// call only under locked fMtx!
auto TryChangeStateCompletion() -> void auto TryChangeStateCompletion() -> void
{ {
bool targetStateReached( if (!fChangeStateOp.IsCompleted() && fTransitionedCount == fState.size()) {
std::all_of(fState.cbegin(), fState.cend(), [&](TopologyStateByTask::value_type i) {
return (i.second.state == fChangeStateTarget) && i.second.initialized;
}));
if (!fChangeStateOp.IsCompleted() && targetStateReached) {
fChangeStateOpTimer.cancel(); fChangeStateOpTimer.cancel();
fChangeStateOp.Complete(MakeTopologyStateFromMap()); fChangeStateOp.Complete(MakeTopologyStateFromMap());
} }
} }
/// call only under locked fMtx!
auto ResetTransitionedCount(DeviceState targetState) -> void
{
fTransitionedCount = 0;
for (const auto& s : fState) {
if (s.second.state == targetState) {
++fTransitionedCount;
}
}
}
/// call only under locked fMtx! /// call only under locked fMtx!
auto GetCurrentStateUnsafe() const -> TopologyState auto GetCurrentStateUnsafe() const -> TopologyState
{ {