LLDB mainline
ThreadedCommunication.cpp
Go to the documentation of this file.
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-exception
6//
7//===----------------------------------------------------------------------===//
8
10
14#include "lldb/Utility/Event.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
36 static ConstString class_name("lldb.communication");
37 return class_name;
38}
39
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) {
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
56}
57
60 "{0} ThreadedCommunication::~ThreadedCommunication (name = {1})",
61 this, GetBroadcasterName());
62}
63
65 SetReadThreadBytesReceivedCallback(nullptr, nullptr);
66 StopReadThread(nullptr);
68}
69
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) {
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
87 // We have a dedicated read thread that is getting data for us
88 size_t cached_bytes = GetCachedBytes(dst, dst_len);
89 if (cached_bytes > 0) {
91 return cached_bytes;
92 }
93 if (timeout && timeout->count() == 0) {
94 if (error_ptr)
95 error_ptr->SetErrorString("Timed out.");
97 return 0;
98 }
99
100 if (!m_connection_sp) {
101 if (error_ptr)
102 error_ptr->SetErrorString("Invalid connection.");
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 our
114 // listener.
115 cached_bytes = GetCachedBytes(dst, dst_len);
116 if (cached_bytes > 0) {
118 return cached_bytes;
119 }
120
121 EventSP event_sp;
122 // Explicitly check for the thread exit, for the same reason.
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->SetErrorString("Timed out.");
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 it
141 // 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 this
154 // thread.
155 return Communication::Read(dst, dst_len, timeout, status, error_ptr);
156}
157
159 std::lock_guard<std::mutex> lock(m_read_thread_mutex);
160
161 if (error_ptr)
162 error_ptr->Clear();
163
165 return true;
166
168 "{0} ThreadedCommunication::StartReadThread ()", this);
169
170 const std::string thread_name =
171 llvm::formatv("<lldb.comm.{0}>", GetBroadcasterName());
172
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(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
189 m_read_thread_enabled = false;
190
192}
193
195 std::lock_guard<std::mutex> lock(m_read_thread_mutex);
196
198 return true;
199
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
212 std::lock_guard<std::mutex> lock(m_read_thread_mutex);
213
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 bytes
225 // that are available so the caller can call again
226 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) {
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) &&
249 return;
250 if (m_callback) {
251 // If the user registered a callback, then call it and do not broadcast
252 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
263}
264
267
268 LLDB_LOG(log, "Communication({0}) thread starting...", this);
269
270 uint8_t buf[1024];
271
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) {
284 break;
285
287 done = true;
288 disconnect = GetCloseOnEOF();
289 break;
290 case eConnectionStatusError: // Check GetError() for details
291 if (error.GetType() == eErrorTypePOSIX && error.GetError() == EIO) {
292 // EIO on a pipe is usually caused by remote shutdown
293 disconnect = GetCloseOnEOF();
294 done = true;
295 }
296 if (error.Fail())
297 LLDB_LOG(log, "error: {0}, status = {1}", error,
299 break;
300 case eConnectionStatusInterrupted: // Synchronization signal from
301 // SynchronizeWithReadThread()
302 // The connection returns eConnectionStatusInterrupted only when there is
303 // no input pending to be read, so we can signal that.
304 BroadcastEvent(eBroadcastBitNoMorePendingInput);
305 break;
306 case eConnectionStatusNoConnection: // No connection
307 case eConnectionStatusLostConnection: // Lost connection while connected to
308 // a valid connection
309 done = true;
310 [[fallthrough]];
311 case eConnectionStatusTimedOut: // Request timed out
312 if (error.Fail())
313 LLDB_LOG(log, "error: {0}, status = {1}", error,
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 ensure
323 // 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 start
326 // waiting on events from us.
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
346 ReadThreadBytesReceived callback, void *callback_baton) {
347 m_callback = callback;
348 m_callback_baton = callback_baton;
349}
350
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.
357 "ThreadedCommunication::SyncronizeWithReadThread"));
358 listener_sp->StartListeningForEvents(this, eBroadcastBitNoMorePendingInput);
359
360 // If the thread is not running, there is no point in synchronizing.
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
373 std::unique_ptr<Connection> connection) {
374 StopReadThread(nullptr);
375 Communication::SetConnection(std::move(connection));
376}
static llvm::raw_ostream & error(Stream &strm)
#define LLDB_LOG(log,...)
The LLDB_LOG* macros defined below are the way to emit log messages.
Definition: Log.h:342
#define LLDB_LOG_ERROR(log, error,...)
Definition: Log.h:365
static void ReadThreadBytesReceived(void *baton, const void *src, size_t src_len)
An event broadcasting class.
Definition: Broadcaster.h:145
void SetEventName(uint32_t event_mask, const char *name)
Set the name for an event bit.
Definition: Broadcaster.h:242
const std::string & GetBroadcasterName()
Get this broadcaster's name.
Definition: Broadcaster.h:222
void BroadcastEventIfUnique(lldb::EventSP &event_sp)
Definition: Broadcaster.h:171
void BroadcastEvent(lldb::EventSP &event_sp)
Broadcast an event which has no associated data.
Definition: Broadcaster.h:167
An abstract communications class.
Definition: Communication.h:39
size_t ReadFromConnection(void *dst, size_t dst_len, const Timeout< std::micro > &timeout, lldb::ConnectionStatus &status, Status *error_ptr)
virtual size_t Read(void *dst, size_t dst_len, const Timeout< std::micro > &timeout, lldb::ConnectionStatus &status, Status *error_ptr)
Read bytes from the current connection.
lldb::ConnectionSP m_connection_sp
The connection that is current in use by this communications class.
virtual void SetConnection(std::unique_ptr< Connection > connection)
Sets the connection that it to be used by this class.
virtual lldb::ConnectionStatus Disconnect(Status *error_ptr=nullptr)
Disconnect the communications connection if one is currently connected.
static std::string ConnectionStatusAsString(lldb::ConnectionStatus status)
A uniqued constant string class.
Definition: ConstString.h:40
Status Join(lldb::thread_result_t *result)
Definition: HostThread.cpp:20
static lldb::ListenerSP MakeListener(const char *name)
Definition: Listener.cpp:385
An error handling class.
Definition: Status.h:44
void Clear()
Clear the object state.
Definition: Status.cpp:167
void SetErrorString(llvm::StringRef err_str)
Set the current error string to err_str.
Definition: Status.cpp:233
static llvm::Expected< HostThread > LaunchThread(llvm::StringRef name, std::function< lldb::thread_result_t()> thread_function, size_t min_stack_byte_size=0)
virtual void AppendBytesToCache(const uint8_t *src, size_t src_len, bool broadcast, lldb::ConnectionStatus status)
Append new bytes that get read from the read thread into the internal object byte cache.
size_t Read(void *dst, size_t dst_len, const Timeout< std::micro > &timeout, lldb::ConnectionStatus &status, Status *error_ptr) override
Read bytes from the current connection.
static ConstString & GetStaticBroadcasterClass()
void SetConnection(std::unique_ptr< Connection > connection) override
Sets the connection that it to be used by this class.
ThreadedCommunication(const char *broadcaster_name)
Construct the ThreadedCommunication object with the specified name for the Broadcaster that this obje...
std::recursive_mutex m_bytes_mutex
A mutex to protect multi-threaded access to the cached bytes.
HostThread m_read_thread
The read thread handle in case we need to cancel the thread.
virtual bool StopReadThread(Status *error_ptr=nullptr)
Stops the read thread by cancelling it.
void SynchronizeWithReadThread()
Wait for the read thread to process all outstanding data.
std::atomic< bool > m_read_thread_did_exit
Whether the read thread is enabled.
lldb::ConnectionStatus m_pass_status
Connection status passthrough from read thread.
lldb::ConnectionStatus Disconnect(Status *error_ptr=nullptr) override
Disconnect the communications connection if one is currently connected.
Status m_pass_error
Error passthrough from read thread.
virtual bool StartReadThread(Status *error_ptr=nullptr)
Starts a read thread whose sole purpose it to read bytes from the current connection.
std::atomic< bool > m_read_thread_enabled
Whether the read thread is enabled.
virtual bool JoinReadThread(Status *error_ptr=nullptr)
std::string m_bytes
A buffer to cache bytes read in the ReadThread function.
size_t GetCachedBytes(void *dst, size_t dst_len)
Get any available bytes from our data cache.
bool ReadThreadIsRunning()
Checks if there is a currently running read thread.
lldb::thread_result_t ReadThread()
The read thread function.
void SetReadThreadBytesReceivedCallback(ReadThreadBytesReceived callback, void *callback_baton)
A class that represents a running process on the host machine.
Definition: SBAttachInfo.h:14
Log * GetLog(Cat mask)
Retrieve the Log object for the channel associated with the given log enum.
Definition: Log.h:314
Definition: SBAddress.h:15
void * thread_result_t
Definition: lldb-types.h:62
ConnectionStatus
Connection Status Types.
@ eConnectionStatusError
Check GetError() for details.
@ eConnectionStatusInterrupted
Interrupted read.
@ eConnectionStatusTimedOut
Request timed out.
@ eConnectionStatusEndOfFile
End-of-file encountered.
@ eConnectionStatusSuccess
Success.
@ eConnectionStatusLostConnection
Lost connection while connected to a valid connection.
@ eConnectionStatusNoConnection
No connection.
@ eErrorTypePOSIX
POSIX error codes.
std::shared_ptr< lldb_private::Event > EventSP
Definition: lldb-forward.h:337
std::shared_ptr< lldb_private::Listener > ListenerSP
Definition: lldb-forward.h:360