Mirror/FairMQ
1
0
Fork 0
mirror of https://github.com/FairRootGroup/FairMQ.git synced 2025-10-15 17:41:45 +00:00
Code Issues Packages Projects Releases Wiki Activity

SDK: Require DDS 2.5.36 and support new Tools API

Browse Source
This commit is contained in:
Alexey Rybalchenko
2019-08-13 19:07:48 +02:00
committed by Dennis Klein
parent 7f0237d97d
commit eaa8f5cbdd
10 changed files with 229 additions and 107 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
fairmq/SDK.h
View File

@@ -10,6 +10,7 @@
#define FAIR_MQ_SDK_H
// IWYU pragma: begin_exports
#include <fairmq/sdk/DDSAgent.h>
#include <fairmq/sdk/DDSInfo.h>
#include <fairmq/sdk/DDSEnvironment.h>
#include <fairmq/sdk/DDSSession.h>

1
fairmq/sdk/CMakeLists.txt
View File

@@ -15,6 +15,7 @@ set(target SDK)
set(SDK_PUBLIC_HEADER_FILES
../SDK.h
DDSAgent.h
DDSEnvironment.h
DDSSession.h
DDSTopology.h

94
fairmq/sdk/DDSAgent.h Normal file
View File

@@ -0,0 +1,94 @@
/********************************************************************************
* 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_DDSSAGENT_H
#define FAIR_MQ_SDK_DDSSAGENT_H
#include <fairmq/sdk/DDSSession.h>
#include <ostream>
#include <string>
#include <chrono>
#include <cstdint>
namespace fair {
namespace mq {
namespace sdk {
/**
* @class DDSAgent <fairmq/sdk/DDSAgent.h>
* @brief Represents a DDS agent
*/
class DDSAgent
{
public:
using Id = uint64_t;
using Pid = uint32_t;
explicit DDSAgent(DDSSession session,
Id id,
Pid pid,
std::string state,
std::string path,
std::string host,
bool lobbyLeader,
std::chrono::milliseconds startupTime,
Id taskId,
std::string username)
: fSession(std::move(session))
, fId(id)
, fPid(pid)
, fState(std::move(state))
, fDDSPath(std::move(path))
, fHost(std::move(host))
, fLobbyLeader(lobbyLeader)
, fStartupTime(startupTime)
, fTaskId(taskId)
, fUsername(std::move(username))
{}
DDSSession GetSession() const { return fSession; }
Id GetId() const { return fId; }
Pid GetPid() const { return fPid; }
std::string GetState() const { return fState; }
std::string GetHost() const { return fHost; }
bool IsLobbyLeader() const { return fLobbyLeader; }
std::chrono::milliseconds GetStartupTime() const { return fStartupTime; }
std::string GetUsername() const { return fUsername; }
friend auto operator<<(std::ostream& os, const DDSAgent& agent) -> std::ostream&
{
return os << "DDSAgent id: " << agent.fId
<< ", pid: " << agent.fPid
<< ", state: " << agent.fState
<< ", path: " << agent.fDDSPath
<< ", host: " << agent.fHost
<< ", lobbyLeader: " << agent.fLobbyLeader
<< ", startupTime: " << agent.fStartupTime.count()
<< ", taskId: " << agent.fTaskId
<< ", username: " << agent.fUsername;
}
private:
DDSSession fSession;
Id fId;
Pid fPid;
std::string fState;
std::string fDDSPath;
std::string fHost;
bool fLobbyLeader;
std::chrono::milliseconds fStartupTime;
Id fTaskId;
std::string fUsername;
};
} // namespace sdk
} // namespace mq
} // namespace fair
#endif /* FAIR_MQ_SDK_DDSSAGENT_H */

167
fairmq/sdk/DDSSession.cxx
View File

@@ -8,6 +8,7 @@
#include "DDSSession.h"
#include <fairmq/sdk/DDSAgent.h>
#include <fairmq/sdk/DDSEnvironment.h>
#include <fairmq/sdk/DDSTopology.h>
#include <fairmq/Tools.h>
@@ -180,76 +181,110 @@ auto DDSSession::SubmitAgents(Quantity agents) -> void
// Requesting to submit 0 agents is not meaningful
assert(agents > 0);
dds::tools_api::SSubmitRequestData submitInfo;
using namespace dds::tools_api;
SSubmitRequestData submitInfo;
submitInfo.m_rms = tools::ToString(GetRMSPlugin());
submitInfo.m_instances = agents;
submitInfo.m_config = GetRMSConfig().string();
tools::Semaphore blocker;
auto submitRequest = dds::tools_api::SSubmitRequest::makeRequest(submitInfo);
submitRequest->setMessageCallback(
[](const dds::tools_api::SMessageResponseData& message) { LOG(debug) << message; });
auto submitRequest = SSubmitRequest::makeRequest(submitInfo);
submitRequest->setMessageCallback([](const SMessageResponseData& message){
LOG(debug) << message.m_msg;
});
submitRequest->setDoneCallback([&]() {
LOG(debug) << agents << " Agents submitted";
blocker.Signal();
});
fImpl->fSession->sendRequest<dds::tools_api::SSubmitRequest>(submitRequest);
fImpl->fSession->sendRequest<SSubmitRequest>(submitRequest);
blocker.Wait();
// perfect
WaitForIdleAgents(agents);
}
auto DDSSession::RequestAgentInfo() -> AgentInfo
auto DDSSession::RequestAgentCount() -> AgentCount
{
dds::tools_api::SAgentInfoRequestData agentInfoInfo;
tools::Semaphore blocker;
AgentInfo info;
auto agentInfoRequest = dds::tools_api::SAgentInfoRequest::makeRequest(agentInfoInfo);
agentInfoRequest->setResponseCallback(
[this, &info](const dds::tools_api::SAgentInfoResponseData& _response) {
if (_response.m_index == 0) {
info.activeAgentsCount = _response.m_activeAgentsCount;
info.idleAgentsCount = _response.m_idleAgentsCount;
info.executingAgentsCount = _response.m_executingAgentsCount;
info.agents.reserve(_response.m_activeAgentsCount);
}
info.agents.emplace_back(*this, _response.m_agentInfo);
});
agentInfoRequest->setMessageCallback(
[](const dds::tools_api::SMessageResponseData& _message) { LOG(debug) << _message; });
agentInfoRequest->setDoneCallback([&]() { blocker.Signal(); });
fImpl->fSession->sendRequest<dds::tools_api::SAgentInfoRequest>(agentInfoRequest);
blocker.Wait();
using namespace dds::tools_api;
return info;
SAgentCountRequest::response_t res;
fImpl->fSession->syncSendRequest<SAgentCountRequest>(SAgentCountRequest::request_t(), res);
AgentCount count;
count.active = res.m_activeAgentsCount;
count.idle = res.m_idleAgentsCount;
count.executing = res.m_executingAgentsCount;
return count;
}
auto DDSSession::RequestAgentInfo() -> std::vector<DDSAgent>
{
using namespace dds::tools_api;
SAgentInfoRequest::responseVector_t res;
fImpl->fSession->syncSendRequest<SAgentInfoRequest>(SAgentInfoRequest::request_t(), res);
std::vector<DDSAgent> agentInfo;
agentInfo.reserve(res.size());
for (const auto& a : res) {
agentInfo.emplace_back(
*this,
a.m_agentID,
a.m_agentPid,
a.m_agentState,
a.m_DDSPath,
a.m_host,
a.m_lobbyLeader,
a.m_startUpTime,
a.m_taskID,
a.m_username
);
}
return agentInfo;
}
auto DDSSession::RequestTaskInfo() -> std::vector<DDSTask>
{
using namespace dds::tools_api;
SAgentInfoRequest::responseVector_t res;
fImpl->fSession->syncSendRequest<SAgentInfoRequest>(SAgentInfoRequest::request_t(), res);
std::vector<DDSTask> taskInfo;
taskInfo.reserve(res.size());
for (auto& a : res) {
taskInfo.emplace_back(a.m_taskID);
}
return taskInfo;
}
auto DDSSession::RequestCommanderInfo() -> CommanderInfo
{
dds::tools_api::SCommanderInfoRequestData commanderInfoInfo;
using namespace dds::tools_api;
SCommanderInfoRequestData commanderInfo;
tools::Semaphore blocker;
std::string error;
auto commanderInfoRequest =
dds::tools_api::SCommanderInfoRequest::makeRequest(commanderInfoInfo);
auto commanderInfoRequest = SCommanderInfoRequest::makeRequest(commanderInfo);
CommanderInfo info;
commanderInfoRequest->setResponseCallback(
[&info](const dds::tools_api::SCommanderInfoResponseData& _response) {
info.pid = _response.m_pid;
info.activeTopologyName = _response.m_activeTopologyName;
});
commanderInfoRequest->setMessageCallback(
[&](const dds::tools_api::SMessageResponseData& _message) {
if (_message.m_severity == dds::intercom_api::EMsgSeverity::error) {
error = _message.m_msg;
blocker.Signal();
} else {
LOG(debug) << _message;
}
});
commanderInfoRequest->setResponseCallback([&info](const SCommanderInfoResponseData& _response) {
info.pid = _response.m_pid;
info.activeTopologyName = _response.m_activeTopologyName;
});
commanderInfoRequest->setMessageCallback([&](const SMessageResponseData& _message) {
if (_message.m_severity == dds::intercom_api::EMsgSeverity::error) {
error = _message.m_msg;
blocker.Signal();
} else {
LOG(debug) << _message.m_msg;
}
});
commanderInfoRequest->setDoneCallback([&]() { blocker.Signal(); });
fImpl->fSession->sendRequest<dds::tools_api::SCommanderInfoRequest>(commanderInfoRequest);
fImpl->fSession->sendRequest<SCommanderInfoRequest>(commanderInfoRequest);
blocker.Wait();
if (!error.empty()) {
@@ -261,38 +296,41 @@ auto DDSSession::RequestCommanderInfo() -> CommanderInfo
auto DDSSession::WaitForExecutingAgents(Quantity minCount) -> void
{
auto info(RequestAgentInfo());
auto count(RequestAgentCount());
int interval(8);
while (info.executingAgentsCount < minCount) {
while (count.executing < minCount) {
std::this_thread::sleep_for(std::chrono::milliseconds(interval));
interval = std::min(256, interval * 2);
info = RequestAgentInfo();
count = RequestAgentCount();
}
}
auto DDSSession::WaitForIdleAgents(Quantity minCount) -> void
{
auto info(RequestAgentInfo());
auto count(RequestAgentCount());
int interval(8);
while (info.idleAgentsCount < minCount) {
while (count.idle < minCount) {
std::this_thread::sleep_for(std::chrono::milliseconds(interval));
interval = std::min(256, interval * 2);
info = RequestAgentInfo();
count = RequestAgentCount();
}
}
auto DDSSession::ActivateTopology(DDSTopology topo) -> void
{
dds::tools_api::STopologyRequestData topologyInfo;
topologyInfo.m_updateType = dds::tools_api::STopologyRequestData::EUpdateType::ACTIVATE;
using namespace dds::tools_api;
STopologyRequestData topologyInfo;
topologyInfo.m_updateType = STopologyRequestData::EUpdateType::ACTIVATE;
topologyInfo.m_topologyFile = topo.GetTopoFile().string();
tools::Semaphore blocker;
auto topologyRequest = dds::tools_api::STopologyRequest::makeRequest(topologyInfo);
topologyRequest->setMessageCallback(
[](const dds::tools_api::SMessageResponseData& _message) { LOG(debug) << _message; });
auto topologyRequest = STopologyRequest::makeRequest(topologyInfo);
topologyRequest->setMessageCallback([](const SMessageResponseData& _message) {
LOG(debug) << _message.m_msg;
});
topologyRequest->setDoneCallback([&]() { blocker.Signal(); });
fImpl->fSession->sendRequest<dds::tools_api::STopologyRequest>(topologyRequest);
fImpl->fSession->sendRequest<STopologyRequest>(topologyRequest);
blocker.Wait();
WaitForExecutingAgents(topo.GetNumRequiredAgents());
@@ -337,21 +375,6 @@ auto operator<<(std::ostream& os, const DDSSession& session) -> std::ostream&
return os << "$DDS_SESSION_ID: " << session.GetId();
}
auto DDSAgent::GetSession() const -> DDSSession
{
return fSession;
}
auto DDSAgent::GetInfoStr() const -> std::string
{
return fInfoStr;
}
auto operator<<(std::ostream& os, const DDSAgent& agent) -> std::ostream&
{
return os << agent.GetInfoStr();
}
} // namespace sdk
} // namespace mq
} // namespace fair

49
fairmq/sdk/DDSSession.h
View File

@@ -21,6 +21,7 @@
#include <stdexcept>
#include <string>
#include <functional>
#include <vector>
namespace fair {
namespace mq {
@@ -47,6 +48,20 @@ class DDSTask
{
public:
using Id = std::uint64_t;
explicit DDSTask(Id id)
: fId(id)
{}
Id GetId() const { return fId; }
friend auto operator<<(std::ostream& os, const DDSTask& task) -> std::ostream&
{
return os << "DDSTask id: " << task.fId;
}
private:
Id fId;
};
class DDSChannel
@@ -84,13 +99,14 @@ class DDSSession
auto StopOnDestruction(bool stop = true) -> void;
auto IsRunning() const -> bool;
auto SubmitAgents(Quantity agents) -> void;
struct AgentInfo {
Quantity idleAgentsCount = 0;
Quantity activeAgentsCount = 0;
Quantity executingAgentsCount = 0;
std::vector<DDSAgent> agents;
struct AgentCount {
Quantity idle = 0;
Quantity active = 0;
Quantity executing = 0;
};
auto RequestAgentInfo() -> AgentInfo;
auto RequestAgentCount() -> AgentCount;
auto RequestAgentInfo() -> std::vector<DDSAgent>;
auto RequestTaskInfo() -> std::vector<DDSTask>;
struct CommanderInfo {
int pid = -1;
std::string activeTopologyName;
@@ -117,27 +133,6 @@ class DDSSession
std::shared_ptr<Impl> fImpl;
};
/**
* @class DDSAgent DDSSession.h <fairmq/sdk/DDSSession.h>
* @brief Represents a DDS agent
*/
class DDSAgent
{
public:
explicit DDSAgent(DDSSession session, std::string infostr)
: fInfoStr(std::move(infostr))
, fSession(std::move(session))
{}
auto GetSession() const -> DDSSession;
auto GetInfoStr() const -> std::string;
friend auto operator<<(std::ostream& os, const DDSAgent& plugin) -> std::ostream&;
private:
std::string fInfoStr;
DDSSession fSession;
};
} // namespace sdk
} // namespace mq
} // namespace fair

8
fairmq/sdk/Topology.cxx
View File

@@ -231,18 +231,18 @@ void Topology::WaitForState()
fExecutionCV.wait(lock);
}
}
LOG(debug) << "WaitForState shutting down";
LOG(debug) << "Topology::WaitForState shutting down";
};
void Topology::AddNewStateEntry(DDSTask::Id senderId, const std::string& state)
void Topology::AddNewStateEntry(DDSTask::Id taskId, const std::string& state)
{
std::size_t pos = state.find("->");
std::string endState = state.substr(pos + 2);
// LOG(debug) << "Adding new state entry: " << senderId << ", " << state << ", end state: " << endState;
// LOG(debug) << "Adding new state entry: " << taskId << ", " << state << ", end state: " << endState;
{
try {
std::unique_lock<std::mutex> lock(fMtx);
fState[senderId] = DeviceStatus{ true, fair::mq::GetState(endState) };
fState[taskId] = DeviceStatus{ true, fair::mq::GetState(endState) };
} catch (const std::exception& e) {
LOG(error) << "Exception in AddNewStateEntry: " << e.what();
}

2
fairmq/sdk/Topology.h
View File

@@ -152,7 +152,7 @@ class Topology
std::string fStateChangeError;
void WaitForState();
void AddNewStateEntry(DDSTask::Id senderId, const std::string& state);
void AddNewStateEntry(DDSTask::Id taskId, const std::string& state);
};
using Topo = Topology;