[lldb] [Core] Pass error/status from Communication::ReadThread()

[mgorny]

Summary

Ensure that the ConnectionStatus and Status from
Communication::ReadThread() is correctly passed to ::Read() in order
to fix further discrepancies between ::Read() calls in non-threaded
and threaded modes.

Sponsored by: The FreeBSD Foundation

[labath]

lldb/unittests/Core/CommunicationTest.cpp

140

I don't understand what is the purpose of this. Can you elaborate?

[mgorny]

lldb/unittests/Core/CommunicationTest.cpp

140

Arrr, I've accidentally moved this test from the MainLoop patch while rebasing. Long story short, StopReadThread() could hang there if we don't trigger the callbacks there.

[mgorny]

Hmm, I see you've updated the tests, so I need to update this patch anyway.

[mgorny]

Rebase, move StopReadThread() test to the other patch.

[labath]

lldb/unittests/Core/CommunicationTest.cpp

17–18

I guess these two are not used anymore.

lldb/include/lldb/Core/Communication.h

protected:
   HostThread m_read_thread; ///< The read thread handle in case we need to
                             ///cancel the 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.
+  Status m_pass_error; ///< Error passthrough from read thread.
   std::mutex
       m_write_mutex; ///< Don't let multiple threads write at the same time...
   std::mutex m_synchronize_mutex;
   ReadThreadBytesReceived m_callback;
   void *m_callback_baton;
   bool m_close_on_eof;
 
   size_t ReadFromConnection(void *dst, size_t dst_len,

lldb/source/Core/Communication.cpp

if (m_read_thread_enabled) {
     while (listener_sp->GetEvent(event_sp, timeout)) {
       const uint32_t event_type = event_sp->GetType();
       if (event_type & eBroadcastBitReadThreadGotBytes) {
         return GetCachedBytes(dst, dst_len);
       }
 
       if (event_type & eBroadcastBitReadThreadDidExit) {
         // If the thread exited of its own accord, it either means it
-        // hit an end-of-file condition or lost connection
-        // (we verified that we had an connection above).
-        if (!m_connection_sp) {
-          if (error_ptr)
-            error_ptr->SetErrorString("Lost connection.");
-          status = eConnectionStatusLostConnection;
-        } else
-          status = eConnectionStatusEndOfFile;
+        // hit an end-of-file condition or an error.
+        status = m_pass_status;
+          *error_ptr = std::move(m_pass_error);
 
         if (GetCloseOnEOF())
           Disconnect(nullptr);
         return 0;
       }
     }
 
     if (error_ptr)
case eConnectionStatusLostConnection: // Lost connection while connected to
       [[fallthrough]];
     case eConnectionStatusTimedOut: // Request timed out
       if (error.Fail())
         LLDB_LOG(log, "error: {0}, status = {1}", error,
                  Communication::ConnectionStatusAsString(status));
       break;
     }
   }
-  log = GetLog(LLDBLog::Communication);
+  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.

lldb/unittests/Core/CommunicationTest.cpp

 //===-- CommunicationTest.cpp ---------------------------------------------===//
 //
 // 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
 //
 //===----------------------------------------------------------------------===//
 
 #include "lldb/Core/Communication.h"
 #include "lldb/Host/Config.h"
 #include "lldb/Host/ConnectionFileDescriptor.h"
 #include "lldb/Host/Pipe.h"
 #include "llvm/Testing/Support/Error.h"
 #include "gtest/gtest.h"
 
+#include <chrono>
+#include <condition_variable>
+#include <mutex>
 #include <thread>
 
 #if LLDB_ENABLE_POSIX
 #include <fcntl.h>
 #endif
 
 using namespace lldb_private;
 
static void CommunicationReadTest(bool use_read_thread) {
   // This read should return EOF.
   pipe.CloseWriteFileDescriptor();
   error.Clear();
   EXPECT_EQ(
       comm.Read(buf, sizeof(buf), std::chrono::seconds(5), status, &error), 0U);
   EXPECT_EQ(status, lldb::eConnectionStatusEndOfFile);
   EXPECT_THAT_ERROR(error.ToError(), llvm::Succeeded());
 
-  // JoinReadThread() should just return immediately since there was no read
+  // JoinReadThread() should just return immediately if there was no read
   // thread started.
   EXPECT_TRUE(comm.JoinReadThread());
+
+  // Test using Communication without a connection.
+  comm.SetConnection(nullptr);
+  if (use_read_thread)
+    ASSERT_TRUE(comm.StartReadThread());
+  error.Clear();
+  EXPECT_EQ(
+      comm.Read(buf, sizeof(buf), std::chrono::seconds(5), status, &error), 0U);
+  EXPECT_EQ(status, lldb::eConnectionStatusNoConnection);
+  EXPECT_THAT_ERROR(error.ToError(), llvm::Failed());
+  EXPECT_TRUE(comm.JoinReadThread());
+
+  // Test using Communication that is disconnected.
+  ASSERT_THAT_ERROR(pipe.CreateNew(/*child_process_inherit=*/false).ToError(),
+                    llvm::Succeeded());
+  comm.SetConnection(std::make_unique<ConnectionFileDescriptor>(
+      pipe.ReleaseReadFileDescriptor(), /*owns_fd=*/true));
+  comm.SetCloseOnEOF(true);
+  ASSERT_EQ(comm.Disconnect(), lldb::eConnectionStatusSuccess);
+  if (use_read_thread)
+    ASSERT_TRUE(comm.StartReadThread());
+  error.Clear();
+  EXPECT_EQ(
+      comm.Read(buf, sizeof(buf), std::chrono::seconds(5), status, &error), 0U);
+  EXPECT_EQ(status, lldb::eConnectionStatusLostConnection);
+  EXPECT_THAT_ERROR(error.ToError(), llvm::Failed());
+  EXPECT_TRUE(comm.JoinReadThread());
 }
 
 TEST(CommunicationTest, Read) {
   CommunicationReadTest(/*use_thread=*/false);
 }
 
 TEST(CommunicationTest, ReadThread) {
   CommunicationReadTest(/*use_thread=*/true);
 }
 
+TEST(CommunicationTest, StopReadThread) {
+  std::condition_variable finished;
+  std::mutex finished_mutex;
+
+  std::thread test_thread{[&finished]() {
+    Pipe pipe;
+    ASSERT_THAT_ERROR(pipe.CreateNew(/*child_process_inherit=*/false).ToError(),
+                      llvm::Succeeded());
+
+    Communication comm("test");
+    comm.SetConnection(std::make_unique<ConnectionFileDescriptor>(
+        pipe.ReleaseReadFileDescriptor(), /*owns_fd=*/true));
+    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(CommunicationTest, SynchronizeWhileClosing) {
   // Set up a communication object reading from a pipe.
   Pipe pipe;
   ASSERT_THAT_ERROR(pipe.CreateNew(/*child_process_inherit=*/false).ToError(),
                     llvm::Succeeded());
 
   Communication comm("test");
   comm.SetConnection(std::make_unique<ConnectionFileDescriptor>(