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 
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   lldb_private::LogIfAnyCategoriesSet(
50       LIBLLDB_LOG_OBJECT | LIBLLDB_LOG_COMMUNICATION,
51       "%p Communication::Communication (name = %s)", this, name);
52 
53   SetEventName(eBroadcastBitDisconnected, "disconnected");
54   SetEventName(eBroadcastBitReadThreadGotBytes, "got bytes");
55   SetEventName(eBroadcastBitReadThreadDidExit, "read thread did exit");
56   SetEventName(eBroadcastBitReadThreadShouldExit, "read thread should exit");
57   SetEventName(eBroadcastBitPacketAvailable, "packet available");
58   SetEventName(eBroadcastBitNoMorePendingInput, "no more pending input");
59 
60   CheckInWithManager();
61 }
62 
63 Communication::~Communication() {
64   lldb_private::LogIfAnyCategoriesSet(
65       LIBLLDB_LOG_OBJECT | LIBLLDB_LOG_COMMUNICATION,
66       "%p Communication::~Communication (name = %s)", this,
67       GetBroadcasterName().AsCString());
68   Clear();
69 }
70 
71 void Communication::Clear() {
72   SetReadThreadBytesReceivedCallback(nullptr, nullptr);
73   Disconnect(nullptr);
74   StopReadThread(nullptr);
75 }
76 
77 ConnectionStatus Communication::Connect(const char *url, Status *error_ptr) {
78   Clear();
79 
80   lldb_private::LogIfAnyCategoriesSet(LIBLLDB_LOG_COMMUNICATION,
81                                       "%p Communication::Connect (url = %s)",
82                                       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 
92 ConnectionStatus Communication::Disconnect(Status *error_ptr) {
93   lldb_private::LogIfAnyCategoriesSet(LIBLLDB_LOG_COMMUNICATION,
94                                       "%p 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 
114 bool Communication::IsConnected() const {
115   lldb::ConnectionSP connection_sp(m_connection_sp);
116   return (connection_sp ? connection_sp->IsConnected() : false);
117 }
118 
119 bool Communication::HasConnection() const {
120   return m_connection_sp.get() != nullptr;
121 }
122 
123 size_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 
171 size_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_private::LogIfAnyCategoriesSet(
177       LIBLLDB_LOG_COMMUNICATION,
178       "%p Communication::Write (src = %p, src_len = %" PRIu64
179       ") connection = %p",
180       this, src, (uint64_t)src_len, connection_sp.get());
181 
182   if (connection_sp)
183     return connection_sp->Write(src, src_len, status, error_ptr);
184 
185   if (error_ptr)
186     error_ptr->SetErrorString("Invalid connection.");
187   status = eConnectionStatusNoConnection;
188   return 0;
189 }
190 
191 bool Communication::StartReadThread(Status *error_ptr) {
192   if (error_ptr)
193     error_ptr->Clear();
194 
195   if (m_read_thread.IsJoinable())
196     return true;
197 
198   lldb_private::LogIfAnyCategoriesSet(
199       LIBLLDB_LOG_COMMUNICATION, "%p Communication::StartReadThread ()", this);
200 
201   char thread_name[1024];
202   snprintf(thread_name, sizeof(thread_name), "<lldb.comm.%s>",
203            GetBroadcasterName().AsCString());
204 
205   m_read_thread_enabled = true;
206   m_read_thread_did_exit = false;
207   m_read_thread = ThreadLauncher::LaunchThread(
208       thread_name, Communication::ReadThread, this, error_ptr);
209   if (!m_read_thread.IsJoinable())
210     m_read_thread_enabled = false;
211   return m_read_thread_enabled;
212 }
213 
214 bool Communication::StopReadThread(Status *error_ptr) {
215   if (!m_read_thread.IsJoinable())
216     return true;
217 
218   lldb_private::LogIfAnyCategoriesSet(
219       LIBLLDB_LOG_COMMUNICATION, "%p Communication::StopReadThread ()", this);
220 
221   m_read_thread_enabled = false;
222 
223   BroadcastEvent(eBroadcastBitReadThreadShouldExit, nullptr);
224 
225   // error = m_read_thread.Cancel();
226 
227   Status error = m_read_thread.Join(nullptr);
228   return error.Success();
229 }
230 
231 bool Communication::JoinReadThread(Status *error_ptr) {
232   if (!m_read_thread.IsJoinable())
233     return true;
234 
235   Status error = m_read_thread.Join(nullptr);
236   return error.Success();
237 }
238 
239 size_t Communication::GetCachedBytes(void *dst, size_t dst_len) {
240   std::lock_guard<std::recursive_mutex> guard(m_bytes_mutex);
241   if (!m_bytes.empty()) {
242     // If DST is nullptr and we have a thread, then return the number of bytes
243     // that are available so the caller can call again
244     if (dst == nullptr)
245       return m_bytes.size();
246 
247     const size_t len = std::min<size_t>(dst_len, m_bytes.size());
248 
249     ::memcpy(dst, m_bytes.c_str(), len);
250     m_bytes.erase(m_bytes.begin(), m_bytes.begin() + len);
251 
252     return len;
253   }
254   return 0;
255 }
256 
257 void Communication::AppendBytesToCache(const uint8_t *bytes, size_t len,
258                                        bool broadcast,
259                                        ConnectionStatus status) {
260   lldb_private::LogIfAnyCategoriesSet(
261       LIBLLDB_LOG_COMMUNICATION,
262       "%p Communication::AppendBytesToCache (src = %p, src_len = %" PRIu64
263       ", broadcast = %i)",
264       this, bytes, (uint64_t)len, broadcast);
265   if ((bytes == nullptr || len == 0) &&
266       (status != lldb::eConnectionStatusEndOfFile))
267     return;
268   if (m_callback) {
269     // If the user registered a callback, then call it and do not broadcast
270     m_callback(m_callback_baton, bytes, len);
271   } else if (bytes != nullptr && len > 0) {
272     std::lock_guard<std::recursive_mutex> guard(m_bytes_mutex);
273     m_bytes.append((const char *)bytes, len);
274     if (broadcast)
275       BroadcastEventIfUnique(eBroadcastBitReadThreadGotBytes);
276   }
277 }
278 
279 size_t Communication::ReadFromConnection(void *dst, size_t dst_len,
280                                          const Timeout<std::micro> &timeout,
281                                          ConnectionStatus &status,
282                                          Status *error_ptr) {
283   lldb::ConnectionSP connection_sp(m_connection_sp);
284   if (connection_sp)
285     return connection_sp->Read(dst, dst_len, timeout, status, error_ptr);
286 
287   if (error_ptr)
288     error_ptr->SetErrorString("Invalid connection.");
289   status = eConnectionStatusNoConnection;
290   return 0;
291 }
292 
293 bool Communication::ReadThreadIsRunning() { return m_read_thread_enabled; }
294 
295 lldb::thread_result_t Communication::ReadThread(lldb::thread_arg_t p) {
296   Communication *comm = (Communication *)p;
297 
298   Log *log(lldb_private::GetLogIfAllCategoriesSet(LIBLLDB_LOG_COMMUNICATION));
299 
300   if (log)
301     log->Printf("%p Communication::ReadThread () thread starting...", p);
302 
303   uint8_t buf[1024];
304 
305   Status error;
306   ConnectionStatus status = eConnectionStatusSuccess;
307   bool done = false;
308   while (!done && comm->m_read_thread_enabled) {
309     size_t bytes_read = comm->ReadFromConnection(
310         buf, sizeof(buf), std::chrono::seconds(5), status, &error);
311     if (bytes_read > 0)
312       comm->AppendBytesToCache(buf, bytes_read, true, status);
313     else if ((bytes_read == 0) && status == eConnectionStatusEndOfFile) {
314       if (comm->GetCloseOnEOF())
315         comm->Disconnect();
316       comm->AppendBytesToCache(buf, bytes_read, true, status);
317     }
318 
319     switch (status) {
320     case eConnectionStatusSuccess:
321       break;
322 
323     case eConnectionStatusEndOfFile:
324       done = true;
325       break;
326     case eConnectionStatusError: // Check GetError() for details
327       if (error.GetType() == eErrorTypePOSIX && error.GetError() == EIO) {
328         // EIO on a pipe is usually caused by remote shutdown
329         comm->Disconnect();
330         done = true;
331       }
332       if (error.Fail())
333         LLDB_LOG(log, "error: {0}, status = {1}", error,
334                  Communication::ConnectionStatusAsCString(status));
335       break;
336     case eConnectionStatusInterrupted: // Synchronization signal from
337                                        // SynchronizeWithReadThread()
338       // The connection returns eConnectionStatusInterrupted only when there is
339       // no input pending to be read, so we can signal that.
340       comm->BroadcastEvent(eBroadcastBitNoMorePendingInput);
341       break;
342     case eConnectionStatusNoConnection:   // No connection
343     case eConnectionStatusLostConnection: // Lost connection while connected to
344                                           // a valid connection
345       done = true;
346       LLVM_FALLTHROUGH;
347     case eConnectionStatusTimedOut: // Request timed out
348       if (error.Fail())
349         LLDB_LOG(log, "error: {0}, status = {1}", error,
350                  Communication::ConnectionStatusAsCString(status));
351       break;
352     }
353   }
354   log = lldb_private::GetLogIfAllCategoriesSet(LIBLLDB_LOG_COMMUNICATION);
355   if (log)
356     log->Printf("%p Communication::ReadThread () thread exiting...", p);
357 
358   comm->m_read_thread_did_exit = true;
359   // Let clients know that this thread is exiting
360   comm->BroadcastEvent(eBroadcastBitNoMorePendingInput);
361   comm->BroadcastEvent(eBroadcastBitReadThreadDidExit);
362   return NULL;
363 }
364 
365 void Communication::SetReadThreadBytesReceivedCallback(
366     ReadThreadBytesReceived callback, void *callback_baton) {
367   m_callback = callback;
368   m_callback_baton = callback_baton;
369 }
370 
371 void Communication::SynchronizeWithReadThread() {
372   // Only one thread can do the synchronization dance at a time.
373   std::lock_guard<std::mutex> guard(m_synchronize_mutex);
374 
375   // First start listening for the synchronization event.
376   ListenerSP listener_sp(
377       Listener::MakeListener("Communication::SyncronizeWithReadThread"));
378   listener_sp->StartListeningForEvents(this, eBroadcastBitNoMorePendingInput);
379 
380   // If the thread is not running, there is no point in synchronizing.
381   if (!m_read_thread_enabled || m_read_thread_did_exit)
382     return;
383 
384   // Notify the read thread.
385   m_connection_sp->InterruptRead();
386 
387   // Wait for the synchronization event.
388   EventSP event_sp;
389   listener_sp->GetEvent(event_sp, llvm::None);
390 }
391 
392 void Communication::SetConnection(Connection *connection) {
393   Disconnect(nullptr);
394   StopReadThread(nullptr);
395   m_connection_sp.reset(connection);
396 }
397 
398 const char *
399 Communication::ConnectionStatusAsCString(lldb::ConnectionStatus status) {
400   switch (status) {
401   case eConnectionStatusSuccess:
402     return "success";
403   case eConnectionStatusError:
404     return "error";
405   case eConnectionStatusTimedOut:
406     return "timed out";
407   case eConnectionStatusNoConnection:
408     return "no connection";
409   case eConnectionStatusLostConnection:
410     return "lost connection";
411   case eConnectionStatusEndOfFile:
412     return "end of file";
413   case eConnectionStatusInterrupted:
414     return "interrupted";
415   }
416 
417   static char unknown_state_string[64];
418   snprintf(unknown_state_string, sizeof(unknown_state_string),
419            "ConnectionStatus = %i", status);
420   return unknown_state_string;
421 }
422