[lldb] Properly protect the Communication class with reader/writer lock

[augusto2112]

Summary

This patch picks up where https://reviews.llvm.org/D157159 left of, but
allows for concurrent reads/writes, but protects setting up and tearing
down the underlying Connection object.

[bulbazord]

lldb/include/lldb/Core/Communication.h

86–90

This isn't necessarily introduced by your patch, but I don't think this lock does much other than make sure nobody can change m_connection_sp in between you calling GetConnection and you actually receiving the Connection pointer. It still comes with all the pitfalls of handing out a raw pointer (instead of a shared_ptr).

As a follow-up or maybe even precursor, we should probably stop handing out the raw connection pointer...

[JDevlieghere]

lldb/include/lldb/Core/Communication.h

172–175

Why do we need both? Can't we use m_connection_mutex in write mode instead of m_write_mutex?

[augusto2112]

lldb/include/lldb/Core/Communication.h

86–90

I agree. I think we should do it after this patch though.

172–175

If we use m_connection_mutex in write mode instead of m_write_mutex then we can't have concurrent reads and writes.

lldb/include/lldb/Core/Communication.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 <mutex>
+#include <shared_mutex>
 #include <string>
 
 namespace lldb_private {
 class Connection;
 class ConstString;
 class Status;
 
 /// \class Communication Communication.h "lldb/Core/Communication.h" An
public:
   /// Construct the Communication object.
   Communication();
 
   /// Destructor.
   ///
   /// The destructor is virtual since this class gets subclassed.
   virtual ~Communication();
 
-  virtual void Clear();
-
   /// Connect using the current connection by passing \a url to its connect
   /// function. string.
   ///
   /// \param[in] url
   ///     A string that contains all information needed by the
   ///     subclass to connect to another client.
   ///
   /// \return
public:
   ///
   /// \return
   ///     \b True if this object is currently connected, \b false
   ///     otherwise.
   bool IsConnected() const;
 
   bool HasConnection() const;
 
-  lldb_private::Connection *GetConnection() { return m_connection_sp.get(); }
+  lldb_private::Connection *GetConnection() {
+    std::shared_lock guard(m_connection_mutex);
+    return m_connection_sp.get();
+  }
 
   /// Read bytes from the current connection.
   ///
   /// If no read thread is running, this function call the connection's
   /// Connection::Read(...) function to get any available.
   ///
   /// \param[in] dst
   ///     A destination buffer that must be at least \a dst_len bytes
   ///     long.
public:
 
   bool GetCloseOnEOF() const { return m_close_on_eof; }
 
   void SetCloseOnEOF(bool b) { m_close_on_eof = b; }
 
 protected:
   lldb::ConnectionSP m_connection_sp; ///< The connection that is current in use
                                       ///by this communications class.
   std::mutex
       m_write_mutex; ///< Don't let multiple threads write at the same time...
+  mutable std::shared_mutex m_connection_mutex;
   bool m_close_on_eof;
 
+  /// Same as read but with m_connection_mutex unlocked.
+  size_t ReadUnlocked(void *dst, size_t dst_len,
+                      const Timeout<std::micro> &timeout,
+                      lldb::ConnectionStatus &status, Status *error_ptr);
+
   size_t ReadFromConnection(void *dst, size_t dst_len,
                             const Timeout<std::micro> &timeout,
                             lldb::ConnectionStatus &status, Status *error_ptr);
 
 private:
+  /// Same as Disconnect but with with m_connection_mutex unlocked.
+  lldb::ConnectionStatus DisconnectUnlocked(Status *error_ptr = nullptr);
+  /// Same as Write but with both m_write_mutex and m_connection_mutex unlocked.
+  size_t WriteUnlocked(const void *src, size_t src_len,
+                       lldb::ConnectionStatus &status, Status *error_ptr);
   Communication(const Communication &) = delete;
   const Communication &operator=(const Communication &) = delete;
 };
 
 } // namespace lldb_private
 
 #endif // LLDB_CORE_COMMUNICATION_H

lldb/include/lldb/Core/ThreadedCommunication.h

public:
   ///     is called.
   ThreadedCommunication(const char *broadcaster_name);
 
   /// Destructor.
   ///
   /// The destructor is virtual since this class gets subclassed.
   ~ThreadedCommunication() override;
 
-  void Clear() override;
-
   /// Disconnect the communications connection if one is currently connected.
   ///
   /// \return
   ///     \b True if the disconnect succeeded, \b false otherwise. The
   ///     internal error object should be filled in with an
   ///     appropriate value based on the result of this function.
   ///
   /// \see Status& Communication::GetError ();

lldb/source/Core/Communication.cpp

 #include <cerrno>
 #include <cinttypes>
 #include <cstdio>
 
 using namespace lldb;
 using namespace lldb_private;
 
 Communication::Communication()
-    : m_connection_sp(), m_write_mutex(), m_close_on_eof(true) {
-}
+    : m_connection_sp(), m_connection_mutex(), m_close_on_eof(true) {}
 
-Communication::~Communication() {
-  Clear();
-}
-
-void Communication::Clear() {
-  Disconnect(nullptr);
-}
+Communication::~Communication() { Disconnect(nullptr); }
 
 ConnectionStatus Communication::Connect(const char *url, Status *error_ptr) {
-  Clear();
+  std::unique_lock guard(m_connection_mutex);
 
   LLDB_LOG(GetLog(LLDBLog::Communication),
            "{0} Communication::Connect (url = {1})", this, url);
 
-  lldb::ConnectionSP connection_sp(m_connection_sp);
-  if (connection_sp)
-    return connection_sp->Connect(url, error_ptr);
+  DisconnectUnlocked();
+
+  if (m_connection_sp)
+    return m_connection_sp->Connect(url, error_ptr);
   if (error_ptr)
     error_ptr->SetErrorString("Invalid connection.");
   return eConnectionStatusNoConnection;
 }
 
 ConnectionStatus Communication::Disconnect(Status *error_ptr) {
+  std::unique_lock guard(m_connection_mutex);
+  return DisconnectUnlocked(error_ptr);
+}
+
+ConnectionStatus Communication::DisconnectUnlocked(Status *error_ptr) {
   LLDB_LOG(GetLog(LLDBLog::Communication), "{0} Communication::Disconnect ()",
            this);
 
-  lldb::ConnectionSP connection_sp(m_connection_sp);
-  if (connection_sp) {
-    ConnectionStatus status = connection_sp->Disconnect(error_ptr);
-    // We currently don't protect connection_sp with any mutex for multi-
-    // threaded environments. So lets not nuke our connection class without
-    // putting some multi-threaded protections in. We also probably don't want
-    // to pay for the overhead it might cause if every time we access the
-    // connection we have to take a lock.
-    //
-    // This unique pointer will cleanup after itself when this object goes
-    // away, so there is no need to currently have it destroy itself
-    // immediately upon disconnect.
-    // connection_sp.reset();
+  if (m_connection_sp) {
+    ConnectionStatus status = m_connection_sp->Disconnect(error_ptr);
     return status;
   }
   return eConnectionStatusNoConnection;
 }
 
 bool Communication::IsConnected() const {
-  lldb::ConnectionSP connection_sp(m_connection_sp);
-  return (connection_sp ? connection_sp->IsConnected() : false);
+  std::shared_lock guard(m_connection_mutex);
+  return (m_connection_sp ? m_connection_sp->IsConnected() : false);
 }
 
 bool Communication::HasConnection() const {
+  std::shared_lock guard(m_connection_mutex);
   return m_connection_sp.get() != nullptr;
 }
 
 size_t Communication::Read(void *dst, size_t dst_len,
                            const Timeout<std::micro> &timeout,
                            ConnectionStatus &status, Status *error_ptr) {
-  Log *log = GetLog(LLDBLog::Communication);
-  LLDB_LOG(
-      log,
-      "this = {0}, dst = {1}, dst_len = {2}, timeout = {3}, connection = {4}",
-      this, dst, dst_len, timeout, m_connection_sp.get());
-
-  return ReadFromConnection(dst, dst_len, timeout, status, error_ptr);
+  std::shared_lock guard(m_connection_mutex);
+  return ReadUnlocked(dst, dst_len, timeout, status, error_ptr);
 }
 
 size_t Communication::Write(const void *src, size_t src_len,
                             ConnectionStatus &status, Status *error_ptr) {
-  lldb::ConnectionSP connection_sp(m_connection_sp);
-
-  std::lock_guard<std::mutex> guard(m_write_mutex);
-  LLDB_LOG(GetLog(LLDBLog::Communication),
-           "{0} Communication::Write (src = {1}, src_len = {2}"
-           ") connection = {3}",
-           this, src, (uint64_t)src_len, connection_sp.get());
-
-  if (connection_sp)
-    return connection_sp->Write(src, src_len, status, error_ptr);
+  // We need to lock the write mutex so no concurrent writes happen, but also
+  // lock the connection mutex so it's not reset mid write. We need both mutexes
+  // because reads and writes from the connection can happen concurrently.
+  std::shared_lock guard(m_connection_mutex);
+  std::lock_guard<std::mutex> guard_write(m_write_mutex);
+  return WriteUnlocked(src, src_len, status, error_ptr);
+}
 
+size_t Communication::WriteUnlocked(const void *src, size_t src_len,
+                                    ConnectionStatus &status,
+                                    Status *error_ptr) {
+  if (!m_connection_sp) {
     if (error_ptr)
       error_ptr->SetErrorString("Invalid connection.");
     status = eConnectionStatusNoConnection;
     return 0;
   }
 
+  LLDB_LOG(GetLog(LLDBLog::Communication),
+           "{0} Communication::Write (src = {1}, src_len = {2}"
+           ") connection = {3}",
+           this, src, (uint64_t)src_len, m_connection_sp.get());
+
+  return m_connection_sp->Write(src, src_len, status, error_ptr);
+}
+
 size_t Communication::WriteAll(const void *src, size_t src_len,
                                ConnectionStatus &status, Status *error_ptr) {
+  std::shared_lock guard(m_connection_mutex);
+  std::lock_guard<std::mutex> guard_write(m_write_mutex);
   size_t total_written = 0;
   do
-    total_written += Write(static_cast<const char *>(src) + total_written,
-                           src_len - total_written, status, error_ptr);
+    total_written +=
+        WriteUnlocked(static_cast<const char *>(src) + total_written,
   while (status == eConnectionStatusSuccess && total_written < src_len);
   return total_written;
 }
 
-size_t Communication::ReadFromConnection(void *dst, size_t dst_len,
-                                         const Timeout<std::micro> &timeout,
-                                         ConnectionStatus &status,
-                                         Status *error_ptr) {
-  lldb::ConnectionSP connection_sp(m_connection_sp);
-  if (connection_sp)
-    return connection_sp->Read(dst, dst_len, timeout, status, error_ptr);
+size_t Communication::ReadUnlocked(void *dst, size_t dst_len,
+  Log *log = GetLog(LLDBLog::Communication);
+  LLDB_LOG(
+      log,
+      "this = {0}, dst = {1}, dst_len = {2}, timeout = {3}, connection = {4}",
+      this, dst, dst_len, timeout, m_connection_sp.get());
+  if (m_connection_sp)
+    return m_connection_sp->Read(dst, dst_len, timeout, status, error_ptr);
 
   if (error_ptr)
     error_ptr->SetErrorString("Invalid connection.");
   status = eConnectionStatusNoConnection;
   return 0;
 }
 
 void Communication::SetConnection(std::unique_ptr<Connection> connection) {
-  Disconnect(nullptr);
+  std::unique_lock guard(m_connection_mutex);
+  DisconnectUnlocked(nullptr);
   m_connection_sp = std::move(connection);
 }
 
 std::string
 Communication::ConnectionStatusAsString(lldb::ConnectionStatus status) {
   switch (status) {
   case eConnectionStatusSuccess:
     return "success";

lldb/source/Core/ThreadedCommunication.cpp

 }
 
 ThreadedCommunication::~ThreadedCommunication() {
   LLDB_LOG(GetLog(LLDBLog::Object | LLDBLog::Communication),
            "{0} ThreadedCommunication::~ThreadedCommunication (name = {1})",
            this, GetBroadcasterName());
 }
 
-void ThreadedCommunication::Clear() {
-  SetReadThreadBytesReceivedCallback(nullptr, nullptr);
-  StopReadThread(nullptr);
-  Communication::Clear();
-}
 
 ConnectionStatus ThreadedCommunication::Disconnect(Status *error_ptr) {
   assert((!m_read_thread_enabled || m_read_thread_did_exit) &&
          "Disconnecting while the read thread is running is racy!");
   return Communication::Disconnect(error_ptr);
 }
 
 size_t ThreadedCommunication::Read(void *dst, size_t dst_len,
                                    const Timeout<std::micro> &timeout,
                                    ConnectionStatus &status,
                                    Status *error_ptr) {
+  std::shared_lock guard(m_connection_mutex);
   Log *log = GetLog(LLDBLog::Communication);
   LLDB_LOG(
       log,
       "this = {0}, dst = {1}, dst_len = {2}, timeout = {3}, connection = {4}",
       this, dst, dst_len, timeout, m_connection_sp.get());
 
   if (m_read_thread_enabled) {
     // We have a dedicated read thread that is getting data for us
if (event_type & eBroadcastBitReadThreadDidExit) {
         Disconnect(nullptr);
       return 0;
     }
     llvm_unreachable("Got unexpected event type!");
   }
 
   // We aren't using a read thread, just read the data synchronously in this
   // thread.
-  return Communication::Read(dst, dst_len, timeout, status, error_ptr);
+  return Communication::ReadUnlocked(dst, dst_len, timeout, status, error_ptr);
 }
 
 bool ThreadedCommunication::StartReadThread(Status *error_ptr) {
   std::lock_guard<std::mutex> lock(m_read_thread_mutex);
 
   if (error_ptr)
     error_ptr->Clear();
 
lldb::thread_result_t ThreadedCommunication::ReadThread() {
   LLDB_LOG(log, "Communication({0}) thread starting...", this);
 
   uint8_t buf[1024];
 
   Status error;
   ConnectionStatus status = eConnectionStatusSuccess;
   bool done = false;
   bool disconnect = false;
+
+  {
+    std::shared_lock guard(m_connection_mutex);
     while (!done && m_read_thread_enabled) {
-    size_t bytes_read = ReadFromConnection(
-        buf, sizeof(buf), std::chrono::seconds(5), status, &error);
+      size_t bytes_read = ReadUnlocked(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:
         break;
 
       case eConnectionStatusEndOfFile:
         done = true;
         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
           disconnect = GetCloseOnEOF();
           done = true;
         }
         if (error.Fail())
           LLDB_LOG(log, "error: {0}, status = {1}", error,
                    ThreadedCommunication::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.
+        // 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
+      case eConnectionStatusLostConnection: // Lost connection while connected
+                                            // to a valid connection
         done = true;
         [[fallthrough]];
       case eConnectionStatusTimedOut: // Request timed out
         if (error.Fail())
           LLDB_LOG(log, "error: {0}, status = {1}", error,
                    ThreadedCommunication::ConnectionStatusAsString(status));
         break;
       }
     }
+  }
   m_pass_status = status;
   m_pass_error = std::move(error);
   LLDB_LOG(log, "Communication({0}) thread exiting...", this);
 
   // Start shutting down. We need to do this in a very specific order to ensure
   // we don't race with threads wanting to read/synchronize with us.
 
   // First, we signal our intent to exit. This ensures no new thread start
void ThreadedCommunication::SynchronizeWithReadThread() {
   ListenerSP listener_sp(Listener::MakeListener(
       "ThreadedCommunication::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.
+    std::shared_lock guard(m_connection_mutex);
+    if (m_connection_sp)
     m_connection_sp->InterruptRead();
+  }
 
   // Wait for the synchronization event.
   EventSP event_sp;
   listener_sp->GetEvent(event_sp, std::nullopt);
 }
 
 void ThreadedCommunication::SetConnection(
     std::unique_ptr<Connection> connection) {
   StopReadThread(nullptr);
   Communication::SetConnection(std::move(connection));
 }

lldb/source/Plugins/Process/MacOSX-Kernel/ProcessKDP.cpp

if (!keep_stopped && m_comm.IsConnected()) {
         else
           log->PutCString(
               "ProcessKDP::DoDetach() connection channel shutdown failed");
       }
       m_comm.Disconnect();
     }
   }
   StopAsyncThread();
-  m_comm.Clear();
 
   SetPrivateState(eStateDetached);
   ResumePrivateStateThread();
 
   // KillDebugserverProcess ();
   return error;
 }