# FairMQ: fairmq/sdk/Topology.h Source File

Topology.h
/********************************************************************************
* Copyright (C) 2019 GSI Helmholtzzentrum fuer Schwerionenforschung GmbH *
* *
* This software is distributed under the terms of the *
* GNU Lesser General Public Licence (LGPL) version 3, *
* copied verbatim in the file "LICENSE" *
********************************************************************************/
#ifndef FAIR_MQ_SDK_TOPOLOGY_H
#define FAIR_MQ_SDK_TOPOLOGY_H
#include <fairmq/sdk/AsioAsyncOp.h>
#include <fairmq/sdk/AsioBase.h>
#include <fairmq/sdk/commands/Commands.h>
#include <fairmq/sdk/DDSCollection.h>
#include <fairmq/sdk/DDSInfo.h>
#include <fairmq/sdk/DDSSession.h>
#include <fairmq/sdk/DDSTask.h>
#include <fairmq/sdk/DDSTopology.h>
#include <fairmq/sdk/Error.h>
#include <fairmq/States.h>
#include <fairmq/tools/Semaphore.h>
#include <fairmq/tools/Unique.h>
#include <fairlogger/Logger.h>
#ifndef FAIR_LOG
#define FAIR_LOG LOG
#endif /* ifndef FAIR_LOG */
#include <asio/associated_executor.hpp>
#include <asio/async_result.hpp>
#include <asio/steady_timer.hpp>
#include <asio/system_executor.hpp>
#include <algorithm>
#include <chrono>
#include <condition_variable>
#include <functional>
#include <map>
#include <memory>
#include <mutex>
#include <ostream>
#include <set>
#include <stdexcept>
#include <string>
#include <thread>
#include <unordered_map>
#include <utility>
#include <vector>
namespace fair {
namespace mq {
namespace sdk {
using DeviceId = std::string;
using DeviceState = fair::mq::State;
using DeviceTransition = fair::mq::Transition;
const std::map<DeviceTransition, DeviceState> expectedState =
{
{ DeviceTransition::InitDevice, DeviceState::InitializingDevice },
{ DeviceTransition::CompleteInit, DeviceState::Initialized },
{ DeviceTransition::Bind, DeviceState::Bound },
{ DeviceTransition::Connect, DeviceState::DeviceReady },
{ DeviceTransition::InitTask, DeviceState::Ready },
{ DeviceTransition::Run, DeviceState::Running },
{ DeviceTransition::Stop, DeviceState::Ready },
{ DeviceTransition::ResetTask, DeviceState::DeviceReady },
{ DeviceTransition::ResetDevice, DeviceState::Idle },
{ DeviceTransition::End, DeviceState::Exiting }
};
struct DeviceStatus
{
bool subscribed_to_state_changes;
DeviceState lastState;
DeviceState state;
DDSTask::Id taskId;
DDSCollection::Id collectionId;
};
using DeviceProperty = std::pair<std::string, std::string>;
using DeviceProperties = std::vector<DeviceProperty>;
using DevicePropertyQuery = std::string;
using FailedDevices = std::set<DeviceId>;
struct GetPropertiesResult
{
struct Device
{
DeviceProperties props;
};
std::unordered_map<DeviceId, Device> devices;
FailedDevices failed;
};
using TopologyState = std::vector<DeviceStatus>;
using TopologyStateIndex = std::unordered_map<DDSTask::Id, int>; // task id -> index in the data vector
using TopologyStateByTask = std::unordered_map<DDSTask::Id, DeviceStatus>;
using TopologyStateByCollection = std::unordered_map<DDSCollection::Id, std::vector<DeviceStatus>>;
using TopologyTransition = fair::mq::Transition;
inline DeviceState AggregateState(const TopologyState& topologyState)
{
DeviceState first = topologyState.begin()->state;
if (std::all_of(topologyState.cbegin(), topologyState.cend(), [&](TopologyState::value_type i) {
return i.state == first;
})) {
return first;
}
throw MixedStateError("State is not uniform");
}
inline bool StateEqualsTo(const TopologyState& topologyState, DeviceState state)
{
return AggregateState(topologyState) == state;
}
inline TopologyStateByCollection GroupByCollectionId(const TopologyState& topologyState)
{
TopologyStateByCollection state;
for (const auto& ds : topologyState) {
if (ds.collectionId != 0) {
state[ds.collectionId].push_back(ds);
}
}
return state;
}
inline TopologyStateByTask GroupByTaskId(const TopologyState& topologyState)
{
TopologyStateByTask state;
for (const auto& ds : topologyState) {
state[ds.taskId] = ds;
}
return state;
}
template <typename Executor, typename Allocator>
class BasicTopology : public AsioBase<Executor, Allocator>
{
public:
BasicTopology(DDSTopology topo, DDSSession session)
: BasicTopology<Executor, Allocator>(asio::system_executor(), std::move(topo), std::move(session))
{}
BasicTopology(const Executor& ex,
DDSTopology topo,
DDSSession session,
Allocator alloc = DefaultAllocator())
: AsioBase<Executor, Allocator>(ex, std::move(alloc))
, fDDSSession(std::move(session))
, fDDSTopo(std::move(topo))
, fStateData()
, fStateIndex()
, fHeartbeatsTimer(asio::system_executor())
, fHeartbeatInterval(600000)
{
makeTopologyState();
std::string activeTopo(fDDSSession.RequestCommanderInfo().activeTopologyName);
std::string givenTopo(fDDSTopo.GetName());
if (activeTopo != givenTopo) {
throw RuntimeError("Given topology ", givenTopo, " is not activated (active: ", activeTopo, ")");
}
SubscribeToCommands();
fDDSSession.StartDDSService();
SubscribeToStateChanges();
}
BasicTopology(const BasicTopology&) = delete;
BasicTopology& operator=(const BasicTopology&) = delete;
BasicTopology(BasicTopology&&) = default;
BasicTopology& operator=(BasicTopology&&) = default;
~BasicTopology()
{
UnsubscribeFromStateChanges();
std::lock_guard<std::mutex> lk(fMtx);
fDDSSession.UnsubscribeFromCommands();
try {
for (auto& op : fChangeStateOps) {
op.second.Complete(MakeErrorCode(ErrorCode::OperationCanceled));
}
} catch (...) {}
}
void SubscribeToStateChanges()
{
// FAIR_LOG(debug) << "Subscribing to state change";
cmd::Cmds cmds(cmd::make<cmd::SubscribeToStateChange>(fHeartbeatInterval.count()));
fDDSSession.SendCommand(cmds.Serialize());
fHeartbeatsTimer.expires_after(fHeartbeatInterval);
fHeartbeatsTimer.async_wait(std::bind(&BasicTopology::SendSubscriptionHeartbeats, this, std::placeholders::_1));
}
void SendSubscriptionHeartbeats(const std::error_code& ec)
{
if (!ec) {
// Timer expired.
fDDSSession.SendCommand(cmd::Cmds(cmd::make<cmd::SubscriptionHeartbeat>(fHeartbeatInterval.count())).Serialize());
// schedule again
fHeartbeatsTimer.expires_after(fHeartbeatInterval);
fHeartbeatsTimer.async_wait(std::bind(&BasicTopology::SendSubscriptionHeartbeats, this, std::placeholders::_1));
} else if (ec == asio::error::operation_aborted) {
// FAIR_LOG(debug) << "Heartbeats timer canceled";
} else {
FAIR_LOG(error) << "Timer error: " << ec;
}
}
void UnsubscribeFromStateChanges()
{
// stop sending heartbeats
fHeartbeatsTimer.cancel();
// unsubscribe from state changes
fDDSSession.SendCommand(cmd::Cmds(cmd::make<cmd::UnsubscribeFromStateChange>()).Serialize());
// wait for all tasks to confirm unsubscription
std::unique_lock<std::mutex> lk(fMtx);
fStateChangeUnsubscriptionCV.wait(lk, [&](){
unsigned int count = std::count_if(fStateIndex.cbegin(), fStateIndex.cend(), [=](const auto& s) {
return fStateData.at(s.second).subscribed_to_state_changes == false;
});
return count == fStateIndex.size();
});
}
void SubscribeToCommands()
{
fDDSSession.SubscribeToCommands([&](const std::string& msg, const std::string& /* condition */, DDSChannel::Id senderId) {
cmd::Cmds inCmds;
inCmds.Deserialize(msg);
// FAIR_LOG(debug) << "Received " << inCmds.Size() << " command(s) with total size of " << msg.length() << " bytes: ";
for (const auto& cmd : inCmds) {
// FAIR_LOG(debug) << " > " << cmd->GetType();
switch (cmd->GetType()) {
case cmd::Type::state_change_subscription:
HandleCmd(static_cast<cmd::StateChangeSubscription&>(*cmd));
break;
case cmd::Type::state_change_unsubscription:
HandleCmd(static_cast<cmd::StateChangeUnsubscription&>(*cmd));
break;
case cmd::Type::state_change:
HandleCmd(static_cast<cmd::StateChange&>(*cmd), senderId);
break;
case cmd::Type::transition_status:
HandleCmd(static_cast<cmd::TransitionStatus&>(*cmd));
break;
case cmd::Type::properties:
HandleCmd(static_cast<cmd::Properties&>(*cmd));
break;
case cmd::Type::properties_set:
HandleCmd(static_cast<cmd::PropertiesSet&>(*cmd));
break;
default:
FAIR_LOG(warn) << "Unexpected/unknown command received: " << cmd->GetType();
FAIR_LOG(warn) << "Origin: " << senderId;
break;
}
}
});
}
auto HandleCmd(cmd::StateChangeSubscription const& cmd) -> void
{
if (cmd.GetResult() == cmd::Result::Ok) {
DDSTask::Id taskId(cmd.GetTaskId());
try {
std::lock_guard<std::mutex> lk(fMtx);
DeviceStatus& task = fStateData.at(fStateIndex.at(taskId));
task.subscribed_to_state_changes = true;
} catch (const std::exception& e) {
FAIR_LOG(error) << "Exception in HandleCmd(cmd::StateChangeSubscription const&): " << e.what();
}
} else {
FAIR_LOG(error) << "State change subscription failed for device: " << cmd.GetDeviceId() << ", task id: " << cmd.GetTaskId();
}
}
auto HandleCmd(cmd::StateChangeUnsubscription const& cmd) -> void
{
if (cmd.GetResult() == cmd::Result::Ok) {
DDSTask::Id taskId(cmd.GetTaskId());
try {
std::unique_lock<std::mutex> lk(fMtx);
DeviceStatus& task = fStateData.at(fStateIndex.at(taskId));
task.subscribed_to_state_changes = false;
lk.unlock();
fStateChangeUnsubscriptionCV.notify_one();
} catch (const std::exception& e) {
FAIR_LOG(error) << "Exception in HandleCmd(cmd::StateChangeUnsubscription const&): " << e.what();
}
} else {
FAIR_LOG(error) << "State change unsubscription failed for device: " << cmd.GetDeviceId() << ", task id: " << cmd.GetTaskId();
}
}
auto HandleCmd(cmd::StateChange const& cmd, DDSChannel::Id const& senderId) -> void
{
if (cmd.GetCurrentState() == DeviceState::Exiting) {
fDDSSession.SendCommand(cmd::Cmds(cmd::make<cmd::StateChangeExitingReceived>()).Serialize(), senderId);
}
DDSTask::Id taskId(cmd.GetTaskId());
try {
std::lock_guard<std::mutex> lk(fMtx);
DeviceStatus& task = fStateData.at(fStateIndex.at(taskId));
task.lastState = cmd.GetLastState();
task.state = cmd.GetCurrentState();
// if the task is exiting, it will not respond to unsubscription request anymore, set it to false now.
if (task.state == DeviceState::Exiting) {
task.subscribed_to_state_changes = false;
}
// FAIR_LOG(debug) << "Updated state entry: taskId=" << taskId << ", state=" << state;
for (auto& op : fChangeStateOps) {
op.second.Update(taskId, cmd.GetCurrentState());
}
for (auto& op : fWaitForStateOps) {
op.second.Update(taskId, cmd.GetLastState(), cmd.GetCurrentState());
}
} catch (const std::exception& e) {
FAIR_LOG(error) << "Exception in HandleCmd(cmd::StateChange const&): " << e.what();
}
}
auto HandleCmd(cmd::TransitionStatus const& cmd) -> void
{
if (cmd.GetResult() != cmd::Result::Ok) {
FAIR_LOG(error) << cmd.GetTransition() << " transition failed for " << cmd.GetDeviceId();
DDSTask::Id taskId(cmd.GetTaskId());
std::lock_guard<std::mutex> lk(fMtx);
for (auto& op : fChangeStateOps) {
if (!op.second.IsCompleted() && op.second.ContainsTask(taskId) &&
fStateData.at(fStateIndex.at(taskId)).state != op.second.GetTargetState()) {
op.second.Complete(MakeErrorCode(ErrorCode::DeviceChangeStateFailed));
}
}
}
}
auto HandleCmd(cmd::Properties const& cmd) -> void
{
std::unique_lock<std::mutex> lk(fMtx);
try {
auto& op(fGetPropertiesOps.at(cmd.GetRequestId()));
lk.unlock();
op.Update(cmd.GetDeviceId(), cmd.GetResult(), cmd.GetProps());
} catch (std::out_of_range& e) {
FAIR_LOG(debug) << "GetProperties operation (request id: " << cmd.GetRequestId()
<< ") not found (probably completed or timed out), "
<< "discarding reply of device " << cmd.GetDeviceId();
}
}
auto HandleCmd(cmd::PropertiesSet const& cmd) -> void
{
std::unique_lock<std::mutex> lk(fMtx);
try {
auto& op(fSetPropertiesOps.at(cmd.GetRequestId()));
lk.unlock();
op.Update(cmd.GetDeviceId(), cmd.GetResult());
} catch (std::out_of_range& e) {
FAIR_LOG(debug) << "SetProperties operation (request id: " << cmd.GetRequestId()
<< ") not found (probably completed or timed out), "
<< "discarding reply of device " << cmd.GetDeviceId();
}
}
using Duration = std::chrono::milliseconds;
using ChangeStateCompletionSignature = void(std::error_code, TopologyState);
private:
struct ChangeStateOp
{
using Id = std::size_t;
using Count = unsigned int;
template<typename Handler>
ChangeStateOp(Id id,
const TopologyTransition transition,
std::vector<DDSTask> tasks,
TopologyState& stateData,
Duration timeout,
std::mutex& mutex,
Executor const & ex,
Allocator const & alloc,
Handler&& handler)
: fId(id)
, fOp(ex, alloc, std::move(handler))
, fStateData(stateData)
, fTimer(ex)
, fCount(0)
, fTasks(std::move(tasks))
, fTargetState(expectedState.at(transition))
, fMtx(mutex)
{
if (timeout > std::chrono::milliseconds(0)) {
fTimer.expires_after(timeout);
fTimer.async_wait([&](std::error_code ec) {
if (!ec) {
std::lock_guard<std::mutex> lk(fMtx);
fOp.Timeout(fStateData);
}
});
}
}
ChangeStateOp() = delete;
ChangeStateOp(const ChangeStateOp&) = delete;
ChangeStateOp& operator=(const ChangeStateOp&) = delete;
ChangeStateOp(ChangeStateOp&&) = default;
ChangeStateOp& operator=(ChangeStateOp&&) = default;
~ChangeStateOp() = default;
auto ResetCount(const TopologyStateIndex& stateIndex, const TopologyState& stateData) -> void
{
fCount = std::count_if(stateIndex.cbegin(), stateIndex.cend(), [=](const auto& s) {
if (ContainsTask(stateData.at(s.second).taskId)) {
return stateData.at(s.second).state == fTargetState;
} else {
return false;
}
});
}
auto Update(const DDSTask::Id taskId, const DeviceState currentState) -> void
{
if (!fOp.IsCompleted() && ContainsTask(taskId)) {
if (currentState == fTargetState) {
++fCount;
}
TryCompletion();
}
}
auto TryCompletion() -> void
{
if (!fOp.IsCompleted() && fCount == fTasks.size()) {
Complete(std::error_code());
}
}
auto Complete(std::error_code ec) -> void
{
fTimer.cancel();
fOp.Complete(ec, fStateData);
}
auto ContainsTask(DDSTask::Id id) -> bool
{
auto it = std::find_if(fTasks.begin(), fTasks.end(), [id](const DDSTask& t) { return t.GetId() == id; });
return it != fTasks.end();
}
bool IsCompleted() { return fOp.IsCompleted(); }
auto GetTargetState() const -> DeviceState { return fTargetState; }
private:
Id const fId;
AsioAsyncOp<Executor, Allocator, ChangeStateCompletionSignature> fOp;
TopologyState& fStateData;
asio::steady_timer fTimer;
Count fCount;
std::vector<DDSTask> fTasks;
DeviceState fTargetState;
std::mutex& fMtx;
};
public:
template<typename CompletionToken>
auto AsyncChangeState(const TopologyTransition transition,
const std::string& path,
Duration timeout,
CompletionToken&& token)
{
return asio::async_initiate<CompletionToken, ChangeStateCompletionSignature>([&](auto handler) {
typename ChangeStateOp::Id const id(tools::UuidHash());
std::lock_guard<std::mutex> lk(fMtx);
for (auto it = begin(fChangeStateOps); it != end(fChangeStateOps);) {
if (it->second.IsCompleted()) {
it = fChangeStateOps.erase(it);
} else {
++it;
}
}
auto p = fChangeStateOps.emplace(
std::piecewise_construct,
std::forward_as_tuple(id),
std::forward_as_tuple(id,
transition,
fDDSTopo.GetTasks(path),
fStateData,
timeout,
fMtx,
AsioBase<Executor, Allocator>::GetExecutor(),
AsioBase<Executor, Allocator>::GetAllocator(),
std::move(handler)));
cmd::Cmds cmds(cmd::make<cmd::ChangeState>(transition));
fDDSSession.SendCommand(cmds.Serialize(), path);
p.first->second.ResetCount(fStateIndex, fStateData);
// TODO: make sure following operation properly queues the completion and not doing it directly out of initiation call.
p.first->second.TryCompletion();
},
token);
}
template<typename CompletionToken>
auto AsyncChangeState(const TopologyTransition transition, CompletionToken&& token)
{
return AsyncChangeState(transition, "", Duration(0), std::move(token));
}
template<typename CompletionToken>
auto AsyncChangeState(const TopologyTransition transition, Duration timeout, CompletionToken&& token)
{
return AsyncChangeState(transition, "", timeout, std::move(token));
}
template<typename CompletionToken>
auto AsyncChangeState(const TopologyTransition transition, const std::string& path, CompletionToken&& token)
{
return AsyncChangeState(transition, path, Duration(0), std::move(token));
}
auto ChangeState(const TopologyTransition transition, const std::string& path = "", Duration timeout = Duration(0))
-> std::pair<std::error_code, TopologyState>
{
tools::SharedSemaphore blocker;
std::error_code ec;
TopologyState state;
AsyncChangeState(transition, path, timeout, [&, blocker](std::error_code _ec, TopologyState _state) mutable {
ec = _ec;
state = _state;
blocker.Signal();
});
blocker.Wait();
return {ec, state};
}
auto ChangeState(const TopologyTransition transition, Duration timeout)
-> std::pair<std::error_code, TopologyState>
{
return ChangeState(transition, "", timeout);
}
auto GetCurrentState() const -> TopologyState
{
std::lock_guard<std::mutex> lk(fMtx);
return fStateData;
}
auto AggregateState() const -> DeviceState { return sdk::AggregateState(GetCurrentState()); }
auto StateEqualsTo(DeviceState state) const -> bool { return sdk::StateEqualsTo(GetCurrentState(), state); }
using WaitForStateCompletionSignature = void(std::error_code);
private:
struct WaitForStateOp
{
using Id = std::size_t;
using Count = unsigned int;
template<typename Handler>
WaitForStateOp(Id id,
DeviceState targetLastState,
DeviceState targetCurrentState,
std::vector<DDSTask> tasks,
Duration timeout,
std::mutex& mutex,
Executor const & ex,
Allocator const & alloc,
Handler&& handler)
: fId(id)
, fOp(ex, alloc, std::move(handler))
, fTimer(ex)
, fCount(0)
, fTasks(std::move(tasks))
, fTargetLastState(targetLastState)
, fTargetCurrentState(targetCurrentState)
, fMtx(mutex)
{
if (timeout > std::chrono::milliseconds(0)) {
fTimer.expires_after(timeout);
fTimer.async_wait([&](std::error_code ec) {
if (!ec) {
std::lock_guard<std::mutex> lk(fMtx);
fOp.Timeout();
}
});
}
}
WaitForStateOp() = delete;
WaitForStateOp(const WaitForStateOp&) = delete;
WaitForStateOp& operator=(const WaitForStateOp&) = delete;
WaitForStateOp(WaitForStateOp&&) = default;
WaitForStateOp& operator=(WaitForStateOp&&) = default;
~WaitForStateOp() = default;
auto ResetCount(const TopologyStateIndex& stateIndex, const TopologyState& stateData) -> void
{
fCount = std::count_if(stateIndex.cbegin(), stateIndex.cend(), [=](const auto& s) {
if (ContainsTask(stateData.at(s.second).taskId)) {
return stateData.at(s.second).state == fTargetCurrentState
&&
(stateData.at(s.second).lastState == fTargetLastState || fTargetLastState == DeviceState::Ok);
} else {
return false;
}
});
}
auto Update(const DDSTask::Id taskId, const DeviceState lastState, const DeviceState currentState) -> void
{
if (!fOp.IsCompleted() && ContainsTask(taskId)) {
if (currentState == fTargetCurrentState &&
(lastState == fTargetLastState ||
fTargetLastState == DeviceState::Ok)) {
++fCount;
}
TryCompletion();
}
}
auto TryCompletion() -> void
{
if (!fOp.IsCompleted() && fCount == fTasks.size()) {
fTimer.cancel();
fOp.Complete();
}
}
bool IsCompleted() { return fOp.IsCompleted(); }
private:
Id const fId;
AsioAsyncOp<Executor, Allocator, WaitForStateCompletionSignature> fOp;
asio::steady_timer fTimer;
Count fCount;
std::vector<DDSTask> fTasks;
DeviceState fTargetLastState;
DeviceState fTargetCurrentState;
std::mutex& fMtx;
auto ContainsTask(DDSTask::Id id) -> bool
{
auto it = std::find_if(fTasks.begin(), fTasks.end(), [id](const DDSTask& t) { return t.GetId() == id; });
return it != fTasks.end();
}
};
public:
template<typename CompletionToken>
auto AsyncWaitForState(const DeviceState targetLastState,
const DeviceState targetCurrentState,
const std::string& path,
Duration timeout,
CompletionToken&& token)
{
return asio::async_initiate<CompletionToken, WaitForStateCompletionSignature>([&](auto handler) {
typename GetPropertiesOp::Id const id(tools::UuidHash());
std::lock_guard<std::mutex> lk(fMtx);
for (auto it = begin(fWaitForStateOps); it != end(fWaitForStateOps);) {
if (it->second.IsCompleted()) {
it = fWaitForStateOps.erase(it);
} else {
++it;
}
}
auto p = fWaitForStateOps.emplace(
std::piecewise_construct,
std::forward_as_tuple(id),
std::forward_as_tuple(id,
targetLastState,
targetCurrentState,
fDDSTopo.GetTasks(path),
timeout,
fMtx,
AsioBase<Executor, Allocator>::GetExecutor(),
AsioBase<Executor, Allocator>::GetAllocator(),
std::move(handler)));
p.first->second.ResetCount(fStateIndex, fStateData);
// TODO: make sure following operation properly queues the completion and not doing it directly out of initiation call.
p.first->second.TryCompletion();
},
token);
}
template<typename CompletionToken>
auto AsyncWaitForState(const DeviceState targetLastState, const DeviceState targetCurrentState, CompletionToken&& token)
{
return AsyncWaitForState(targetLastState, targetCurrentState, "", Duration(0), std::move(token));
}
template<typename CompletionToken>
auto AsyncWaitForState(const DeviceState targetCurrentState, CompletionToken&& token)
{
return AsyncWaitForState(DeviceState::Ok, targetCurrentState, "", Duration(0), std::move(token));
}
auto WaitForState(const DeviceState targetLastState, const DeviceState targetCurrentState, const std::string& path = "", Duration timeout = Duration(0))
-> std::error_code
{
tools::SharedSemaphore blocker;
std::error_code ec;
AsyncWaitForState(targetLastState, targetCurrentState, path, timeout, [&, blocker](std::error_code _ec) mutable {
ec = _ec;
blocker.Signal();
});
blocker.Wait();
return ec;
}
auto WaitForState(const DeviceState targetCurrentState, const std::string& path = "", Duration timeout = Duration(0))
-> std::error_code
{
return WaitForState(DeviceState::Ok, targetCurrentState, path, timeout);
}
using GetPropertiesCompletionSignature = void(std::error_code, GetPropertiesResult);
private:
struct GetPropertiesOp
{
using Id = std::size_t;
using GetCount = unsigned int;
template<typename Handler>
GetPropertiesOp(Id id,
GetCount expectedCount,
Duration timeout,
std::mutex& mutex,
Executor const & ex,
Allocator const & alloc,
Handler&& handler)
: fId(id)
, fOp(ex, alloc, std::move(handler))
, fTimer(ex)
, fCount(0)
, fExpectedCount(expectedCount)
, fMtx(mutex)
{
if (timeout > std::chrono::milliseconds(0)) {
fTimer.expires_after(timeout);
fTimer.async_wait([&](std::error_code ec) {
if (!ec) {
std::lock_guard<std::mutex> lk(fMtx);
fOp.Timeout(fResult);
}
});
}
// FAIR_LOG(debug) << "GetProperties " << fId << " with expected count of " << fExpectedCount << " started.";
}
GetPropertiesOp() = delete;
GetPropertiesOp(const GetPropertiesOp&) = delete;
GetPropertiesOp& operator=(const GetPropertiesOp&) = delete;
GetPropertiesOp(GetPropertiesOp&&) = default;
GetPropertiesOp& operator=(GetPropertiesOp&&) = default;
~GetPropertiesOp() = default;
auto Update(const std::string& deviceId, cmd::Result result, DeviceProperties props) -> void
{
std::lock_guard<std::mutex> lk(fMtx);
if (cmd::Result::Ok != result) {
fResult.failed.insert(deviceId);
} else {
fResult.devices.insert({deviceId, {std::move(props)}});
}
++fCount;
TryCompletion();
}
bool IsCompleted() { return fOp.IsCompleted(); }
private:
Id const fId;
AsioAsyncOp<Executor, Allocator, GetPropertiesCompletionSignature> fOp;
asio::steady_timer fTimer;
GetCount fCount;
GetCount const fExpectedCount;
GetPropertiesResult fResult;
std::mutex& fMtx;
auto TryCompletion() -> void
{
if (!fOp.IsCompleted() && fCount == fExpectedCount) {
fTimer.cancel();
if (fResult.failed.size() > 0) {
fOp.Complete(MakeErrorCode(ErrorCode::DeviceGetPropertiesFailed), std::move(fResult));
} else {
fOp.Complete(std::move(fResult));
}
}
}
};
public:
template<typename CompletionToken>
auto AsyncGetProperties(DevicePropertyQuery const& query,
const std::string& path,
Duration timeout,
CompletionToken&& token)
{
return asio::async_initiate<CompletionToken, GetPropertiesCompletionSignature>(
[&](auto handler) {
typename GetPropertiesOp::Id const id(tools::UuidHash());
std::lock_guard<std::mutex> lk(fMtx);
for (auto it = begin(fGetPropertiesOps); it != end(fGetPropertiesOps);) {
if (it->second.IsCompleted()) {
it = fGetPropertiesOps.erase(it);
} else {
++it;
}
}
fGetPropertiesOps.emplace(
std::piecewise_construct,
std::forward_as_tuple(id),
std::forward_as_tuple(id,
fDDSTopo.GetTasks(path).size(),
timeout,
fMtx,
AsioBase<Executor, Allocator>::GetExecutor(),
AsioBase<Executor, Allocator>::GetAllocator(),
std::move(handler)));
cmd::Cmds const cmds(cmd::make<cmd::GetProperties>(id, query));
fDDSSession.SendCommand(cmds.Serialize(), path);
},
token);
}
template<typename CompletionToken>
auto AsyncGetProperties(DevicePropertyQuery const& query, CompletionToken&& token)
{
return AsyncGetProperties(query, "", Duration(0), std::move(token));
}
auto GetProperties(DevicePropertyQuery const& query, const std::string& path = "", Duration timeout = Duration(0))
-> std::pair<std::error_code, GetPropertiesResult>
{
tools::SharedSemaphore blocker;
std::error_code ec;
GetPropertiesResult result;
AsyncGetProperties(query, path, timeout, [&, blocker](std::error_code _ec, GetPropertiesResult _result) mutable {
ec = _ec;
result = _result;
blocker.Signal();
});
blocker.Wait();
return {ec, result};
}
using SetPropertiesCompletionSignature = void(std::error_code, FailedDevices);
private:
struct SetPropertiesOp
{
using Id = std::size_t;
using SetCount = unsigned int;
template<typename Handler>
SetPropertiesOp(Id id,
SetCount expectedCount,
Duration timeout,
std::mutex& mutex,
Executor const & ex,
Allocator const & alloc,
Handler&& handler)
: fId(id)
, fOp(ex, alloc, std::move(handler))
, fTimer(ex)
, fCount(0)
, fExpectedCount(expectedCount)
, fFailedDevices()
, fMtx(mutex)
{
if (timeout > std::chrono::milliseconds(0)) {
fTimer.expires_after(timeout);
fTimer.async_wait([&](std::error_code ec) {
if (!ec) {
std::lock_guard<std::mutex> lk(fMtx);
fOp.Timeout(fFailedDevices);
}
});
}
// FAIR_LOG(debug) << "SetProperties " << fId << " with expected count of " << fExpectedCount << " started.";
}
SetPropertiesOp() = delete;
SetPropertiesOp(const SetPropertiesOp&) = delete;
SetPropertiesOp& operator=(const SetPropertiesOp&) = delete;
SetPropertiesOp(SetPropertiesOp&&) = default;
SetPropertiesOp& operator=(SetPropertiesOp&&) = default;
~SetPropertiesOp() = default;
auto Update(const std::string& deviceId, cmd::Result result) -> void
{
std::lock_guard<std::mutex> lk(fMtx);
if (cmd::Result::Ok != result) {
fFailedDevices.insert(deviceId);
}
++fCount;
TryCompletion();
}
bool IsCompleted() { return fOp.IsCompleted(); }
private:
Id const fId;
AsioAsyncOp<Executor, Allocator, SetPropertiesCompletionSignature> fOp;
asio::steady_timer fTimer;
SetCount fCount;
SetCount const fExpectedCount;
FailedDevices fFailedDevices;
std::mutex& fMtx;
auto TryCompletion() -> void
{
if (!fOp.IsCompleted() && fCount == fExpectedCount) {
fTimer.cancel();
if (fFailedDevices.size() > 0) {
fOp.Complete(MakeErrorCode(ErrorCode::DeviceSetPropertiesFailed), fFailedDevices);
} else {
fOp.Complete(fFailedDevices);
}
}
}
};
public:
template<typename CompletionToken>
auto AsyncSetProperties(const DeviceProperties& props,
const std::string& path,
Duration timeout,
CompletionToken&& token)
{
return asio::async_initiate<CompletionToken, SetPropertiesCompletionSignature>(
[&](auto handler) {
typename SetPropertiesOp::Id const id(tools::UuidHash());
std::lock_guard<std::mutex> lk(fMtx);
for (auto it = begin(fGetPropertiesOps); it != end(fGetPropertiesOps);) {
if (it->second.IsCompleted()) {
it = fGetPropertiesOps.erase(it);
} else {
++it;
}
}
fSetPropertiesOps.emplace(
std::piecewise_construct,
std::forward_as_tuple(id),
std::forward_as_tuple(id,
fDDSTopo.GetTasks(path).size(),
timeout,
fMtx,
AsioBase<Executor, Allocator>::GetExecutor(),
AsioBase<Executor, Allocator>::GetAllocator(),
std::move(handler)));
cmd::Cmds const cmds(cmd::make<cmd::SetProperties>(id, props));
fDDSSession.SendCommand(cmds.Serialize(), path);
},
token);
}
template<typename CompletionToken>
auto AsyncSetProperties(DeviceProperties const & props, CompletionToken&& token)
{
return AsyncSetProperties(props, "", Duration(0), std::move(token));
}
auto SetProperties(DeviceProperties const& properties, const std::string& path = "", Duration timeout = Duration(0))
-> std::pair<std::error_code, FailedDevices>
{
tools::SharedSemaphore blocker;
std::error_code ec;
FailedDevices failed;
AsyncSetProperties(properties, path, timeout, [&, blocker](std::error_code _ec, FailedDevices _failed) mutable {
ec = _ec;
failed = _failed;
blocker.Signal();
});
blocker.Wait();
return {ec, failed};
}
Duration GetHeartbeatInterval() const { return fHeartbeatInterval; }
void SetHeartbeatInterval(Duration duration) { fHeartbeatInterval = duration; }
private:
using TransitionedCount = unsigned int;
DDSSession fDDSSession;
DDSTopology fDDSTopo;
TopologyState fStateData;
TopologyStateIndex fStateIndex;
mutable std::mutex fMtx;
std::condition_variable fStateChangeUnsubscriptionCV;
asio::steady_timer fHeartbeatsTimer;
Duration fHeartbeatInterval;
std::unordered_map<typename ChangeStateOp::Id, ChangeStateOp> fChangeStateOps;
std::unordered_map<typename WaitForStateOp::Id, WaitForStateOp> fWaitForStateOps;
std::unordered_map<typename SetPropertiesOp::Id, SetPropertiesOp> fSetPropertiesOps;
std::unordered_map<typename GetPropertiesOp::Id, GetPropertiesOp> fGetPropertiesOps;
auto makeTopologyState() -> void
{
fStateData.reserve(fDDSTopo.GetTasks().size());
int index = 0;
for (const auto& task : fDDSTopo.GetTasks()) {
fStateData.push_back(DeviceStatus{false, DeviceState::Ok, DeviceState::Ok, task.GetId(), task.GetCollectionId()});
fStateIndex.emplace(task.GetId(), index);
index++;
}
}
auto GetCurrentStateUnsafe() const -> TopologyState
{
return fStateData;
}
};
using Topology = BasicTopology<DefaultExecutor, DefaultAllocator>;
using Topo = Topology;
auto MakeTopology(dds::topology_api::CTopology nativeTopo,
std::shared_ptr<dds::tools_api::CSession> nativeSession,
DDSEnv env = {}) -> Topology;
} // namespace sdk
} // namespace mq
} // namespace fair
#endif /* FAIR_MQ_SDK_TOPOLOGY_H */
fair::mq::sdk::BasicTopology
Represents a FairMQ topology.
Definition: Topology.h:154
fair::mq::sdk::DDSSession
Represents a DDS session.
Definition: DDSSession.h:56
fair::mq::sdk::BasicTopology::AsyncChangeState
auto AsyncChangeState(const TopologyTransition transition, CompletionToken &&token)
Initiate state transition on all FairMQ devices in this topology.
Definition: Topology.h:637
fair::mq::sdk::GetPropertiesResult
Definition: Topology.h:87
fair::mq::sdk::AsioBase::GetExecutor
auto GetExecutor() const noexcept -> ExecutorType
Get associated I/O executor.
Definition: AsioBase.h:41
fair::mq::sdk::cmd::StateChangeSubscription
Definition: Commands.h:233
fair::mq::sdk::AsioAsyncOp< Executor, Allocator, ChangeStateCompletionSignature >
fair::mq::sdk::BasicTopology::AsyncSetProperties
auto AsyncSetProperties(DeviceProperties const &props, CompletionToken &&token)
Initiate property update on selected FairMQ devices in this topology.
Definition: Topology.h:1191
fair::mq::sdk::cmd::TransitionStatus
Definition: Commands.h:189
fair::mq::sdk::AsioBase::GetAllocator
auto GetAllocator() const noexcept -> AllocatorType
Get associated default allocator.
Definition: AsioBase.h:46
fair::mq::sdk::AsioBase
Base for creating Asio-enabled I/O objects.
Definition: AsioBase.h:35
std
Definition: Error.h:56
fair::mq::sdk::BasicTopology::AsyncWaitForState
auto AsyncWaitForState(const DeviceState targetCurrentState, CompletionToken &&token)
Initiate waiting for selected FairMQ devices to reach given current state in this topology...
Definition: Topology.h:874
fair::mq::sdk::BasicTopology::AsyncWaitForState
auto AsyncWaitForState(const DeviceState targetLastState, const DeviceState targetCurrentState, CompletionToken &&token)
Initiate waiting for selected FairMQ devices to reach given last & current state in this topology...
Definition: Topology.h:863
fair::mq::sdk::GetPropertiesResult::Device
Definition: Topology.h:89
fair::mq::sdk::BasicTopology::WaitForState
auto WaitForState(const DeviceState targetLastState, const DeviceState targetCurrentState, const std::string &path="", Duration timeout=Duration(0)) -> std::error_code
Wait for selected FairMQ devices to reach given last & current state in this topology.
Definition: Topology.h:885
fair::mq::sdk::BasicTopology::AsyncWaitForState
auto AsyncWaitForState(const DeviceState targetLastState, const DeviceState targetCurrentState, const std::string &path, Duration timeout, CompletionToken &&token)
Initiate waiting for selected FairMQ devices to reach given last & current state in this topology...
Definition: Topology.h:818
fair::mq::sdk::DDSTask
Represents a DDS task.
Definition: DDSTask.h:25
fair::mq::sdk::BasicTopology::AsyncChangeState
auto AsyncChangeState(const TopologyTransition transition, Duration timeout, CompletionToken &&token)
Initiate state transition on all FairMQ devices in this topology with a timeout.
Definition: Topology.h:649
fair::mq::sdk::cmd::Properties
Definition: Commands.h:303
fair::mq::sdk::BasicTopology::AsyncGetProperties
auto AsyncGetProperties(DevicePropertyQuery const &query, CompletionToken &&token)
Initiate property query on selected FairMQ devices in this topology.
Definition: Topology.h:1038
fair::mq::sdk::BasicTopology::AsyncChangeState
auto AsyncChangeState(const TopologyTransition transition, const std::string &path, Duration timeout, CompletionToken &&token)
Initiate state transition on all FairMQ devices in this topology.
Definition: Topology.h:589
fair::mq::sdk::BasicTopology::ChangeState
auto ChangeState(const TopologyTransition transition, const std::string &path="", Duration timeout=Duration(0)) -> std::pair< std::error_code, TopologyState >
Perform state transition on FairMQ devices in this topology for a specified topology path...
Definition: Topology.h:671
fair::mq::sdk::BasicTopology::ChangeState
auto ChangeState(const TopologyTransition transition, Duration timeout) -> std::pair< std::error_code, TopologyState >
Perform state transition on all FairMQ devices in this topology with a timeout.
Definition: Topology.h:690
fair::mq::sdk::DeviceStatus
Definition: Topology.h:73
fair::mq::sdk::MixedStateError
Definition: Error.h:28
fair::mq::sdk::BasicTopology::AsyncSetProperties
auto AsyncSetProperties(const DeviceProperties &props, const std::string &path, Duration timeout, CompletionToken &&token)
Initiate property update on selected FairMQ devices in this topology.
Definition: Topology.h:1149
fair::mq::sdk::BasicTopology::AsyncGetProperties
auto AsyncGetProperties(DevicePropertyQuery const &query, const std::string &path, Duration timeout, CompletionToken &&token)
Initiate property query on selected FairMQ devices in this topology.
Definition: Topology.h:996
fair::mq::sdk::DDSTopology::GetTasks
auto GetTasks(const std::string &="") const -> std::vector< DDSTask >
Get list of tasks in this topology, optionally matching provided path.
Definition: DDSTopology.cxx:67
fair::mq::sdk::cmd::PropertiesSet
Definition: Commands.h:329
fair::mq::sdk::cmd::Cmds
Definition: Commands.h:356
fair::mq::tools::SharedSemaphore
A simple copyable blocking semaphore.
Definition: Semaphore.h:45
fair::mq::sdk::BasicTopology::GetCurrentState
auto GetCurrentState() const -> TopologyState
Returns the current state of the topology.
Definition: Topology.h:698
fair::mq::sdk::BasicTopology::BasicTopology
BasicTopology(const Executor &ex, DDSTopology topo, DDSSession session, Allocator alloc=DefaultAllocator())
(Re)Construct a FairMQ topology from an existing DDS topology
Definition: Topology.h:169
fair::mq::sdk::BasicTopology::GetProperties
auto GetProperties(DevicePropertyQuery const &query, const std::string &path="", Duration timeout=Duration(0)) -> std::pair< std::error_code, GetPropertiesResult >
Query properties on selected FairMQ devices in this topology.
Definition: Topology.h:1048
fair::mq::sdk::RuntimeError
Definition: Error.h:20
fair::mq::sdk::cmd::StateChange
Definition: Commands.h:277
fair
Tools for interfacing containers to the transport via polymorphic allocators.
Definition: DeviceRunner.h:23
fair::mq::sdk::BasicTopology::SetProperties
auto SetProperties(DeviceProperties const &properties, const std::string &path="", Duration timeout=Duration(0)) -> std::pair< std::error_code, FailedDevices >
Set properties on selected FairMQ devices in this topology.
Definition: Topology.h:1201
fair::mq::sdk::BasicTopology::BasicTopology
BasicTopology(DDSTopology topo, DDSSession session)
(Re)Construct a FairMQ topology from an existing DDS topology
Definition: Topology.h:160
fair::mq::sdk::BasicTopology::WaitForState
auto WaitForState(const DeviceState targetCurrentState, const std::string &path="", Duration timeout=Duration(0)) -> std::error_code
Wait for selected FairMQ devices to reach given current state in this topology.
Definition: Topology.h:903
fair::mq::sdk::DDSEnvironment
Sets up the DDS environment (object helper)
Definition: DDSEnvironment.h:24
asio
Definition: Traits.h:16
fair::mq::sdk::cmd::StateChangeUnsubscription
Definition: Commands.h:255
fair::mq::sdk::DDSTopology
Represents a DDS topology.
Definition: DDSTopology.h:29
fair::mq::sdk::BasicTopology::AsyncChangeState
auto AsyncChangeState(const TopologyTransition transition, const std::string &path, CompletionToken &&token)
Initiate state transition on all FairMQ devices in this topology with a timeout.
Definition: Topology.h:661

privacy