• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
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 
34 using namespace lldb;
35 using namespace lldb_private;
36 
GetStaticBroadcasterClass()37 ConstString &Communication::GetStaticBroadcasterClass() {
38   static ConstString class_name("lldb.communication");
39   return class_name;
40 }
41 
Communication(const char * name)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 
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 
~Communication()64 Communication::~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 
Clear()72 void Communication::Clear() {
73   SetReadThreadBytesReceivedCallback(nullptr, nullptr);
74   StopReadThread(nullptr);
75   Disconnect(nullptr);
76 }
77 
Connect(const char * url,Status * error_ptr)78 ConnectionStatus 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 
Disconnect(Status * error_ptr)92 ConnectionStatus 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 
IsConnected() const116 bool Communication::IsConnected() const {
117   lldb::ConnectionSP connection_sp(m_connection_sp);
118   return (connection_sp ? connection_sp->IsConnected() : false);
119 }
120 
HasConnection() const121 bool Communication::HasConnection() const {
122   return m_connection_sp.get() != nullptr;
123 }
124 
Read(void * dst,size_t dst_len,const Timeout<std::micro> & timeout,ConnectionStatus & status,Status * error_ptr)125 size_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 
Write(const void * src,size_t src_len,ConnectionStatus & status,Status * error_ptr)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);
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 
StartReadThread(Status * error_ptr)192 bool 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 
StopReadThread(Status * error_ptr)227 bool 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 
JoinReadThread(Status * error_ptr)244 bool 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 
GetCachedBytes(void * dst,size_t dst_len)252 size_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 
AppendBytesToCache(const uint8_t * bytes,size_t len,bool broadcast,ConnectionStatus status)270 void 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 
ReadFromConnection(void * dst,size_t dst_len,const Timeout<std::micro> & timeout,ConnectionStatus & status,Status * error_ptr)291 size_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 
ReadThreadIsRunning()305 bool Communication::ReadThreadIsRunning() { return m_read_thread_enabled; }
306 
ReadThread(lldb::thread_arg_t p)307 lldb::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 
SetReadThreadBytesReceivedCallback(ReadThreadBytesReceived callback,void * callback_baton)386 void Communication::SetReadThreadBytesReceivedCallback(
387     ReadThreadBytesReceived callback, void *callback_baton) {
388   m_callback = callback;
389   m_callback_baton = callback_baton;
390 }
391 
SynchronizeWithReadThread()392 void 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 
SetConnection(std::unique_ptr<Connection> connection)413 void Communication::SetConnection(std::unique_ptr<Connection> connection) {
414   Disconnect(nullptr);
415   StopReadThread(nullptr);
416   m_connection_sp = std::move(connection);
417 }
418 
419 std::string
ConnectionStatusAsString(lldb::ConnectionStatus status)420 Communication::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