[lldb] [Core] Reimplement Communication::ReadThread using MainLoop

[mgorny]

Summary

Use MainLoop to implement Communication::ReadThread() instead of
blocking reads. Most importantly, this makes it possible to use
the newly implemented TriggerPendingCallbacks() mechanism to immediately
interrupt waiting for new data. This in turn will make it possible to
enable the possibility of writing to the descriptor using the same
thread.

Sponsored by: The FreeBSD Foundation

[mgorny]

Unfortunately, many tests are failing or timing out, so need to investigate more.

[mgorny]

Update Communication::SynchronizeWithReadThread(), fixing tests.

[mgorny]

Rebase/update for the new error passing mechanism.

[labath]

lldb/source/Core/Communication.cpp

427

Have you considered putting this code (some version of it) inside InterruptRead? Basically replacing the select call inside BytesAvailable with something MainLoop-based ?

[mgorny]

Rebase, move StopReadThread() test here.

[mgorny]

lldb/source/Core/Communication.cpp

427

To be honest, I've been considering removing InterruptRead() entirely, now that the read loop is triggered by available input rather than read-with-timeout. However, it's still used by editline support.

That said, I'm not sure what's your idea, given that Connection does not have awareness of Communication that's using it. I suppose I could pass the MainLoop instance as a parameter but we'd still have to maintain a separate version for editline support, and we only have a single caller here.

[labath]

lldb/source/Core/Communication.cpp

427

To be honest, I've been considering removing InterruptRead() entirely, now that the read loop is triggered by available input rather than read-with-timeout. However, it's still used by editline support.

If we could do that, then it would be even better. And it looks like it should be possible to use a MainLoop instance inside the Editline class, instead of the built-in interruption support.

That said, I'm not sure what's your idea, given that Connection does not have awareness of Communication that's using it. I suppose I could pass the MainLoop instance as a parameter but we'd still have to maintain a separate version for editline support, and we only have a single caller here.

I don't claim to have thought this out completely, but I was imagining that the MainLoop instance would be internal to the Connection class. The external interface of the Connection class would remain unchanged (so the Communication class would not need to change), and the InterruptRead function would be implemented using the MainLoop functionality.

[mgorny]

lldb/source/Core/Communication.cpp

427

That said, I'm not sure what's your idea, given that Connection does not have awareness of Communication that's using it. I suppose I could pass the MainLoop instance as a parameter but we'd still have to maintain a separate version for editline support, and we only have a single caller here.

I don't claim to have thought this out completely, but I was imagining that the MainLoop instance would be internal to the Connection class. The external interface of the Connection class would remain unchanged (so the Communication class would not need to change), and the InterruptRead function would be implemented using the MainLoop functionality.

So basically make ConnectionFileDescriptor::BytesAvailable() use a main loop to wait for some event, and make ConnectionFileDescriptor::InterruptRead() interrupt that?

I suppose both options are feasible but I don't have sufficient foresight to tell which one is better. That said, I have no clue about ConnectionGenericFileWindows and I feel that using MainLoop would make it possible to reduce code duplication between it and CFD. Though I'd preferred that someone with access to a Windows environment done that.

[labath]

lldb/source/Core/Communication.cpp

427

So basically make ConnectionFileDescriptor::BytesAvailable() use a main loop to wait for some event, and make ConnectionFileDescriptor::InterruptRead() interrupt that?

Correct.

I suppose both options are feasible but I don't have sufficient foresight to tell which one is better. That said, I have no clue about ConnectionGenericFileWindows and I feel that using MainLoop would make it possible to reduce code duplication between it and CFD. Though I'd preferred that someone with access to a Windows environment done that.

That is my hope as well, but it's not going to be very easy, since the ReadFile (used by CGFW) api requires you to actually attempt to read some data in order to wait for the result) . That said, I'm not sure this really speaks in favour of one approach over the other. If e.g. we are able to use the same MainLoop code for both CFD and CGFW, then we could probably move that code over to a common base class.

What I would like to avoid is to increase the number of places that do selects. So, if you can port editline code to something mainloop-based and remove the InterruptRead, then that's fine by me, and we can figure out what to do with windows later.

[mgorny]

Rebase on top of D131160.

[mgorny]

lldb/source/Core/Communication.cpp

427

What I would like to avoid is to increase the number of places that do selects. So, if you can port editline code to something mainloop-based and remove the InterruptRead, then that's fine by me, and we can figure out what to do with windows later.

Well, I suppose I'll add that to my TODO and maybe I'll manage to get to it before I retire ;-). Also note that the proposed gdb-remote interrupt changes rely on InterruptRead(), so I'd have to change that as well. However, I think it's better to focus on multiprocess first, as that definitely will influence the final shape of how we do async there.

lldb/include/lldb/Core/Communication.h

 //===-- Communication.h -----------------------------------------*- C++ -*-===//
 //
 // Part of the LLVM Project, under the Apache License v2.0 with LLVM Exceptions.
 // See https://llvm.org/LICENSE.txt for license information.
 // SPDX-License-Identifier: Apache-2.0 WITH LLVM-exception
 //
 //===----------------------------------------------------------------------===//
 
 #ifndef LLDB_CORE_COMMUNICATION_H
 #define LLDB_CORE_COMMUNICATION_H
 
 #include "lldb/Host/HostThread.h"
+#include "lldb/Host/MainLoop.h"
 #include "lldb/Utility/Broadcaster.h"
 #include "lldb/Utility/Timeout.h"
 #include "lldb/lldb-defines.h"
 #include "lldb/lldb-enumerations.h"
 #include "lldb/lldb-forward.h"
 #include "lldb/lldb-types.h"
 
 #include <atomic>
ConstString &GetBroadcasterClass() const override {
     return GetStaticBroadcasterClass();
   }
 
 protected:
   lldb::ConnectionSP m_connection_sp; ///< The connection that is current in use
                                       ///by this communications class.
   HostThread m_read_thread; ///< The read thread handle in case we need to
                             ///cancel the thread.
+  std::unique_ptr<MainLoop> m_io_loop; ///< The loop instance used by
+                                       ///the read thread.
   std::atomic<bool> m_read_thread_enabled;
   std::atomic<bool> m_read_thread_did_exit;
   std::string
       m_bytes; ///< A buffer to cache bytes read in the ReadThread function.
   std::recursive_mutex m_bytes_mutex; ///< A mutex to protect multi-threaded
                                       ///access to the cached bytes.
   lldb::ConnectionStatus m_pass_status; ///< Connection status passthrough
                                         ///from read thread.

lldb/source/Core/Communication.cpp

bool Communication::StartReadThread(Status *error_ptr) {
   LLDB_LOG(GetLog(LLDBLog::Communication),
            "{0} Communication::StartReadThread ()", this);
 
   const std::string thread_name =
       llvm::formatv("<lldb.comm.{0}>", GetBroadcasterName());
 
   m_read_thread_enabled = true;
   m_read_thread_did_exit = false;
+  // Allocate the I/O loop in main thread to avoid races.
+  m_io_loop.reset(new MainLoop());
   auto maybe_thread = ThreadLauncher::LaunchThread(
       thread_name, [this] { return ReadThread(); });
   if (maybe_thread) {
     m_read_thread = *maybe_thread;
   } else {
     if (error_ptr)
       *error_ptr = Status(maybe_thread.takeError());
     else {
bool Communication::StopReadThread(Status *error_ptr) {
 
   LLDB_LOG(GetLog(LLDBLog::Communication),
            "{0} Communication::StopReadThread ()", this);
 
   m_read_thread_enabled = false;
 
   BroadcastEvent(eBroadcastBitReadThreadShouldExit, nullptr);
 
-  // error = m_read_thread.Cancel();
+  m_io_loop->AddPendingCallback(
+      [](MainLoopBase &loop) { loop.RequestTermination(); });
+  m_io_loop->TriggerPendingCallbacks();
 
   Status error = m_read_thread.Join(nullptr);
+  if (error.Success())
+    m_io_loop.reset(nullptr);
   return error.Success();
 }
 
 bool Communication::JoinReadThread(Status *error_ptr) {
   if (!m_read_thread.IsJoinable())
     return true;
 
   Status error = m_read_thread.Join(nullptr);

 
 bool Communication::ReadThreadIsRunning() { return m_read_thread_enabled; }
 
 lldb::thread_result_t Communication::ReadThread() {
   Log *log = GetLog(LLDBLog::Communication);
 
   LLDB_LOG(log, "Communication({0}) thread starting...", this);
 
-  uint8_t buf[1024];
-
-  Status error;
-  ConnectionStatus status = eConnectionStatusSuccess;
-  bool done = false;
   bool disconnect = false;
-  while (!done && m_read_thread_enabled) {
+  ConnectionStatus status = eConnectionStatusSuccess;
+  Status error;
+  if (IsConnected()) {
+    Status loop_error;
+    auto handle = m_io_loop->RegisterReadObject(
+        m_connection_sp->GetReadObject(),
+        [this, &disconnect, &status, &error](MainLoopBase &loop) {
+          Log *log = GetLog(LLDBLog::Communication);
+
+          if (!m_read_thread_enabled) {
+            loop.RequestTermination();
+            return;
+          }
+
+          uint8_t buf[1024];
           size_t bytes_read = ReadFromConnection(
               buf, sizeof(buf), std::chrono::seconds(5), status, &error);
           if (bytes_read > 0 || status == eConnectionStatusEndOfFile)
             AppendBytesToCache(buf, bytes_read, true, status);
 
           switch (status) {
           case eConnectionStatusSuccess:
+          case eConnectionStatusInterrupted:
             break;
 
           case eConnectionStatusEndOfFile:
-      done = true;
+            loop.RequestTermination();
             disconnect = GetCloseOnEOF();
             break;
           case eConnectionStatusError: // Check GetError() for details
             if (error.GetType() == eErrorTypePOSIX && error.GetError() == EIO) {
               // EIO on a pipe is usually caused by remote shutdown
+              loop.RequestTermination();
               disconnect = GetCloseOnEOF();
-        done = true;
             }
             if (error.Fail())
               LLDB_LOG(log, "error: {0}, status = {1}", error,
                        Communication::ConnectionStatusAsString(status));
             break;
-    case eConnectionStatusInterrupted: // Synchronization signal from
-                                       // SynchronizeWithReadThread()
-      // The connection returns eConnectionStatusInterrupted only when there is
-      // no input pending to be read, so we can signal that.
-      BroadcastEvent(eBroadcastBitNoMorePendingInput);
-      break;
           case eConnectionStatusNoConnection:   // No connection
-    case eConnectionStatusLostConnection: // Lost connection while connected to
-                                          // a valid connection
-      done = true;
+          case eConnectionStatusLostConnection: // Lost connection while
+                                                // connected to a valid
+                                                // connection
+            loop.RequestTermination();
             [[fallthrough]];
           case eConnectionStatusTimedOut: // Request timed out
             if (error.Fail())
               LLDB_LOG(log, "error: {0}, status = {1}", error,
                        Communication::ConnectionStatusAsString(status));
             break;
           }
+        },
+        loop_error);
+    if (loop_error.Success())
+      loop_error = m_io_loop->Run();
+    if (!loop_error.Success()) {
+      error = std::move(loop_error);
+      status = lldb::eConnectionStatusError;
+    }
+  } else {
+    if (m_connection_sp)
+      status = lldb::eConnectionStatusLostConnection;
+    else
+      status = lldb::eConnectionStatusNoConnection;
+    error.SetErrorString(ConnectionStatusAsString(status));
   }
   m_pass_status = status;
   m_pass_error = std::move(error);
+
   LLDB_LOG(log, "Communication({0}) thread exiting...", this);
 
   // Handle threads wishing to synchronize with us.
   {
     // Prevent new ones from showing up.
     m_read_thread_did_exit = true;
 
     // Unblock any existing thread waiting for the synchronization signal.
ListenerSP listener_sp(
       Listener::MakeListener("Communication::SyncronizeWithReadThread"));
   listener_sp->StartListeningForEvents(this, eBroadcastBitNoMorePendingInput);
 
   // If the thread is not running, there is no point in synchronizing.
   if (!m_read_thread_enabled || m_read_thread_did_exit)
     return;
 
   // Notify the read thread.
-  m_connection_sp->InterruptRead();
+  m_io_loop->AddPendingCallback([this](MainLoopBase &loop) {
+    BroadcastEvent(eBroadcastBitNoMorePendingInput);
+  });
+  m_io_loop->TriggerPendingCallbacks();
 
   // Wait for the synchronization event.
   EventSP event_sp;
   listener_sp->GetEvent(event_sp, llvm::None);
 }
 
 void Communication::SetConnection(std::unique_ptr<Connection> connection) {
   Disconnect(nullptr);

lldb/unittests/Core/CommunicationTest.cpp

 TEST_F(CommunicationTest, Read) {
   CommunicationReadTest(/*use_thread=*/false);
 }
 
 TEST_F(CommunicationTest, ReadThread) {
   CommunicationReadTest(/*use_thread=*/true);
 }
 
+TEST_F(CommunicationTest, StopReadThread) {
+  std::condition_variable finished;
+  std::mutex finished_mutex;
+
+  std::thread test_thread{[&]() {
+    std::unique_ptr<TCPSocket> a, b;
+    ASSERT_TRUE(CreateTCPConnectedSockets("localhost", &a, &b));
+
+    Communication comm("test");
+    comm.SetConnection(std::make_unique<ConnectionFileDescriptor>(b.release()));
+    comm.SetCloseOnEOF(true);
+
+    ASSERT_TRUE(comm.StartReadThread());
+    ASSERT_TRUE(comm.StopReadThread());
+    finished.notify_all();
+  }};
+
+  // StopReadThread() can hang, so force an external timeout.
+  std::unique_lock<std::mutex> lock{finished_mutex};
+  ASSERT_EQ(finished.wait_for(lock, std::chrono::seconds(3)),
+            std::cv_status::no_timeout);
+  test_thread.join();
+}
+
 TEST_F(CommunicationTest, SynchronizeWhileClosing) {
   std::unique_ptr<TCPSocket> a, b;
   ASSERT_TRUE(CreateTCPConnectedSockets("localhost", &a, &b));
 
   Communication comm("test");
   comm.SetConnection(std::make_unique<ConnectionFileDescriptor>(b.release()));
   comm.SetCloseOnEOF(true);
   ASSERT_TRUE(comm.StartReadThread());