Communication.cpp revision 360784 
1//===-- Communication.cpp ---------------------------------------*- C++ -*-===//
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  Disconnect(nullptr);
75  StopReadThread(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  lldb::ConnectionSP connection_sp(m_connection_sp);
97  if (connection_sp) {
98    ConnectionStatus status = connection_sp->Disconnect(error_ptr);
99    // We currently don't protect connection_sp with any mutex for multi-
100    // threaded environments. So lets not nuke our connection class without
101    // putting some multi-threaded protections in. We also probably don't want
102    // to pay for the overhead it might cause if every time we access the
103    // connection we have to take a lock.
104    //
105    // This unique pointer will cleanup after itself when this object goes
106    // away, so there is no need to currently have it destroy itself
107    // immediately upon disconnect.
108    // connection_sp.reset();
109    return status;
110  }
111  return eConnectionStatusNoConnection;
112}
113
114bool Communication::IsConnected() const {
115  lldb::ConnectionSP connection_sp(m_connection_sp);
116  return (connection_sp ? connection_sp->IsConnected() : false);
117}
118
119bool Communication::HasConnection() const {
120  return m_connection_sp.get() != nullptr;
121}
122
123size_t Communication::Read(void *dst, size_t dst_len,
124                           const Timeout<std::micro> &timeout,
125                           ConnectionStatus &status, Status *error_ptr) {
126  Log *log = GetLogIfAllCategoriesSet(LIBLLDB_LOG_COMMUNICATION);
127  LLDB_LOG(
128      log,
129      "this = {0}, dst = {1}, dst_len = {2}, timeout = {3}, connection = {4}",
130      this, dst, dst_len, timeout, m_connection_sp.get());
131
132  if (m_read_thread_enabled) {
133    // We have a dedicated read thread that is getting data for us
134    size_t cached_bytes = GetCachedBytes(dst, dst_len);
135    if (cached_bytes > 0 || (timeout && timeout->count() == 0)) {
136      status = eConnectionStatusSuccess;
137      return cached_bytes;
138    }
139
140    if (!m_connection_sp) {
141      if (error_ptr)
142        error_ptr->SetErrorString("Invalid connection.");
143      status = eConnectionStatusNoConnection;
144      return 0;
145    }
146
147    ListenerSP listener_sp(Listener::MakeListener("Communication::Read"));
148    listener_sp->StartListeningForEvents(
149        this, eBroadcastBitReadThreadGotBytes | eBroadcastBitReadThreadDidExit);
150    EventSP event_sp;
151    while (listener_sp->GetEvent(event_sp, timeout)) {
152      const uint32_t event_type = event_sp->GetType();
153      if (event_type & eBroadcastBitReadThreadGotBytes) {
154        return GetCachedBytes(dst, dst_len);
155      }
156
157      if (event_type & eBroadcastBitReadThreadDidExit) {
158        if (GetCloseOnEOF())
159          Disconnect(nullptr);
160        break;
161      }
162    }
163    return 0;
164  }
165
166  // We aren't using a read thread, just read the data synchronously in this
167  // thread.
168  return ReadFromConnection(dst, dst_len, timeout, status, error_ptr);
169}
170
171size_t Communication::Write(const void *src, size_t src_len,
172                            ConnectionStatus &status, Status *error_ptr) {
173  lldb::ConnectionSP connection_sp(m_connection_sp);
174
175  std::lock_guard<std::mutex> guard(m_write_mutex);
176  LLDB_LOG(lldb_private::GetLogIfAllCategoriesSet(LIBLLDB_LOG_COMMUNICATION),
177           "{0} Communication::Write (src = {1}, src_len = %" PRIu64
178           ") connection = {2}",
179           this, src, (uint64_t)src_len, connection_sp.get());
180
181  if (connection_sp)
182    return connection_sp->Write(src, src_len, status, error_ptr);
183
184  if (error_ptr)
185    error_ptr->SetErrorString("Invalid connection.");
186  status = eConnectionStatusNoConnection;
187  return 0;
188}
189
190bool Communication::StartReadThread(Status *error_ptr) {
191  if (error_ptr)
192    error_ptr->Clear();
193
194  if (m_read_thread.IsJoinable())
195    return true;
196
197  LLDB_LOG(lldb_private::GetLogIfAllCategoriesSet(LIBLLDB_LOG_COMMUNICATION),
198           "{0} Communication::StartReadThread ()", this);
199
200  char thread_name[1024];
201  snprintf(thread_name, sizeof(thread_name), "<lldb.comm.%s>",
202           GetBroadcasterName().AsCString());
203
204  m_read_thread_enabled = true;
205  m_read_thread_did_exit = false;
206  auto maybe_thread = ThreadLauncher::LaunchThread(
207      thread_name, Communication::ReadThread, this);
208  if (maybe_thread) {
209    m_read_thread = *maybe_thread;
210  } else {
211    if (error_ptr)
212      *error_ptr = Status(maybe_thread.takeError());
213    else {
214      LLDB_LOG(lldb_private::GetLogIfAllCategoriesSet(LIBLLDB_LOG_HOST),
215               "failed to launch host thread: {}",
216               llvm::toString(maybe_thread.takeError()));
217    }
218  }
219
220  if (!m_read_thread.IsJoinable())
221    m_read_thread_enabled = false;
222
223  return m_read_thread_enabled;
224}
225
226bool Communication::StopReadThread(Status *error_ptr) {
227  if (!m_read_thread.IsJoinable())
228    return true;
229
230  LLDB_LOG(lldb_private::GetLogIfAllCategoriesSet(LIBLLDB_LOG_COMMUNICATION),
231           "{0} Communication::StopReadThread ()", this);
232
233  m_read_thread_enabled = false;
234
235  BroadcastEvent(eBroadcastBitReadThreadShouldExit, nullptr);
236
237  // error = m_read_thread.Cancel();
238
239  Status error = m_read_thread.Join(nullptr);
240  return error.Success();
241}
242
243bool Communication::JoinReadThread(Status *error_ptr) {
244  if (!m_read_thread.IsJoinable())
245    return true;
246
247  Status error = m_read_thread.Join(nullptr);
248  return error.Success();
249}
250
251size_t Communication::GetCachedBytes(void *dst, size_t dst_len) {
252  std::lock_guard<std::recursive_mutex> guard(m_bytes_mutex);
253  if (!m_bytes.empty()) {
254    // If DST is nullptr and we have a thread, then return the number of bytes
255    // that are available so the caller can call again
256    if (dst == nullptr)
257      return m_bytes.size();
258
259    const size_t len = std::min<size_t>(dst_len, m_bytes.size());
260
261    ::memcpy(dst, m_bytes.c_str(), len);
262    m_bytes.erase(m_bytes.begin(), m_bytes.begin() + len);
263
264    return len;
265  }
266  return 0;
267}
268
269void Communication::AppendBytesToCache(const uint8_t *bytes, size_t len,
270                                       bool broadcast,
271                                       ConnectionStatus status) {
272  LLDB_LOG(lldb_private::GetLogIfAllCategoriesSet(LIBLLDB_LOG_COMMUNICATION),
273           "{0} Communication::AppendBytesToCache (src = {1}, src_len = {2}, "
274           "broadcast = {3})",
275           this, bytes, (uint64_t)len, broadcast);
276  if ((bytes == nullptr || len == 0) &&
277      (status != lldb::eConnectionStatusEndOfFile))
278    return;
279  if (m_callback) {
280    // If the user registered a callback, then call it and do not broadcast
281    m_callback(m_callback_baton, bytes, len);
282  } else if (bytes != nullptr && len > 0) {
283    std::lock_guard<std::recursive_mutex> guard(m_bytes_mutex);
284    m_bytes.append((const char *)bytes, len);
285    if (broadcast)
286      BroadcastEventIfUnique(eBroadcastBitReadThreadGotBytes);
287  }
288}
289
290size_t Communication::ReadFromConnection(void *dst, size_t dst_len,
291                                         const Timeout<std::micro> &timeout,
292                                         ConnectionStatus &status,
293                                         Status *error_ptr) {
294  lldb::ConnectionSP connection_sp(m_connection_sp);
295  if (connection_sp)
296    return connection_sp->Read(dst, dst_len, timeout, status, error_ptr);
297
298  if (error_ptr)
299    error_ptr->SetErrorString("Invalid connection.");
300  status = eConnectionStatusNoConnection;
301  return 0;
302}
303
304bool Communication::ReadThreadIsRunning() { return m_read_thread_enabled; }
305
306lldb::thread_result_t Communication::ReadThread(lldb::thread_arg_t p) {
307  Communication *comm = (Communication *)p;
308
309  Log *log(lldb_private::GetLogIfAllCategoriesSet(LIBLLDB_LOG_COMMUNICATION));
310
311  LLDB_LOGF(log, "%p Communication::ReadThread () thread starting...", p);
312
313  uint8_t buf[1024];
314
315  Status error;
316  ConnectionStatus status = eConnectionStatusSuccess;
317  bool done = false;
318  while (!done && comm->m_read_thread_enabled) {
319    size_t bytes_read = comm->ReadFromConnection(
320        buf, sizeof(buf), std::chrono::seconds(5), status, &error);
321    if (bytes_read > 0)
322      comm->AppendBytesToCache(buf, bytes_read, true, status);
323    else if ((bytes_read == 0) && status == eConnectionStatusEndOfFile) {
324      if (comm->GetCloseOnEOF())
325        comm->Disconnect();
326      comm->AppendBytesToCache(buf, bytes_read, true, status);
327    }
328
329    switch (status) {
330    case eConnectionStatusSuccess:
331      break;
332
333    case eConnectionStatusEndOfFile:
334      done = true;
335      break;
336    case eConnectionStatusError: // Check GetError() for details
337      if (error.GetType() == eErrorTypePOSIX && error.GetError() == EIO) {
338        // EIO on a pipe is usually caused by remote shutdown
339        comm->Disconnect();
340        done = true;
341      }
342      if (error.Fail())
343        LLDB_LOG(log, "error: {0}, status = {1}", error,
344                 Communication::ConnectionStatusAsCString(status));
345      break;
346    case eConnectionStatusInterrupted: // Synchronization signal from
347                                       // SynchronizeWithReadThread()
348      // The connection returns eConnectionStatusInterrupted only when there is
349      // no input pending to be read, so we can signal that.
350      comm->BroadcastEvent(eBroadcastBitNoMorePendingInput);
351      break;
352    case eConnectionStatusNoConnection:   // No connection
353    case eConnectionStatusLostConnection: // Lost connection while connected to
354                                          // a valid connection
355      done = true;
356      LLVM_FALLTHROUGH;
357    case eConnectionStatusTimedOut: // Request timed out
358      if (error.Fail())
359        LLDB_LOG(log, "error: {0}, status = {1}", error,
360                 Communication::ConnectionStatusAsCString(status));
361      break;
362    }
363  }
364  log = lldb_private::GetLogIfAllCategoriesSet(LIBLLDB_LOG_COMMUNICATION);
365  if (log)
366    LLDB_LOGF(log, "%p Communication::ReadThread () thread exiting...", p);
367
368  comm->m_read_thread_did_exit = true;
369  // Let clients know that this thread is exiting
370  comm->BroadcastEvent(eBroadcastBitNoMorePendingInput);
371  comm->BroadcastEvent(eBroadcastBitReadThreadDidExit);
372  return {};
373}
374
375void Communication::SetReadThreadBytesReceivedCallback(
376    ReadThreadBytesReceived callback, void *callback_baton) {
377  m_callback = callback;
378  m_callback_baton = callback_baton;
379}
380
381void Communication::SynchronizeWithReadThread() {
382  // Only one thread can do the synchronization dance at a time.
383  std::lock_guard<std::mutex> guard(m_synchronize_mutex);
384
385  // First start listening for the synchronization event.
386  ListenerSP listener_sp(
387      Listener::MakeListener("Communication::SyncronizeWithReadThread"));
388  listener_sp->StartListeningForEvents(this, eBroadcastBitNoMorePendingInput);
389
390  // If the thread is not running, there is no point in synchronizing.
391  if (!m_read_thread_enabled || m_read_thread_did_exit)
392    return;
393
394  // Notify the read thread.
395  m_connection_sp->InterruptRead();
396
397  // Wait for the synchronization event.
398  EventSP event_sp;
399  listener_sp->GetEvent(event_sp, llvm::None);
400}
401
402void Communication::SetConnection(Connection *connection) {
403  Disconnect(nullptr);
404  StopReadThread(nullptr);
405  m_connection_sp.reset(connection);
406}
407
408const char *
409Communication::ConnectionStatusAsCString(lldb::ConnectionStatus status) {
410  switch (status) {
411  case eConnectionStatusSuccess:
412    return "success";
413  case eConnectionStatusError:
414    return "error";
415  case eConnectionStatusTimedOut:
416    return "timed out";
417  case eConnectionStatusNoConnection:
418    return "no connection";
419  case eConnectionStatusLostConnection:
420    return "lost connection";
421  case eConnectionStatusEndOfFile:
422    return "end of file";
423  case eConnectionStatusInterrupted:
424    return "interrupted";
425  }
426
427  static char unknown_state_string[64];
428  snprintf(unknown_state_string, sizeof(unknown_state_string),
429           "ConnectionStatus = %i", status);
430  return unknown_state_string;
431}
432