Add some visibility on which rank to rank connections created by Gloo

Summary:
Created a starter task for myself to get familiar with Gloo diff work flow. In recent debugging on prod Gloo issues, we found it 'not-so-easy' to answer 1) when
and how many Gloo PGs are created. 2) For each rank in each PG, how many TCP connections are
established. First adding a simple log at the TCP context layer to expose the rank to rank
connectivity during the creation of Full Mesh.

Gloo logging seems inconvenient now, thinking we might introduce
other OSS logging libs in the future

Reviewed By: XilunWu

Differential Revision: D52897536

fbshipit-source-id: c4b78ee7c2a83534c704a55c39766b08d98810f9

Author: shuqiangzhang

gloo/transport/ibverbs/pair.cc

@@ -346,6 +346,12 @@ void Pair::recv(
       "Unbound buffers not supported yet for ibverbs transport");
 }
 
+// place holder for future use
+bool Pair::isConnected() {
+   GLOO_THROW_INVALID_OPERATION_EXCEPTION(
+      "isConnected not supported yet for ibverbs transport");
+}
+
 // handleCompletionEvent is called by the device thread when it
 // received an event for this pair's completion queue on its
 // completion channel.

gloo/transport/ibverbs/pair.h

@@ -67,6 +67,8 @@ class Pair : public ::gloo::transport::Pair {
   virtual std::unique_ptr<::gloo::transport::Buffer>
   createRecvBuffer(int slot, void* ptr, size_t size) override;
 
+  virtual bool isConnected() override;
+
   // Send from the specified buffer to remote side of pair.
   virtual void send(
       transport::UnboundBuffer* tbuf,

gloo/transport/pair.h

@@ -36,6 +36,8 @@ class Pair {
   virtual std::unique_ptr<Buffer>
   createRecvBuffer(int slot, void* ptr, size_t size) = 0;
 
+  virtual bool isConnected() = 0;
+
   // Send from the specified buffer to remote side of pair.
   virtual void send(
       UnboundBuffer* buf,

gloo/transport/tcp/context.cc

@@ -9,6 +9,7 @@
 #include "gloo/transport/tcp/context.h"
 
 #include <cstring>
+#include <iostream>
 
 #include "gloo/common/error.h"
 #include "gloo/common/logging.h"
@@ -117,6 +118,8 @@ void Context::createAndConnectAllPairs(IStore &store) {
     const auto& pair = getPair(i);
     pair->setLocalRank(localRank);
   }
+
+  printConnectivityInfo();
 }
 
 std::unique_ptr<transport::Pair>& Context::createPair(int rank) {
@@ -132,6 +135,49 @@ std::unique_ptr<transport::UnboundBuffer> Context::createUnboundBuffer(
   return std::unique_ptr<transport::UnboundBuffer>(buf);
 }
 
+std::vector<int> Context::getConnectedPeerRanks() const {
+  std::vector<int> result;
+  GLOO_ENFORCE(size == pairs_.size());
+  for (int i = 0; i < size; i++) {
+    if (pairs_.at(i)->isConnected() && i != rank) {
+      result.push_back(i);
+    }
+  }
+  return result;
+}
+
+std::vector<int> Context::getUnConnectedPeerRanks() const {
+  std::vector<int> result;
+  GLOO_ENFORCE(size == pairs_.size());
+  for (int i = 0; i < size; i++) {
+    if (!pairs_.at(i)->isConnected() && i != rank) {
+      result.push_back(i);
+    }
+  }
+  return result;
+}
+
+void Context::printConnectivityInfo() const {
+  int numConnectedPeers = getConnectedPeerRanks().size();
+  std::cout << "[Gloo] Rank "  << rank << " is connected to "
+            << numConnectedPeers << " peer ranks. "
+            << "Expected number of connected peer ranks is : " << size - 1
+            << std::endl;
+
+  if (numConnectedPeers != size - 1) {
+    std::vector<int> unConnectedPeers = getUnConnectedPeerRanks();
+    std::cout << "[Gloo] Rank " << rank << " is NOT connected to: [";
+    for (int i = 0; i < unConnectedPeers.size(); i++) {
+      if (i != unConnectedPeers.size() - 1) {
+        std::cout << unConnectedPeers[i] << ", ";
+      } else {
+        std::cout << unConnectedPeers[i];
+      }
+    }
+    std::cout << "]" << std::endl;
+  }
+}
+
 void Context::recvFromAny(
     UnboundBuffer* buf,
     uint64_t slot,

gloo/transport/tcp/context.h

@@ -86,6 +86,20 @@ class Context : public ::gloo::transport::Context,
   // out. All pairs should be signaled and closed in that event.
   void signalException(const std::string& msg);
 
+  // Returns a sorted list of connected peer ranks, excluding self.
+  // Normally all peer ranks should be connected at the end of
+  // createAndConnectAllPairs
+  // peer rank concept is introduced at the transport/tcp/pair level, so
+  // this method is defined at the same level instead of the parent contexts
+  std::vector<int> getConnectedPeerRanks() const;
+
+  // Returns a sorted list of unconnected and peer ranks, excluding self.
+  // Normally empty at the end of createAndConnectAllPairs
+  std::vector<int> getUnConnectedPeerRanks() const;
+
+  // a helper function to print out rank to rank connectivity information
+  void printConnectivityInfo() const;
+
   friend class ContextMutator;
 
   friend class UnboundBuffer;

gloo/transport/tcp/pair.cc

@@ -95,6 +95,11 @@ const Address& Pair::address() const {
   return self_;
 }
 
+bool Pair::isConnected() {
+  std::lock_guard<std::mutex> lock(m_);
+  return state_ == CONNECTED;
+}
+
 void Pair::connect(const std::vector<char>& bytes) {
   const auto peer = Address(bytes);
 

gloo/transport/tcp/pair.h

@@ -146,6 +146,8 @@ class Pair : public ::gloo::transport::Pair, public Handler {
 
   void close() override;
 
+  bool isConnected() override;
+
  protected:
   // Refer to parent context using raw pointer. This could be a
   // weak_ptr, seeing as the context class is a shared_ptr, but: