| //===-- 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/Listener.h" |
| #include "lldb/Utility/Log.h" |
| #include "lldb/Utility/Logging.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(lldb_private::GetLogIfAnyCategoriesSet(LIBLLDB_LOG_OBJECT | |
| LIBLLDB_LOG_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() { |
| LLDB_LOG(lldb_private::GetLogIfAnyCategoriesSet(LIBLLDB_LOG_OBJECT | |
| LIBLLDB_LOG_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(lldb_private::GetLogIfAllCategoriesSet(LIBLLDB_LOG_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(lldb_private::GetLogIfAllCategoriesSet(LIBLLDB_LOG_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. |
| // |
| // 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); |
| } |
| |
| bool Communication::HasConnection() const { |
| return m_connection_sp.get() != nullptr; |
| } |
| |
| size_t Communication::Read(void *dst, size_t dst_len, |
| const Timeout<std::micro> &timeout, |
| ConnectionStatus &status, Status *error_ptr) { |
| Log *log = GetLogIfAllCategoriesSet(LIBLLDB_LOG_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 || (timeout && timeout->count() == 0)) { |
| status = eConnectionStatusSuccess; |
| return cached_bytes; |
| } |
| |
| 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 (GetCloseOnEOF()) |
| Disconnect(nullptr); |
| break; |
| } |
| } |
| 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(lldb_private::GetLogIfAllCategoriesSet(LIBLLDB_LOG_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; |
| } |
| |
| bool Communication::StartReadThread(Status *error_ptr) { |
| if (error_ptr) |
| error_ptr->Clear(); |
| |
| if (m_read_thread.IsJoinable()) |
| return true; |
| |
| LLDB_LOG(lldb_private::GetLogIfAllCategoriesSet(LIBLLDB_LOG_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, Communication::ReadThread, this); |
| if (maybe_thread) { |
| m_read_thread = *maybe_thread; |
| } else { |
| if (error_ptr) |
| *error_ptr = Status(maybe_thread.takeError()); |
| else { |
| LLDB_LOG(lldb_private::GetLogIfAllCategoriesSet(LIBLLDB_LOG_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(lldb_private::GetLogIfAllCategoriesSet(LIBLLDB_LOG_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(lldb_private::GetLogIfAllCategoriesSet(LIBLLDB_LOG_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(lldb::thread_arg_t p) { |
| Communication *comm = (Communication *)p; |
| |
| Log *log(lldb_private::GetLogIfAllCategoriesSet(LIBLLDB_LOG_COMMUNICATION)); |
| |
| LLDB_LOGF(log, "%p Communication::ReadThread () thread starting...", p); |
| |
| uint8_t buf[1024]; |
| |
| Status error; |
| ConnectionStatus status = eConnectionStatusSuccess; |
| bool done = false; |
| bool disconnect = false; |
| while (!done && comm->m_read_thread_enabled) { |
| size_t bytes_read = comm->ReadFromConnection( |
| buf, sizeof(buf), std::chrono::seconds(5), status, &error); |
| if (bytes_read > 0 || status == eConnectionStatusEndOfFile) |
| comm->AppendBytesToCache(buf, bytes_read, true, status); |
| |
| switch (status) { |
| case eConnectionStatusSuccess: |
| break; |
| |
| case eConnectionStatusEndOfFile: |
| done = true; |
| disconnect = comm->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 = comm->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. |
| comm->BroadcastEvent(eBroadcastBitNoMorePendingInput); |
| break; |
| case eConnectionStatusNoConnection: // No connection |
| case eConnectionStatusLostConnection: // Lost connection while connected to |
| // a valid connection |
| done = true; |
| LLVM_FALLTHROUGH; |
| case eConnectionStatusTimedOut: // Request timed out |
| if (error.Fail()) |
| LLDB_LOG(log, "error: {0}, status = {1}", error, |
| Communication::ConnectionStatusAsString(status)); |
| break; |
| } |
| } |
| log = lldb_private::GetLogIfAllCategoriesSet(LIBLLDB_LOG_COMMUNICATION); |
| if (log) |
| LLDB_LOGF(log, "%p Communication::ReadThread () thread exiting...", p); |
| |
| // Handle threads wishing to synchronize with us. |
| { |
| // Prevent new ones from showing up. |
| comm->m_read_thread_did_exit = true; |
| |
| // Unblock any existing thread waiting for the synchronization signal. |
| comm->BroadcastEvent(eBroadcastBitNoMorePendingInput); |
| |
| // Wait for the thread to finish... |
| std::lock_guard<std::mutex> guard(comm->m_synchronize_mutex); |
| // ... and disconnect. |
| if (disconnect) |
| comm->Disconnect(); |
| } |
| |
| // Let clients know that this thread is exiting |
| comm->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"; |
| 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); |
| } |