1//===-- Communication.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
9#include "lldb/Core/Communication.h"
10
11#include "lldb/Host/HostThread.h"
12#include "lldb/Host/ThreadLauncher.h"
13#include "lldb/Utility/Connection.h"
14#include "lldb/Utility/ConstString.h"
15#include "lldb/Utility/Event.h"
16#include "lldb/Utility/Listener.h"
17#include "lldb/Utility/Log.h"
18#include "lldb/Utility/Logging.h"
19#include "lldb/Utility/Status.h"
20
21#include "llvm/ADT/None.h"
22#include "llvm/ADT/Optional.h"
23#include "llvm/Support/Compiler.h"
24
25#include <algorithm>
26#include <chrono>
27#include <cstring>
28#include <memory>
29
30#include <errno.h>
31#include <inttypes.h>
32#include <stdio.h>
33
34using namespace lldb;
35using namespace lldb_private;
36
37ConstString &Communication::GetStaticBroadcasterClass() {
38 static ConstString class_name("lldb.communication");
39 return class_name;
40}
41
42Communication::Communication(const char *name)
43 : Broadcaster(nullptr, name), m_connection_sp(),
44 m_read_thread_enabled(false), m_read_thread_did_exit(false), m_bytes(),
45 m_bytes_mutex(), m_write_mutex(), m_synchronize_mutex(),
46 m_callback(nullptr), m_callback_baton(nullptr), m_close_on_eof(true)
47
48{
49
50 LLDB_LOG(lldb_private::GetLogIfAnyCategoriesSet(LIBLLDB_LOG_OBJECT |
51 LIBLLDB_LOG_COMMUNICATION),
52 "{0} Communication::Communication (name = {1})", this, name);
53
54 SetEventName(eBroadcastBitDisconnected, "disconnected");
55 SetEventName(eBroadcastBitReadThreadGotBytes, "got bytes");
56 SetEventName(eBroadcastBitReadThreadDidExit, "read thread did exit");
57 SetEventName(eBroadcastBitReadThreadShouldExit, "read thread should exit");
58 SetEventName(eBroadcastBitPacketAvailable, "packet available");
59 SetEventName(eBroadcastBitNoMorePendingInput, "no more pending input");
60
61 CheckInWithManager();
62}
63
64Communication::~Communication() {
65 LLDB_LOG(lldb_private::GetLogIfAnyCategoriesSet(LIBLLDB_LOG_OBJECT |
66 LIBLLDB_LOG_COMMUNICATION),
67 "{0} Communication::~Communication (name = {1})", this,
68 GetBroadcasterName().AsCString());
69 Clear();
70}
71
72void Communication::Clear() {
73 SetReadThreadBytesReceivedCallback(nullptr, nullptr);
74 StopReadThread(nullptr);
75 Disconnect(nullptr);
76}
77
78ConnectionStatus Communication::Connect(const char *url, Status *error_ptr) {
79 Clear();
80
81 LLDB_LOG(lldb_private::GetLogIfAllCategoriesSet(LIBLLDB_LOG_COMMUNICATION),
82 "{0} Communication::Connect (url = {1})", this, url);
83
84 lldb::ConnectionSP connection_sp(m_connection_sp);
85 if (connection_sp)
86 return connection_sp->Connect(url, error_ptr);
87 if (error_ptr)
88 error_ptr->SetErrorString("Invalid connection.");
89 return eConnectionStatusNoConnection;
90}
91
92ConnectionStatus Communication::Disconnect(Status *error_ptr) {
93 LLDB_LOG(lldb_private::GetLogIfAllCategoriesSet(LIBLLDB_LOG_COMMUNICATION),
94 "{0} Communication::Disconnect ()", this);
95
96 assert((!m_read_thread_enabled || m_read_thread_did_exit) &&
97 "Disconnecting while the read thread is running is racy!");
98 lldb::ConnectionSP connection_sp(m_connection_sp);
99 if (connection_sp) {
100 ConnectionStatus status = connection_sp->Disconnect(error_ptr);
101 // We currently don't protect connection_sp with any mutex for multi-
102 // threaded environments. So lets not nuke our connection class without
103 // putting some multi-threaded protections in. We also probably don't want
104 // to pay for the overhead it might cause if every time we access the
105 // connection we have to take a lock.
106 //
107 // This unique pointer will cleanup after itself when this object goes
108 // away, so there is no need to currently have it destroy itself
109 // immediately upon disconnect.
110 // connection_sp.reset();
111 return status;
112 }
113 return eConnectionStatusNoConnection;
114}
115
116bool Communication::IsConnected() const {
117 lldb::ConnectionSP connection_sp(m_connection_sp);
118 return (connection_sp ? connection_sp->IsConnected() : false);
119}
120
121bool Communication::HasConnection() const {
122 return m_connection_sp.get() != nullptr;
123}
124
125size_t Communication::Read(void *dst, size_t dst_len,
126 const Timeout<std::micro> &timeout,
127 ConnectionStatus &status, Status *error_ptr) {
128 Log *log = GetLogIfAllCategoriesSet(LIBLLDB_LOG_COMMUNICATION);
129 LLDB_LOG(
130 log,
131 "this = {0}, dst = {1}, dst_len = {2}, timeout = {3}, connection = {4}",
132 this, dst, dst_len, timeout, m_connection_sp.get());
133
134 if (m_read_thread_enabled) {
135 // We have a dedicated read thread that is getting data for us
136 size_t cached_bytes = GetCachedBytes(dst, dst_len);
137 if (cached_bytes > 0 || (timeout && timeout->count() == 0)) {
138 status = eConnectionStatusSuccess;
139 return cached_bytes;
140 }
141
142 if (!m_connection_sp) {
143 if (error_ptr)
144 error_ptr->SetErrorString("Invalid connection.");
145 status = eConnectionStatusNoConnection;
146 return 0;
147 }
148
149 ListenerSP listener_sp(Listener::MakeListener("Communication::Read"));
150 listener_sp->StartListeningForEvents(
151 this, eBroadcastBitReadThreadGotBytes | eBroadcastBitReadThreadDidExit);
152 EventSP event_sp;
153 while (listener_sp->GetEvent(event_sp, timeout)) {
154 const uint32_t event_type = event_sp->GetType();
155 if (event_type & eBroadcastBitReadThreadGotBytes) {
156 return GetCachedBytes(dst, dst_len);
157 }
158
159 if (event_type & eBroadcastBitReadThreadDidExit) {
160 if (GetCloseOnEOF())
161 Disconnect(nullptr);
162 break;
163 }
164 }
165 return 0;
166 }
167
168 // We aren't using a read thread, just read the data synchronously in this
169 // thread.
170 return ReadFromConnection(dst, dst_len, timeout, status, error_ptr);
171}
172
173size_t Communication::Write(const void *src, size_t src_len,
174 ConnectionStatus &status, Status *error_ptr) {
175 lldb::ConnectionSP connection_sp(m_connection_sp);
176
177 std::lock_guard<std::mutex> guard(m_write_mutex);
178 LLDB_LOG(lldb_private::GetLogIfAllCategoriesSet(LIBLLDB_LOG_COMMUNICATION),
179 "{0} Communication::Write (src = {1}, src_len = %" PRIu64
180 ") connection = {2}",
181 this, src, (uint64_t)src_len, connection_sp.get());
182
183 if (connection_sp)
184 return connection_sp->Write(src, src_len, status, error_ptr);
185
186 if (error_ptr)
187 error_ptr->SetErrorString("Invalid connection.");
188 status = eConnectionStatusNoConnection;
189 return 0;
190}
191
192bool Communication::StartReadThread(Status *error_ptr) {
193 if (error_ptr)
194 error_ptr->Clear();
195
196 if (m_read_thread.IsJoinable())
197 return true;
198
199 LLDB_LOG(lldb_private::GetLogIfAllCategoriesSet(LIBLLDB_LOG_COMMUNICATION),
200 "{0} Communication::StartReadThread ()", this);
201
202 const std::string thread_name =
203 llvm::formatv("<lldb.comm.{0}>", GetBroadcasterName());
204
205 m_read_thread_enabled = true;
206 m_read_thread_did_exit = false;
207 auto maybe_thread = ThreadLauncher::LaunchThread(
208 thread_name, Communication::ReadThread, this);
209 if (maybe_thread) {
210 m_read_thread = *maybe_thread;
211 } else {
212 if (error_ptr)
213 *error_ptr = Status(maybe_thread.takeError());
214 else {
215 LLDB_LOG(lldb_private::GetLogIfAllCategoriesSet(LIBLLDB_LOG_HOST),
216 "failed to launch host thread: {}",
217 llvm::toString(maybe_thread.takeError()));
218 }
219 }
220
221 if (!m_read_thread.IsJoinable())
222 m_read_thread_enabled = false;
223
224 return m_read_thread_enabled;
225}
226
227bool Communication::StopReadThread(Status *error_ptr) {
228 if (!m_read_thread.IsJoinable())
229 return true;
230
231 LLDB_LOG(lldb_private::GetLogIfAllCategoriesSet(LIBLLDB_LOG_COMMUNICATION),
232 "{0} Communication::StopReadThread ()", this);
233
234 m_read_thread_enabled = false;
235
236 BroadcastEvent(eBroadcastBitReadThreadShouldExit, nullptr);
237
238 // error = m_read_thread.Cancel();
239
240 Status error = m_read_thread.Join(nullptr);
241 return error.Success();
242}
243
244bool Communication::JoinReadThread(Status *error_ptr) {
245 if (!m_read_thread.IsJoinable())
246 return true;
247
248 Status error = m_read_thread.Join(nullptr);
249 return error.Success();
250}
251
252size_t Communication::GetCachedBytes(void *dst, size_t dst_len) {
253 std::lock_guard<std::recursive_mutex> guard(m_bytes_mutex);
254 if (!m_bytes.empty()) {
255 // If DST is nullptr and we have a thread, then return the number of bytes
256 // that are available so the caller can call again
257 if (dst == nullptr)
258 return m_bytes.size();
259
260 const size_t len = std::min<size_t>(dst_len, m_bytes.size());
261
262 ::memcpy(dst, m_bytes.c_str(), len);
263 m_bytes.erase(m_bytes.begin(), m_bytes.begin() + len);
264
265 return len;
266 }
267 return 0;
268}
269
270void Communication::AppendBytesToCache(const uint8_t *bytes, size_t len,
271 bool broadcast,
272 ConnectionStatus status) {
273 LLDB_LOG(lldb_private::GetLogIfAllCategoriesSet(LIBLLDB_LOG_COMMUNICATION),
274 "{0} Communication::AppendBytesToCache (src = {1}, src_len = {2}, "
275 "broadcast = {3})",
276 this, bytes, (uint64_t)len, broadcast);
277 if ((bytes == nullptr || len == 0) &&
278 (status != lldb::eConnectionStatusEndOfFile))
279 return;
280 if (m_callback) {
281 // If the user registered a callback, then call it and do not broadcast
282 m_callback(m_callback_baton, bytes, len);
283 } else if (bytes != nullptr && len > 0) {
284 std::lock_guard<std::recursive_mutex> guard(m_bytes_mutex);
285 m_bytes.append((const char *)bytes, len);
286 if (broadcast)
287 BroadcastEventIfUnique(eBroadcastBitReadThreadGotBytes);
288 }
289}
290
291size_t Communication::ReadFromConnection(void *dst, size_t dst_len,
292 const Timeout<std::micro> &timeout,
293 ConnectionStatus &status,
294 Status *error_ptr) {
295 lldb::ConnectionSP connection_sp(m_connection_sp);
296 if (connection_sp)
297 return connection_sp->Read(dst, dst_len, timeout, status, error_ptr);
298
299 if (error_ptr)
300 error_ptr->SetErrorString("Invalid connection.");
301 status = eConnectionStatusNoConnection;
302 return 0;
303}
304
305bool Communication::ReadThreadIsRunning() { return m_read_thread_enabled; }
306
307lldb::thread_result_t Communication::ReadThread(lldb::thread_arg_t p) {
308 Communication *comm = (Communication *)p;
309
310 Log *log(lldb_private::GetLogIfAllCategoriesSet(LIBLLDB_LOG_COMMUNICATION));
311
312 LLDB_LOGF(log, "%p Communication::ReadThread () thread starting...", p);
313
314 uint8_t buf[1024];
315
316 Status error;
317 ConnectionStatus status = eConnectionStatusSuccess;
318 bool done = false;
319 bool disconnect = false;
320 while (!done && comm->m_read_thread_enabled) {
321 size_t bytes_read = comm->ReadFromConnection(
322 buf, sizeof(buf), std::chrono::seconds(5), status, &error);
323 if (bytes_read > 0 || status == eConnectionStatusEndOfFile)
324 comm->AppendBytesToCache(buf, bytes_read, true, status);
325
326 switch (status) {
327 case eConnectionStatusSuccess:
328 break;
329
330 case eConnectionStatusEndOfFile:
331 done = true;
332 disconnect = comm->GetCloseOnEOF();
333 break;
334 case eConnectionStatusError: // Check GetError() for details
335 if (error.GetType() == eErrorTypePOSIX && error.GetError() == EIO) {
336 // EIO on a pipe is usually caused by remote shutdown
337 disconnect = comm->GetCloseOnEOF();
338 done = true;
339 }
340 if (error.Fail())
341 LLDB_LOG(log, "error: {0}, status = {1}", error,
342 Communication::ConnectionStatusAsString(status));
343 break;
344 case eConnectionStatusInterrupted: // Synchronization signal from
345 // SynchronizeWithReadThread()
346 // The connection returns eConnectionStatusInterrupted only when there is
347 // no input pending to be read, so we can signal that.
348 comm->BroadcastEvent(eBroadcastBitNoMorePendingInput);
349 break;
350 case eConnectionStatusNoConnection: // No connection
351 case eConnectionStatusLostConnection: // Lost connection while connected to
352 // a valid connection
353 done = true;
354 LLVM_FALLTHROUGH;
355 case eConnectionStatusTimedOut: // Request timed out
356 if (error.Fail())
357 LLDB_LOG(log, "error: {0}, status = {1}", error,
358 Communication::ConnectionStatusAsString(status));
359 break;
360 }
361 }
362 log = lldb_private::GetLogIfAllCategoriesSet(LIBLLDB_LOG_COMMUNICATION);
363 if (log)
364 LLDB_LOGF(log, "%p Communication::ReadThread () thread exiting...", p);
365
366 // Handle threads wishing to synchronize with us.
367 {
368 // Prevent new ones from showing up.
369 comm->m_read_thread_did_exit = true;
370
371 // Unblock any existing thread waiting for the synchronization signal.
372 comm->BroadcastEvent(eBroadcastBitNoMorePendingInput);
373
374 // Wait for the thread to finish...
375 std::lock_guard<std::mutex> guard(comm->m_synchronize_mutex);
376 // ... and disconnect.
377 if (disconnect)
378 comm->Disconnect();
379 }
380
381 // Let clients know that this thread is exiting
382 comm->BroadcastEvent(eBroadcastBitReadThreadDidExit);
383 return {};
384}
385
386void Communication::SetReadThreadBytesReceivedCallback(
387 ReadThreadBytesReceived callback, void *callback_baton) {
388 m_callback = callback;
389 m_callback_baton = callback_baton;
390}
391
392void Communication::SynchronizeWithReadThread() {
393 // Only one thread can do the synchronization dance at a time.
394 std::lock_guard<std::mutex> guard(m_synchronize_mutex);
395
396 // First start listening for the synchronization event.
397 ListenerSP listener_sp(
398 Listener::MakeListener("Communication::SyncronizeWithReadThread"));
399 listener_sp->StartListeningForEvents(this, eBroadcastBitNoMorePendingInput);
400
401 // If the thread is not running, there is no point in synchronizing.
402 if (!m_read_thread_enabled || m_read_thread_did_exit)
403 return;
404
405 // Notify the read thread.
406 m_connection_sp->InterruptRead();
407
408 // Wait for the synchronization event.
409 EventSP event_sp;
410 listener_sp->GetEvent(event_sp, llvm::None);
411}
412
413void Communication::SetConnection(std::unique_ptr<Connection> connection) {
414 Disconnect(nullptr);
415 StopReadThread(nullptr);
416 m_connection_sp = std::move(connection);
417}
418
419std::string
420Communication::ConnectionStatusAsString(lldb::ConnectionStatus status) {
421 switch (status) {
422 case eConnectionStatusSuccess:
423 return "success";
424 case eConnectionStatusError:
425 return "error";
426 case eConnectionStatusTimedOut:
427 return "timed out";
428 case eConnectionStatusNoConnection:
429 return "no connection";
430 case eConnectionStatusLostConnection:
431 return "lost connection";
432 case eConnectionStatusEndOfFile:
433 return "end of file";
434 case eConnectionStatusInterrupted:
435 return "interrupted";
436 }
437
438 return "@" + std::to_string(status);
439}
440