[lldb] Improving synchronization of MainLoopWindows. (#147438)

This should improve synchronizing the MainLoopWindows monitor thread
with the main loop state.

This uses the `m_ready` and `m_event` event handles to manage when the
Monitor thread continues and adds new tests to cover additional use
cases.

I believe this should fix #147291 but it is hard to ensure a race
condition is fixed without running the CI on multiple
machines/configurations.

---------

Co-authored-by: Pavel Labath <pavel@labath.sk>

Author: ashgti

lldb/source/Host/windows/MainLoopWindows.cpp

@@ -12,16 +12,16 @@
 #include "lldb/Host/windows/windows.h"
 #include "lldb/Utility/Status.h"
 #include "llvm/Config/llvm-config.h"
-#include "llvm/Support/Casting.h"
 #include "llvm/Support/WindowsError.h"
 #include <algorithm>
 #include <cassert>
-#include <cerrno>
-#include <csignal>
 #include <ctime>
 #include <io.h>
+#include <synchapi.h>
 #include <thread>
 #include <vector>
+#include <winbase.h>
+#include <winerror.h>
 #include <winsock2.h>
 
 using namespace lldb;
@@ -42,11 +42,12 @@ namespace {
 class PipeEvent : public MainLoopWindows::IOEvent {
 public:
   explicit PipeEvent(HANDLE handle)
-      : IOEvent(CreateEventW(NULL, /*bManualReset=*/FALSE,
+      : IOEvent(CreateEventW(NULL, /*bManualReset=*/TRUE,
                              /*bInitialState=*/FALSE, NULL)),
-        m_handle(handle), m_ready(CreateEventW(NULL, /*bManualReset=*/FALSE,
+        m_handle(handle), m_ready(CreateEventW(NULL, /*bManualReset=*/TRUE,
                                                /*bInitialState=*/FALSE, NULL)) {
     assert(m_event && m_ready);
+    m_monitor_thread = std::thread(&PipeEvent::Monitor, this);
   }
 
   ~PipeEvent() override {
@@ -65,15 +66,27 @@ class PipeEvent : public MainLoopWindows::IOEvent {
   }
 
   void WillPoll() override {
-    if (!m_monitor_thread.joinable())
-      m_monitor_thread = std::thread(&PipeEvent::Monitor, this);
+    if (WaitForSingleObject(m_event, /*dwMilliseconds=*/0) != WAIT_TIMEOUT) {
+      // The thread has already signalled that the data is available. No need
+      // for further polling until we consume that event.
+      return;
+    }
+    if (WaitForSingleObject(m_ready, /*dwMilliseconds=*/0) != WAIT_TIMEOUT) {
+      // The thread is already waiting for data to become available.
+      return;
+    }
+    // Start waiting.
+    SetEvent(m_ready);
   }
 
-  void Disarm() override { SetEvent(m_ready); }
+  void Disarm() override { ResetEvent(m_event); }
 
   /// Monitors the handle performing a zero byte read to determine when data is
   /// avaiable.
   void Monitor() {
+    // Wait until the MainLoop tells us to start.
+    WaitForSingleObject(m_ready, INFINITE);
+
     do {
       char buf[1];
       DWORD bytes_read = 0;
@@ -110,7 +123,11 @@ class PipeEvent : public MainLoopWindows::IOEvent {
         continue;
       }
 
+      // Notify that data is available on the pipe. It's important to set this
+      // before clearing m_ready to avoid a race with WillPoll.
       SetEvent(m_event);
+      // Stop polling until we're told to resume.
+      ResetEvent(m_ready);
 
       // Wait until the current read is consumed before doing the next read.
       WaitForSingleObject(m_ready, INFINITE);

lldb/unittests/Host/MainLoopTest.cpp

@@ -10,6 +10,7 @@
 #include "TestingSupport/SubsystemRAII.h"
 #include "lldb/Host/ConnectionFileDescriptor.h"
 #include "lldb/Host/FileSystem.h"
+#include "lldb/Host/MainLoopBase.h"
 #include "lldb/Host/PseudoTerminal.h"
 #include "lldb/Host/common/TCPSocket.h"
 #include "llvm/Config/llvm-config.h" // for LLVM_ON_UNIX
@@ -64,7 +65,7 @@ class MainLoopTest : public testing::Test {
 };
 } // namespace
 
-TEST_F(MainLoopTest, ReadObject) {
+TEST_F(MainLoopTest, ReadSocketObject) {
   char X = 'X';
   size_t len = sizeof(X);
   ASSERT_TRUE(socketpair[0]->Write(&X, len).Success());
@@ -101,6 +102,144 @@ TEST_F(MainLoopTest, ReadPipeObject) {
   ASSERT_EQ(1u, callback_count);
 }
 
+TEST_F(MainLoopTest, MultipleReadsPipeObject) {
+  Pipe pipe;
+
+  ASSERT_TRUE(pipe.CreateNew().Success());
+
+  MainLoop loop;
+
+  std::future<void> async_writer = std::async(std::launch::async, [&] {
+    for (int i = 0; i < 5; ++i) {
+      std::this_thread::sleep_for(std::chrono::milliseconds(500));
+      char X = 'X';
+      size_t len = sizeof(X);
+      ASSERT_THAT_EXPECTED(pipe.Write(&X, len), llvm::HasValue(1));
+    }
+  });
+
+  Status error;
+  lldb::FileSP file = std::make_shared<NativeFile>(
+      pipe.GetReadFileDescriptor(), File::eOpenOptionReadOnly, false);
+  auto handle = loop.RegisterReadObject(
+      file,
+      [&](MainLoopBase &loop) {
+        callback_count++;
+        if (callback_count == 5)
+          loop.RequestTermination();
+
+        // Read some data to ensure the handle is not in a readable state.
+        char buf[1024] = {0};
+        size_t len = sizeof(buf);
+        ASSERT_THAT_ERROR(file->Read(buf, len).ToError(), llvm::Succeeded());
+        EXPECT_EQ(len, 1);
+        EXPECT_EQ(buf[0], 'X');
+      },
+      error);
+  ASSERT_TRUE(error.Success());
+  ASSERT_TRUE(handle);
+  ASSERT_TRUE(loop.Run().Success());
+  ASSERT_EQ(5u, callback_count);
+  async_writer.wait();
+}
+
+TEST_F(MainLoopTest, PipeDelayBetweenRegisterAndRun) {
+  Pipe pipe;
+
+  ASSERT_TRUE(pipe.CreateNew().Success());
+
+  MainLoop loop;
+
+  Status error;
+  lldb::FileSP file = std::make_shared<NativeFile>(
+      pipe.GetReadFileDescriptor(), File::eOpenOptionReadOnly, false);
+  auto handle = loop.RegisterReadObject(
+      file,
+      [&](MainLoopBase &loop) {
+        callback_count++;
+
+        // Read some data to ensure the handle is not in a readable state.
+        char buf[1024] = {0};
+        size_t len = sizeof(buf);
+        ASSERT_THAT_ERROR(file->Read(buf, len).ToError(), llvm::Succeeded());
+        EXPECT_EQ(len, 2);
+        EXPECT_EQ(buf[0], 'X');
+        EXPECT_EQ(buf[1], 'X');
+      },
+      error);
+  auto cb = [&](MainLoopBase &) {
+    callback_count++;
+    char X = 'X';
+    size_t len = sizeof(X);
+    // Write twice and ensure we coalesce into a single read.
+    ASSERT_THAT_EXPECTED(pipe.Write(&X, len), llvm::HasValue(1));
+    ASSERT_THAT_EXPECTED(pipe.Write(&X, len), llvm::HasValue(1));
+  };
+  // Add a write that triggers a read events.
+  loop.AddCallback(cb, std::chrono::milliseconds(500));
+  loop.AddCallback([](MainLoopBase &loop) { loop.RequestTermination(); },
+                   std::chrono::milliseconds(1000));
+  ASSERT_TRUE(error.Success());
+  ASSERT_TRUE(handle);
+
+  // Write between RegisterReadObject / Run should NOT invoke the callback.
+  cb(loop);
+  ASSERT_EQ(1u, callback_count);
+
+  ASSERT_TRUE(loop.Run().Success());
+  ASSERT_EQ(4u, callback_count);
+}
+
+TEST_F(MainLoopTest, NoSelfTriggersDuringPipeHandler) {
+  Pipe pipe;
+
+  ASSERT_TRUE(pipe.CreateNew().Success());
+
+  MainLoop loop;
+
+  Status error;
+  lldb::FileSP file = std::make_shared<NativeFile>(
+      pipe.GetReadFileDescriptor(), File::eOpenOptionReadOnly, false);
+  auto handle = loop.RegisterReadObject(
+      file,
+      [&](MainLoopBase &lop) {
+        callback_count++;
+
+        char X = 'Y';
+        size_t len = sizeof(X);
+        // writes / reads during the handler callback should NOT trigger itself.
+        ASSERT_THAT_EXPECTED(pipe.Write(&X, len), llvm::HasValue(1));
+
+        char buf[1024] = {0};
+        len = sizeof(buf);
+        ASSERT_THAT_ERROR(file->Read(buf, len).ToError(), llvm::Succeeded());
+        EXPECT_EQ(len, 2);
+        EXPECT_EQ(buf[0], 'X');
+        EXPECT_EQ(buf[1], 'Y');
+
+        if (callback_count == 2)
+          loop.RequestTermination();
+      },
+      error);
+  // Add a write that triggers a read event.
+  loop.AddPendingCallback([&](MainLoopBase &) {
+    char X = 'X';
+    size_t len = sizeof(X);
+    ASSERT_THAT_EXPECTED(pipe.Write(&X, len), llvm::HasValue(1));
+  });
+  loop.AddCallback(
+      [&](MainLoopBase &) {
+        char X = 'X';
+        size_t len = sizeof(X);
+        ASSERT_THAT_EXPECTED(pipe.Write(&X, len), llvm::HasValue(1));
+      },
+      std::chrono::milliseconds(500));
+  ASSERT_TRUE(error.Success());
+  ASSERT_TRUE(handle);
+  ASSERT_TRUE(loop.Run().Success());
+  ASSERT_EQ(2u, callback_count);
+}
+
 TEST_F(MainLoopTest, NoSpuriousPipeReads) {
   Pipe pipe;