Update to new DDS API, bump required version

CMakeLists.txt

@@ -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()

examples/dds/fairmq-start-ex-dds.sh.in

@@ -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,28 +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

fairmq/Plugin.h

@@ -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); }
 

fairmq/PluginServices.cxx

@@ -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};

fairmq/PluginServices.h

@@ -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

fairmq/plugins/DDS/DDS.cxx

@@ -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: {

fairmq/plugins/DDS/DDS.h

@@ -133,7 +133,6 @@ class DDS : public Plugin
     ~DDS();
 
   private:
-    auto StaticControl() -> void;
     auto WaitForExitingAck() -> void;
     auto StartWorkerThread() -> void;
 
@@ -154,14 +153,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;
 

fairmq/plugins/DDS/runDDSCommandUI.cxx

@@ -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 {

fairmq/sdk/DDSAgent.h

@@ -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;
 };
 

fairmq/sdk/DDSSession.cxx

@@ -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;