Implement some PMIx C++ bindings

pmix::init()
pmix::finalize()
pmix::publish()
pmix::initialized()
pmix::get_version()

and supporting data structures.

fairmq/plugins/PMIx/CMakeLists.txt

@@ -7,7 +7,11 @@
 ################################################################################
 
 set(plugin FairMQPlugin_pmix)
-add_library(${plugin} SHARED ${CMAKE_CURRENT_SOURCE_DIR}/PMIx.cxx ${CMAKE_CURRENT_SOURCE_DIR}/PMIx.h)
+add_library(${plugin} SHARED
+  ${CMAKE_CURRENT_SOURCE_DIR}/PMIxPlugin.cxx
+  ${CMAKE_CURRENT_SOURCE_DIR}/PMIxPlugin.h
+  ${CMAKE_CURRENT_SOURCE_DIR}/PMIx.hpp
+)
 target_link_libraries(${plugin} FairMQ PMIx::libpmix)
 target_include_directories(${plugin} PRIVATE ${CMAKE_CURRENT_BINARY_DIR})
 set_target_properties(${plugin} PROPERTIES

fairmq/plugins/PMIx/PMIx.cxx

@@ -1,60 +0,0 @@
-/********************************************************************************
- *    Copyright (C) 2017 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"                       *
- ********************************************************************************/
-
-#include "PMIx.h"
-#include <FairMQLogger.h>
-#include <fairmq/Tools.h>
-#include <stdexcept>
-
-namespace fair
-{
-namespace mq
-{
-namespace plugins
-{
-
-PMIx::PMIx(const std::string& name,
-           const Plugin::Version version,
-           const std::string& maintainer,
-           const std::string& homepage,
-           PluginServices* pluginServices)
-    : Plugin(name, version, maintainer, homepage, pluginServices)
-    , fPid(getpid())
-{
-    auto rc = PMIx_Init(&fPMIxProc, NULL, 0);
-    if (rc != PMIX_SUCCESS) {
-        throw std::runtime_error(tools::ToString("Client ns ", fPMIxProc.nspace,
-                                          " rank ", fPMIxProc.rank,
-                                          " pid ", fPid,
-                                          ": PMIx_Init failed: ", rc));
-    }
-
-    LOG(info) << "Client ns " << fPMIxProc.nspace << " rank " << fPMIxProc.rank << " pid " << fPid
-              << ": Running";
-}
-
-PMIx::~PMIx()
-{
-    LOG(info) << "Client ns " << fPMIxProc.nspace << " rank " << fPMIxProc.rank << " pid " << fPid
-              << ": Finalizing";
-
-    auto rc = PMIx_Finalize(NULL, 0);
-    if (rc != PMIX_SUCCESS) {
-        throw std::runtime_error(tools::ToString("Client ns ", fPMIxProc.nspace,
-                                          " rank ", fPMIxProc.rank,
-                                          " pid ", fPid,
-                                          ": PMIx_Finalize failed: ", rc));
-    }
-
-    LOG(info) << "Client ns " << fPMIxProc.nspace << " rank " << fPMIxProc.rank << " pid " << fPid
-              << ": PMIx_Finalize successfully completed";
-}
-
-} /* namespace plugins */
-} /* namespace mq */
-} /* namespace fair */

fairmq/plugins/PMIx/PMIx.hpp

@@ -0,0 +1,152 @@
+/********************************************************************************
+ *    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 PMIX_HPP
+#define PMIX_HPP
+
+#include <cstring>
+#include <limits>
+#include <memory>
+#include <ostream>
+#include <pmix.h>
+#include <stdexcept>
+#include <string.h>
+#include <type_traits>
+#include <utility>
+#include <vector>
+#include <FairMQLogger.h>
+
+// C++ PMIx v2.1 API
+namespace pmix
+{
+
+struct runtime_error : std::runtime_error
+{
+    using std::runtime_error::runtime_error;
+};
+
+using status = pmix_status_t;
+
+using nspace = pmix_nspace_t;
+
+using key = pmix_key_t;
+
+using data_type = pmix_data_type_t;
+
+struct rank
+{
+    enum named : pmix_rank_t
+    {
+        undef = PMIX_RANK_UNDEF,
+        wildcard = PMIX_RANK_WILDCARD,
+        local_node = PMIX_RANK_LOCAL_NODE
+    };
+    rank(pmix_rank_t r)
+        : m_value(r)
+    {}
+    operator pmix_rank_t() { return m_value; }
+
+  private:
+    pmix_rank_t m_value;
+};
+
+struct proc : pmix_proc_t
+{
+    proc() { PMIX_PROC_CONSTRUCT(static_cast<pmix_proc_t*>(this)); }
+    ~proc() { PMIX_PROC_DESTRUCT(static_cast<pmix_proc_t*>(this)); }
+    proc(pmix::nspace ns, pmix::rank r)
+    {
+        PMIX_PROC_LOAD(static_cast<pmix_proc_t*>(this), ns, static_cast<pmix_rank_t>(r));
+    }
+
+    friend std::ostream& operator<<(std::ostream& os, const proc& p)
+    {
+        return os << "nspace=" << p.nspace << ",rank=" << p.rank;
+    }
+};
+
+struct value : pmix_value_t
+{
+    value() { PMIX_VALUE_CONSTRUCT(static_cast<pmix_value_t*>(this)); }
+    ~value() { /*PMIX_VALUE_DESTRUCT(static_cast<pmix_value_t*>(this));*/ }
+
+    // template<typename T>
+    // value(const T* val, data_type dt)
+    // {
+        // PMIX_VALUE_LOAD(static_cast<pmix_value_t*>(this), const_cast<void*>(val), dt);
+    // }
+    template<typename T>
+    value(T)
+    {
+        throw runtime_error("Given value type not supported or not yet implemented.");
+    }
+    value(const char* val)
+    {
+        PMIX_VALUE_LOAD(static_cast<pmix_value_t*>(this), const_cast<char*>(val), PMIX_STRING);
+    }
+    value(const std::string& val)
+    {
+        PMIX_VALUE_LOAD(
+            static_cast<pmix_value_t*>(this), const_cast<char*>(val.c_str()), PMIX_STRING);
+    }
+};
+
+struct info : pmix_info_t
+{
+    info() { PMIX_INFO_CONSTRUCT(static_cast<pmix_info_t*>(this)); }
+    ~info() { PMIX_INFO_DESTRUCT(static_cast<pmix_info_t*>(this)); }
+
+    template<typename... Args>
+    info(const std::string& k, Args&&... args)
+    {
+        (void)strncpy(key, k.c_str(), PMIX_MAX_KEYLEN);
+        flags = 0;
+        value = pmix::value(std::forward<Args>(args)...);
+    }
+};
+
+auto init(const std::vector<info>& info = {}) -> proc
+{
+    proc res;
+    status rc;
+
+    rc = PMIx_Init(&res, const_cast<pmix::info*>(info.data()), info.size());
+    if (rc != PMIX_SUCCESS) {
+        throw runtime_error("pmix::init() failed: rc=" + rc);
+    }
+
+    return res;
+}
+
+auto initialized() -> bool { return !!PMIx_Initialized(); }
+
+auto get_version() -> const char* { return PMIx_Get_version(); }
+
+auto finalize(const std::vector<info>& info = {}) -> void
+{
+    status rc;
+
+    rc = PMIx_Finalize(info.data(), info.size());
+    if (rc != PMIX_SUCCESS) {
+        throw runtime_error("pmix::finalize() failed: rc=" + rc);
+    }
+}
+
+auto publish(const std::vector<info>& info) -> void
+{
+    status rc;
+
+    rc = PMIx_Publish(info.data(), info.size());
+    if (rc != PMIX_SUCCESS) {
+        throw runtime_error("pmix::publish() failed: rc=" + rc);
+    }
+}
+
+} /* namespace pmix */
+
+#endif /* PMIX_HPP */

fairmq/plugins/PMIx/PMIxPlugin.cxx

@@ -0,0 +1,124 @@
+/********************************************************************************
+ *    Copyright (C) 2017 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"                       *
+ ********************************************************************************/
+
+#include "PMIxPlugin.h"
+
+#include <boost/algorithm/string/join.hpp>
+#include <fairmq/Tools.h>
+#include <stdexcept>
+
+namespace fair
+{
+namespace mq
+{
+namespace plugins
+{
+
+PMIxPlugin::PMIxPlugin(const std::string& name,
+                       const Plugin::Version version,
+                       const std::string& maintainer,
+                       const std::string& homepage,
+                       PluginServices* pluginServices)
+    : Plugin(name, version, maintainer, homepage, pluginServices)
+    , fPid(getpid())
+{
+    SubscribeToDeviceStateChange([&](DeviceState newState) {
+        switch (newState) {
+            case DeviceState::InitializingDevice:
+                if (!pmix::initialized()) {
+                    fProc = pmix::init();
+                    LOG(debug) << PMIxClient() << " pmix::init() OK: " << fProc
+                               << ",version=" << pmix::get_version();
+                }
+
+                FillChannelContainers();
+
+                PublishBoundChannels();
+
+                // pmix_proc_t proc;
+                // rc = PMIx_Fence(&proc, 1, NULL, 0)
+                // fence
+
+                // lookup
+
+                // fence
+                break;
+        case DeviceState::Exiting:
+            UnsubscribeFromDeviceStateChange();
+            break;
+        default:
+            break;
+        }
+    });
+}
+
+PMIxPlugin::~PMIxPlugin()
+{
+    LOG(debug) << PMIxClient() << " Finalizing PMIx session... (On success, logs seen by the RTE will stop here.)";
+
+    while (pmix::initialized()) {
+        try {
+            pmix::finalize();
+            LOG(debug) << PMIxClient() << " pmix::finalize() OK";
+        } catch (const pmix::runtime_error& e) {
+            LOG(debug) << PMIxClient() << " pmix::finalize() failed: " << e.what();
+        }
+    }
+}
+
+auto PMIxPlugin::FillChannelContainers() -> void
+{
+    try {
+        std::unordered_map<std::string, int> channelInfo(GetChannelInfo());
+
+        // fill binding and connecting chans
+        for (const auto& c : channelInfo) {
+            std::string methodKey{"chans." + c.first + "." + std::to_string(c.second - 1)
+                                  + ".method"};
+            if (GetProperty<std::string>(methodKey) == "bind") {
+                fBindingChannels.insert(std::make_pair(c.first, std::vector<std::string>()));
+                for (int i = 0; i < c.second; ++i) {
+                    fBindingChannels.at(c.first).push_back(GetProperty<std::string>(
+                        std::string{"chans." + c.first + "." + std::to_string(i) + ".address"}));
+                }
+            } else if (GetProperty<std::string>(methodKey) == "connect") {
+                fConnectingChannels.insert(std::make_pair(c.first, ConnectingChannel()));
+                LOG(debug) << "preparing to connect: " << c.first << " with " << c.second
+                           << " sub-channels.";
+                for (int i = 0; i < c.second; ++i) {
+                    fConnectingChannels.at(c.first).fSubChannelAddresses.push_back(std::string());
+                }
+            } else {
+                LOG(error) << "Cannot update address configuration. Channel method (bind/connect) "
+                              "not specified.";
+                return;
+            }
+        }
+    } catch (const std::exception& e) {
+        LOG(error) << "Error filling channel containers: " << e.what();
+    }
+}
+
+auto PMIxPlugin::PublishBoundChannels() -> void
+{
+    std::vector<pmix::info> infos;
+    infos.reserve(fBindingChannels.size());
+
+    for (const auto& channel : fBindingChannels) {
+        std::string joined = boost::algorithm::join(channel.second, ",");
+        infos.emplace_back(channel.first, joined);
+    }
+
+    pmix::publish(infos);
+    LOG(debug) << PMIxClient() << " pmix::publish() OK: published "
+               << fBindingChannels.size() << " binding channels.";
+}
+
+} /* namespace plugins */
+} /* namespace mq */
+} /* namespace fair */

fairmq/plugins/PMIx/PMIxPlugin.h

@@ -9,12 +9,20 @@
 #ifndef FAIR_MQ_PLUGINS_PMIX
 #define FAIR_MQ_PLUGINS_PMIX
 
+#include "PMIx.hpp"
+
 #include <fairmq/Plugin.h>
 #include <fairmq/Version.h>
+#include <FairMQLogger.h>
 
-#include <pmix.h>
+#include <string>
+#include <sstream>
+#include <stdexcept>
+#include <string>
 #include <sys/types.h>
 #include <unistd.h>
+#include <unordered_map>
+#include <vector>
 
 namespace fair
 {
@@ -23,19 +31,41 @@ namespace mq
 namespace plugins
 {
 
-class PMIx : public Plugin
+struct ConnectingChannel
+{
+    ConnectingChannel()
+        : fSubChannelAddresses()
+        , fValues()
+    {}
+
+    std::vector<std::string> fSubChannelAddresses;
+    std::unordered_map<uint64_t, std::string> fValues;
+};
+
+class PMIxPlugin : public Plugin
 {
   public:
-    PMIx(const std::string& name,
+    PMIxPlugin(const std::string& name,
          const Plugin::Version version,
          const std::string& maintainer,
          const std::string& homepage,
          PluginServices* pluginServices);
-    ~PMIx();
+    ~PMIxPlugin();
+    auto PMIxClient() const -> std::string 
+    {
+        std::stringstream ss;
+        ss << "PMIx client(pid=" << fPid << ")";
+        return ss.str();
+    }
 
   private:
-    pmix_proc_t fPMIxProc;
+    pmix::proc fProc;
     pid_t fPid;
+    std::unordered_map<std::string, std::vector<std::string>> fBindingChannels;
+    std::unordered_map<std::string, ConnectingChannel> fConnectingChannels;
+
+    auto FillChannelContainers() -> void;
+    auto PublishBoundChannels() -> void;
 };
 
 Plugin::ProgOptions PMIxProgramOptions()
@@ -47,7 +77,7 @@ Plugin::ProgOptions PMIxProgramOptions()
 }
 
 REGISTER_FAIRMQ_PLUGIN(
-    PMIx,                                        // Class name
+    PMIxPlugin,                                  // Class name
     pmix,                                        // Plugin name (string, lower case chars only)
     (Plugin::Version{FAIRMQ_VERSION_MAJOR,
                      FAIRMQ_VERSION_MINOR,
@@ -61,4 +91,4 @@ REGISTER_FAIRMQ_PLUGIN(
 } /* namespace mq */
 } /* namespace fair */
 
-#endif /* FAIR_MQ_PLUGINS_DDS */
+#endif /* FAIR_MQ_PLUGINS_PMIX */