[lldb] [Core] Split read thread support into ThreadedCommunication

[mgorny]

Summary

Split the read thread support from Communication into a dedicated
ThreadedCommunication subclass. The read thread support is used only
by a subset of Communication consumers, and it adds a lot of complexity
to the base class. Furthermore, having a dedicated subclass makes it
clear whether a particular consumer needs to account for the possibility
of read thread being running or not.

The modules currently calling StartReadThread() are updated to use
ThreadedCommunication. The remaining modules use the simplified
Communication class.

SBCommunication is changed to use ThreadedCommunication in order
to avoid changing the public API.

CommunicationKDP is updated in order to (hopefully) compile with
the new code. However, I do not have a Darwin box to test it, so I've
limited the changes to the bare minimum.

GDBRemoteCommunication is updated to become a Broadcaster directly.
Since it does not inherit from ThreadedCommunication, its event
support no longer collides with the one used for read thread and can
be implemented cleanly. The support for
eBroadcastBitReadThreadDidExit is removed from the code -- since
the read thread was not used, this event was never reported.

Sponsored by: The FreeBSD Foundation

[labath]

I like this a lot.

lldb/source/Core/CMakeLists.txt

57

I think you're missing this file.

[mgorny]

Add the missing file.

[mgorny]

lldb/source/Core/CMakeLists.txt

57

Indeed I were. Thanks for noticing this before I did git clean -dfx ;-).

[kastiglione]

CommunicationKDP is updated in order to (hopefully) compile with the new code. However, I do not have a Darwin box to test it, so I've limited the changes to the bare minimum.

Compilation fails:

/Users/buildslave/jenkins/workspace/lldb-cmake/llvm-project/lldb/source/Plugins/Process/MacOSX-Kernel/CommunicationKDP.cpp:232:47: error: unknown type name 'm_bytes_mutex'
  std::lock_guard<std::recursive_mutex> guard(m_bytes_mutex);

Should we revert?

[kastiglione]

I created D133365, which solves the compilation issue but I don't know whether it's semantically correct.

lldb/include/lldb/API/SBCommunication.h

public:
 
   bool SetReadThreadBytesReceivedCallback(ReadThreadBytesReceived callback,
                                           void *callback_baton);
 
 private:
   SBCommunication(const SBCommunication &) = delete;
   const SBCommunication &operator=(const SBCommunication &) = delete;
 
-  lldb_private::Communication *m_opaque = nullptr;
+  lldb_private::ThreadedCommunication *m_opaque = nullptr;
   bool m_opaque_owned = false;
 };
 
 } // namespace lldb
 
 #endif // LLDB_API_SBCOMMUNICATION_H

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/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>
 #include <mutex>
-#include <ratio>
 #include <string>
 
-#include <cstddef>
-#include <cstdint>
-
 namespace lldb_private {
 class Connection;
 class ConstString;
 class Status;
 
 /// \class Communication Communication.h "lldb/Core/Communication.h" An
 /// abstract communications class.
 ///
 /// Communication is an class that handles data communication between two data
 /// sources. It uses a Connection class to do the real communication. This
 /// approach has a couple of advantages: it allows a single instance of this
 /// class to be used even though its connection can change. Connections could
 /// negotiate for different connections based on abilities like starting with
-/// Bluetooth and negotiating up to WiFi if available. It also allows this
-/// class to be subclassed by any interfaces that don't want to give bytes but
-/// want to validate and give out packets. This can be done by overriding:
-///
-/// AppendBytesToCache (const uint8_t *src, size_t src_len, bool broadcast);
-///
-/// Communication inherits from Broadcaster which means it can be used in
-/// conjunction with Listener to wait for multiple broadcaster objects and
-/// multiple events from each of those objects. Communication defines a set of
-/// pre-defined event bits (see enumerations definitions that start with
-/// "eBroadcastBit" below).
-///
-/// There are two modes in which communications can occur:
-///     \li single-threaded
-///     \li multi-threaded
-///
-/// In single-threaded mode, all reads and writes happen synchronously on the
-/// calling thread.
-///
-/// In multi-threaded mode, a read thread is spawned that continually reads
-/// data and caches any received bytes. To start the read thread clients call:
-///
-///     bool Communication::StartReadThread (Status *);
-///
-/// If true is returned a read thread has been spawned that will continually
-/// execute a call to the pure virtual DoRead function:
-///
-///     size_t Communication::ReadFromConnection (void *, size_t, uint32_t);
-///
-/// When bytes are received the data gets cached in \a m_bytes and this class
-/// will broadcast a \b eBroadcastBitReadThreadGotBytes event. Clients that
-/// want packet based communication should override AppendBytesToCache. The
-/// subclasses can choose to call the built in AppendBytesToCache with the \a
-/// broadcast parameter set to false. This will cause the \b
-/// eBroadcastBitReadThreadGotBytes event not get broadcast, and then the
-/// subclass can post a \b eBroadcastBitPacketAvailable event when a full
-/// packet of data has been received.
-///
-/// If the connection is disconnected a \b eBroadcastBitDisconnected event
-/// gets broadcast. If the read thread exits a \b
-/// eBroadcastBitReadThreadDidExit event will be broadcast. Clients can also
-/// post a \b eBroadcastBitReadThreadShouldExit event to this object which
-/// will cause the read thread to exit.
-class Communication : public Broadcaster {
+/// Bluetooth and negotiating up to WiFi if available.
+///
+/// When using this class, all reads and writes happen synchronously on the
+/// calling thread. There is also a ThreadedCommunication class that supports
+/// multi-threaded mode.
+class Communication {
 public:
-  FLAGS_ANONYMOUS_ENUM(){
-      eBroadcastBitDisconnected =
-          (1u << 0), ///< Sent when the communications connection is lost.
-      eBroadcastBitReadThreadGotBytes =
-          (1u << 1), ///< Sent by the read thread when bytes become available.
-      eBroadcastBitReadThreadDidExit =
-          (1u
-           << 2), ///< Sent by the read thread when it exits to inform clients.
-      eBroadcastBitReadThreadShouldExit =
-          (1u << 3), ///< Sent by clients that need to cancel the read thread.
-      eBroadcastBitPacketAvailable =
-          (1u << 4), ///< Sent when data received makes a complete packet.
-      eBroadcastBitNoMorePendingInput = (1u << 5), ///< Sent by the read thread
-                                                   ///to indicate all pending
-                                                   ///input has been processed.
-      kLoUserBroadcastBit =
-          (1u << 16), ///< Subclasses can used bits 31:16 for any needed events.
-      kHiUserBroadcastBit = (1u << 31),
-      eAllEventBits = 0xffffffff};
-
-  typedef void (*ReadThreadBytesReceived)(void *baton, const void *src,
-                                          size_t src_len);
-
-  /// Construct the Communication object with the specified name for the
-  /// Broadcaster that this object inherits from.
-  ///
-  /// \param[in] broadcaster_name
-  ///     The name of the broadcaster object.  This name should be as
-  ///     complete as possible to uniquely identify this object. The
-  ///     broadcaster name can be updated after the connect function
-  ///     is called.
-  Communication(const char *broadcaster_name);
+  /// Construct the Communication object.
+  Communication();
 
   /// Destructor.
   ///
   /// The destructor is virtual since this class gets subclassed.
-  ~Communication() override;
+  virtual ~Communication();
 
-  void Clear();
+  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.
   ///
public:
   ///
   /// \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 ();
   /// \see bool Connection::Disconnect ();
-  lldb::ConnectionStatus Disconnect(Status *error_ptr = nullptr);
+  virtual lldb::ConnectionStatus Disconnect(Status *error_ptr = nullptr);
 
   /// Check if the connection is valid.
   ///
   /// \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(); }
 
   /// 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.
   ///
-  /// If a read thread has been started, this function will check for any
-  /// cached bytes that have already been read and return any currently
-  /// available bytes. If no bytes are cached, it will wait for the bytes to
-  /// become available by listening for the \a eBroadcastBitReadThreadGotBytes
-  /// event. If this function consumes all of the bytes in the cache, it will
-  /// reset the \a eBroadcastBitReadThreadGotBytes event bit.
-  ///
   /// \param[in] dst
   ///     A destination buffer that must be at least \a dst_len bytes
   ///     long.
   ///
   /// \param[in] dst_len
   ///     The number of bytes to attempt to read, and also the max
   ///     number of bytes that can be placed into \a dst.
   ///
   /// \param[in] timeout
   ///     A timeout value or llvm::None for no timeout.
   ///
   /// \return
   ///     The number of bytes actually read.
   ///
   /// \see size_t Connection::Read (void *, size_t);
-  size_t Read(void *dst, size_t dst_len, const Timeout<std::micro> &timeout,
+  virtual size_t Read(void *dst, size_t dst_len,
+                      const Timeout<std::micro> &timeout,
                       lldb::ConnectionStatus &status, Status *error_ptr);
 
   /// The actual write function that attempts to write to the communications
   /// protocol.
   ///
   /// Subclasses must override this function.
   ///
   /// \param[in] src
   ///     A source buffer that must be at least \a src_len bytes
public:
   /// connection without any interruption to the client. It also allows the
   /// Communication class to be subclassed for packet based communication.
   ///
   /// \param[in] connection
   ///     A connection that this class will own and destroy.
   ///
   /// \see
   ///     class Connection
-  void SetConnection(std::unique_ptr<Connection> connection);
-
-  /// Starts a read thread whose sole purpose it to read bytes from the
-  /// current connection. This function will call connection's read function:
-  ///
-  /// size_t Connection::Read (void *, size_t);
-  ///
-  /// When bytes are read and cached, this function will call:
-  ///
-  /// Communication::AppendBytesToCache (const uint8_t * bytes, size_t len,
-  /// bool
-  /// broadcast);
-  ///
-  /// Subclasses should override this function if they wish to override the
-  /// default action of caching the bytes and broadcasting a \b
-  /// eBroadcastBitReadThreadGotBytes event.
-  ///
-  /// \return
-  ///     \b True if the read thread was successfully started, \b
-  ///     false otherwise.
-  ///
-  /// \see size_t Connection::Read (void *, size_t);
-  /// \see void Communication::AppendBytesToCache (const uint8_t * bytes,
-  ///                                              size_t len, bool broadcast);
-  virtual bool StartReadThread(Status *error_ptr = nullptr);
-
-  /// Stops the read thread by cancelling it.
-  ///
-  /// \return
-  ///     \b True if the read thread was successfully canceled, \b
-  ///     false otherwise.
-  virtual bool StopReadThread(Status *error_ptr = nullptr);
-
-  virtual bool JoinReadThread(Status *error_ptr = nullptr);
-  /// Checks if there is a currently running read thread.
-  ///
-  /// \return
-  ///     \b True if the read thread is running, \b false otherwise.
-  bool ReadThreadIsRunning();
-
-  /// The read thread function. This function will call the "DoRead"
-  /// function continuously and wait for data to become available. When data
-  /// is received it will append the available data to the internal cache and
-  /// broadcast a \b eBroadcastBitReadThreadGotBytes event.
-  ///
-  /// \param[in] comm_ptr
-  ///     A pointer to an instance of this class.
-  ///
-  /// \return
-  ///     \b NULL.
-  ///
-  /// \see void Communication::ReadThreadGotBytes (const uint8_t *, size_t);
-  lldb::thread_result_t ReadThread();
-
-  void SetReadThreadBytesReceivedCallback(ReadThreadBytesReceived callback,
-                                          void *callback_baton);
-
-  /// Wait for the read thread to process all outstanding data.
-  ///
-  /// After this function returns, the read thread has processed all data that
-  /// has been waiting in the Connection queue.
-  ///
-  void SynchronizeWithReadThread();
+  virtual void SetConnection(std::unique_ptr<Connection> connection);
 
   static std::string ConnectionStatusAsString(lldb::ConnectionStatus status);
 
   bool GetCloseOnEOF() const { return m_close_on_eof; }
 
   void SetCloseOnEOF(bool b) { m_close_on_eof = b; }
 
-  static ConstString &GetStaticBroadcasterClass();
-
-  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::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,
                             const Timeout<std::micro> &timeout,
                             lldb::ConnectionStatus &status, Status *error_ptr);
 
-  /// Append new bytes that get read from the read thread into the internal
-  /// object byte cache. This will cause a \b eBroadcastBitReadThreadGotBytes
-  /// event to be broadcast if \a broadcast is true.
-  ///
-  /// Subclasses can override this function in order to inspect the received
-  /// data and check if a packet is available.
-  ///
-  /// Subclasses can also still call this function from the overridden method
-  /// to allow the caching to correctly happen and suppress the broadcasting
-  /// of the \a eBroadcastBitReadThreadGotBytes event by setting \a broadcast
-  /// to false.
-  ///
-  /// \param[in] src
-  ///     A source buffer that must be at least \a src_len bytes
-  ///     long.
-  ///
-  /// \param[in] src_len
-  ///     The number of bytes to append to the cache.
-  virtual void AppendBytesToCache(const uint8_t *src, size_t src_len,
-                                  bool broadcast,
-                                  lldb::ConnectionStatus status);
-
-  /// Get any available bytes from our data cache. If this call empties the
-  /// data cache, the \b eBroadcastBitReadThreadGotBytes event will be reset
-  /// to signify no more bytes are available.
-  ///
-  /// \param[in] dst
-  ///     A destination buffer that must be at least \a dst_len bytes
-  ///     long.
-  ///
-  /// \param[in] dst_len
-  ///     The number of bytes to attempt to read from the cache,
-  ///     and also the max number of bytes that can be placed into
-  ///     \a dst.
-  ///
-  /// \return
-  ///     The number of bytes extracted from the data cache.
-  size_t GetCachedBytes(void *dst, size_t dst_len);
-
 private:
   Communication(const Communication &) = delete;
   const Communication &operator=(const Communication &) = delete;
 };
 
 } // namespace lldb_private
 
 #endif // LLDB_CORE_COMMUNICATION_H

lldb/include/lldb/Core/ThreadedCommunication.h

-//===-- Communication.h -----------------------------------------*- C++ -*-===//
+//===-- ThreadedCommunication.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
+#ifndef LLDB_CORE_THREADEDCOMMUNICATION_H
+#define LLDB_CORE_THREADEDCOMMUNICATION_H
 
+#include "lldb/Core/Communication.h"
 #include "lldb/Host/HostThread.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>
 #include <mutex>
-#include <ratio>
 #include <string>
 
 #include <cstddef>
 #include <cstdint>
 
 namespace lldb_private {
-class Connection;
-class ConstString;
-class Status;
-
-/// \class Communication Communication.h "lldb/Core/Communication.h" An
-/// abstract communications class.
-///
-/// Communication is an class that handles data communication between two data
-/// sources. It uses a Connection class to do the real communication. This
-/// approach has a couple of advantages: it allows a single instance of this
-/// class to be used even though its connection can change. Connections could
-/// negotiate for different connections based on abilities like starting with
-/// Bluetooth and negotiating up to WiFi if available. It also allows this
-/// class to be subclassed by any interfaces that don't want to give bytes but
-/// want to validate and give out packets. This can be done by overriding:
-///
-/// AppendBytesToCache (const uint8_t *src, size_t src_len, bool broadcast);
-///
-/// Communication inherits from Broadcaster which means it can be used in
-/// conjunction with Listener to wait for multiple broadcaster objects and
-/// multiple events from each of those objects. Communication defines a set of
-/// pre-defined event bits (see enumerations definitions that start with
-/// "eBroadcastBit" below).
-///
-/// There are two modes in which communications can occur:
-///     \li single-threaded
-///     \li multi-threaded
-///
-/// In single-threaded mode, all reads and writes happen synchronously on the
-/// calling thread.
-///
-/// In multi-threaded mode, a read thread is spawned that continually reads
-/// data and caches any received bytes. To start the read thread clients call:
+
+/// \class ThreadedCommunication ThreadedCommunication.h
+/// "lldb/Core/ThreadedCommunication.h" Variation of Communication that
+/// supports threaded reads.
+///
+/// ThreadedCommunication enhances the base Communication class with support
+/// for multi-threaded mode.  In this mode, a read thread is spawned that
+/// continually reads data and caches any received bytes. To start the read
+/// thread clients call:
 ///
-///     bool Communication::StartReadThread (Status *);
+///     bool ThreadedCommunication::StartReadThread (Status *);
 ///
 /// If true is returned a read thread has been spawned that will continually
 /// execute a call to the pure virtual DoRead function:
 ///
 ///     size_t Communication::ReadFromConnection (void *, size_t, uint32_t);
 ///
 /// When bytes are received the data gets cached in \a m_bytes and this class
 /// will broadcast a \b eBroadcastBitReadThreadGotBytes event. Clients that
 /// want packet based communication should override AppendBytesToCache. The
 /// subclasses can choose to call the built in AppendBytesToCache with the \a
 /// broadcast parameter set to false. This will cause the \b
 /// eBroadcastBitReadThreadGotBytes event not get broadcast, and then the
 /// subclass can post a \b eBroadcastBitPacketAvailable event when a full
 /// packet of data has been received.
 ///
 /// If the connection is disconnected a \b eBroadcastBitDisconnected event
 /// gets broadcast. If the read thread exits a \b
 /// eBroadcastBitReadThreadDidExit event will be broadcast. Clients can also
 /// post a \b eBroadcastBitReadThreadShouldExit event to this object which
 /// will cause the read thread to exit.
-class Communication : public Broadcaster {
+///
+/// ThreadedCommunication inherits from Broadcaster which means it can be used
+/// in conjunction with Listener to wait for multiple broadcaster objects and
+/// multiple events from each of those objects. ThreadedCommunication defines a
+/// set of pre-defined event bits (see enumerations definitions that start with
+/// "eBroadcastBit" below).
+class ThreadedCommunication : public Communication, public Broadcaster {
+  using Communication::Communication;
+
 public:
   FLAGS_ANONYMOUS_ENUM(){
       eBroadcastBitDisconnected =
           (1u << 0), ///< Sent when the communications connection is lost.
       eBroadcastBitReadThreadGotBytes =
           (1u << 1), ///< Sent by the read thread when bytes become available.
       eBroadcastBitReadThreadDidExit =
           (1u
            << 2), ///< Sent by the read thread when it exits to inform clients.
       eBroadcastBitReadThreadShouldExit =
           (1u << 3), ///< Sent by clients that need to cancel the read thread.
       eBroadcastBitPacketAvailable =
           (1u << 4), ///< Sent when data received makes a complete packet.
       eBroadcastBitNoMorePendingInput = (1u << 5), ///< Sent by the read thread
-                                                   ///to indicate all pending
-                                                   ///input has been processed.
-      kLoUserBroadcastBit =
-          (1u << 16), ///< Subclasses can used bits 31:16 for any needed events.
-      kHiUserBroadcastBit = (1u << 31),
-      eAllEventBits = 0xffffffff};
+                                                   /// to indicate all pending
+                                                   /// input has been processed.
+  };
 
   typedef void (*ReadThreadBytesReceived)(void *baton, const void *src,
                                           size_t src_len);
 
-  /// Construct the Communication object with the specified name for the
+  /// Construct the ThreadedCommunication object with the specified name for the
   /// Broadcaster that this object inherits from.
   ///
   /// \param[in] broadcaster_name
   ///     The name of the broadcaster object.  This name should be as
   ///     complete as possible to uniquely identify this object. The
   ///     broadcaster name can be updated after the connect function
   ///     is called.
-  Communication(const char *broadcaster_name);
+  ThreadedCommunication(const char *broadcaster_name);
 
   /// Destructor.
   ///
   /// The destructor is virtual since this class gets subclassed.
-  ~Communication() override;
+  ~ThreadedCommunication() override;
 
-  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
-  ///     \b True if the connect 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 ();
-  /// \see bool Connection::Connect (const char *url);
-  lldb::ConnectionStatus Connect(const char *url, Status *error_ptr);
+  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 ();
   /// \see bool Connection::Disconnect ();
-  lldb::ConnectionStatus Disconnect(Status *error_ptr = nullptr);
-
-  /// Check if the connection is valid.
-  ///
-  /// \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::ConnectionStatus Disconnect(Status *error_ptr = nullptr) override;
 
   /// 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.
   ///
   /// If a read thread has been started, this function will check for any
   /// cached bytes that have already been read and return any currently
public:
   /// \param[in] timeout
   ///     A timeout value or llvm::None for no timeout.
   ///
   /// \return
   ///     The number of bytes actually read.
   ///
   /// \see size_t Connection::Read (void *, size_t);
   size_t Read(void *dst, size_t dst_len, const Timeout<std::micro> &timeout,
-              lldb::ConnectionStatus &status, Status *error_ptr);
-
-  /// The actual write function that attempts to write to the communications
-  /// protocol.
-  ///
-  /// Subclasses must override this function.
-  ///
-  /// \param[in] src
-  ///     A source buffer that must be at least \a src_len bytes
-  ///     long.
-  ///
-  /// \param[in] src_len
-  ///     The number of bytes to attempt to write, and also the
-  ///     number of bytes are currently available in \a src.
-  ///
-  /// \return
-  ///     The number of bytes actually Written.
-  size_t Write(const void *src, size_t src_len, lldb::ConnectionStatus &status,
-               Status *error_ptr);
-
-  /// Repeatedly attempt writing until either \a src_len bytes are written
-  /// or a permanent failure occurs.
-  ///
-  /// \param[in] src
-  ///     A source buffer that must be at least \a src_len bytes
-  ///     long.
-  ///
-  /// \param[in] src_len
-  ///     The number of bytes to attempt to write, and also the
-  ///     number of bytes are currently available in \a src.
-  ///
-  /// \return
-  ///     The number of bytes actually Written.
-  size_t WriteAll(const void *src, size_t src_len,
-                  lldb::ConnectionStatus &status, Status *error_ptr);
+              lldb::ConnectionStatus &status, Status *error_ptr) override;
 
   /// Sets the connection that it to be used by this class.
   ///
   /// By making a communication class that uses different connections it
   /// allows a single communication interface to negotiate and change its
   /// connection without any interruption to the client. It also allows the
   /// Communication class to be subclassed for packet based communication.
   ///
   /// \param[in] connection
   ///     A connection that this class will own and destroy.
   ///
   /// \see
   ///     class Connection
-  void SetConnection(std::unique_ptr<Connection> connection);
+  void SetConnection(std::unique_ptr<Connection> connection) override;
 
   /// Starts a read thread whose sole purpose it to read bytes from the
   /// current connection. This function will call connection's read function:
   ///
   /// size_t Connection::Read (void *, size_t);
   ///
   /// When bytes are read and cached, this function will call:
   ///
public:
 
   /// Wait for the read thread to process all outstanding data.
   ///
   /// After this function returns, the read thread has processed all data that
   /// has been waiting in the Connection queue.
   ///
   void SynchronizeWithReadThread();
 
-  static std::string ConnectionStatusAsString(lldb::ConnectionStatus status);
-
-  bool GetCloseOnEOF() const { return m_close_on_eof; }
-
-  void SetCloseOnEOF(bool b) { m_close_on_eof = b; }
-
   static ConstString &GetStaticBroadcasterClass();
 
   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.
+                            /// 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.
+  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...
+                                        /// from read thread.
+  Status m_pass_error;                  ///< Error passthrough from read thread.
   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,
-                            const Timeout<std::micro> &timeout,
-                            lldb::ConnectionStatus &status, Status *error_ptr);
 
   /// Append new bytes that get read from the read thread into the internal
   /// object byte cache. This will cause a \b eBroadcastBitReadThreadGotBytes
   /// event to be broadcast if \a broadcast is true.
   ///
   /// Subclasses can override this function in order to inspect the received
   /// data and check if a packet is available.
   ///
protected:
   ///     and also the max number of bytes that can be placed into
   ///     \a dst.
   ///
   /// \return
   ///     The number of bytes extracted from the data cache.
   size_t GetCachedBytes(void *dst, size_t dst_len);
 
 private:
-  Communication(const Communication &) = delete;
-  const Communication &operator=(const Communication &) = delete;
+  ThreadedCommunication(const ThreadedCommunication &) = delete;
+  const ThreadedCommunication &
+  operator=(const ThreadedCommunication &) = delete;
 };
 
 } // namespace lldb_private
 
-#endif // LLDB_CORE_COMMUNICATION_H
+#endif // LLDB_CORE_THREADEDCOMMUNICATION_H

lldb/include/lldb/Interpreter/ScriptInterpreter.h

 //===-- ScriptInterpreter.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_INTERPRETER_SCRIPTINTERPRETER_H
 #define LLDB_INTERPRETER_SCRIPTINTERPRETER_H
 
 #include "lldb/API/SBData.h"
 #include "lldb/API/SBError.h"
 #include "lldb/API/SBMemoryRegionInfo.h"
 #include "lldb/Breakpoint/BreakpointOptions.h"
-#include "lldb/Core/Communication.h"
 #include "lldb/Core/PluginInterface.h"
 #include "lldb/Core/SearchFilter.h"
 #include "lldb/Core/StreamFile.h"
+#include "lldb/Core/ThreadedCommunication.h"
 #include "lldb/Host/PseudoTerminal.h"
 #include "lldb/Interpreter/ScriptedProcessInterface.h"
 #include "lldb/Utility/Broadcaster.h"
 #include "lldb/Utility/Status.h"
 #include "lldb/Utility/StructuredData.h"
 #include "lldb/lldb-private.h"
 
 namespace lldb_private {

 private:
   ScriptInterpreterIORedirect(std::unique_ptr<File> input,
                               std::unique_ptr<File> output);
   ScriptInterpreterIORedirect(Debugger &debugger, CommandReturnObject *result);
 
   lldb::FileSP m_input_file_sp;
   lldb::StreamFileSP m_output_file_sp;
   lldb::StreamFileSP m_error_file_sp;
-  Communication m_communication;
+  ThreadedCommunication m_communication;
   bool m_disconnect;
 };
 
 class ScriptInterpreter : public PluginInterface {
 public:
   enum ScriptReturnType {
     eScriptReturnTypeCharPtr,
     eScriptReturnTypeBool,

lldb/include/lldb/Target/Process.h

 #include <list>
 #include <memory>
 #include <mutex>
 #include <string>
 #include <unordered_set>
 #include <vector>
 
 #include "lldb/Breakpoint/BreakpointSiteList.h"
-#include "lldb/Core/Communication.h"
 #include "lldb/Core/LoadedModuleInfoList.h"
 #include "lldb/Core/PluginInterface.h"
 #include "lldb/Core/ThreadSafeValue.h"
+#include "lldb/Core/ThreadedCommunication.h"
 #include "lldb/Core/UserSettingsController.h"
 #include "lldb/Host/HostThread.h"
 #include "lldb/Host/ProcessLaunchInfo.h"
 #include "lldb/Host/ProcessRunLock.h"
 #include "lldb/Symbol/ObjectFile.h"
 #include "lldb/Target/ExecutionContextScope.h"
 #include "lldb/Target/InstrumentationRuntime.h"
 #include "lldb/Target/Memory.h"
lldb::DynamicCheckerFunctionsUP m_dynamic_checkers_up; ///< The functions used
                                                          /// data that
                                                          /// expressions use.
   lldb::OperatingSystemUP m_os_up;
   lldb::SystemRuntimeUP m_system_runtime_up;
   lldb::UnixSignalsSP
       m_unix_signals_sp; /// This is the current signal set for this process.
   lldb::ABISP m_abi_sp;
   lldb::IOHandlerSP m_process_input_reader;
-  Communication m_stdio_communication;
+  ThreadedCommunication m_stdio_communication;
   std::recursive_mutex m_stdio_communication_mutex;
   bool m_stdin_forward; /// Remember if stdin must be forwarded to remote debug
                         /// server
   std::string m_stdout_data;
   std::string m_stderr_data;
   std::recursive_mutex m_profile_data_comm_mutex;
   std::vector<std::string> m_profile_data;
   Predicate<uint32_t> m_iohandler_sync;

lldb/include/lldb/lldb-forward.h

 class ThreadPlanStepInstruction;
 class ThreadPlanStepOut;
 class ThreadPlanStepOverBreakpoint;
 class ThreadPlanStepRange;
 class ThreadPlanStepThrough;
 class ThreadPlanTracer;
 class ThreadSpec;
 class ThreadPostMortemTrace;
+class ThreadedCommunication;
 class Trace;
 class TraceCursor;
 class TraceExporter;
 class Type;
 class TypeAndOrName;
 class TypeCategoryImpl;
 class TypeCategoryMap;
 class TypeEnumMemberImpl;

lldb/source/API/SBCommunication.cpp

 //===-- SBCommunication.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/API/SBCommunication.h"
 #include "lldb/API/SBBroadcaster.h"
-#include "lldb/Core/Communication.h"
+#include "lldb/Core/ThreadedCommunication.h"
 #include "lldb/Host/ConnectionFileDescriptor.h"
 #include "lldb/Host/Host.h"
 #include "lldb/Utility/Instrumentation.h"
 
 using namespace lldb;
 using namespace lldb_private;
 
 SBCommunication::SBCommunication() { LLDB_INSTRUMENT_VA(this); }
 
 SBCommunication::SBCommunication(const char *broadcaster_name)
-    : m_opaque(new Communication(broadcaster_name)), m_opaque_owned(true) {
+    : m_opaque(new ThreadedCommunication(broadcaster_name)),
+      m_opaque_owned(true) {
   LLDB_INSTRUMENT_VA(this, broadcaster_name);
 }
 
 SBCommunication::~SBCommunication() {
   if (m_opaque && m_opaque_owned)
     delete m_opaque;
   m_opaque = nullptr;
   m_opaque_owned = false;
SBBroadcaster SBCommunication::GetBroadcaster() {
 
   SBBroadcaster broadcaster(m_opaque, false);
   return broadcaster;
 }
 
 const char *SBCommunication::GetBroadcasterClass() {
   LLDB_INSTRUMENT();
 
-  return Communication::GetStaticBroadcasterClass().AsCString();
+  return ThreadedCommunication::GetStaticBroadcasterClass().AsCString();
 }

lldb/source/Core/CMakeLists.txt

add_lldb_library(lldbCore
   Progress.cpp
   RichManglingContext.cpp
   SearchFilter.cpp
   Section.cpp
   SourceLocationSpec.cpp
   SourceManager.cpp
   StreamAsynchronousIO.cpp
   StreamFile.cpp
+  ThreadedCommunication.cpp
   UserSettingsController.cpp
   Value.cpp
   ValueObject.cpp
   ValueObjectCast.cpp
   ValueObjectChild.cpp
   ValueObjectConstResult.cpp
   ValueObjectConstResultCast.cpp
   ValueObjectConstResultChild.cpp

lldb/source/Core/Communication.cpp

 //===-- Communication.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/HostThread.h"
-#include "lldb/Host/ThreadLauncher.h"
 #include "lldb/Utility/Connection.h"
-#include "lldb/Utility/ConstString.h"
-#include "lldb/Utility/Event.h"
 #include "lldb/Utility/LLDBLog.h"
-#include "lldb/Utility/Listener.h"
 #include "lldb/Utility/Log.h"
 #include "lldb/Utility/Status.h"
 
-#include "llvm/ADT/None.h"
-#include "llvm/ADT/Optional.h"
 #include "llvm/Support/Compiler.h"
 
 #include <algorithm>
-#include <chrono>
 #include <cstring>
 #include <memory>
 
 #include <cerrno>
 #include <cinttypes>
 #include <cstdio>
 
 using namespace lldb;
 using namespace lldb_private;
 
-ConstString &Communication::GetStaticBroadcasterClass() {
-  static ConstString class_name("lldb.communication");
-  return class_name;
-}
-
-Communication::Communication(const char *name)
-    : Broadcaster(nullptr, name), m_connection_sp(),
-      m_read_thread_enabled(false), m_read_thread_did_exit(false), m_bytes(),
-      m_bytes_mutex(), m_write_mutex(), m_synchronize_mutex(),
-      m_callback(nullptr), m_callback_baton(nullptr), m_close_on_eof(true)
-
-{
-
-  LLDB_LOG(GetLog(LLDBLog::Object | LLDBLog::Communication),
-           "{0} Communication::Communication (name = {1})", this, name);
-
-  SetEventName(eBroadcastBitDisconnected, "disconnected");
-  SetEventName(eBroadcastBitReadThreadGotBytes, "got bytes");
-  SetEventName(eBroadcastBitReadThreadDidExit, "read thread did exit");
-  SetEventName(eBroadcastBitReadThreadShouldExit, "read thread should exit");
-  SetEventName(eBroadcastBitPacketAvailable, "packet available");
-  SetEventName(eBroadcastBitNoMorePendingInput, "no more pending input");
-
-  CheckInWithManager();
+Communication::Communication()
+    : m_connection_sp(), m_write_mutex(), m_close_on_eof(true) {
 }
 
 Communication::~Communication() {
-  LLDB_LOG(GetLog(LLDBLog::Object | LLDBLog::Communication),
-           "{0} Communication::~Communication (name = {1})", this,
-           GetBroadcasterName().AsCString());
   Clear();
 }
 
 void Communication::Clear() {
-  SetReadThreadBytesReceivedCallback(nullptr, nullptr);
-  StopReadThread(nullptr);
   Disconnect(nullptr);
 }
 
 ConnectionStatus Communication::Connect(const char *url, Status *error_ptr) {
   Clear();
 
   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);
   if (error_ptr)
     error_ptr->SetErrorString("Invalid connection.");
   return eConnectionStatusNoConnection;
 }
 
 ConnectionStatus Communication::Disconnect(Status *error_ptr) {
   LLDB_LOG(GetLog(LLDBLog::Communication), "{0} Communication::Disconnect ()",
            this);
 
-  assert((!m_read_thread_enabled || m_read_thread_did_exit) &&
-         "Disconnecting while the read thread is running is racy!");
   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.
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());
 
-  if (m_read_thread_enabled) {
-    // We have a dedicated read thread that is getting data for us
-    size_t cached_bytes = GetCachedBytes(dst, dst_len);
-    if (cached_bytes > 0) {
-      status = eConnectionStatusSuccess;
-      return cached_bytes;
-    }
-    if (timeout && timeout->count() == 0) {
-      if (error_ptr)
-        error_ptr->SetErrorString("Timed out.");
-      status = eConnectionStatusTimedOut;
-      return 0;
-    }
-
-    if (!m_connection_sp) {
-      if (error_ptr)
-        error_ptr->SetErrorString("Invalid connection.");
-      status = eConnectionStatusNoConnection;
-      return 0;
-    }
-
-    ListenerSP listener_sp(Listener::MakeListener("Communication::Read"));
-    listener_sp->StartListeningForEvents(
-        this, eBroadcastBitReadThreadGotBytes | eBroadcastBitReadThreadDidExit);
-    EventSP event_sp;
-    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 an error.
-        status = m_pass_status;
-        if (error_ptr)
-          *error_ptr = std::move(m_pass_error);
-
-        if (GetCloseOnEOF())
-          Disconnect(nullptr);
-        return 0;
-      }
-    }
-
-    if (error_ptr)
-      error_ptr->SetErrorString("Timed out.");
-    status = eConnectionStatusTimedOut;
-    return 0;
-  }
-
-  // We aren't using a read thread, just read the data synchronously in this
-  // thread.
   return ReadFromConnection(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);
size_t Communication::WriteAll(const void *src, size_t src_len,
   size_t total_written = 0;
   do
     total_written += Write(static_cast<const char *>(src) + total_written,
                            src_len - total_written, status, error_ptr);
   while (status == eConnectionStatusSuccess && total_written < src_len);
   return total_written;
 }
 
-bool Communication::StartReadThread(Status *error_ptr) {
-  if (error_ptr)
-    error_ptr->Clear();
-
-  if (m_read_thread.IsJoinable())
-    return true;
-
-  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;
-  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 {
-      LLDB_LOG(GetLog(LLDBLog::Host), "failed to launch host thread: {}",
-               llvm::toString(maybe_thread.takeError()));
-    }
-  }
-
-  if (!m_read_thread.IsJoinable())
-    m_read_thread_enabled = false;
-
-  return m_read_thread_enabled;
-}
-
-bool Communication::StopReadThread(Status *error_ptr) {
-  if (!m_read_thread.IsJoinable())
-    return true;
-
-  LLDB_LOG(GetLog(LLDBLog::Communication),
-           "{0} Communication::StopReadThread ()", this);
-
-  m_read_thread_enabled = false;
-
-  BroadcastEvent(eBroadcastBitReadThreadShouldExit, nullptr);
-
-  // error = m_read_thread.Cancel();
-
-  Status error = m_read_thread.Join(nullptr);
-  return error.Success();
-}
-
-bool Communication::JoinReadThread(Status *error_ptr) {
-  if (!m_read_thread.IsJoinable())
-    return true;
-
-  Status error = m_read_thread.Join(nullptr);
-  return error.Success();
-}
-
-size_t Communication::GetCachedBytes(void *dst, size_t dst_len) {
-  std::lock_guard<std::recursive_mutex> guard(m_bytes_mutex);
-  if (!m_bytes.empty()) {
-    // If DST is nullptr and we have a thread, then return the number of bytes
-    // that are available so the caller can call again
-    if (dst == nullptr)
-      return m_bytes.size();
-
-    const size_t len = std::min<size_t>(dst_len, m_bytes.size());
-
-    ::memcpy(dst, m_bytes.c_str(), len);
-    m_bytes.erase(m_bytes.begin(), m_bytes.begin() + len);
-
-    return len;
-  }
-  return 0;
-}
-
-void Communication::AppendBytesToCache(const uint8_t *bytes, size_t len,
-                                       bool broadcast,
-                                       ConnectionStatus status) {
-  LLDB_LOG(GetLog(LLDBLog::Communication),
-           "{0} Communication::AppendBytesToCache (src = {1}, src_len = {2}, "
-           "broadcast = {3})",
-           this, bytes, (uint64_t)len, broadcast);
-  if ((bytes == nullptr || len == 0) &&
-      (status != lldb::eConnectionStatusEndOfFile))
-    return;
-  if (m_callback) {
-    // If the user registered a callback, then call it and do not broadcast
-    m_callback(m_callback_baton, bytes, len);
-  } else if (bytes != nullptr && len > 0) {
-    std::lock_guard<std::recursive_mutex> guard(m_bytes_mutex);
-    m_bytes.append((const char *)bytes, len);
-    if (broadcast)
-      BroadcastEventIfUnique(eBroadcastBitReadThreadGotBytes);
-  }
-}
-
 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);
 
   if (error_ptr)
     error_ptr->SetErrorString("Invalid connection.");
   status = eConnectionStatusNoConnection;
   return 0;
 }
 
-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) {
-    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:
-      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,
-                 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;
-      [[fallthrough]];
-    case eConnectionStatusTimedOut: // Request timed out
-      if (error.Fail())
-        LLDB_LOG(log, "error: {0}, status = {1}", error,
-                 Communication::ConnectionStatusAsString(status));
-      break;
-    }
-  }
-  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.
-    BroadcastEvent(eBroadcastBitNoMorePendingInput);
-
-    // Wait for the thread to finish...
-    std::lock_guard<std::mutex> guard(m_synchronize_mutex);
-    // ... and disconnect.
-    if (disconnect)
-      Disconnect();
-  }
-
-  // Let clients know that this thread is exiting
-  BroadcastEvent(eBroadcastBitReadThreadDidExit);
-  return {};
-}
-
-void Communication::SetReadThreadBytesReceivedCallback(
-    ReadThreadBytesReceived callback, void *callback_baton) {
-  m_callback = callback;
-  m_callback_baton = callback_baton;
-}
-
-void Communication::SynchronizeWithReadThread() {
-  // Only one thread can do the synchronization dance at a time.
-  std::lock_guard<std::mutex> guard(m_synchronize_mutex);
-
-  // First start listening for the synchronization event.
-  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();
-
-  // 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);
-  StopReadThread(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

-//===-- Communication.cpp -------------------------------------------------===//
+//===-- ThreadedCommunication.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/Core/ThreadedCommunication.h"
 
-#include "lldb/Host/HostThread.h"
 #include "lldb/Host/ThreadLauncher.h"
 #include "lldb/Utility/Connection.h"
 #include "lldb/Utility/ConstString.h"
 #include "lldb/Utility/Event.h"
 #include "lldb/Utility/LLDBLog.h"
 #include "lldb/Utility/Listener.h"
 #include "lldb/Utility/Log.h"
 #include "lldb/Utility/Status.h"
 
 #include "llvm/ADT/None.h"
-#include "llvm/ADT/Optional.h"
 #include "llvm/Support/Compiler.h"
 
 #include <algorithm>
 #include <chrono>
 #include <cstring>
 #include <memory>
 
 #include <cerrno>
 #include <cinttypes>
 #include <cstdio>
 
 using namespace lldb;
 using namespace lldb_private;
 
-ConstString &Communication::GetStaticBroadcasterClass() {
+ConstString &ThreadedCommunication::GetStaticBroadcasterClass() {
   static ConstString class_name("lldb.communication");
   return class_name;
 }
 
-Communication::Communication(const char *name)
-    : Broadcaster(nullptr, name), m_connection_sp(),
-      m_read_thread_enabled(false), m_read_thread_did_exit(false), m_bytes(),
-      m_bytes_mutex(), m_write_mutex(), m_synchronize_mutex(),
-      m_callback(nullptr), m_callback_baton(nullptr), m_close_on_eof(true)
-
-{
-
+ThreadedCommunication::ThreadedCommunication(const char *name)
+    : Communication(), Broadcaster(nullptr, name), m_read_thread_enabled(false),
+      m_read_thread_did_exit(false), m_bytes(), m_bytes_mutex(),
+      m_synchronize_mutex(), m_callback(nullptr), m_callback_baton(nullptr) {
   LLDB_LOG(GetLog(LLDBLog::Object | LLDBLog::Communication),
-           "{0} Communication::Communication (name = {1})", this, name);
+           "{0} ThreadedCommunication::ThreadedCommunication (name = {1})",
+           this, name);
 
   SetEventName(eBroadcastBitDisconnected, "disconnected");
   SetEventName(eBroadcastBitReadThreadGotBytes, "got bytes");
   SetEventName(eBroadcastBitReadThreadDidExit, "read thread did exit");
   SetEventName(eBroadcastBitReadThreadShouldExit, "read thread should exit");
   SetEventName(eBroadcastBitPacketAvailable, "packet available");
   SetEventName(eBroadcastBitNoMorePendingInput, "no more pending input");
 
   CheckInWithManager();
 }
 
-Communication::~Communication() {
+ThreadedCommunication::~ThreadedCommunication() {
   LLDB_LOG(GetLog(LLDBLog::Object | LLDBLog::Communication),
-           "{0} Communication::~Communication (name = {1})", this,
-           GetBroadcasterName().AsCString());
-  Clear();
+           "{0} ThreadedCommunication::~ThreadedCommunication (name = {1})",
+           this, GetBroadcasterName().AsCString());
 }
 
-void Communication::Clear() {
+void ThreadedCommunication::Clear() {
   SetReadThreadBytesReceivedCallback(nullptr, nullptr);
   StopReadThread(nullptr);
-  Disconnect(nullptr);
+  Communication::Clear();
 }
 
-ConnectionStatus Communication::Connect(const char *url, Status *error_ptr) {
-  Clear();
-
-  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);
-  if (error_ptr)
-    error_ptr->SetErrorString("Invalid connection.");
-  return eConnectionStatusNoConnection;
-}
-
-ConnectionStatus Communication::Disconnect(Status *error_ptr) {
-  LLDB_LOG(GetLog(LLDBLog::Communication), "{0} Communication::Disconnect ()",
-           this);
-
+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!");
-  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();
-    return status;
-  }
-  return eConnectionStatusNoConnection;
-}
-
-bool Communication::IsConnected() const {
-  lldb::ConnectionSP connection_sp(m_connection_sp);
-  return (connection_sp ? connection_sp->IsConnected() : false);
+  return Communication::Disconnect(error_ptr);
 }
 
-bool Communication::HasConnection() const {
-  return m_connection_sp.get() != nullptr;
-}
-
-size_t Communication::Read(void *dst, size_t dst_len,
+size_t ThreadedCommunication::Read(void *dst, size_t dst_len,
                                    const Timeout<std::micro> &timeout,
-                           ConnectionStatus &status, Status *error_ptr) {
+                                   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());
 
   if (m_read_thread_enabled) {
     // We have a dedicated read thread that is getting data for us
if (m_read_thread_enabled) {
 
     if (!m_connection_sp) {
       if (error_ptr)
         error_ptr->SetErrorString("Invalid connection.");
       status = eConnectionStatusNoConnection;
       return 0;
     }
 
-    ListenerSP listener_sp(Listener::MakeListener("Communication::Read"));
+    ListenerSP listener_sp(
+        Listener::MakeListener("ThreadedCommunication::Read"));
     listener_sp->StartListeningForEvents(
         this, eBroadcastBitReadThreadGotBytes | eBroadcastBitReadThreadDidExit);
     EventSP event_sp;
     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 (m_read_thread_enabled) {
     if (error_ptr)
       error_ptr->SetErrorString("Timed out.");
     status = eConnectionStatusTimedOut;
     return 0;
   }
 
   // We aren't using a read thread, just read the data synchronously in this
   // thread.
-  return ReadFromConnection(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);
-
-  if (error_ptr)
-    error_ptr->SetErrorString("Invalid connection.");
-  status = eConnectionStatusNoConnection;
-  return 0;
-}
-
-size_t Communication::WriteAll(const void *src, size_t src_len,
-                               ConnectionStatus &status, Status *error_ptr) {
-  size_t total_written = 0;
-  do
-    total_written += Write(static_cast<const char *>(src) + total_written,
-                           src_len - total_written, status, error_ptr);
-  while (status == eConnectionStatusSuccess && total_written < src_len);
-  return total_written;
+  return Communication::Read(dst, dst_len, timeout, status, error_ptr);
 }
 
-bool Communication::StartReadThread(Status *error_ptr) {
+bool ThreadedCommunication::StartReadThread(Status *error_ptr) {
   if (error_ptr)
     error_ptr->Clear();
 
   if (m_read_thread.IsJoinable())
     return true;
 
   LLDB_LOG(GetLog(LLDBLog::Communication),
-           "{0} Communication::StartReadThread ()", this);
+           "{0} ThreadedCommunication::StartReadThread ()", this);
 
   const std::string thread_name =
       llvm::formatv("<lldb.comm.{0}>", GetBroadcasterName());
 
   m_read_thread_enabled = true;
   m_read_thread_did_exit = false;
   auto maybe_thread = ThreadLauncher::LaunchThread(
       thread_name, [this] { return ReadThread(); });
bool ThreadedCommunication::StartReadThread(Status *error_ptr) {
   }
 
   if (!m_read_thread.IsJoinable())
     m_read_thread_enabled = false;
 
   return m_read_thread_enabled;
 }
 
-bool Communication::StopReadThread(Status *error_ptr) {
+bool ThreadedCommunication::StopReadThread(Status *error_ptr) {
   if (!m_read_thread.IsJoinable())
     return true;
 
   LLDB_LOG(GetLog(LLDBLog::Communication),
-           "{0} Communication::StopReadThread ()", this);
+           "{0} ThreadedCommunication::StopReadThread ()", this);
 
   m_read_thread_enabled = false;
 
   BroadcastEvent(eBroadcastBitReadThreadShouldExit, nullptr);
 
   // error = m_read_thread.Cancel();
 
   Status error = m_read_thread.Join(nullptr);
   return error.Success();
 }
 
-bool Communication::JoinReadThread(Status *error_ptr) {
+bool ThreadedCommunication::JoinReadThread(Status *error_ptr) {
   if (!m_read_thread.IsJoinable())
     return true;
 
   Status error = m_read_thread.Join(nullptr);
   return error.Success();
 }
 
-size_t Communication::GetCachedBytes(void *dst, size_t dst_len) {
+size_t ThreadedCommunication::GetCachedBytes(void *dst, size_t dst_len) {
   std::lock_guard<std::recursive_mutex> guard(m_bytes_mutex);
   if (!m_bytes.empty()) {
     // If DST is nullptr and we have a thread, then return the number of bytes
     // that are available so the caller can call again
     if (dst == nullptr)
       return m_bytes.size();
 
     const size_t len = std::min<size_t>(dst_len, m_bytes.size());
 
     ::memcpy(dst, m_bytes.c_str(), len);
     m_bytes.erase(m_bytes.begin(), m_bytes.begin() + len);
 
     return len;
   }
   return 0;
 }
 
-void Communication::AppendBytesToCache(const uint8_t *bytes, size_t len,
+void ThreadedCommunication::AppendBytesToCache(const uint8_t *bytes, size_t len,
                                                bool broadcast,
                                                ConnectionStatus status) {
   LLDB_LOG(GetLog(LLDBLog::Communication),
-           "{0} Communication::AppendBytesToCache (src = {1}, src_len = {2}, "
+           "{0} ThreadedCommunication::AppendBytesToCache (src = {1}, src_len "
+           "= {2}, "
            "broadcast = {3})",
            this, bytes, (uint64_t)len, broadcast);
   if ((bytes == nullptr || len == 0) &&
       (status != lldb::eConnectionStatusEndOfFile))
     return;
   if (m_callback) {
     // If the user registered a callback, then call it and do not broadcast
     m_callback(m_callback_baton, bytes, len);
   } else if (bytes != nullptr && len > 0) {
     std::lock_guard<std::recursive_mutex> guard(m_bytes_mutex);
     m_bytes.append((const char *)bytes, len);
     if (broadcast)
       BroadcastEventIfUnique(eBroadcastBitReadThreadGotBytes);
   }
 }
 
-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);
-
-  if (error_ptr)
-    error_ptr->SetErrorString("Invalid connection.");
-  status = eConnectionStatusNoConnection;
-  return 0;
+bool ThreadedCommunication::ReadThreadIsRunning() {
+  return m_read_thread_enabled;
 }
 
-bool Communication::ReadThreadIsRunning() { return m_read_thread_enabled; }
-
-lldb::thread_result_t Communication::ReadThread() {
+lldb::thread_result_t ThreadedCommunication::ReadThread() {
   Log *log = GetLog(LLDBLog::Communication);
 
   LLDB_LOG(log, "Communication({0}) thread starting...", this);
 
   uint8_t buf[1024];
 
   Status error;
   ConnectionStatus status = eConnectionStatusSuccess;
while (!done && m_read_thread_enabled) {
     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,
-                 Communication::ConnectionStatusAsString(status));
+                 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.
       BroadcastEvent(eBroadcastBitNoMorePendingInput);
       break;
     case eConnectionStatusNoConnection:   // No 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,
-                 Communication::ConnectionStatusAsString(status));
+                 ThreadedCommunication::ConnectionStatusAsString(status));
       break;
     }
   }
   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.
if (disconnect)
       Disconnect();
   }
 
   // Let clients know that this thread is exiting
   BroadcastEvent(eBroadcastBitReadThreadDidExit);
   return {};
 }
 
-void Communication::SetReadThreadBytesReceivedCallback(
+void ThreadedCommunication::SetReadThreadBytesReceivedCallback(
     ReadThreadBytesReceived callback, void *callback_baton) {
   m_callback = callback;
   m_callback_baton = callback_baton;
 }
 
-void Communication::SynchronizeWithReadThread() {
+void ThreadedCommunication::SynchronizeWithReadThread() {
   // Only one thread can do the synchronization dance at a time.
   std::lock_guard<std::mutex> guard(m_synchronize_mutex);
 
   // First start listening for the synchronization event.
-  ListenerSP listener_sp(
-      Listener::MakeListener("Communication::SyncronizeWithReadThread"));
+  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.
   m_connection_sp->InterruptRead();
 
   // 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);
+void ThreadedCommunication::SetConnection(
+    std::unique_ptr<Connection> connection) {
   StopReadThread(nullptr);
-  m_connection_sp = std::move(connection);
-}
-
-std::string
-Communication::ConnectionStatusAsString(lldb::ConnectionStatus status) {
-  switch (status) {
-  case eConnectionStatusSuccess:
-    return "success";
-  case eConnectionStatusError:
-    return "error";
-  case eConnectionStatusTimedOut:
-    return "timed out";
-  case eConnectionStatusNoConnection:
-    return "no connection";
-  case eConnectionStatusLostConnection:
-    return "lost connection";
-  case eConnectionStatusEndOfFile:
-    return "end of file";
-  case eConnectionStatusInterrupted:
-    return "interrupted";
-  }
-
-  return "@" + std::to_string(status);
+  Communication::SetConnection(std::move(connection));
 }

lldb/source/Plugins/Process/MacOSX-Kernel/CommunicationKDP.h

 #include "lldb/Core/Communication.h"
 #include "lldb/Core/StreamBuffer.h"
 #include "lldb/Utility/Listener.h"
 #include "lldb/Utility/Predicate.h"
 #include "lldb/lldb-private.h"
 
 class CommunicationKDP : public lldb_private::Communication {
 public:
-  enum { eBroadcastBitRunPacketSent = kLoUserBroadcastBit };
-
   const static uint32_t kMaxPacketSize = 1200;
   const static uint32_t kMaxDataSize = 1024;
   typedef lldb_private::StreamBuffer<4096> PacketStreamType;
   enum CommandType {
     KDP_CONNECT = 0u,
     KDP_DISCONNECT,
     KDP_HOSTINFO,
     KDP_VERSION,

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

 
 #include "ProcessKDPLog.h"
 
 using namespace lldb;
 using namespace lldb_private;
 
 // CommunicationKDP constructor
 CommunicationKDP::CommunicationKDP(const char *comm_name)
-    : Communication(comm_name), m_addr_byte_size(4),
+    : Communication(), m_addr_byte_size(4),
       m_byte_order(eByteOrderLittle), m_packet_timeout(5), m_sequence_mutex(),
       m_is_running(false), m_session_key(0u), m_request_sequence_id(0u),
       m_exception_sequence_id(0u), m_kdp_version_version(0u),
       m_kdp_version_feature(0u), m_kdp_hostinfo_cpu_mask(0u),
       m_kdp_hostinfo_cpu_type(0u), m_kdp_hostinfo_cpu_subtype(0u) {}
 
 // Destructor
 CommunicationKDP::~CommunicationKDP() {

lldb/source/Plugins/Process/gdb-remote/GDBRemoteCommunication.h

 enum GDBErrno {
 #define HANDLE_ERRNO(name, value) GDB_##name = value,
 #include "Plugins/Process/gdb-remote/GDBRemoteErrno.def"
   GDB_EUNKNOWN = 9999
 };
 
 class ProcessGDBRemote;
 
-class GDBRemoteCommunication : public Communication {
+class GDBRemoteCommunication : public Communication, public Broadcaster {
 public:
   enum {
-    eBroadcastBitRunPacketSent = kLoUserBroadcastBit,
+    eBroadcastBitRunPacketSent = (1u << 0),
   };
 
   enum class PacketType { Invalid = 0, Standard, Notify };
 
   enum class PacketResult {
     Success = 0,        // Success
     ErrorSendFailed,    // Status sending the packet
     ErrorSendAck,       // Didn't get an ack back after sending a packet
protected:
   uint32_t m_echo_number;
   LazyBool m_supports_qEcho;
   GDBRemoteCommunicationHistory m_history;
   bool m_send_acks;
   bool m_is_platform; // Set to true if this class represents a platform,
                       // false if this class represents a debug session for
                       // a single process
 
+  std::string m_bytes;
+  std::recursive_mutex m_bytes_mutex;
   CompressionType m_compression_type;
 
   PacketResult SendPacketNoLock(llvm::StringRef payload);
   PacketResult SendNotificationPacketNoLock(llvm::StringRef notify_type,
                                             std::deque<std::string>& queue,
                                             llvm::StringRef payload);
   PacketResult SendRawPacketNoLock(llvm::StringRef payload,
                                    bool skip_ack = false);

lldb/source/Plugins/Process/gdb-remote/GDBRemoteCommunication.cpp

 
 using namespace lldb;
 using namespace lldb_private;
 using namespace lldb_private::process_gdb_remote;
 
 // GDBRemoteCommunication constructor
 GDBRemoteCommunication::GDBRemoteCommunication(const char *comm_name,
                                                const char *listener_name)
-    : Communication(comm_name),
+    : Communication(), Broadcaster(nullptr, comm_name),
 #ifdef LLDB_CONFIGURATION_DEBUG
       m_packet_timeout(1000),
 #else
       m_packet_timeout(1),
 #endif
       m_echo_number(0), m_supports_qEcho(eLazyBoolCalculate), m_history(512),
       m_send_acks(true), m_is_platform(false),
       m_compression_type(CompressionType::None), m_listen_url() {

lldb/source/Plugins/Process/gdb-remote/GDBRemoteCommunicationServerLLGS.cpp

 
 // GDBRemoteCommunicationServerLLGS constructor
 GDBRemoteCommunicationServerLLGS::GDBRemoteCommunicationServerLLGS(
     MainLoop &mainloop, const NativeProcessProtocol::Factory &process_factory)
     : GDBRemoteCommunicationServerCommon("gdb-remote.server",
                                          "gdb-remote.server.rx_packet"),
       m_mainloop(mainloop), m_process_factory(process_factory),
       m_current_process(nullptr), m_continue_process(nullptr),
-      m_stdio_communication("process.stdio") {
+      m_stdio_communication() {
   RegisterPacketHandlers();
 }
 
 void GDBRemoteCommunicationServerLLGS::RegisterPacketHandlers() {
   RegisterMemberFunctionHandler(StringExtractorGDBRemote::eServerPacketType_C,
                                 &GDBRemoteCommunicationServerLLGS::Handle_C);
   RegisterMemberFunctionHandler(StringExtractorGDBRemote::eServerPacketType_c,
                                 &GDBRemoteCommunicationServerLLGS::Handle_c);

lldb/source/Plugins/Process/gdb-remote/ProcessGDBRemote.cpp

ProcessGDBRemote::ProcessGDBRemote(lldb::TargetSP target_sp,
   if (m_async_listener_sp->StartListeningForEvents(
           &m_async_broadcaster, async_event_mask) != async_event_mask) {
     LLDB_LOGF(log,
               "ProcessGDBRemote::%s failed to listen for "
               "m_async_broadcaster events",
               __FUNCTION__);
   }
 
-  const uint32_t gdb_event_mask = Communication::eBroadcastBitReadThreadDidExit;
-  if (m_async_listener_sp->StartListeningForEvents(
-          &m_gdb_comm, gdb_event_mask) != gdb_event_mask) {
-    LLDB_LOGF(log,
-              "ProcessGDBRemote::%s failed to listen for m_gdb_comm events",
-              __FUNCTION__);
-  }
-
   const uint64_t timeout_seconds =
       GetGlobalPluginProperties().GetPacketTimeout();
   if (timeout_seconds > 0)
     m_gdb_comm.SetPacketTimeout(std::chrono::seconds(timeout_seconds));
 
   m_use_g_packet_for_reading =
       GetGlobalPluginProperties().GetUseGPacketForReading();
 }
if (m_async_listener_sp->GetEvent(event_sp, llvm::None)) {
         default:
           LLDB_LOGF(log,
                     "ProcessGDBRemote::%s(pid = %" PRIu64
                     ") got unknown event 0x%8.8x",
                     __FUNCTION__, GetID(), event_type);
           done = true;
           break;
         }
-      } else if (event_sp->BroadcasterIs(&m_gdb_comm)) {
-        switch (event_type) {
-        case Communication::eBroadcastBitReadThreadDidExit:
-          SetExitStatus(-1, "lost connection");
-          done = true;
-          break;
-
-        default:
-          LLDB_LOGF(log,
-                    "ProcessGDBRemote::%s(pid = %" PRIu64
-                    ") got unknown event 0x%8.8x",
-                    __FUNCTION__, GetID(), event_type);
-          done = true;
-          break;
-        }
       }
     } else {
       LLDB_LOGF(log,
                 "ProcessGDBRemote::%s(pid = %" PRIu64
                 ") listener.WaitForEvent (NULL, event_sp) => false",
                 __FUNCTION__, GetID());
       done = true;
     }

lldb/source/Plugins/ScriptInterpreter/Python/ScriptInterpreterPython.cpp

 #include "ScriptInterpreterPythonImpl.h"
 #include "ScriptedProcessPythonInterface.h"
 
 #include "lldb/API/SBError.h"
 #include "lldb/API/SBFrame.h"
 #include "lldb/API/SBValue.h"
 #include "lldb/Breakpoint/StoppointCallbackContext.h"
 #include "lldb/Breakpoint/WatchpointOptions.h"
-#include "lldb/Core/Communication.h"
 #include "lldb/Core/Debugger.h"
 #include "lldb/Core/PluginManager.h"
+#include "lldb/Core/ThreadedCommunication.h"
 #include "lldb/Core/ValueObject.h"
 #include "lldb/DataFormatters/TypeSummary.h"
 #include "lldb/Host/FileSystem.h"
 #include "lldb/Host/HostInfo.h"
 #include "lldb/Host/Pipe.h"
 #include "lldb/Interpreter/CommandInterpreter.h"
 #include "lldb/Interpreter/CommandReturnObject.h"
 #include "lldb/Target/Thread.h"

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/Core/ThreadedCommunication.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 "TestingSupport/Host/SocketTestUtilities.h"
 #include "TestingSupport/SubsystemRAII.h"
 

 static void CommunicationReadTest(bool use_read_thread) {
   std::unique_ptr<TCPSocket> a, b;
   ASSERT_TRUE(CreateTCPConnectedSockets("localhost", &a, &b));
 
   size_t num_bytes = 4;
   ASSERT_THAT_ERROR(a->Write("test", num_bytes).ToError(), llvm::Succeeded());
   ASSERT_EQ(num_bytes, 4U);
 
-  Communication comm("test");
+  ThreadedCommunication comm("test");
   comm.SetConnection(std::make_unique<ConnectionFileDescriptor>(b.release()));
   comm.SetCloseOnEOF(true);
 
   if (use_read_thread) {
     ASSERT_TRUE(comm.StartReadThread());
   }
 
   // This read should wait for the data to become available and return it.

 TEST_F(CommunicationTest, ReadThread) {
   CommunicationReadTest(/*use_thread=*/true);
 }
 
 TEST_F(CommunicationTest, SynchronizeWhileClosing) {
   std::unique_ptr<TCPSocket> a, b;
   ASSERT_TRUE(CreateTCPConnectedSockets("localhost", &a, &b));
 
-  Communication comm("test");
+  ThreadedCommunication comm("test");
   comm.SetConnection(std::make_unique<ConnectionFileDescriptor>(b.release()));
   comm.SetCloseOnEOF(true);
   ASSERT_TRUE(comm.StartReadThread());
 
   // Ensure that we can safely synchronize with the read thread while it is
   // closing the read end (in response to us closing the write end).
   ASSERT_THAT_ERROR(a->Close().ToError(), llvm::Succeeded());
   comm.SynchronizeWithReadThread();
TEST_F(CommunicationTest, WriteAll) {
   // write.
   int write_fd = pipe.ReleaseWriteFileDescriptor();
   int flags = fcntl(write_fd, F_GETFL);
   ASSERT_NE(flags, -1);
   ASSERT_NE(fcntl(write_fd, F_SETFL, flags | O_NONBLOCK), -1);
 
   ConnectionFileDescriptor read_conn{pipe.ReleaseReadFileDescriptor(),
                                      /*owns_fd=*/true};
-  Communication write_comm("test");
+  Communication write_comm;
   write_comm.SetConnection(
       std::make_unique<ConnectionFileDescriptor>(write_fd, /*owns_fd=*/true));
 
   std::thread read_thread{[&read_conn]() {
     // Read using a smaller buffer to increase chances of partial write.
     char buf[128 * 1024];
     lldb::ConnectionStatus conn_status;