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/LLDBLog.h"
17 #include "lldb/Utility/Listener.h"
18 #include "lldb/Utility/Log.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 <cerrno>
31 #include <cinttypes>
32 #include <cstdio>
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 
50   LLDB_LOG(GetLog(LLDBLog::Object | LLDBLog::Communication),
51            "{0} Communication::Communication (name = {1})", 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_LOG(GetLog(LLDBLog::Object | LLDBLog::Communication),
65            "{0} Communication::~Communication (name = {1})", this,
66            GetBroadcasterName().AsCString());
67   Clear();
68 }
69 
70 void Communication::Clear() {
71   SetReadThreadBytesReceivedCallback(nullptr, nullptr);
72   StopReadThread(nullptr);
73   Disconnect(nullptr);
74 }
75 
76 ConnectionStatus Communication::Connect(const char *url, Status *error_ptr) {
77   Clear();
78 
79   LLDB_LOG(GetLog(LLDBLog::Communication),
80            "{0} Communication::Connect (url = {1})", this, url);
81 
82   lldb::ConnectionSP connection_sp(m_connection_sp);
83   if (connection_sp)
84     return connection_sp->Connect(url, error_ptr);
85   if (error_ptr)
86     error_ptr->SetErrorString("Invalid connection.");
87   return eConnectionStatusNoConnection;
88 }
89 
90 ConnectionStatus Communication::Disconnect(Status *error_ptr) {
91   LLDB_LOG(GetLog(LLDBLog::Communication), "{0} Communication::Disconnect ()",
92            this);
93 
94   assert((!m_read_thread_enabled || m_read_thread_did_exit) &&
95          "Disconnecting while the read thread is running is racy!");
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 = GetLog(LLDBLog::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_LOG(GetLog(LLDBLog::Communication),
177            "{0} Communication::Write (src = {1}, src_len = {2}"
178            ") connection = {3}",
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 
190 size_t Communication::WriteAll(const void *src, size_t src_len,
191                                ConnectionStatus &status, Status *error_ptr) {
192   size_t total_written = 0;
193   do
194     total_written += Write(static_cast<const char *>(src) + total_written,
195                            src_len - total_written, status, error_ptr);
196   while (status == eConnectionStatusSuccess && total_written < src_len);
197   return total_written;
198 }
199 
200 bool Communication::StartReadThread(Status *error_ptr) {
201   if (error_ptr)
202     error_ptr->Clear();
203 
204   if (m_read_thread.IsJoinable())
205     return true;
206 
207   LLDB_LOG(GetLog(LLDBLog::Communication),
208            "{0} Communication::StartReadThread ()", this);
209 
210   const std::string thread_name =
211       llvm::formatv("<lldb.comm.{0}>", GetBroadcasterName());
212 
213   m_read_thread_enabled = true;
214   m_read_thread_did_exit = false;
215   auto maybe_thread = ThreadLauncher::LaunchThread(
216       thread_name, Communication::ReadThread, this);
217   if (maybe_thread) {
218     m_read_thread = *maybe_thread;
219   } else {
220     if (error_ptr)
221       *error_ptr = Status(maybe_thread.takeError());
222     else {
223       LLDB_LOG(GetLog(LLDBLog::Host), "failed to launch host thread: {}",
224                llvm::toString(maybe_thread.takeError()));
225     }
226   }
227 
228   if (!m_read_thread.IsJoinable())
229     m_read_thread_enabled = false;
230 
231   return m_read_thread_enabled;
232 }
233 
234 bool Communication::StopReadThread(Status *error_ptr) {
235   if (!m_read_thread.IsJoinable())
236     return true;
237 
238   LLDB_LOG(GetLog(LLDBLog::Communication),
239            "{0} Communication::StopReadThread ()", this);
240 
241   m_read_thread_enabled = false;
242 
243   BroadcastEvent(eBroadcastBitReadThreadShouldExit, nullptr);
244 
245   // error = m_read_thread.Cancel();
246 
247   Status error = m_read_thread.Join(nullptr);
248   return error.Success();
249 }
250 
251 bool Communication::JoinReadThread(Status *error_ptr) {
252   if (!m_read_thread.IsJoinable())
253     return true;
254 
255   Status error = m_read_thread.Join(nullptr);
256   return error.Success();
257 }
258 
259 size_t Communication::GetCachedBytes(void *dst, size_t dst_len) {
260   std::lock_guard<std::recursive_mutex> guard(m_bytes_mutex);
261   if (!m_bytes.empty()) {
262     // If DST is nullptr and we have a thread, then return the number of bytes
263     // that are available so the caller can call again
264     if (dst == nullptr)
265       return m_bytes.size();
266 
267     const size_t len = std::min<size_t>(dst_len, m_bytes.size());
268 
269     ::memcpy(dst, m_bytes.c_str(), len);
270     m_bytes.erase(m_bytes.begin(), m_bytes.begin() + len);
271 
272     return len;
273   }
274   return 0;
275 }
276 
277 void Communication::AppendBytesToCache(const uint8_t *bytes, size_t len,
278                                        bool broadcast,
279                                        ConnectionStatus status) {
280   LLDB_LOG(GetLog(LLDBLog::Communication),
281            "{0} Communication::AppendBytesToCache (src = {1}, src_len = {2}, "
282            "broadcast = {3})",
283            this, bytes, (uint64_t)len, broadcast);
284   if ((bytes == nullptr || len == 0) &&
285       (status != lldb::eConnectionStatusEndOfFile))
286     return;
287   if (m_callback) {
288     // If the user registered a callback, then call it and do not broadcast
289     m_callback(m_callback_baton, bytes, len);
290   } else if (bytes != nullptr && len > 0) {
291     std::lock_guard<std::recursive_mutex> guard(m_bytes_mutex);
292     m_bytes.append((const char *)bytes, len);
293     if (broadcast)
294       BroadcastEventIfUnique(eBroadcastBitReadThreadGotBytes);
295   }
296 }
297 
298 size_t Communication::ReadFromConnection(void *dst, size_t dst_len,
299                                          const Timeout<std::micro> &timeout,
300                                          ConnectionStatus &status,
301                                          Status *error_ptr) {
302   lldb::ConnectionSP connection_sp(m_connection_sp);
303   if (connection_sp)
304     return connection_sp->Read(dst, dst_len, timeout, status, error_ptr);
305 
306   if (error_ptr)
307     error_ptr->SetErrorString("Invalid connection.");
308   status = eConnectionStatusNoConnection;
309   return 0;
310 }
311 
312 bool Communication::ReadThreadIsRunning() { return m_read_thread_enabled; }
313 
314 lldb::thread_result_t Communication::ReadThread(lldb::thread_arg_t p) {
315   Communication *comm = (Communication *)p;
316 
317   Log *log = GetLog(LLDBLog::Communication);
318 
319   LLDB_LOGF(log, "%p Communication::ReadThread () thread starting...", p);
320 
321   uint8_t buf[1024];
322 
323   Status error;
324   ConnectionStatus status = eConnectionStatusSuccess;
325   bool done = false;
326   bool disconnect = false;
327   while (!done && comm->m_read_thread_enabled) {
328     size_t bytes_read = comm->ReadFromConnection(
329         buf, sizeof(buf), std::chrono::seconds(5), status, &error);
330     if (bytes_read > 0 || status == eConnectionStatusEndOfFile)
331       comm->AppendBytesToCache(buf, bytes_read, true, status);
332 
333     switch (status) {
334     case eConnectionStatusSuccess:
335       break;
336 
337     case eConnectionStatusEndOfFile:
338       done = true;
339       disconnect = comm->GetCloseOnEOF();
340       break;
341     case eConnectionStatusError: // Check GetError() for details
342       if (error.GetType() == eErrorTypePOSIX && error.GetError() == EIO) {
343         // EIO on a pipe is usually caused by remote shutdown
344         disconnect = comm->GetCloseOnEOF();
345         done = true;
346       }
347       if (error.Fail())
348         LLDB_LOG(log, "error: {0}, status = {1}", error,
349                  Communication::ConnectionStatusAsString(status));
350       break;
351     case eConnectionStatusInterrupted: // Synchronization signal from
352                                        // SynchronizeWithReadThread()
353       // The connection returns eConnectionStatusInterrupted only when there is
354       // no input pending to be read, so we can signal that.
355       comm->BroadcastEvent(eBroadcastBitNoMorePendingInput);
356       break;
357     case eConnectionStatusNoConnection:   // No connection
358     case eConnectionStatusLostConnection: // Lost connection while connected to
359                                           // a valid connection
360       done = true;
361       LLVM_FALLTHROUGH;
362     case eConnectionStatusTimedOut: // Request timed out
363       if (error.Fail())
364         LLDB_LOG(log, "error: {0}, status = {1}", error,
365                  Communication::ConnectionStatusAsString(status));
366       break;
367     }
368   }
369   log = GetLog(LLDBLog::Communication);
370   if (log)
371     LLDB_LOGF(log, "%p Communication::ReadThread () thread exiting...", p);
372 
373   // Handle threads wishing to synchronize with us.
374   {
375     // Prevent new ones from showing up.
376     comm->m_read_thread_did_exit = true;
377 
378     // Unblock any existing thread waiting for the synchronization signal.
379     comm->BroadcastEvent(eBroadcastBitNoMorePendingInput);
380 
381     // Wait for the thread to finish...
382     std::lock_guard<std::mutex> guard(comm->m_synchronize_mutex);
383     // ... and disconnect.
384     if (disconnect)
385       comm->Disconnect();
386   }
387 
388   // Let clients know that this thread is exiting
389   comm->BroadcastEvent(eBroadcastBitReadThreadDidExit);
390   return {};
391 }
392 
393 void Communication::SetReadThreadBytesReceivedCallback(
394     ReadThreadBytesReceived callback, void *callback_baton) {
395   m_callback = callback;
396   m_callback_baton = callback_baton;
397 }
398 
399 void Communication::SynchronizeWithReadThread() {
400   // Only one thread can do the synchronization dance at a time.
401   std::lock_guard<std::mutex> guard(m_synchronize_mutex);
402 
403   // First start listening for the synchronization event.
404   ListenerSP listener_sp(
405       Listener::MakeListener("Communication::SyncronizeWithReadThread"));
406   listener_sp->StartListeningForEvents(this, eBroadcastBitNoMorePendingInput);
407 
408   // If the thread is not running, there is no point in synchronizing.
409   if (!m_read_thread_enabled || m_read_thread_did_exit)
410     return;
411 
412   // Notify the read thread.
413   m_connection_sp->InterruptRead();
414 
415   // Wait for the synchronization event.
416   EventSP event_sp;
417   listener_sp->GetEvent(event_sp, llvm::None);
418 }
419 
420 void Communication::SetConnection(std::unique_ptr<Connection> connection) {
421   Disconnect(nullptr);
422   StopReadThread(nullptr);
423   m_connection_sp = std::move(connection);
424 }
425 
426 std::string
427 Communication::ConnectionStatusAsString(lldb::ConnectionStatus status) {
428   switch (status) {
429   case eConnectionStatusSuccess:
430     return "success";
431   case eConnectionStatusError:
432     return "error";
433   case eConnectionStatusTimedOut:
434     return "timed out";
435   case eConnectionStatusNoConnection:
436     return "no connection";
437   case eConnectionStatusLostConnection:
438     return "lost connection";
439   case eConnectionStatusEndOfFile:
440     return "end of file";
441   case eConnectionStatusInterrupted:
442     return "interrupted";
443   }
444 
445   return "@" + std::to_string(status);
446 }
447