Mover selection

ecmwf · Nov 15, 2023 · 47d3df0 · 47d3df0
1 parent c871e6a
commit 47d3df0
Show file tree

Hide file tree

Showing 6 changed files with 29 additions and 3 deletions.
diff --git a/src/eckit/io/MoverTransferSelection.cc b/src/eckit/io/MoverTransferSelection.cc
@@ -33,7 +33,7 @@ void MoverTransferSelection::updateCost(const std::string& name, const eckit::Le
 
 
 void MoverTransferSelection::updateCost(const eckit::NodeInfo& node, const eckit::Length& length) {
-    updateCost(node.name(), length);
+    updateCost(node.node(), length);
 }
 
 void MoverTransferSelection::requiredMoverAttributes(const std::set<std::string>& attrs) {
@@ -120,10 +120,11 @@ NodeInfo MoverTransferSelection::selectedMover() {
 }
 
 void MoverTransferSelection::preferredMover(const NodeInfo& node) {
-    preferredMover(node.name());
+    preferredMover(node.node());
 }
 
 void MoverTransferSelection::preferredMover(const std::string& name) {
+    Log::info() << "MoverTransferSelection::preferredMover " << name << std::endl;
     preferredMover_ = name;
 }
 

diff --git a/src/eckit/io/TCPHandle.cc b/src/eckit/io/TCPHandle.cc
@@ -9,6 +9,8 @@
  */
 
 #include "eckit/io/TCPHandle.h"
+#include "eckit/io/MoverTransferSelection.h"
+#include "eckit/io/cluster/ClusterNodes.h"
 
 //----------------------------------------------------------------------------------------------------------------------
 
@@ -84,6 +86,16 @@ std::string TCPHandle::title() const {
     return os.str();
 }
 
+void TCPHandle::selectMover(eckit::MoverTransferSelection& selection, bool read) const {
+    // If we use a callback server that is collocated on a mover node
+    // we want to use that mover
+    NodeInfo node;
+    if (ClusterNodes::lookUpHost("mover", host_, node)) {
+        selection.preferredMover(node);
+    }
+}
+
+
 //----------------------------------------------------------------------------------------------------------------------
 
 }  // namespace eckit
diff --git a/src/eckit/io/TCPHandle.h b/src/eckit/io/TCPHandle.h
@@ -53,6 +53,8 @@ class TCPHandle : public DataHandle {
 
     bool canSeek() const override { return false; }
 
+    virtual void selectMover(eckit::MoverTransferSelection&, bool) const override;
+
     // From Streamable
 
     void encode(Stream&) const override;

diff --git a/src/eckit/io/cluster/ClusterNodes.cc b/src/eckit/io/cluster/ClusterNodes.cc
@@ -22,6 +22,7 @@
 #include "eckit/memory/Zero.h"
 #include "eckit/thread/AutoLock.h"
 #include "eckit/utils/Clock.h"
+#include "eckit/net/IPAddress.h"
 
 namespace eckit {
 
@@ -488,8 +489,10 @@ bool ClusterNodes::lookUpHost(const std::string& type, const std::string& host,
     pthread_once(&once, init);
     AutoLock<NodeArray> lock(*nodeArray);
 
+    auto ip = eckit::net::IPAddress::hostAddress(host);
+
     for (const auto& k : *nodeArray) {
-        if (k.active() && type == k.type() && host == k.host()) {
+        if (k.active() && type == k.type() && ip == eckit::net::IPAddress::hostAddress(k.host())) {
             result = k.asNodeInfo();
             return true;
         }

diff --git a/src/eckit/net/IPAddress.cc b/src/eckit/net/IPAddress.cc
@@ -17,6 +17,7 @@
 #include <sys/socket.h>
 #include <unistd.h>
 #include <ostream>
+#include <string.h>
 
 namespace eckit::net {
 
@@ -45,6 +46,11 @@ std::string IPAddress::asString() const {
     return inet_ntoa(address_);
 }
 
+bool IPAddress::operator==(const IPAddress& other) const {
+    return ::memcmp(&address_, &other.address_, sizeof(address_)) == 0;
+}
+
+
 IPAddress IPAddress::hostAddress(const std::string& hostname) {
     struct hostent* hostEntry = gethostbyname(hostname.c_str());
     ASSERT(hostEntry);

diff --git a/src/eckit/net/IPAddress.h b/src/eckit/net/IPAddress.h
@@ -37,6 +37,8 @@ class IPAddress {
     static IPAddress myIPAddress();
     static IPAddress hostAddress(const std::string& hostname);
 
+    bool operator==(const IPAddress& other) const;
+
 private:
     // Members