LLDB  mainline
Communication.cpp
Go to the documentation of this file.
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 
10 
11 #include "lldb/Host/HostThread.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 
34 using namespace lldb;
35 using namespace lldb_private;
36 
37 ConstString &Communication::GetStaticBroadcasterClass() {
38  static ConstString class_name("lldb.communication");
39  return class_name;
40 }
41 
42 Communication::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 
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 
62 }
63 
67  "{0} Communication::~Communication (name = {1})", this,
68  GetBroadcasterName().AsCString());
69  Clear();
70 }
71 
73  SetReadThreadBytesReceivedCallback(nullptr, nullptr);
74  StopReadThread(nullptr);
75  Disconnect(nullptr);
76 }
77 
78 ConnectionStatus Communication::Connect(const char *url, Status *error_ptr) {
79  Clear();
80 
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.");
90 }
91 
94  "{0} Communication::Disconnect ()", this);
95 
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  }
114 }
115 
117  lldb::ConnectionSP connection_sp(m_connection_sp);
118  return (connection_sp ? connection_sp->IsConnected() : false);
119 }
120 
122  return m_connection_sp.get() != nullptr;
123 }
124 
125 size_t Communication::Read(void *dst, size_t dst_len,
126  const Timeout<std::micro> &timeout,
127  ConnectionStatus &status, Status *error_ptr) {
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.");
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 
173 size_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);
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.");
189  return 0;
190 }
191 
193  if (error_ptr)
194  error_ptr->Clear();
195 
197  return true;
198 
200  "{0} Communication::StartReadThread ()", this);
201 
202  char thread_name[1024];
203  snprintf(thread_name, sizeof(thread_name), "<lldb.comm.%s>",
204  GetBroadcasterName().AsCString());
205 
206  m_read_thread_enabled = true;
207  m_read_thread_did_exit = false;
208  auto maybe_thread = ThreadLauncher::LaunchThread(
209  thread_name, Communication::ReadThread, this);
210  if (maybe_thread) {
211  m_read_thread = *maybe_thread;
212  } else {
213  if (error_ptr)
214  *error_ptr = Status(maybe_thread.takeError());
215  else {
217  "failed to launch host thread: {}",
218  llvm::toString(maybe_thread.takeError()));
219  }
220  }
221 
222  if (!m_read_thread.IsJoinable())
223  m_read_thread_enabled = false;
224 
225  return m_read_thread_enabled;
226 }
227 
229  if (!m_read_thread.IsJoinable())
230  return true;
231 
233  "{0} Communication::StopReadThread ()", this);
234 
235  m_read_thread_enabled = false;
236 
237  BroadcastEvent(eBroadcastBitReadThreadShouldExit, nullptr);
238 
239  // error = m_read_thread.Cancel();
240 
241  Status error = m_read_thread.Join(nullptr);
242  return error.Success();
243 }
244 
246  if (!m_read_thread.IsJoinable())
247  return true;
248 
249  Status error = m_read_thread.Join(nullptr);
250  return error.Success();
251 }
252 
253 size_t Communication::GetCachedBytes(void *dst, size_t dst_len) {
254  std::lock_guard<std::recursive_mutex> guard(m_bytes_mutex);
255  if (!m_bytes.empty()) {
256  // If DST is nullptr and we have a thread, then return the number of bytes
257  // that are available so the caller can call again
258  if (dst == nullptr)
259  return m_bytes.size();
260 
261  const size_t len = std::min<size_t>(dst_len, m_bytes.size());
262 
263  ::memcpy(dst, m_bytes.c_str(), len);
264  m_bytes.erase(m_bytes.begin(), m_bytes.begin() + len);
265 
266  return len;
267  }
268  return 0;
269 }
270 
271 void Communication::AppendBytesToCache(const uint8_t *bytes, size_t len,
272  bool broadcast,
273  ConnectionStatus status) {
275  "{0} Communication::AppendBytesToCache (src = {1}, src_len = {2}, "
276  "broadcast = {3})",
277  this, bytes, (uint64_t)len, broadcast);
278  if ((bytes == nullptr || len == 0) &&
280  return;
281  if (m_callback) {
282  // If the user registered a callback, then call it and do not broadcast
283  m_callback(m_callback_baton, bytes, len);
284  } else if (bytes != nullptr && len > 0) {
285  std::lock_guard<std::recursive_mutex> guard(m_bytes_mutex);
286  m_bytes.append((const char *)bytes, len);
287  if (broadcast)
288  BroadcastEventIfUnique(eBroadcastBitReadThreadGotBytes);
289  }
290 }
291 
292 size_t Communication::ReadFromConnection(void *dst, size_t dst_len,
293  const Timeout<std::micro> &timeout,
294  ConnectionStatus &status,
295  Status *error_ptr) {
296  lldb::ConnectionSP connection_sp(m_connection_sp);
297  if (connection_sp)
298  return connection_sp->Read(dst, dst_len, timeout, status, error_ptr);
299 
300  if (error_ptr)
301  error_ptr->SetErrorString("Invalid connection.");
303  return 0;
304 }
305 
307 
309  Communication *comm = (Communication *)p;
310 
312 
313  LLDB_LOGF(log, "%p Communication::ReadThread () thread starting...", p);
314 
315  uint8_t buf[1024];
316 
317  Status error;
319  bool done = false;
320  bool disconnect = false;
321  while (!done && comm->m_read_thread_enabled) {
322  size_t bytes_read = comm->ReadFromConnection(
323  buf, sizeof(buf), std::chrono::seconds(5), status, &error);
324  if (bytes_read > 0 || status == eConnectionStatusEndOfFile)
325  comm->AppendBytesToCache(buf, bytes_read, true, status);
326 
327  switch (status) {
329  break;
330 
332  done = true;
333  disconnect = comm->GetCloseOnEOF();
334  break;
335  case eConnectionStatusError: // Check GetError() for details
336  if (error.GetType() == eErrorTypePOSIX && error.GetError() == EIO) {
337  // EIO on a pipe is usually caused by remote shutdown
338  disconnect = comm->GetCloseOnEOF();
339  done = true;
340  }
341  if (error.Fail())
342  LLDB_LOG(log, "error: {0}, status = {1}", error,
344  break;
345  case eConnectionStatusInterrupted: // Synchronization signal from
346  // SynchronizeWithReadThread()
347  // The connection returns eConnectionStatusInterrupted only when there is
348  // no input pending to be read, so we can signal that.
349  comm->BroadcastEvent(eBroadcastBitNoMorePendingInput);
350  break;
351  case eConnectionStatusNoConnection: // No connection
352  case eConnectionStatusLostConnection: // Lost connection while connected to
353  // a valid connection
354  done = true;
355  LLVM_FALLTHROUGH;
356  case eConnectionStatusTimedOut: // Request timed out
357  if (error.Fail())
358  LLDB_LOG(log, "error: {0}, status = {1}", error,
360  break;
361  }
362  }
364  if (log)
365  LLDB_LOGF(log, "%p Communication::ReadThread () thread exiting...", p);
366 
367  // Handle threads wishing to synchronize with us.
368  {
369  // Prevent new ones from showing up.
370  comm->m_read_thread_did_exit = true;
371 
372  // Unblock any existing thread waiting for the synchronization signal.
373  comm->BroadcastEvent(eBroadcastBitNoMorePendingInput);
374 
375  // Wait for the thread to finish...
376  std::lock_guard<std::mutex> guard(comm->m_synchronize_mutex);
377  // ... and disconnect.
378  if (disconnect)
379  comm->Disconnect();
380  }
381 
382  // Let clients know that this thread is exiting
383  comm->BroadcastEvent(eBroadcastBitReadThreadDidExit);
384  return {};
385 }
386 
388  ReadThreadBytesReceived callback, void *callback_baton) {
389  m_callback = callback;
390  m_callback_baton = callback_baton;
391 }
392 
394  // Only one thread can do the synchronization dance at a time.
395  std::lock_guard<std::mutex> guard(m_synchronize_mutex);
396 
397  // First start listening for the synchronization event.
398  ListenerSP listener_sp(
399  Listener::MakeListener("Communication::SyncronizeWithReadThread"));
400  listener_sp->StartListeningForEvents(this, eBroadcastBitNoMorePendingInput);
401 
402  // If the thread is not running, there is no point in synchronizing.
404  return;
405 
406  // Notify the read thread.
407  m_connection_sp->InterruptRead();
408 
409  // Wait for the synchronization event.
410  EventSP event_sp;
411  listener_sp->GetEvent(event_sp, llvm::None);
412 }
413 
414 void Communication::SetConnection(std::unique_ptr<Connection> connection) {
415  Disconnect(nullptr);
416  StopReadThread(nullptr);
417  m_connection_sp = std::move(connection);
418 }
419 
420 const char *
422  switch (status) {
424  return "success";
426  return "error";
428  return "timed out";
430  return "no connection";
432  return "lost connection";
434  return "end of file";
436  return "interrupted";
437  }
438 
439  static char unknown_state_string[64];
440  snprintf(unknown_state_string, sizeof(unknown_state_string),
441  "ConnectionStatus = %i", status);
442  return unknown_state_string;
443 }
void SynchronizeWithReadThread()
Wait for the read thread to process all outstanding data.
void * thread_arg_t
Definition: lldb-types.h:61
#define LIBLLDB_LOG_OBJECT
Definition: Logging.h:25
void SetEventName(uint32_t event_mask, const char *name)
Set the name for an event bit.
Definition: Broadcaster.h:337
A class that represents a running process on the host machine.
virtual bool StopReadThread(Status *error_ptr=nullptr)
Stops the read thread by cancelling it.
lldb::ConnectionStatus Disconnect(Status *error_ptr=nullptr)
Disconnect the communications connection if one is currently connected.
virtual bool StartReadThread(Status *error_ptr=nullptr)
Starts a read thread whose sole purpose it to read bytes from the current connection.
std::mutex m_write_mutex
Don&#39;t let multiple threads write at the same time...
HostThread m_read_thread
The read thread handle in case we need to cancel the thread.
void(* ReadThreadBytesReceived)(void *baton, const void *src, size_t src_len)
void SetConnection(std::unique_ptr< Connection > connection)
Sets the connection that it to be used by this class.
End-of-file encountered.
lldb::ConnectionSP m_connection_sp
The connection that is current in use by this communications class.
An event broadcasting class.
Definition: Broadcaster.h:242
virtual bool JoinReadThread(Status *error_ptr=nullptr)
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...
POSIX error codes.
#define LLDB_LOG(log,...)
The LLDB_LOG* macros defined below are the way to emit log messages.
Definition: Log.h:242
void BroadcastEvent(lldb::EventSP &event_sp)
Broadcast an event which has no associated data.
Definition: Broadcaster.h:262
static lldb::ListenerSP MakeListener(const char *name)
Definition: Listener.cpp:464
void BroadcastEventIfUnique(lldb::EventSP &event_sp)
Definition: Broadcaster.h:266
#define LIBLLDB_LOG_HOST
Definition: Logging.h:28
llvm::StringRef toString(Record::Kind K)
static llvm::raw_ostream & error(Stream &strm)
Log * GetLogIfAllCategoriesSet(uint32_t mask)
Definition: Logging.cpp:58
ReadThreadBytesReceived m_callback
void Clear()
Clear the object state.
Definition: Status.cpp:168
static lldb::thread_result_t ReadThread(lldb::thread_arg_t comm_ptr)
The static read thread function.
std::atomic< bool > m_read_thread_did_exit
~Communication() override
Destructor.
void SetErrorString(llvm::StringRef err_str)
Set the current error string to err_str.
Definition: Status.cpp:242
std::atomic< bool > m_read_thread_enabled
static llvm::Expected< HostThread > LaunchThread(llvm::StringRef name, lldb::thread_func_t thread_function, lldb::thread_arg_t thread_arg, size_t min_stack_byte_size=0)
bool Success() const
Test for success condition.
Definition: Status.cpp:288
Status Join(lldb::thread_result_t *result)
Definition: HostThread.cpp:20
void SetReadThreadBytesReceivedCallback(ReadThreadBytesReceived callback, void *callback_baton)
bool IsConnected() const
Check if the connection is valid.
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.
An abstract communications class.
Definition: Communication.h:84
ConstString GetBroadcasterName()
Get the NULL terminated C string name of this Broadcaster object.
Definition: Broadcaster.h:317
#define LLDB_LOGF(log,...)
Definition: Log.h:249
lldb::ErrorType GetType() const
Access the error type.
Definition: Status.cpp:178
Check GetError() for details.
A uniqued constant string class.
Definition: ConstString.h:40
size_t ReadFromConnection(void *dst, size_t dst_len, const Timeout< std::micro > &timeout, lldb::ConnectionStatus &status, Status *error_ptr)
bool Fail() const
Test for error condition.
Definition: Status.cpp:182
size_t GetCachedBytes(void *dst, size_t dst_len)
Get any available bytes from our data cache.
Log * GetLogIfAnyCategoriesSet(uint32_t mask)
Definition: Logging.cpp:62
Definition: SBAddress.h:15
std::string m_bytes
A buffer to cache bytes read in the ReadThread function.
Lost connection while connected to a valid connection.
bool ReadThreadIsRunning()
Checks if there is a currently running read thread.
#define LIBLLDB_LOG_COMMUNICATION
Definition: Logging.h:26
ConnectionStatus
Connection Status Types.
ValueType GetError() const
Access the error value.
Definition: Status.cpp:175
std::recursive_mutex m_bytes_mutex
A mutex to protect multi-threaded access to the cached bytes.
lldb::ConnectionStatus Connect(const char *url, Status *error_ptr)
Connect using the current connection by passing url to its connect function.
static const char * ConnectionStatusAsCString(lldb::ConnectionStatus status)
An error handling class.
Definition: Status.h:44
void * thread_result_t
Definition: lldb-types.h:62
size_t Write(const void *src, size_t src_len, lldb::ConnectionStatus &status, Status *error_ptr)
The actual write function that attempts to write to the communications protocol.