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

Update to new DDS API, bump required version

Browse Source
This commit is contained in:
Alexey Rybalchenko 2019-12-04 12:48:50 +01:00
parent 5607d47664
commit f31be6d7a1
4 changed files with 26 additions and 40 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
CMakeLists.txt
View File

@ -86,7 +86,7 @@ if(BUILD_NANOMSG_TRANSPORT)
endif()
if(BUILD_SDK)
set(required_dds_version 2.5.46)
set(required_dds_version 2.5.100)
else()
set(required_dds_version 2.4)
endif()

34
examples/dds/fairmq-start-ex-dds.sh.in
View File

@ -30,14 +30,14 @@ echo "SESSION ID: ${DDS_SESSION_ID}"
trap "cleanup ${DDS_SESSION_ID}" EXIT
requiredNofAgents=12
requiredNofSlots=12
if [[ "$plugin" == "ssh" ]]; then
dds-submit -r ${plugin} -c @DATA_DIR@/ex-dds-hosts.cfg
else
dds-submit -r ${plugin} -n ${requiredNofAgents}
dds-submit -r ${plugin} --slots ${requiredNofSlots}
fi
echo "...waiting for ${requiredNofAgents} idle agents..."
@WAIT_COMMAND@ ${requiredNofAgents}
echo "...waiting for ${requiredNofSlots} slots..."
@WAIT_COMMAND@ ${requiredNofSlots}
topologyFile=@DATA_DIR@/ex-dds-topology.xml
echo "TOPOLOGY FILE: ${topologyFile}"
@ -48,29 +48,29 @@ echo "TOPOLOGY FILE: ${topologyFile}"
# dds-info --active-topology
dds-topology --activate ${topologyFile}
# dds-info --active-topology
# dds-info --wait-for-executing-agents ${requiredNofAgents}
# dds-info --wait-for-executing-agents ${requiredNofSlots}
sleep 1
echo "------------------------"
echo "...waiting for Topology to finish..."
# TODO Retrieve number of devices from DDS topology API instead of having the user pass it explicitely
fairmq-dds-command-ui -w "IDLE" -n ${requiredNofAgents}
fairmq-dds-command-ui -c i -w "INITIALIZING DEVICE" -n ${requiredNofAgents}
fairmq-dds-command-ui -c k -w "INITIALIZED" -n ${requiredNofAgents}
fairmq-dds-command-ui -c b -w "BOUND" -n ${requiredNofAgents}
fairmq-dds-command-ui -c x -w "DEVICE READY" -n ${requiredNofAgents}
fairmq-dds-command-ui -c j -w "READY" -n ${requiredNofAgents}
fairmq-dds-command-ui -w "IDLE" -n ${requiredNofSlots}
fairmq-dds-command-ui -c i -w "INITIALIZING DEVICE" -n ${requiredNofSlots}
fairmq-dds-command-ui -c k -w "INITIALIZED" -n ${requiredNofSlots}
fairmq-dds-command-ui -c b -w "BOUND" -n ${requiredNofSlots}
fairmq-dds-command-ui -c x -w "DEVICE READY" -n ${requiredNofSlots}
fairmq-dds-command-ui -c j -w "READY" -n ${requiredNofSlots}
fairmq-dds-command-ui -c r
sampler_and_sink="main/(Sampler|Sink)"
# processors="main/ProcessorGroup/Processor"
fairmq-dds-command-ui -p $sampler_and_sink -w "RUNNING->READY" -n 2
echo "...$sampler_and_sink are READY, sending shutdown..."
fairmq-dds-command-ui -c s -w "RUNNING->READY" -n ${requiredNofAgents}
fairmq-dds-command-ui -c t -w "DEVICE READY" -n ${requiredNofAgents}
fairmq-dds-command-ui -c d -w "IDLE" -n ${requiredNofAgents}
fairmq-dds-command-ui -c q -w "EXITING" -n ${requiredNofAgents}
echo "...waiting for ${requiredNofAgents} idle agents..."
@WAIT_COMMAND@ ${requiredNofAgents}
fairmq-dds-command-ui -c s -w "RUNNING->READY" -n ${requiredNofSlots}
fairmq-dds-command-ui -c t -w "DEVICE READY" -n ${requiredNofSlots}
fairmq-dds-command-ui -c d -w "IDLE" -n ${requiredNofSlots}
fairmq-dds-command-ui -c q -w "EXITING" -n ${requiredNofSlots}
echo "...waiting for ${requiredNofSlots} slots..."
@WAIT_COMMAND@ ${requiredNofSlots}
echo "------------------------"
# TODO Uncomment once DDS 2.6 is released

14
fairmq/sdk/DDSAgent.h
View File

@ -33,32 +33,24 @@ class DDSAgent
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; }
std::string GetDDSPath() const { return fDDSPath; }
bool IsLobbyLeader() const { return fLobbyLeader; }
std::chrono::milliseconds GetStartupTime() const { return fStartupTime; }
std::string GetUsername() const { return fUsername; }
@ -66,12 +58,9 @@ class DDSAgent
{
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;
}
@ -79,12 +68,9 @@ class DDSAgent
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;
};

16
fairmq/sdk/DDSSession.cxx
View File

@ -183,7 +183,8 @@ auto DDSSession::SubmitAgents(Quantity agents) -> void
SSubmitRequestData submitInfo;
submitInfo.m_rms = tools::ToString(GetRMSPlugin());
submitInfo.m_instances = agents;
submitInfo.m_instances = 1;
submitInfo.m_slots = agents; // TODO new api: get slots from agents
submitInfo.m_config = GetRMSConfig().string();
tools::SharedSemaphore blocker;
@ -210,9 +211,9 @@ auto DDSSession::RequestAgentCount() -> AgentCount
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;
count.active = res.m_activeSlotsCount;
count.idle = res.m_idleSlotsCount;
count.executing = res.m_executingSlotsCount;
return count;
}
@ -231,13 +232,11 @@ auto DDSSession::RequestAgentInfo() -> std::vector<DDSAgent>
*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
// a.m_nSlots
);
}
@ -254,7 +253,8 @@ auto DDSSession::RequestTaskInfo() -> std::vector<DDSTask>
std::vector<DDSTask> taskInfo;
taskInfo.reserve(res.size());
for (auto& a : res) {
taskInfo.emplace_back(a.m_taskID, 0);
//taskInfo.emplace_back(a.m_taskID, 0);
taskInfo.emplace_back(0, 0);
}
return taskInfo;