Add pmix::lookup binding and cleanup

fairmq/plugins/PMIx/PMIx.hpp

@@ -131,11 +131,27 @@ struct info : pmix_info_t
         PMIX_VALUE_XFER(rc, lhs, static_cast<pmix_value_t*>(&rhs));
 
         if (rc != PMIX_SUCCESS) {
-            throw runtime_error("pmix::info ctor failed: rc=" + rc);
+            throw runtime_error("pmix::info ctor failed: rc=" + std::to_string(rc));
         }
     }
 };
 
+struct pdata : pmix_pdata_t
+{
+    pdata() { PMIX_PDATA_CONSTRUCT(static_cast<pmix_pdata_t*>(this)); }
+    ~pdata() { PMIX_PDATA_DESTRUCT(static_cast<pmix_pdata_t*>(this)); }
+    pdata(const pdata& rhs)
+    {
+        PMIX_PDATA_XFER(static_cast<pmix_pdata_t*>(this),
+                        static_cast<pmix_pdata_t*>(const_cast<pdata*>(&rhs)));
+    }
+
+    auto set_key(const std::string& new_key) -> void
+    {
+        (void)strncpy(key, new_key.c_str(), PMIX_MAX_KEYLEN);
+    }
+};
+
 auto init(const std::vector<info>& info = {}) -> proc
 {
     proc res;
@@ -143,7 +159,7 @@ auto init(const std::vector<info>& info = {}) -> proc
 
     rc = PMIx_Init(&res, const_cast<pmix::info*>(info.data()), info.size());
     if (rc != PMIX_SUCCESS) {
-        throw runtime_error("pmix::init() failed: rc=" + rc);
+        throw runtime_error("pmix::init() failed: rc=" + std::to_string(rc));
     }
 
     return res;
@@ -159,7 +175,7 @@ auto finalize(const std::vector<info>& info = {}) -> void
 
     rc = PMIx_Finalize(info.data(), info.size());
     if (rc != PMIX_SUCCESS) {
-        throw runtime_error("pmix::finalize() failed: rc=" + rc);
+        throw runtime_error("pmix::finalize() failed: rc=" + std::to_string(rc));
     }
 }
 
@@ -169,7 +185,7 @@ auto publish(const std::vector<info>& info) -> void
 
     rc = PMIx_Publish(info.data(), info.size());
     if (rc != PMIX_SUCCESS) {
-        throw runtime_error("pmix::publish() failed: rc=" + rc);
+        throw runtime_error("pmix::publish() failed: rc=" + std::to_string(rc));
     }
 }
 
@@ -179,10 +195,19 @@ auto fence(const std::vector<proc>& procs = {}, const std::vector<info>& info =
 
     rc = PMIx_Fence(procs.data(), procs.size(), info.data(), info.size());
     if (rc != PMIX_SUCCESS) {
-        throw runtime_error("pmix::fence() failed: rc=" + rc);
+        throw runtime_error("pmix::fence() failed: rc=" + std::to_string(rc));
     }
 }
 
+auto lookup(std::vector<pdata>& pdata, const std::vector<info>& info = {}) -> void
+{
+    status rc;
+
+    rc = PMIx_Lookup(pdata.data(), pdata.size(), info.data(), info.size());
+    if (rc != PMIX_SUCCESS) {
+        throw runtime_error("pmix::lookup() failed: rc=" + std::to_string(rc));
+    }
+}
 
 } /* namespace pmix */
 

fairmq/plugins/PMIx/PMIxPlugin.cxx

@@ -30,24 +30,10 @@ PMIxPlugin::PMIxPlugin(const std::string& name,
     SubscribeToDeviceStateChange([&](DeviceState newState) {
         switch (newState) {
             case DeviceState::InitializingDevice:
-                if (!pmix::initialized()) {
+                Init();
-                    fProc = pmix::init();
+                Publish();
-                    LOG(debug) << PMIxClient() << " pmix::init() OK: " << fProc
+                Fence();
-                               << ",version=" << pmix::get_version();
+                Lookup();
-                }
-
-                FillChannelContainers();
-
-                PublishBoundChannels();
-
-                {
-                    pmix::proc all(fProc);
-                    all.rank = pmix::rank::wildcard;
-
-                    pmix::fence({all});
-                }
-
-                // lookup
                 break;
         case DeviceState::Exiting:
             UnsubscribeFromDeviceStateChange();
@@ -60,8 +46,6 @@ PMIxPlugin::PMIxPlugin(const std::string& name,
 
 PMIxPlugin::~PMIxPlugin()
 {
-    LOG(debug) << PMIxClient() << " Finalizing PMIx session... (On success, logs seen by the RTE will stop here.)";
-
     while (pmix::initialized()) {
         try {
             pmix::finalize();
@@ -72,52 +56,83 @@ PMIxPlugin::~PMIxPlugin()
     }
 }
 
-auto PMIxPlugin::FillChannelContainers() -> void
+auto PMIxPlugin::PMIxClient() const -> std::string
 {
-    try {
+    std::stringstream ss;
-        std::unordered_map<std::string, int> channelInfo(GetChannelInfo());
+    ss << "PMIx client(pid=" << fPid << ")";
+    return ss.str();
+}
 
-        // fill binding and connecting chans
+auto PMIxPlugin::Init() -> void
-        for (const auto& c : channelInfo) {
+{
-            std::string methodKey{"chans." + c.first + "." + std::to_string(c.second - 1)
+    if (!pmix::initialized()) {
-                                  + ".method"};
+        fProc = pmix::init();
-            if (GetProperty<std::string>(methodKey) == "bind") {
+        LOG(debug) << PMIxClient() << " pmix::init() OK: " << fProc
-                fBindingChannels.insert(std::make_pair(c.first, std::vector<std::string>()));
+                   << ",version=" << pmix::get_version();
-                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
+auto PMIxPlugin::Publish() -> void
 {
-    std::vector<pmix::info> infos;
+    auto channels(GetChannelInfo());
-    infos.reserve(fBindingChannels.size());
+    std::vector<pmix::info> info;
 
-    for (const auto& channel : fBindingChannels) {
+    for (const auto& c : channels) {
-        std::string joined = boost::algorithm::join(channel.second, ",");
+        std::string methodKey{"chans." + c.first + "." + std::to_string(c.second - 1) + ".method"};
-        infos.emplace_back(channel.first, joined);
+        if (GetProperty<std::string>(methodKey) == "bind") {
+            for (int i = 0; i < c.second; ++i) {
+                std::string addressKey{"chans." + c.first + "." + std::to_string(i) + ".address"};
+                info.emplace_back(addressKey, GetProperty<std::string>(addressKey));
+            }
+        }
     }
 
-    pmix::publish(infos);
+    if (info.size() > 0) {
-    LOG(debug) << PMIxClient() << " pmix::publish() OK: published "
+        pmix::publish(info);
-               << fBindingChannels.size() << " binding channels.";
+        LOG(debug) << PMIxClient() << " pmix::publish() OK: published "
+                   << info.size() << " binding channels.";
+    }
+}
+
+auto PMIxPlugin::Fence() -> void
+{
+    pmix::proc all(fProc);
+    all.rank = pmix::rank::wildcard;
+
+    pmix::fence({all});
+}
+
+auto PMIxPlugin::Lookup() -> void
+{
+    auto channels(GetChannelInfo());
+    std::vector<pmix::pdata> pdata;
+
+    for (const auto& c : channels) {
+        std::string methodKey{"chans." + c.first + "." + std::to_string(c.second - 1) + ".method"};
+        if (GetProperty<std::string>(methodKey) == "connect") {
+            for (int i = 0; i < c.second; ++i) {
+                std::string addressKey{"chans." + c.first + "." + std::to_string(i) + ".address"};
+                pdata.emplace_back();
+                pdata.back().set_key(addressKey);
+            }
+        }
+    }
+
+    if (pdata.size() > 0) {
+        pmix::lookup(pdata);
+        LOG(debug) << PMIxClient() << " pmix::lookup() OK";
+    }
+
+    for (const auto& p : pdata) {
+        if (p.value.type == PMIX_UNDEF) {
+            LOG(debug) << PMIxClient() << " pmix::lookup() not found: key=" << p.key;
+        } else if (p.value.type == PMIX_STRING) {
+            SetProperty<std::string>(p.key, p.value.data.string);
+            LOG(debug) << PMIxClient() << " pmix::lookup() found: key=" << p.key << ",value=" << p.value.data.string;
+        } else {
+            LOG(debug) << PMIxClient() << " pmix::lookup() wrong type returned: key=" << p.key << ",type=" << p.value.type;
+        }
+    }
 }
 
 } /* namespace plugins */

fairmq/plugins/PMIx/PMIxPlugin.h

@@ -21,7 +21,6 @@
 #include <string>
 #include <sys/types.h>
 #include <unistd.h>
-#include <unordered_map>
 #include <vector>
 
 namespace fair
@@ -31,17 +30,6 @@ namespace mq
 namespace plugins
 {
 
-struct ConnectingChannel
-{
-    ConnectingChannel()
-        : fSubChannelAddresses()
-        , fValues()
-    {}
-
-    std::vector<std::string> fSubChannelAddresses;
-    std::unordered_map<uint64_t, std::string> fValues;
-};
-
 class PMIxPlugin : public Plugin
 {
   public:
@@ -51,21 +39,16 @@ class PMIxPlugin : public Plugin
          const std::string& homepage,
          PluginServices* pluginServices);
     ~PMIxPlugin();
-    auto PMIxClient() const -> std::string 
+    auto PMIxClient() const -> std::string;
-    {
-        std::stringstream ss;
-        ss << "PMIx client(pid=" << fPid << ")";
-        return ss.str();
-    }
 
   private:
     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 Init() -> void;
-    auto PublishBoundChannels() -> void;
+    auto Publish() -> void;
+    auto Fence() -> void;
+    auto Lookup() -> void;
 };
 
 Plugin::ProgOptions PMIxProgramOptions()