1 files changed, 376 insertions, 0 deletions
diff --git a/contrib/llvm-project/lldb/source/Core/ThreadedCommunication.cpp b/contrib/llvm-project/lldb/source/Core/ThreadedCommunication.cpp
new file mode 100644
index 000000000000..2f3dada3ac93
--- /dev/null
+++ b/contrib/llvm-project/lldb/source/Core/ThreadedCommunication.cpp
@@ -0,0 +1,376 @@
+//===-- 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/ThreadedCommunication.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/Support/Compiler.h"
+
+#include <algorithm>
+#include <chrono>
+#include <cstring>
+#include <memory>
+#include <shared_mutex>
+
+#include <cerrno>
+#include <cinttypes>
+#include <cstdio>
+
+using namespace lldb;
+using namespace lldb_private;
+
+llvm::StringRef ThreadedCommunication::GetStaticBroadcasterClass() {
+  static constexpr llvm::StringLiteral class_name("lldb.communication");
+  return class_name;
+}
+
+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} 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();
+}
+
+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) {
+  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;
+    }
+
+    // No data yet, we have to start listening.
+    ListenerSP listener_sp(
+        Listener::MakeListener("ThreadedCommunication::Read"));
+    listener_sp->StartListeningForEvents(
+        this, eBroadcastBitReadThreadGotBytes | eBroadcastBitReadThreadDidExit);
+
+    // Re-check for data, as it might have arrived while we were setting up our
+    // listener.
+    cached_bytes = GetCachedBytes(dst, dst_len);
+    if (cached_bytes > 0) {
+      status = eConnectionStatusSuccess;
+      return cached_bytes;
+    }
+
+    EventSP event_sp;
+    // Explicitly check for the thread exit, for the same reason.
+    if (m_read_thread_did_exit) {
+      // We've missed the event, lets just conjure one up.
+      event_sp = std::make_shared<Event>(eBroadcastBitReadThreadDidExit);
+    } else {
+      if (!listener_sp->GetEvent(event_sp, timeout)) {
+        if (error_ptr)
+          error_ptr->SetErrorString("Timed out.");
+        status = eConnectionStatusTimedOut;
+        return 0;
+      }
+    }
+    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;
+    }
+    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);
+}
+
+bool ThreadedCommunication::StartReadThread(Status *error_ptr) {
+  std::lock_guard<std::mutex> lock(m_read_thread_mutex);
+
+  if (error_ptr)
+    error_ptr->Clear();
+
+  if (m_read_thread.IsJoinable())
+    return true;
+
+  LLDB_LOG(GetLog(LLDBLog::Communication),
+           "{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(); });
+  if (maybe_thread) {
+    m_read_thread = *maybe_thread;
+  } else {
+    if (error_ptr)
+      *error_ptr = Status(maybe_thread.takeError());
+    else {
+      LLDB_LOG_ERROR(GetLog(LLDBLog::Host), maybe_thread.takeError(),
+                     "failed to launch host thread: {0}");
+    }
+  }
+
+  if (!m_read_thread.IsJoinable())
+    m_read_thread_enabled = false;
+
+  return m_read_thread_enabled;
+}
+
+bool ThreadedCommunication::StopReadThread(Status *error_ptr) {
+  std::lock_guard<std::mutex> lock(m_read_thread_mutex);
+
+  if (!m_read_thread.IsJoinable())
+    return true;
+
+  LLDB_LOG(GetLog(LLDBLog::Communication),
+           "{0} ThreadedCommunication::StopReadThread ()", this);
+
+  m_read_thread_enabled = false;
+
+  BroadcastEvent(eBroadcastBitReadThreadShouldExit, nullptr);
+
+  Status error = m_read_thread.Join(nullptr);
+  return error.Success();
+}
+
+bool ThreadedCommunication::JoinReadThread(Status *error_ptr) {
+  std::lock_guard<std::mutex> lock(m_read_thread_mutex);
+
+  if (!m_read_thread.IsJoinable())
+    return true;
+
+  Status error = m_read_thread.Join(nullptr);
+  return error.Success();
+}
+
+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 ThreadedCommunication::AppendBytesToCache(const uint8_t *bytes, size_t len,
+                                               bool broadcast,
+                                               ConnectionStatus status) {
+  LLDB_LOG(GetLog(LLDBLog::Communication),
+           "{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);
+  }
+}
+
+bool ThreadedCommunication::ReadThreadIsRunning() {
+  return m_read_thread_enabled;
+}
+
+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;
+  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,
+                 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,
+                 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
+  // waiting on events from us.
+  m_read_thread_did_exit = true;
+
+  // Unblock any existing thread waiting for the synchronization signal.
+  BroadcastEvent(eBroadcastBitNoMorePendingInput);
+
+  {
+    // Wait for the synchronization thread to finish...
+    std::lock_guard<std::mutex> guard(m_synchronize_mutex);
+    // ... and disconnect.
+    if (disconnect)
+      Disconnect();
+  }
+
+  // Finally, unblock any readers waiting for us to exit.
+  BroadcastEvent(eBroadcastBitReadThreadDidExit);
+  return {};
+}
+
+void ThreadedCommunication::SetReadThreadBytesReceivedCallback(
+    ReadThreadBytesReceived callback, void *callback_baton) {
+  m_callback = callback;
+  m_callback_baton = callback_baton;
+}
+
+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(
+      "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, std::nullopt);
+}
+
+void ThreadedCommunication::SetConnection(
+    std::unique_ptr<Connection> connection) {
+  StopReadThread(nullptr);
+  Communication::SetConnection(std::move(connection));
+}