377 lines · cpp
1//===-- ThreadedCommunication.cpp -----------------------------------------===//2//3// Part of the LLVM Project, under the Apache License v2.0 with LLVM Exceptions.4// See https://llvm.org/LICENSE.txt for license information.5// SPDX-License-Identifier: Apache-2.0 WITH LLVM-exception6//7//===----------------------------------------------------------------------===//8 9#include "lldb/Core/ThreadedCommunication.h"10 11#include "lldb/Host/ThreadLauncher.h"12#include "lldb/Utility/Connection.h"13#include "lldb/Utility/ConstString.h"14#include "lldb/Utility/Event.h"15#include "lldb/Utility/LLDBLog.h"16#include "lldb/Utility/Listener.h"17#include "lldb/Utility/Log.h"18#include "lldb/Utility/Status.h"19 20#include "llvm/Support/Compiler.h"21 22#include <algorithm>23#include <chrono>24#include <cstring>25#include <memory>26#include <shared_mutex>27 28#include <cerrno>29#include <cinttypes>30#include <cstdio>31 32using namespace lldb;33using namespace lldb_private;34 35llvm::StringRef ThreadedCommunication::GetStaticBroadcasterClass() {36 static constexpr llvm::StringLiteral class_name("lldb.communication");37 return class_name;38}39 40ThreadedCommunication::ThreadedCommunication(const char *name)41 : Communication(), Broadcaster(nullptr, name), m_read_thread_enabled(false),42 m_read_thread_did_exit(false), m_bytes(), m_bytes_mutex(),43 m_synchronize_mutex(), m_callback(nullptr), m_callback_baton(nullptr) {44 LLDB_LOG(GetLog(LLDBLog::Object | LLDBLog::Communication),45 "{0} ThreadedCommunication::ThreadedCommunication (name = {1})",46 this, name);47 48 SetEventName(eBroadcastBitDisconnected, "disconnected");49 SetEventName(eBroadcastBitReadThreadGotBytes, "got bytes");50 SetEventName(eBroadcastBitReadThreadDidExit, "read thread did exit");51 SetEventName(eBroadcastBitReadThreadShouldExit, "read thread should exit");52 SetEventName(eBroadcastBitPacketAvailable, "packet available");53 SetEventName(eBroadcastBitNoMorePendingInput, "no more pending input");54 55 CheckInWithManager();56}57 58ThreadedCommunication::~ThreadedCommunication() {59 LLDB_LOG(GetLog(LLDBLog::Object | LLDBLog::Communication),60 "{0} ThreadedCommunication::~ThreadedCommunication (name = {1})",61 this, GetBroadcasterName());62}63 64void ThreadedCommunication::Clear() {65 SetReadThreadBytesReceivedCallback(nullptr, nullptr);66 StopReadThread(nullptr);67 Communication::Clear();68}69 70ConnectionStatus ThreadedCommunication::Disconnect(Status *error_ptr) {71 assert((!m_read_thread_enabled || m_read_thread_did_exit) &&72 "Disconnecting while the read thread is running is racy!");73 return Communication::Disconnect(error_ptr);74}75 76size_t ThreadedCommunication::Read(void *dst, size_t dst_len,77 const Timeout<std::micro> &timeout,78 ConnectionStatus &status,79 Status *error_ptr) {80 Log *log = GetLog(LLDBLog::Communication);81 LLDB_LOG(82 log,83 "this = {0}, dst = {1}, dst_len = {2}, timeout = {3}, connection = {4}",84 this, dst, dst_len, timeout, m_connection_sp.get());85 86 if (m_read_thread_enabled) {87 // We have a dedicated read thread that is getting data for us88 size_t cached_bytes = GetCachedBytes(dst, dst_len);89 if (cached_bytes > 0) {90 status = eConnectionStatusSuccess;91 return cached_bytes;92 }93 if (timeout && timeout->count() == 0) {94 if (error_ptr)95 *error_ptr = Status::FromErrorString("Timed out.");96 status = eConnectionStatusTimedOut;97 return 0;98 }99 100 if (!m_connection_sp) {101 if (error_ptr)102 *error_ptr = Status::FromErrorString("Invalid connection.");103 status = eConnectionStatusNoConnection;104 return 0;105 }106 107 // No data yet, we have to start listening.108 ListenerSP listener_sp(109 Listener::MakeListener("ThreadedCommunication::Read"));110 listener_sp->StartListeningForEvents(111 this, eBroadcastBitReadThreadGotBytes | eBroadcastBitReadThreadDidExit);112 113 // Re-check for data, as it might have arrived while we were setting up our114 // listener.115 cached_bytes = GetCachedBytes(dst, dst_len);116 if (cached_bytes > 0) {117 status = eConnectionStatusSuccess;118 return cached_bytes;119 }120 121 EventSP event_sp;122 // Explicitly check for the thread exit, for the same reason.123 if (m_read_thread_did_exit) {124 // We've missed the event, lets just conjure one up.125 event_sp = std::make_shared<Event>(eBroadcastBitReadThreadDidExit);126 } else {127 if (!listener_sp->GetEvent(event_sp, timeout)) {128 if (error_ptr)129 *error_ptr = Status::FromErrorString("Timed out.");130 status = eConnectionStatusTimedOut;131 return 0;132 }133 }134 const uint32_t event_type = event_sp->GetType();135 if (event_type & eBroadcastBitReadThreadGotBytes) {136 return GetCachedBytes(dst, dst_len);137 }138 139 if (event_type & eBroadcastBitReadThreadDidExit) {140 // If the thread exited of its own accord, it either means it141 // hit an end-of-file condition or an error.142 status = m_pass_status;143 if (error_ptr)144 *error_ptr = std::move(m_pass_error);145 146 if (GetCloseOnEOF())147 Disconnect(nullptr);148 return 0;149 }150 llvm_unreachable("Got unexpected event type!");151 }152 153 // We aren't using a read thread, just read the data synchronously in this154 // thread.155 return Communication::Read(dst, dst_len, timeout, status, error_ptr);156}157 158bool ThreadedCommunication::StartReadThread(Status *error_ptr) {159 std::lock_guard<std::mutex> lock(m_read_thread_mutex);160 161 if (error_ptr)162 error_ptr->Clear();163 164 if (m_read_thread.IsJoinable())165 return true;166 167 LLDB_LOG(GetLog(LLDBLog::Communication),168 "{0} ThreadedCommunication::StartReadThread ()", this);169 170 const std::string thread_name =171 llvm::formatv("<lldb.comm.{0}>", GetBroadcasterName());172 173 m_read_thread_enabled = true;174 m_read_thread_did_exit = false;175 auto maybe_thread = ThreadLauncher::LaunchThread(176 thread_name, [this] { return ReadThread(); });177 if (maybe_thread) {178 m_read_thread = *maybe_thread;179 } else {180 if (error_ptr)181 *error_ptr = Status::FromError(maybe_thread.takeError());182 else {183 LLDB_LOG_ERROR(GetLog(LLDBLog::Host), maybe_thread.takeError(),184 "failed to launch host thread: {0}");185 }186 }187 188 if (!m_read_thread.IsJoinable())189 m_read_thread_enabled = false;190 191 return m_read_thread_enabled;192}193 194bool ThreadedCommunication::StopReadThread(Status *error_ptr) {195 std::lock_guard<std::mutex> lock(m_read_thread_mutex);196 197 if (!m_read_thread.IsJoinable())198 return true;199 200 LLDB_LOG(GetLog(LLDBLog::Communication),201 "{0} ThreadedCommunication::StopReadThread ()", this);202 203 m_read_thread_enabled = false;204 205 BroadcastEvent(eBroadcastBitReadThreadShouldExit, nullptr);206 207 Status error = m_read_thread.Join(nullptr);208 return error.Success();209}210 211bool ThreadedCommunication::JoinReadThread(Status *error_ptr) {212 std::lock_guard<std::mutex> lock(m_read_thread_mutex);213 214 if (!m_read_thread.IsJoinable())215 return true;216 217 Status error = m_read_thread.Join(nullptr);218 return error.Success();219}220 221size_t ThreadedCommunication::GetCachedBytes(void *dst, size_t dst_len) {222 std::lock_guard<std::recursive_mutex> guard(m_bytes_mutex);223 if (!m_bytes.empty()) {224 // If DST is nullptr and we have a thread, then return the number of bytes225 // that are available so the caller can call again226 if (dst == nullptr)227 return m_bytes.size();228 229 const size_t len = std::min<size_t>(dst_len, m_bytes.size());230 231 ::memcpy(dst, m_bytes.c_str(), len);232 m_bytes.erase(m_bytes.begin(), m_bytes.begin() + len);233 234 return len;235 }236 return 0;237}238 239void ThreadedCommunication::AppendBytesToCache(const uint8_t *bytes, size_t len,240 bool broadcast,241 ConnectionStatus status) {242 LLDB_LOG(GetLog(LLDBLog::Communication),243 "{0} ThreadedCommunication::AppendBytesToCache (src = {1}, src_len "244 "= {2}, "245 "broadcast = {3})",246 this, bytes, (uint64_t)len, broadcast);247 if ((bytes == nullptr || len == 0) &&248 (status != lldb::eConnectionStatusEndOfFile))249 return;250 if (m_callback) {251 // If the user registered a callback, then call it and do not broadcast252 m_callback(m_callback_baton, bytes, len);253 } else if (bytes != nullptr && len > 0) {254 std::lock_guard<std::recursive_mutex> guard(m_bytes_mutex);255 m_bytes.append((const char *)bytes, len);256 if (broadcast)257 BroadcastEventIfUnique(eBroadcastBitReadThreadGotBytes);258 }259}260 261bool ThreadedCommunication::ReadThreadIsRunning() {262 return m_read_thread_enabled;263}264 265lldb::thread_result_t ThreadedCommunication::ReadThread() {266 Log *log = GetLog(LLDBLog::Communication);267 268 LLDB_LOG(log, "Communication({0}) thread starting...", this);269 270 uint8_t buf[1024];271 272 Status error;273 ConnectionStatus status = eConnectionStatusSuccess;274 bool done = false;275 bool disconnect = false;276 while (!done && m_read_thread_enabled) {277 size_t bytes_read = ReadFromConnection(278 buf, sizeof(buf), std::chrono::seconds(5), status, &error);279 if (bytes_read > 0 || status == eConnectionStatusEndOfFile)280 AppendBytesToCache(buf, bytes_read, true, status);281 282 switch (status) {283 case eConnectionStatusSuccess:284 break;285 286 case eConnectionStatusEndOfFile:287 done = true;288 disconnect = GetCloseOnEOF();289 break;290 case eConnectionStatusError: // Check GetError() for details291 if (error.GetType() == eErrorTypePOSIX && error.GetError() == EIO) {292 // EIO on a pipe is usually caused by remote shutdown293 disconnect = GetCloseOnEOF();294 done = true;295 }296 if (error.Fail())297 LLDB_LOG(log, "error: {0}, status = {1}", error,298 ThreadedCommunication::ConnectionStatusAsString(status));299 break;300 case eConnectionStatusInterrupted: // Synchronization signal from301 // SynchronizeWithReadThread()302 // The connection returns eConnectionStatusInterrupted only when there is303 // no input pending to be read, so we can signal that.304 BroadcastEvent(eBroadcastBitNoMorePendingInput);305 break;306 case eConnectionStatusNoConnection: // No connection307 case eConnectionStatusLostConnection: // Lost connection while connected to308 // a valid connection309 done = true;310 [[fallthrough]];311 case eConnectionStatusTimedOut: // Request timed out312 if (error.Fail())313 LLDB_LOG(log, "error: {0}, status = {1}", error,314 ThreadedCommunication::ConnectionStatusAsString(status));315 break;316 }317 }318 m_pass_status = status;319 m_pass_error = std::move(error);320 LLDB_LOG(log, "Communication({0}) thread exiting...", this);321 322 // Start shutting down. We need to do this in a very specific order to ensure323 // we don't race with threads wanting to read/synchronize with us.324 325 // First, we signal our intent to exit. This ensures no new thread start326 // waiting on events from us.327 m_read_thread_did_exit = true;328 329 // Unblock any existing thread waiting for the synchronization signal.330 BroadcastEvent(eBroadcastBitNoMorePendingInput);331 332 {333 // Wait for the synchronization thread to finish...334 std::lock_guard<std::mutex> guard(m_synchronize_mutex);335 // ... and disconnect.336 if (disconnect)337 Disconnect();338 }339 340 // Finally, unblock any readers waiting for us to exit.341 BroadcastEvent(eBroadcastBitReadThreadDidExit);342 return {};343}344 345void ThreadedCommunication::SetReadThreadBytesReceivedCallback(346 ReadThreadBytesReceived callback, void *callback_baton) {347 m_callback = callback;348 m_callback_baton = callback_baton;349}350 351void ThreadedCommunication::SynchronizeWithReadThread() {352 // Only one thread can do the synchronization dance at a time.353 std::lock_guard<std::mutex> guard(m_synchronize_mutex);354 355 // First start listening for the synchronization event.356 ListenerSP listener_sp(Listener::MakeListener(357 "ThreadedCommunication::SyncronizeWithReadThread"));358 listener_sp->StartListeningForEvents(this, eBroadcastBitNoMorePendingInput);359 360 // If the thread is not running, there is no point in synchronizing.361 if (!m_read_thread_enabled || m_read_thread_did_exit)362 return;363 364 // Notify the read thread.365 m_connection_sp->InterruptRead();366 367 // Wait for the synchronization event.368 EventSP event_sp;369 listener_sp->GetEvent(event_sp, std::nullopt);370}371 372void ThreadedCommunication::SetConnection(373 std::unique_ptr<Connection> connection) {374 StopReadThread(nullptr);375 Communication::SetConnection(std::move(connection));376}377