Revert "[c10d] Separate monitoring thread into a class in PGNCCL (pytorch#153977)"

This reverts commit 0db9c64.

Reverted pytorch#153977 on behalf of https://github.com/izaitsevfb due to breaks lots of jobs internally, safer to revert, see D75628917 ([comment](pytorch#153977 (comment)))

Author: pytorchmergebot

test/cpp/c10d/ProcessGroupNCCLErrorsTest.cpp

@@ -179,12 +179,7 @@ class ProcessGroupNCCLNoHeartbeatCaught
       int rank,
       int size,
       c10::intrusive_ptr<c10d::ProcessGroupNCCL::Options> opts)
-      : ProcessGroupNCCLTimedOutErrors(store, rank, size, std::move(opts)) {
-    // Override the heartbeat monitor function to make sure that we capture
-    // the exception in the monitor thread because we cannot try-catch it in
-    // the main thread and we set a flag for the main thread to check.
-    heartbeatMonitor_ = std::make_unique<TestHeartbeatMonitor>(this);
-  }
+      : ProcessGroupNCCLTimedOutErrors(store, rank, size, std::move(opts)) {}
 
   std::mutex& getWatchdogMutex() {
     return workMetaListMutex_;
@@ -200,22 +195,18 @@ class ProcessGroupNCCLNoHeartbeatCaught
     asyncDebugDump.wait();
   }
 
-  class TestHeartbeatMonitor : public c10d::ProcessGroupNCCL::HeartbeatMonitor {
-   public:
-    using HeartbeatMonitor::HeartbeatMonitor;
-
-    void runLoop() override {
-      try {
-        c10d::ProcessGroupNCCL::HeartbeatMonitor::runLoop();
-      } catch (std::runtime_error& e) {
-        // Safe cast because we know it's a ProcessGroupNCCLNoHeartbeatCaught
-        auto* pg = static_cast<ProcessGroupNCCLNoHeartbeatCaught*>(pg_);
-        pg->hasMonitorThreadCaughtError_ = true;
-      }
+ protected:
+  // Override the heartbeat monitor function to make sure that we capture
+  // the exception in the monitor thread because we cannot try-catch it in
+  // the main thread and we set a flag for the main thread to check.
+  void heartbeatMonitor() override {
+    try {
+      c10d::ProcessGroupNCCL::heartbeatMonitor();
+    } catch (std::runtime_error& e) {
+      hasMonitorThreadCaughtError_ = true;
     }
-  };
+  }
 
- protected:
   // It's really hard to unit test std::abort. So we override it instead.
   // Commented this override, we do see process aborted with core dump without
   // this override.