mirror of
https://github.com/FairRootGroup/FairMQ.git
synced 2025-10-15 09:31:45 +00:00
Compare commits
9 Commits
Author | SHA1 | Date | |
---|---|---|---|
|
e6dede492e | ||
|
f195eeac66 | ||
|
4d1e7b9cdb | ||
|
50be386191 | ||
|
f31be6d7a1 | ||
|
5607d47664 | ||
|
0f4595b8c1 | ||
|
b0b271d1f4 | ||
|
073f5e5c0e |
@@ -85,21 +85,15 @@ if(BUILD_NANOMSG_TRANSPORT)
|
||||
)
|
||||
endif()
|
||||
|
||||
if(BUILD_SDK)
|
||||
set(required_dds_version 2.5.46)
|
||||
else()
|
||||
set(required_dds_version 2.4)
|
||||
endif()
|
||||
|
||||
if(BUILD_SDK_COMMANDS)
|
||||
find_package2(PRIVATE Flatbuffers REQUIRED)
|
||||
endif()
|
||||
|
||||
if(BUILD_DDS_PLUGIN OR BUILD_SDK)
|
||||
find_package2(PRIVATE DDS REQUIRED
|
||||
VERSION ${required_dds_version}
|
||||
VERSION 3.0
|
||||
)
|
||||
set(DDS_Boost_COMPONENTS system log log_setup)
|
||||
set(DDS_Boost_COMPONENTS system log log_setup regex filesystem thread)
|
||||
set(DDS_Boost_VERSION 1.67)
|
||||
endif()
|
||||
|
||||
|
2
Dart.sh
2
Dart.sh
@@ -65,7 +65,7 @@ if [ "$1" == "alfa_ci" ]; then
|
||||
export ctest_model=Experimental
|
||||
elif [ "$1" == "codecov" ]; then
|
||||
export ctest_model=Profile
|
||||
export do_codecov_upload=1
|
||||
export do_codecov_upload=0
|
||||
else
|
||||
export ctest_model=$1
|
||||
fi
|
||||
|
@@ -28,12 +28,6 @@ target_link_libraries(fairmq-ex-dds-sink PRIVATE ExampleDDSLib)
|
||||
|
||||
add_custom_target(ExampleDDS DEPENDS fairmq-ex-dds-sampler fairmq-ex-dds-processor fairmq-ex-dds-sink)
|
||||
|
||||
if(DDS_VERSION VERSION_LESS 2.5.25)
|
||||
set(WAIT_COMMAND "dds-info --wait-for-idle-agents")
|
||||
else()
|
||||
set(WAIT_COMMAND "dds-info --idle-count --wait")
|
||||
endif()
|
||||
|
||||
set(BIN_DIR ${CMAKE_CURRENT_BINARY_DIR}:${CMAKE_BINARY_DIR}/fairmq/plugins/DDS)
|
||||
set(DATA_DIR ${CMAKE_CURRENT_BINARY_DIR})
|
||||
configure_file(${CMAKE_CURRENT_SOURCE_DIR}/ex-dds-topology.xml ${CMAKE_CURRENT_BINARY_DIR}/ex-dds-topology.xml @ONLY)
|
||||
@@ -47,7 +41,6 @@ if(DDS_FOUND)
|
||||
add_test(NAME Example.DDS.localhost COMMAND ${CMAKE_CURRENT_BINARY_DIR}/fairmq-start-ex-dds.sh localhost)
|
||||
set_tests_properties(Example.DDS.localhost PROPERTIES
|
||||
TIMEOUT 15
|
||||
RUN_SERIAL true
|
||||
PASS_REGULAR_EXPRESSION "Example successful"
|
||||
)
|
||||
endif()
|
||||
|
@@ -30,52 +30,50 @@ 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} idle slots..."
|
||||
dds-info --idle-count --wait ${requiredNofSlots}
|
||||
|
||||
topologyFile=@DATA_DIR@/ex-dds-topology.xml
|
||||
echo "TOPOLOGY FILE: ${topologyFile}"
|
||||
# TODO Uncomment once DDS 2.6 is released
|
||||
# echo "TOPOLOGY NAME: $(dds-topology --disable-validation --topology-name ${topologyFile})"
|
||||
echo "TOPOLOGY NAME: $(dds-topology --disable-validation --topology-name ${topologyFile})"
|
||||
|
||||
# TODO Uncomment once DDS 2.6 is released
|
||||
# dds-info --active-topology
|
||||
dds-info --active-topology
|
||||
dds-topology --activate ${topologyFile}
|
||||
# dds-info --active-topology
|
||||
# dds-info --wait-for-executing-agents ${requiredNofAgents}
|
||||
sleep 1
|
||||
dds-info --active-topology
|
||||
echo "...waiting for ${requiredNofSlots} executing slots..."
|
||||
dds-info --executing-count --wait ${requiredNofSlots}
|
||||
|
||||
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} idle slots..."
|
||||
dds-info --idle-count --wait ${requiredNofSlots}
|
||||
echo "------------------------"
|
||||
|
||||
# TODO Uncomment once DDS 2.6 is released
|
||||
# dds-info --active-topology
|
||||
dds-info --active-topology
|
||||
dds-topology --stop
|
||||
# dds-info --active-topology
|
||||
dds-info --active-topology
|
||||
|
||||
dds-agent-cmd getlog -a
|
||||
logDir="${wrkDir}/logs"
|
||||
|
@@ -58,6 +58,12 @@ if(BUILD_FAIRMQ OR BUILD_SDK)
|
||||
${TOOLS_PUBLIC_HEADER_FILES}
|
||||
)
|
||||
target_compile_definitions(${target} PUBLIC BOOST_ERROR_CODE_HEADER_ONLY)
|
||||
# workaround https://github.com/boostorg/asio/commit/43874d5497414c67655d901e48c939ef01337edb
|
||||
if( Boost_VERSION VERSION_LESS 1.69
|
||||
AND CMAKE_CXX_COMPILER_ID STREQUAL AppleClang
|
||||
AND CMAKE_CXX_COMPILER_VERSION VERSION_GREATER_EQUAL 10.0.1)
|
||||
target_compile_definitions(${target} PUBLIC BOOST_ASIO_HAS_STD_STRING_VIEW)
|
||||
endif()
|
||||
target_include_directories(${target}
|
||||
PUBLIC
|
||||
$<BUILD_INTERFACE:${CMAKE_SOURCE_DIR}>
|
||||
|
@@ -60,10 +60,10 @@ class Plugin
|
||||
friend auto operator!=(const Plugin& lhs, const Plugin& rhs) -> bool { return !(lhs == rhs); }
|
||||
friend auto operator<<(std::ostream& os, const Plugin& p) -> std::ostream&
|
||||
{
|
||||
return os << "'" << p.GetName() << "', "
|
||||
<< "version '" << p.GetVersion() << "', "
|
||||
return os << "'" << p.GetName() << "', "
|
||||
<< "version '" << p.GetVersion() << "', "
|
||||
<< "maintainer '" << p.GetMaintainer() << "', "
|
||||
<< "homepage '" << p.GetHomepage() << "'";
|
||||
<< "homepage '" << p.GetHomepage() << "'";
|
||||
}
|
||||
static auto NoProgramOptions() -> ProgOptions { return boost::none; }
|
||||
|
||||
@@ -80,7 +80,6 @@ class Plugin
|
||||
auto StealDeviceControl() -> void { fPluginServices->StealDeviceControl(fkName); };
|
||||
auto ReleaseDeviceControl() -> void { fPluginServices->ReleaseDeviceControl(fkName); };
|
||||
auto ChangeDeviceState(const DeviceStateTransition next) -> bool { return fPluginServices->ChangeDeviceState(fkName, next); }
|
||||
void TransitionDeviceStateTo(const DeviceState state) { return fPluginServices->TransitionDeviceStateTo(fkName, state); }
|
||||
auto SubscribeToDeviceStateChange(std::function<void(DeviceState)> callback) -> void { fPluginServices->SubscribeToDeviceStateChange(fkName, callback); }
|
||||
auto UnsubscribeFromDeviceStateChange() -> void { fPluginServices->UnsubscribeFromDeviceStateChange(fkName); }
|
||||
|
||||
|
@@ -28,22 +28,6 @@ auto PluginServices::ChangeDeviceState(const string& controller, const DeviceSta
|
||||
}
|
||||
}
|
||||
|
||||
void PluginServices::TransitionDeviceStateTo(const std::string& controller, DeviceState state)
|
||||
{
|
||||
lock_guard<mutex> lock{fDeviceControllerMutex};
|
||||
|
||||
if (!fDeviceController) fDeviceController = controller;
|
||||
|
||||
if (fDeviceController == controller) {
|
||||
fDevice.TransitionTo(state);
|
||||
} else {
|
||||
throw DeviceControlError{tools::ToString(
|
||||
"Plugin '", controller, "' is not allowed to change device states. ",
|
||||
"Currently, plugin '", *fDeviceController, "' has taken control."
|
||||
)};
|
||||
}
|
||||
}
|
||||
|
||||
auto PluginServices::TakeDeviceControl(const string& controller) -> void
|
||||
{
|
||||
lock_guard<mutex> lock{fDeviceControllerMutex};
|
||||
|
@@ -124,8 +124,6 @@ class PluginServices
|
||||
/// If the device control role has not been taken yet, calling this function will take over control implicitely.
|
||||
auto ChangeDeviceState(const std::string& controller, const DeviceStateTransition next) -> bool;
|
||||
|
||||
void TransitionDeviceStateTo(const std::string& controller, DeviceState state);
|
||||
|
||||
/// @brief Subscribe with a callback to device state changes
|
||||
/// @param subscriber id
|
||||
/// @param callback
|
||||
|
@@ -37,16 +37,6 @@ DDS::DDS(const string& name,
|
||||
const string& homepage,
|
||||
PluginServices* pluginServices)
|
||||
: Plugin(name, version, maintainer, homepage, pluginServices)
|
||||
, fTransitions({"INIT DEVICE",
|
||||
"COMPLETE INIT",
|
||||
"BIND",
|
||||
"CONNECT",
|
||||
"INIT TASK",
|
||||
"RUN",
|
||||
"STOP",
|
||||
"RESET TASK",
|
||||
"RESET DEVICE",
|
||||
"END"})
|
||||
, fCurrentState(DeviceState::Idle)
|
||||
, fLastState(DeviceState::Idle)
|
||||
, fDeviceTerminationRequested(false)
|
||||
@@ -71,15 +61,13 @@ DDS::DDS(const string& name,
|
||||
}
|
||||
|
||||
auto control = GetProperty<string>("control");
|
||||
bool staticMode(false);
|
||||
if (control == "static") {
|
||||
LOG(debug) << "Running DDS controller: static";
|
||||
staticMode = true;
|
||||
LOG(error) << "DDS Plugin: static mode is not supported";
|
||||
throw invalid_argument("DDS Plugin: static mode is not supported");
|
||||
} else if (control == "dynamic" || control == "external" || control == "interactive") {
|
||||
LOG(debug) << "Running DDS controller: external";
|
||||
} else {
|
||||
LOG(error) << "Unrecognized control mode '" << control << "' requested. " << "Ignoring and falling back to static control mode.";
|
||||
staticMode = true;
|
||||
LOG(error) << "Unrecognized control mode '" << control << "' requested. " << "Ignoring and starting in external control mode.";
|
||||
}
|
||||
|
||||
SubscribeForCustomCommands();
|
||||
@@ -87,7 +75,6 @@ DDS::DDS(const string& name,
|
||||
|
||||
// subscribe to device state changes, pushing new state changes into the event queue
|
||||
SubscribeToDeviceStateChange([&](DeviceState newState) {
|
||||
fStateQueue.Push(newState);
|
||||
switch (newState) {
|
||||
case DeviceState::Bound:
|
||||
// Receive addresses of connecting channels from DDS
|
||||
@@ -130,11 +117,7 @@ DDS::DDS(const string& name,
|
||||
}
|
||||
});
|
||||
|
||||
if (staticMode) {
|
||||
fControllerThread = thread(&DDS::StaticControl, this);
|
||||
} else {
|
||||
StartWorkerThread();
|
||||
}
|
||||
StartWorkerThread();
|
||||
|
||||
fDDS.Start();
|
||||
} catch (PluginServices::DeviceControlError& e) {
|
||||
@@ -166,26 +149,6 @@ auto DDS::WaitForExitingAck() -> void
|
||||
[this]() { return fExitingAckedByLastExternalController; });
|
||||
}
|
||||
|
||||
auto DDS::StaticControl() -> void
|
||||
{
|
||||
try {
|
||||
TransitionDeviceStateTo(DeviceState::Running);
|
||||
|
||||
// wait until stop signal
|
||||
unique_lock<mutex> lock(fStopMutex);
|
||||
while (!fDeviceTerminationRequested) {
|
||||
fStopCondition.wait_for(lock, chrono::seconds(1));
|
||||
}
|
||||
LOG(debug) << "Stopping DDS plugin static controller";
|
||||
} catch (DeviceErrorState&) {
|
||||
ReleaseDeviceControl();
|
||||
} catch (exception& e) {
|
||||
ReleaseDeviceControl();
|
||||
LOG(error) << "Error: " << e.what() << "\n";
|
||||
return;
|
||||
}
|
||||
}
|
||||
|
||||
auto DDS::FillChannelContainers() -> void
|
||||
{
|
||||
try {
|
||||
@@ -373,6 +336,7 @@ auto DDS::SubscribeForCustomCommands() -> void
|
||||
inCmds.Deserialize(cmdStr);
|
||||
|
||||
for (const auto& cmd : inCmds) {
|
||||
// LOG(info) << "Received command type: '" << cmd->GetType() << "' from " << senderId;
|
||||
switch (cmd->GetType()) {
|
||||
case Type::check_state: {
|
||||
fDDS.Send(Cmds(make<CurrentState>(id, GetCurrentDeviceState())).Serialize(), to_string(senderId));
|
||||
@@ -387,6 +351,10 @@ auto DDS::SubscribeForCustomCommands() -> void
|
||||
Cmds outCmds(make<TransitionStatus>(id, Result::Failure, transition));
|
||||
fDDS.Send(outCmds.Serialize(), to_string(senderId));
|
||||
}
|
||||
{
|
||||
lock_guard<mutex> lock{fStateChangeSubscriberMutex};
|
||||
fLastExternalController = senderId;
|
||||
}
|
||||
}
|
||||
break;
|
||||
case Type::dump_config: {
|
||||
|
@@ -13,8 +13,7 @@
|
||||
#include <fairmq/StateQueue.h>
|
||||
#include <fairmq/Version.h>
|
||||
|
||||
#include <DDS/dds_env_prop.h>
|
||||
#include <DDS/dds_intercom.h>
|
||||
#include <dds/dds.h>
|
||||
|
||||
#include <boost/asio/executor.hpp>
|
||||
#include <boost/asio/executor_work_guard.hpp>
|
||||
@@ -133,7 +132,6 @@ class DDS : public Plugin
|
||||
~DDS();
|
||||
|
||||
private:
|
||||
auto StaticControl() -> void;
|
||||
auto WaitForExitingAck() -> void;
|
||||
auto StartWorkerThread() -> void;
|
||||
|
||||
@@ -154,14 +152,8 @@ class DDS : public Plugin
|
||||
std::unordered_map<std::string, int> fI;
|
||||
std::unordered_map<std::string, IofN> fIofN;
|
||||
|
||||
std::mutex fStopMutex;
|
||||
std::condition_variable fStopCondition;
|
||||
|
||||
const std::set<std::string> fTransitions;
|
||||
|
||||
std::thread fControllerThread;
|
||||
DeviceState fCurrentState, fLastState;
|
||||
fair::mq::StateQueue fStateQueue;
|
||||
|
||||
std::atomic<bool> fDeviceTerminationRequested;
|
||||
|
||||
|
@@ -9,7 +9,7 @@
|
||||
#include <fairmq/sdk/commands/Commands.h>
|
||||
#include <fairmq/States.h>
|
||||
|
||||
#include <DDS/dds_intercom.h>
|
||||
#include <dds/dds.h>
|
||||
|
||||
#include <boost/program_options.hpp>
|
||||
|
||||
@@ -74,7 +74,8 @@ void printControlsHelp()
|
||||
cout << "To quit press Ctrl+C" << endl;
|
||||
}
|
||||
|
||||
void sendCommand(const string& commandIn, const string& topologyPath, CCustomCmd& ddsCustomCmd) {
|
||||
void sendCommand(const string& commandIn, const string& topologyPath, CCustomCmd& ddsCustomCmd)
|
||||
{
|
||||
char c;
|
||||
string command(commandIn);
|
||||
TerminalConfig tconfig;
|
||||
@@ -159,8 +160,6 @@ struct WaitMode
|
||||
|
||||
void Run(const chrono::milliseconds& timeout, const string& topologyPath, CCustomCmd& ddsCustomCmd, unsigned int numDevices, const string& command = "")
|
||||
{
|
||||
StateSubscription stateSubscription(topologyPath, ddsCustomCmd);
|
||||
|
||||
if (command != "") {
|
||||
sendCommand(command, topologyPath, ddsCustomCmd);
|
||||
}
|
||||
@@ -288,13 +287,19 @@ int main(int argc, char* argv[])
|
||||
}
|
||||
break;
|
||||
case Type::transition_status: {
|
||||
// if (static_cast<TransitionStatus&>(*cmd).GetResult() == Result::Ok) {
|
||||
// cout << "Device " << static_cast<TransitionStatus&>(*cmd).GetDeviceId() << " started to transition with " << static_cast<TransitionStatus&>(*cmd).GetTransition() << endl;
|
||||
// } else {
|
||||
// cout << "Device " << static_cast<TransitionStatus&>(*cmd).GetDeviceId() << " cannot transition with " << static_cast<TransitionStatus&>(*cmd).GetTransition() << endl;
|
||||
// }
|
||||
if (static_cast<TransitionStatus&>(*cmd).GetResult() == Result::Ok) {
|
||||
cout << "Device " << static_cast<TransitionStatus&>(*cmd).GetDeviceId() << " started to transition with " << static_cast<TransitionStatus&>(*cmd).GetTransition() << endl;
|
||||
} else {
|
||||
cout << "Device " << static_cast<TransitionStatus&>(*cmd).GetDeviceId() << " cannot transition with " << static_cast<TransitionStatus&>(*cmd).GetTransition() << endl;
|
||||
}
|
||||
}
|
||||
break;
|
||||
case Type::current_state:
|
||||
cout << "Device " << static_cast<CurrentState&>(*cmd).GetDeviceId() << " is in " << static_cast<CurrentState&>(*cmd).GetCurrentState() << " state" << endl;
|
||||
break;
|
||||
case Type::config:
|
||||
cout << "Received config for device " << static_cast<Config&>(*cmd).GetDeviceId() << ":\n" << static_cast<Config&>(*cmd).GetConfig() << endl;
|
||||
break;
|
||||
default:
|
||||
cout << "Unexpected/unknown command received: " << cmd->GetType() << endl;
|
||||
cout << "Origin: " << senderId << endl;
|
||||
@@ -305,6 +310,8 @@ int main(int argc, char* argv[])
|
||||
|
||||
service.start(sessionID);
|
||||
|
||||
StateSubscription stateSubscription(topologyPath, ddsCustomCmd);
|
||||
|
||||
if (targetState == "") {
|
||||
sendCommand(command, topologyPath, ddsCustomCmd);
|
||||
} else {
|
||||
|
@@ -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;
|
||||
};
|
||||
|
||||
|
@@ -8,15 +8,11 @@
|
||||
|
||||
#include "DDSEnvironment.h"
|
||||
|
||||
#include <cstdlib>
|
||||
#include <dds/dds.h>
|
||||
#include <fairlogger/Logger.h>
|
||||
#include <fairmq/Tools.h>
|
||||
#include <fairmq/sdk/DDSInfo.h>
|
||||
|
||||
#include <fairlogger/Logger.h>
|
||||
|
||||
#include <DDS/Tools.h>
|
||||
#include <DDS/dds_intercom.h>
|
||||
|
||||
#include <cstdlib>
|
||||
#include <sstream>
|
||||
#include <utility>
|
||||
|
||||
|
@@ -8,11 +8,11 @@
|
||||
|
||||
#include "DDSSession.h"
|
||||
|
||||
#include <DDS/Tools.h>
|
||||
#include <boost/process.hpp>
|
||||
#include <boost/uuid/uuid_io.hpp>
|
||||
#include <cassert>
|
||||
#include <cstdlib>
|
||||
#include <dds/dds.h>
|
||||
#include <fairlogger/Logger.h>
|
||||
#include <fairmq/Tools.h>
|
||||
#include <fairmq/sdk/DDSAgent.h>
|
||||
@@ -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,9 @@ 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);
|
||||
(void)a;
|
||||
taskInfo.emplace_back(0, 0);
|
||||
}
|
||||
|
||||
return taskInfo;
|
||||
|
@@ -9,18 +9,14 @@
|
||||
#include "DDSTopology.h"
|
||||
|
||||
#include <boost/range/iterator_range.hpp>
|
||||
|
||||
#include <fairmq/sdk/DDSEnvironment.h>
|
||||
#include <fairmq/Tools.h>
|
||||
|
||||
#include <dds/dds.h>
|
||||
#include <fairlogger/Logger.h>
|
||||
|
||||
#include <DDS/Topology.h>
|
||||
|
||||
#include <fairmq/Tools.h>
|
||||
#include <fairmq/sdk/DDSEnvironment.h>
|
||||
#include <memory>
|
||||
#include <sstream>
|
||||
#include <stdexcept>
|
||||
#include <utility>
|
||||
#include <memory>
|
||||
|
||||
namespace fair {
|
||||
namespace mq {
|
||||
|
@@ -8,8 +8,7 @@
|
||||
|
||||
#include "Topology.h"
|
||||
|
||||
#include <DDS/Tools.h>
|
||||
#include <DDS/Topology.h>
|
||||
#include <dds/dds.h>
|
||||
|
||||
namespace fair {
|
||||
namespace mq {
|
||||
|
@@ -8,8 +8,7 @@
|
||||
|
||||
#include "Fixtures.h"
|
||||
|
||||
#include <DDS/Topology.h>
|
||||
#include <DDS/Tools.h>
|
||||
#include <dds/dds.h>
|
||||
|
||||
namespace {
|
||||
|
||||
|
@@ -9,8 +9,7 @@
|
||||
#include "Fixtures.h"
|
||||
|
||||
#include <asio.hpp>
|
||||
#include <DDS/Topology.h>
|
||||
#include <DDS/Tools.h>
|
||||
#include <dds/dds.h>
|
||||
#include <fairmq/sdk/Topology.h>
|
||||
#include <fairmq/Tools.h>
|
||||
|
||||
|
Reference in New Issue
Block a user