Enable clang-tidy on torch/csrc/distributed (pytorch#139043)

Fixes #ISSUE_NUMBER

Pull Request resolved: pytorch#139043
Approved by: https://github.com/Skylion007

Author: cyyever

.lintrunner.toml

@@ -214,10 +214,6 @@ include_patterns = [
     'torch/csrc/*.cpp',
     'torch/csrc/**/*.h',
     'torch/csrc/**/*.cpp',
-    'torch/csrc/distributed/autograd/**/*.cpp',
-    'torch/csrc/distributed/autograd/**/*.h',
-    'torch/csrc/distributed/rpc/**/*.cpp',
-    'torch/csrc/distributed/rpc/**/*.h',
     'torch/csrc/jit/serialization/*.h',
     'torch/csrc/jit/serialization/*.cpp',
 ]
@@ -245,7 +241,11 @@ exclude_patterns = [
     'torch/csrc/api/include/torch/linalg.h',
     'torch/csrc/api/include/torch/nn/pimpl-inl.h',
     'torch/csrc/autograd/generated/**',
-    'torch/csrc/distributed/**/*',
+    'torch/csrc/distributed/**/*.cu',
+    'torch/csrc/distributed/c10d/CUDASymmetricMemory-inl.h',
+    'torch/csrc/distributed/c10d/ProcessGroupNCCL.cpp',
+    'torch/csrc/distributed/c10d/WinSockUtils.hpp',
+    'torch/csrc/distributed/c10d/quantization/quantization_gpu.h',
     'torch/csrc/dynamo/eval_frame.h',
     'torch/csrc/inductor/aoti_torch/c/shim.h',
     'torch/csrc/jit/**/*',

torch/csrc/distributed/c10d/TCPStore.cpp

@@ -5,32 +5,16 @@
 #include <torch/csrc/distributed/c10d/Backoff.hpp>
 #include <torch/csrc/distributed/c10d/TCPStore.hpp>
 #include <torch/csrc/distributed/c10d/TCPStoreBackend.hpp>
+#include <torch/csrc/distributed/c10d/Utils.hpp>
 #include <torch/csrc/distributed/c10d/logging.h>
 
-#include <fcntl.h>
 #include <chrono>
 #include <fstream>
-#include <random>
+#include <optional>
 #include <thread>
 #include <unordered_map>
 #include <utility>
 
-#ifdef _WIN32
-#include <io.h>
-#include <winsock2.h>
-#else
-#include <poll.h>
-#include <unistd.h>
-#endif
-
-#ifdef _WIN32
-#include <torch/csrc/distributed/c10d/WinSockUtils.hpp>
-#else
-#include <torch/csrc/distributed/c10d/UnixSockUtils.hpp>
-#endif
-
-#include <torch/csrc/distributed/c10d/socket.h>
-
 namespace c10d {
 namespace detail {
 
@@ -143,11 +127,10 @@ class TCPClient {
     }
   }
   template <typename T>
-  bool receiveValueWithTimeout(T& t, std::chrono::milliseconds timeout) {
+  std::optional<T> receiveValueWithTimeout(std::chrono::milliseconds timeout) {
     if (!socket_.waitForInput(timeout))
-      return false;
-    t = tcputil::recvValue<T>(socket_.handle());
-    return true;
+      return {};
+    return tcputil::recvValue<T>(socket_.handle());
   }
   void setTimeout(std::chrono::milliseconds value);
 
@@ -200,8 +183,10 @@ void TCPClient::setTimeout(std::chrono::milliseconds value) {
 
 class SendBuffer {
   // ethernet mtu 1500 - 40 (ip v6 header) - 20 (tcp header)
+  // NOLINTNEXTLINE(cppcoreguidelines-avoid-const-or-ref-data-members)
   const size_t FLUSH_WATERMARK = 1440;
   std::vector<uint8_t> buffer;
+  // NOLINTNEXTLINE(cppcoreguidelines-avoid-const-or-ref-data-members)
   detail::TCPClient& client;
 
   void maybeFlush() {
@@ -557,10 +542,10 @@ void TCPStore::doWait(
     buffer.flush();
   }
 
-  detail::WaitResponseType response;
-  if (client_->receiveValueWithTimeout<detail::WaitResponseType>(
-          response, timeout)) {
-    if (response != detail::WaitResponseType::STOP_WAITING) {
+  auto response_opt =
+      client_->receiveValueWithTimeout<detail::WaitResponseType>(timeout);
+  if (response_opt.has_value()) {
+    if (response_opt != detail::WaitResponseType::STOP_WAITING) {
       TORCH_CHECK(false, "Stop_waiting response is expected");
     }
     return;
@@ -572,7 +557,7 @@ void TCPStore::doWait(
     buffer.flush();
   }
 
-  response = client_->receiveValue<detail::WaitResponseType>();
+  auto response = client_->receiveValue<detail::WaitResponseType>();
   // this can happen if the server responds before we cancel, just ignore it
   if (response != detail::WaitResponseType::WAIT_CANCELED) {
     if (response != detail::WaitResponseType::STOP_WAITING) {
@@ -639,7 +624,7 @@ void TCPStore::multiSet(
   const std::lock_guard<std::mutex> lock(activeOpLock_);
 
   detail::SendBuffer buffer(*client_, detail::QueryType::MULTI_SET);
-  buffer.appendValue<std::int64_t>(keys.size());
+  buffer.appendValue<std::int64_t>(static_cast<int64_t>(keys.size()));
   for (auto i : c10::irange(keys.size())) {
     buffer.appendString(keyPrefix_ + keys[i]);
     buffer.appendBytes(values[i]);

torch/csrc/distributed/c10d/TCPStoreBackend.cpp

@@ -1,20 +1,10 @@
 
 #include <c10/util/irange.h>
-#include <fcntl.h>
 #include <algorithm>
 #include <array>
-#include <system_error>
 #include <unordered_map>
 #include <utility>
 
-#ifdef _WIN32
-#include <io.h>
-#include <winsock2.h>
-#else
-#include <poll.h>
-#include <unistd.h>
-#endif
-
 #include <c10/util/thread_name.h>
 #include <torch/csrc/distributed/c10d/TCPStoreBackend.hpp>
 #include <torch/csrc/distributed/c10d/logging.h>
@@ -111,7 +101,7 @@ class TCPStoreMasterDaemon : public BackgroundThread {
   const std::chrono::milliseconds checkTimeout_ = std::chrono::milliseconds{10};
   HANDLE ghStopEvent_{};
 #else
-  std::array<int, 2> controlPipeFd_{{-1, -1}};
+  std::array<int, 2> controlPipeFd_{-1, -1};
 #endif
 };
 
@@ -217,8 +207,10 @@ void TCPStoreMasterDaemon::queryFds(std::vector<struct pollfd>& fds) {
       // we hit an exception here.
       clearSocketWaitState(fds[fdIdx].fd);
 
-      fds.erase(fds.begin() + fdIdx);
-      sockets_.erase(sockets_.begin() + fdIdx - CONNECT_SOCKET_OFFSET);
+      fds.erase(fds.begin() + static_cast<std::ptrdiff_t>(fdIdx));
+      sockets_.erase(
+          sockets_.begin() + static_cast<std::ptrdiff_t>(fdIdx) -
+          CONNECT_SOCKET_OFFSET);
       --fdIdx;
       continue;
     }
@@ -256,7 +248,7 @@ void TCPStoreMasterDaemon::clearSocketWaitState(int socket) {
 // or, in the case of wait
 // type of query | number of args | size of arg1 | arg1 | ...
 void TCPStoreMasterDaemon::query(int socket) {
-  QueryType qt;
+  QueryType qt{};
   tcputil::recvBytes<QueryType>(socket, &qt, 1);
 
   if (isMiscellaneousSocket(socket)) {
@@ -401,13 +393,13 @@ void TCPStoreMasterDaemon::getHandler(int socket) const {
 }
 
 void TCPStoreMasterDaemon::getNumKeysHandler(int socket) const {
-  tcputil::sendValue<int64_t>(socket, tcpStore_.size());
+  tcputil::sendValue<size_t>(socket, tcpStore_.size());
 }
 
 void TCPStoreMasterDaemon::deleteHandler(int socket) {
   std::string key = tcputil::recvString(socket);
   auto numDeleted = tcpStore_.erase(key);
-  tcputil::sendValue<int64_t>(socket, numDeleted);
+  tcputil::sendValue<size_t>(socket, numDeleted);
 }
 
 void TCPStoreMasterDaemon::checkHandler(int socket) const {

torch/csrc/distributed/c10d/TCPStoreLibUvBackend.cpp

@@ -2,7 +2,6 @@
 #include <deque>
 #include <exception>
 #include <memory>
-#include <ostream>
 #include <unordered_map>
 #include <unordered_set>
 #include <utility>
@@ -35,15 +34,15 @@ Other callbacks don't provide exception safety so avoid there.
 // This controls how many un-accepted TCP connections can be waiting in the
 // backlog. This should be at least world size to avoid issues on init. We set
 // it to -1 to use the host max value which is controlled by `soconnmax`.
-#define DEFAULT_BACKLOG -1
-#define MAX_KEY_COUNT (128 * 1024)
-#define MAX_STRING_LEN (8 * 1024)
-#define MAX_PAYLOAD_LEN (8 * 1024 * 1024)
+auto constexpr DEFAULT_BACKLOG = -1;
+auto constexpr MAX_KEY_COUNT = size_t(128 * 1024);
+auto constexpr MAX_STRING_LEN = 8 * 1024;
+auto constexpr MAX_PAYLOAD_LEN = 8 * 1024 * 1024;
 
 // This controls the preferred size for buffers.
 // Too small and we'll need multiple buffers for one request
 // Too big and we might taxing malloc
-#define ALLOC_BUFFER_SIZE ((size_t)4000)
+auto constexpr ALLOC_BUFFER_SIZE = size_t(4000);
 class UvHandle : public c10::intrusive_ptr_target {
  public:
   ~UvHandle() override = default;
@@ -105,7 +104,8 @@ class UvTcpSocket : public UvHandle {
       uv_handle_t* handle,
       size_t suggested_size,
       uv_buf_t* buf) {
-    suggested_size = std::min(suggested_size, (size_t)ALLOC_BUFFER_SIZE);
+    suggested_size = std::min(suggested_size, ALLOC_BUFFER_SIZE);
+    // NOLINTNEXTLINE(cppcoreguidelines-no-malloc)
     buf->base = (char*)malloc(suggested_size);
     buf->len = suggested_size;
   }
@@ -486,6 +486,7 @@ class ChunkedStream {
 
   void append(uv_buf_t buf) {
     if (buf.len == 0) {
+      // NOLINTNEXTLINE(cppcoreguidelines-no-malloc)
       free(buf.base);
     } else {
       capacity += buf.len;
@@ -597,6 +598,7 @@ class ChunkedStream {
     }
 
     for (size_t i = 0; i < buff_idx; ++i) {
+      // NOLINTNEXTLINE(cppcoreguidelines-no-malloc)
       free(buffers[0].base);
       capacity -= buffers[0].len;
       buffers.pop_front();
@@ -1259,12 +1261,14 @@ const std::vector<uint8_t>& LibUVStoreDaemon::compareAndSet(
     if (expectedValue.empty()) {
       tcpStore_[key] = newValue;
       wakeupWaitingClients(key);
+      // NOLINTNEXTLINE(bugprone-return-const-ref-from-parameter)
       return newValue;
     } else {
       // TODO: This code path is not ideal as we are "lying" to the caller in
       // case the key does not exist. We should come up with a working solution.
       // It might make more sense to return ""
       wakeupWaitingClients(key);
+      // NOLINTNEXTLINE(bugprone-return-const-ref-from-parameter)
       return expectedValue;
     }
   } else {
@@ -1326,11 +1330,11 @@ bool LibUVStoreDaemon::waitKeys(
 }
 
 int64_t LibUVStoreDaemon::size() {
-  return tcpStore_.size();
+  return static_cast<int64_t>(tcpStore_.size());
 }
 
 int64_t LibUVStoreDaemon::deleteKey(const std::string& key) {
-  return tcpStore_.erase(key);
+  return static_cast<int64_t>(tcpStore_.erase(key));
 }
 
 void LibUVStoreDaemon::append(