Back out "fbcode/gloo/transport/uv"

Summary: This code isn't used internally at Meta but is possibly used by open source community.

Reviewed By: fduwjj

Differential Revision: D65232750

fbshipit-source-id: 03a9c34288c8872ad94a94ce82e703f558e976af

Author: c-p-i-o

gloo/transport/uv/address.cc

@@ -0,0 +1,89 @@
+/**
+ * Copyright (c) 2019-present, Facebook, Inc.
+ * All rights reserved.
+ *
+ * This source code is licensed under the BSD-style license found in the
+ * LICENSE file in the root directory of this source tree.
+ */
+
+#include <gloo/transport/uv/address.h>
+
+#include <string.h>
+
+#include <uv.h>
+
+#include <gloo/common/logging.h>
+
+namespace gloo {
+namespace transport {
+namespace uv {
+
+Address::Address(struct sockaddr_storage ss, sequence_type seq) {
+  impl_.ss = std::move(ss);
+  impl_.seq = seq;
+}
+
+Address::Address(const std::vector<char>& bytes) {
+  GLOO_ENFORCE_EQ(sizeof(impl_), bytes.size());
+  memcpy(&impl_, bytes.data(), sizeof(impl_));
+}
+
+Address::Address(const Address& other)
+    : Address(other.impl_.ss, other.impl_.seq) {}
+
+std::vector<char> Address::bytes() const {
+  std::lock_guard<std::mutex> lock(m_);
+  std::vector<char> bytes(sizeof(impl_));
+  memcpy(bytes.data(), &impl_, sizeof(impl_));
+  return bytes;
+}
+
+Address& Address::operator=(Address&& other) {
+  std::lock_guard<std::mutex> lock(m_);
+  impl_.ss = std::move(other.impl_.ss);
+  impl_.seq = other.impl_.seq;
+  return *this;
+}
+
+Address& Address::operator=(const Address& other) {
+  std::lock_guard<std::mutex> lock(m_);
+  impl_.ss = other.impl_.ss;
+  impl_.seq = other.impl_.seq;
+  return *this;
+}
+
+std::string Address::str() const {
+  char str[INET6_ADDRSTRLEN + 128];
+  int port = 0;
+
+  str[0] = '[';
+  if (impl_.ss.ss_family == AF_INET) {
+    auto in = (struct sockaddr_in*)&impl_.ss;
+    uv_ip4_name(in, str + 1, sizeof(str) - 1);
+    port = in->sin_port;
+  } else if (impl_.ss.ss_family == AF_INET6) {
+    auto in6 = (struct sockaddr_in6*)&impl_.ss;
+    uv_ip6_name(in6, str + 1, sizeof(str) - 1);
+    port = in6->sin6_port;
+  } else {
+    snprintf(str + 1, sizeof(str) - 1, "none");
+  }
+
+  size_t len = strlen(str);
+  if (port > 0) {
+    len += snprintf(str + len, sizeof(str) - len, "]:%d", port);
+  } else {
+    len += snprintf(str + len, sizeof(str) - len, "]");
+  }
+
+  // Append sequence number if one is set.
+  if (impl_.seq != SIZE_MAX) {
+    len += snprintf(str + len, sizeof(str) - len, "$%d", impl_.seq);
+  }
+
+  return str;
+}
+
+} // namespace uv
+} // namespace transport
+} // namespace gloo

gloo/transport/uv/address.h

@@ -0,0 +1,77 @@
+/**
+ * Copyright (c) 2019-present, Facebook, Inc.
+ * All rights reserved.
+ *
+ * This source code is licensed under the BSD-style license found in the
+ * LICENSE file in the root directory of this source tree.
+ */
+
+#pragma once
+
+#ifdef _WIN32
+#include "gloo/common/win.h"
+#else
+#include <sys/socket.h>
+#endif
+
+#include <mutex>
+
+#include <gloo/transport/address.h>
+
+namespace gloo {
+namespace transport {
+namespace uv {
+
+class Address : public ::gloo::transport::Address {
+ public:
+  using sequence_type = int;
+
+  Address() {}
+
+  Address(struct sockaddr_storage ss, sequence_type seq = -1);
+
+  explicit Address(const std::vector<char>&);
+
+  Address& operator=(Address&& other);
+  Address& operator=(const Address& other);
+  Address(const Address& other);
+
+  virtual std::vector<char> bytes() const override;
+
+  virtual std::string str() const override;
+
+  const struct sockaddr_storage& getSockaddr() const {
+    return impl_.ss;
+  }
+
+  sequence_type getSeq() const {
+    return impl_.seq;
+  }
+
+  Address withSeq(sequence_type seq) const {
+    return Address(impl_.ss, seq);
+  }
+
+ protected:
+  // Encapsulate fields such that it is trivially copyable. This class
+  // is not trivially copyable itself (because it is a subclass?).
+  struct Impl {
+    // IP address of the listening socket.
+    struct sockaddr_storage ss;
+
+    // Sequence number of this address.
+    // If this is equal to -1, the address is assumed to
+    // represent the listening socket of a device. The sequence number
+    // must be set before it can be used by a pair.
+    sequence_type seq = -1;
+  };
+
+  static_assert(std::is_trivially_copyable<Impl>::value, "!");
+
+  Impl impl_;
+  mutable std::mutex m_;
+};
+
+} // namespace uv
+} // namespace transport
+} // namespace gloo

gloo/transport/uv/common.h

@@ -0,0 +1,37 @@
+/**
+ * Copyright (c) 2019-present, Facebook, Inc.
+ * All rights reserved.
+ *
+ * This source code is licensed under the BSD-style license found in the
+ * LICENSE file in the root directory of this source tree.
+ */
+
+#pragma once
+
+#include <stdio.h>
+
+#include <uv.h>
+
+#define UV_CHECK(rv, prefix)  \
+  {                           \
+    if ((rv) != 0) {          \
+      fprintf(                \
+          stderr,             \
+          "[%s:%d] %s: %s\n", \
+          __FILE__,           \
+          __LINE__,           \
+          prefix,             \
+          uv_strerror(rv));   \
+    }                         \
+  }                           \
+  while (0)                   \
+    ;
+
+#define FAIL(...)                                                       \
+  {                                                                     \
+    const auto __str = ::gloo::MakeString(__VA_ARGS__);                 \
+    fprintf(stderr, "[%s:%d] %s\n", __FILE__, __LINE__, __str.c_str()); \
+    abort();                                                            \
+  }                                                                     \
+  while (0)                                                             \
+    ;

gloo/transport/uv/context.cc

@@ -0,0 +1,145 @@
+/**
+ * Copyright (c) 2019-present, Facebook, Inc.
+ * All rights reserved.
+ *
+ * This source code is licensed under the BSD-style license found in the
+ * LICENSE file in the root directory of this source tree.
+ */
+
+#include <gloo/transport/uv/context.h>
+
+#include <gloo/common/error.h>
+#include <gloo/common/logging.h>
+#include <gloo/transport/uv/device.h>
+#include <gloo/transport/uv/pair.h>
+#include <gloo/transport/uv/unbound_buffer.h>
+
+namespace gloo {
+namespace transport {
+namespace uv {
+
+Context::Context(std::shared_ptr<Device> device, int rank, int size)
+    : ::gloo::transport::Context(rank, size), device_(std::move(device)) {}
+
+Context::~Context() {
+  // Pairs refer to device by raw pointer.
+  // Ensure they are destructed before the device.
+  pairs_.clear();
+  device_.reset();
+}
+
+std::unique_ptr<transport::Pair>& Context::createPair(int rank) {
+  pairs_[rank] = std::unique_ptr<transport::Pair>(
+      new uv::Pair(this, device_.get(), rank, getTimeout()));
+  return pairs_[rank];
+}
+
+std::unique_ptr<transport::UnboundBuffer> Context::createUnboundBuffer(
+    void* ptr,
+    size_t size) {
+  auto buf = new uv::UnboundBuffer(shared_from_this(), ptr, size);
+  return std::unique_ptr<transport::UnboundBuffer>(buf);
+}
+
+void Context::recvFromAny(
+    UnboundBuffer* buf,
+    uint64_t slot,
+    size_t offset,
+    size_t nbytes,
+    std::vector<int> srcRanks) {
+  for (;;) {
+    // Find rank of pair we can attempt a recv from
+    auto rank = recvFromAnyFindRank(buf, slot, offset, nbytes, srcRanks);
+    if (rank == -1) {
+      return;
+    }
+    // Try recv from returned rank
+    auto ptr = pairs_[rank].get();
+    GLOO_ENFORCE(ptr != nullptr);
+    auto pair = dynamic_cast<Pair*>(ptr);
+    GLOO_ENFORCE(pair != nullptr);
+    if (pair->tryRecv(buf, slot, offset, nbytes)) {
+      return;
+    }
+  }
+}
+
+int Context::recvFromAnyFindRank(
+    UnboundBuffer* buf,
+    uint64_t slot,
+    size_t offset,
+    size_t nbytes,
+    const std::vector<int>& srcRanks) {
+  std::unique_lock<std::mutex> lock(mutex_);
+
+  // See if there is a remote pending send that can fulfill this recv.
+  auto it = findPendingOperations(slot);
+  if (it != pendingOperations_.end()) {
+    auto& pendingOperation = *it;
+
+    // Out of all remote pending sends, find the first one
+    // that exists in the set of eligible ranks.
+    for (const auto rank : pendingOperation.getSendList()) {
+      for (const auto srcRank : srcRanks) {
+        if (rank == srcRank) {
+          // We've found a rank that could fulfill this recv.
+          //
+          // The caller of this function will try and attempt a recv,
+          // which will remove this remote pending send operation,
+          // if it's still around.
+          //
+          return rank;
+        }
+      }
+    }
+  }
+
+  // No candidates; register buffer for recv
+  pendingRecv_[slot].emplace_back(
+      buf->getWeakNonOwningPtr(),
+      offset,
+      nbytes,
+      std::unordered_set<int>(srcRanks.begin(), srcRanks.end()));
+  return -1;
+}
+
+// Allowed to be called only by ContextMutator::findRecvFromAny,
+// where the context lock is already held.
+bool Context::findRecvFromAny(
+    uint64_t slot,
+    int rank,
+    WeakNonOwningPtr<UnboundBuffer>* buf,
+    size_t* offset,
+    size_t* nbytes) {
+  // See if there is a pending recv for this slot.
+  auto pit = pendingRecv_.find(slot);
+  if (pit != pendingRecv_.end()) {
+    auto& recvs = pit->second;
+
+    // Iterate over available buffers to find a match.
+    for (auto rit = recvs.begin(); rit != recvs.end(); rit++) {
+      const auto& ranks = std::get<3>(*rit);
+
+      // If the rank of this peer is in the set of acceptable ranks for
+      // this slot we can proceed and return the buffer to recv into.
+      if (ranks.count(rank) > 0) {
+        // Capture values to return.
+        *buf = std::get<0>(*rit);
+        *offset = std::get<1>(*rit);
+        *nbytes = std::get<2>(*rit);
+        // Cleanup.
+        recvs.erase(rit);
+        if (recvs.empty()) {
+          pendingRecv_.erase(pit);
+        }
+        return true;
+      }
+    }
+  }
+
+  return false;
+}
+
+} // namespace uv
+} // namespace transport
+} // namespace gloo