1 //===-- Communication.cpp ---------------------------------------*- C++ -*-===//
2 //
3 //                     The LLVM Compiler Infrastructure
4 //
5 // This file is distributed under the University of Illinois Open Source
6 // License. See LICENSE.TXT for details.
7 //
8 //===----------------------------------------------------------------------===//
9 
10 #include "lldb/Core/Communication.h"
11 
12 #include "lldb/Core/Event.h"
13 #include "lldb/Core/Listener.h"
14 #include "lldb/Host/HostThread.h"
15 #include "lldb/Host/ThreadLauncher.h"
16 #include "lldb/Utility/Connection.h"
17 #include "lldb/Utility/ConstString.h" // for ConstString
18 #include "lldb/Utility/Log.h"
19 #include "lldb/Utility/Logging.h" // for LogIfAnyCategoriesSet, LIBLLDB...
20 #include "lldb/Utility/Status.h"  // for Status
21 
22 #include "llvm/ADT/None.h"         // for None
23 #include "llvm/ADT/Optional.h"     // for Optional
24 #include "llvm/Support/Compiler.h" // for LLVM_FALLTHROUGH
25 
26 #include <algorithm> // for min
27 #include <chrono>    // for duration, seconds
28 #include <cstring>
29 #include <memory> // for shared_ptr
30 
31 #include <errno.h>    // for EIO
32 #include <inttypes.h> // for PRIu64
33 #include <stdio.h>    // for snprintf
34 
35 using namespace lldb;
36 using namespace lldb_private;
37 
38 ConstString &Communication::GetStaticBroadcasterClass() {
39   static ConstString class_name("lldb.communication");
40   return class_name;
41 }
42 
43 Communication::Communication(const char *name)
44     : Broadcaster(nullptr, name), m_connection_sp(),
45       m_read_thread_enabled(false), m_read_thread_did_exit(false), m_bytes(),
46       m_bytes_mutex(), m_write_mutex(), m_synchronize_mutex(),
47       m_callback(nullptr), m_callback_baton(nullptr), m_close_on_eof(true)
48 
49 {
50   lldb_private::LogIfAnyCategoriesSet(
51       LIBLLDB_LOG_OBJECT | LIBLLDB_LOG_COMMUNICATION,
52       "%p Communication::Communication (name = %s)", 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 
64 Communication::~Communication() {
65   lldb_private::LogIfAnyCategoriesSet(
66       LIBLLDB_LOG_OBJECT | LIBLLDB_LOG_COMMUNICATION,
67       "%p Communication::~Communication (name = %s)", this,
68       GetBroadcasterName().AsCString());
69   Clear();
70 }
71 
72 void Communication::Clear() {
73   SetReadThreadBytesReceivedCallback(nullptr, nullptr);
74   Disconnect(nullptr);
75   StopReadThread(nullptr);
76 }
77 
78 ConnectionStatus Communication::Connect(const char *url, Status *error_ptr) {
79   Clear();
80 
81   lldb_private::LogIfAnyCategoriesSet(LIBLLDB_LOG_COMMUNICATION,
82                                       "%p Communication::Connect (url = %s)",
83                                       this, url);
84 
85   lldb::ConnectionSP connection_sp(m_connection_sp);
86   if (connection_sp)
87     return connection_sp->Connect(url, error_ptr);
88   if (error_ptr)
89     error_ptr->SetErrorString("Invalid connection.");
90   return eConnectionStatusNoConnection;
91 }
92 
93 ConnectionStatus Communication::Disconnect(Status *error_ptr) {
94   lldb_private::LogIfAnyCategoriesSet(LIBLLDB_LOG_COMMUNICATION,
95                                       "%p Communication::Disconnect ()", this);
96 
97   lldb::ConnectionSP connection_sp(m_connection_sp);
98   if (connection_sp) {
99     ConnectionStatus status = connection_sp->Disconnect(error_ptr);
100     // We currently don't protect connection_sp with any mutex for multi-
101     // threaded environments. So lets not nuke our connection class without
102     // putting some multi-threaded protections in. We also probably don't want
103     // to pay for the overhead it might cause if every time we access the
104     // connection we have to take a lock.
105     //
106     // This unique pointer will cleanup after itself when this object goes
107     // away, so there is no need to currently have it destroy itself
108     // immediately upon disconnect.
109     // connection_sp.reset();
110     return status;
111   }
112   return eConnectionStatusNoConnection;
113 }
114 
115 bool Communication::IsConnected() const {
116   lldb::ConnectionSP connection_sp(m_connection_sp);
117   return (connection_sp ? connection_sp->IsConnected() : false);
118 }
119 
120 bool Communication::HasConnection() const {
121   return m_connection_sp.get() != nullptr;
122 }
123 
124 size_t Communication::Read(void *dst, size_t dst_len,
125                            const Timeout<std::micro> &timeout,
126                            ConnectionStatus &status, Status *error_ptr) {
127   Log *log = GetLogIfAllCategoriesSet(LIBLLDB_LOG_COMMUNICATION);
128   LLDB_LOG(
129       log,
130       "this = {0}, dst = {1}, dst_len = {2}, timeout = {3}, connection = {4}",
131       this, dst, dst_len, timeout, m_connection_sp.get());
132 
133   if (m_read_thread_enabled) {
134     // We have a dedicated read thread that is getting data for us
135     size_t cached_bytes = GetCachedBytes(dst, dst_len);
136     if (cached_bytes > 0 || (timeout && timeout->count() == 0)) {
137       status = eConnectionStatusSuccess;
138       return cached_bytes;
139     }
140 
141     if (!m_connection_sp) {
142       if (error_ptr)
143         error_ptr->SetErrorString("Invalid connection.");
144       status = eConnectionStatusNoConnection;
145       return 0;
146     }
147 
148     ListenerSP listener_sp(Listener::MakeListener("Communication::Read"));
149     listener_sp->StartListeningForEvents(
150         this, eBroadcastBitReadThreadGotBytes | eBroadcastBitReadThreadDidExit);
151     EventSP event_sp;
152     while (listener_sp->GetEvent(event_sp, timeout)) {
153       const uint32_t event_type = event_sp->GetType();
154       if (event_type & eBroadcastBitReadThreadGotBytes) {
155         return GetCachedBytes(dst, dst_len);
156       }
157 
158       if (event_type & eBroadcastBitReadThreadDidExit) {
159         if (GetCloseOnEOF())
160           Disconnect(nullptr);
161         break;
162       }
163     }
164     return 0;
165   }
166 
167   // We aren't using a read thread, just read the data synchronously in this
168   // thread.
169   return ReadFromConnection(dst, dst_len, timeout, status, error_ptr);
170 }
171 
172 size_t Communication::Write(const void *src, size_t src_len,
173                             ConnectionStatus &status, Status *error_ptr) {
174   lldb::ConnectionSP connection_sp(m_connection_sp);
175 
176   std::lock_guard<std::mutex> guard(m_write_mutex);
177   lldb_private::LogIfAnyCategoriesSet(
178       LIBLLDB_LOG_COMMUNICATION,
179       "%p Communication::Write (src = %p, src_len = %" PRIu64
180       ") connection = %p",
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 
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_private::LogIfAnyCategoriesSet(
200       LIBLLDB_LOG_COMMUNICATION, "%p 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   m_read_thread = ThreadLauncher::LaunchThread(
209       thread_name, Communication::ReadThread, this, error_ptr);
210   if (!m_read_thread.IsJoinable())
211     m_read_thread_enabled = false;
212   return m_read_thread_enabled;
213 }
214 
215 bool Communication::StopReadThread(Status *error_ptr) {
216   if (!m_read_thread.IsJoinable())
217     return true;
218 
219   lldb_private::LogIfAnyCategoriesSet(
220       LIBLLDB_LOG_COMMUNICATION, "%p Communication::StopReadThread ()", this);
221 
222   m_read_thread_enabled = false;
223 
224   BroadcastEvent(eBroadcastBitReadThreadShouldExit, nullptr);
225 
226   // error = m_read_thread.Cancel();
227 
228   Status error = m_read_thread.Join(nullptr);
229   return error.Success();
230 }
231 
232 bool Communication::JoinReadThread(Status *error_ptr) {
233   if (!m_read_thread.IsJoinable())
234     return true;
235 
236   Status error = m_read_thread.Join(nullptr);
237   return error.Success();
238 }
239 
240 size_t Communication::GetCachedBytes(void *dst, size_t dst_len) {
241   std::lock_guard<std::recursive_mutex> guard(m_bytes_mutex);
242   if (!m_bytes.empty()) {
243     // If DST is nullptr and we have a thread, then return the number of bytes
244     // that are available so the caller can call again
245     if (dst == nullptr)
246       return m_bytes.size();
247 
248     const size_t len = std::min<size_t>(dst_len, m_bytes.size());
249 
250     ::memcpy(dst, m_bytes.c_str(), len);
251     m_bytes.erase(m_bytes.begin(), m_bytes.begin() + len);
252 
253     return len;
254   }
255   return 0;
256 }
257 
258 void Communication::AppendBytesToCache(const uint8_t *bytes, size_t len,
259                                        bool broadcast,
260                                        ConnectionStatus status) {
261   lldb_private::LogIfAnyCategoriesSet(
262       LIBLLDB_LOG_COMMUNICATION,
263       "%p Communication::AppendBytesToCache (src = %p, src_len = %" PRIu64
264       ", broadcast = %i)",
265       this, bytes, (uint64_t)len, broadcast);
266   if ((bytes == nullptr || len == 0) &&
267       (status != lldb::eConnectionStatusEndOfFile))
268     return;
269   if (m_callback) {
270     // If the user registered a callback, then call it and do not broadcast
271     m_callback(m_callback_baton, bytes, len);
272   } else if (bytes != nullptr && len > 0) {
273     std::lock_guard<std::recursive_mutex> guard(m_bytes_mutex);
274     m_bytes.append((const char *)bytes, len);
275     if (broadcast)
276       BroadcastEventIfUnique(eBroadcastBitReadThreadGotBytes);
277   }
278 }
279 
280 size_t Communication::ReadFromConnection(void *dst, size_t dst_len,
281                                          const Timeout<std::micro> &timeout,
282                                          ConnectionStatus &status,
283                                          Status *error_ptr) {
284   lldb::ConnectionSP connection_sp(m_connection_sp);
285   if (connection_sp)
286     return connection_sp->Read(dst, dst_len, timeout, status, error_ptr);
287 
288   if (error_ptr)
289     error_ptr->SetErrorString("Invalid connection.");
290   status = eConnectionStatusNoConnection;
291   return 0;
292 }
293 
294 bool Communication::ReadThreadIsRunning() { return m_read_thread_enabled; }
295 
296 lldb::thread_result_t Communication::ReadThread(lldb::thread_arg_t p) {
297   Communication *comm = (Communication *)p;
298 
299   Log *log(lldb_private::GetLogIfAllCategoriesSet(LIBLLDB_LOG_COMMUNICATION));
300 
301   if (log)
302     log->Printf("%p Communication::ReadThread () thread starting...", p);
303 
304   uint8_t buf[1024];
305 
306   Status error;
307   ConnectionStatus status = eConnectionStatusSuccess;
308   bool done = false;
309   while (!done && comm->m_read_thread_enabled) {
310     size_t bytes_read = comm->ReadFromConnection(
311         buf, sizeof(buf), std::chrono::seconds(5), status, &error);
312     if (bytes_read > 0)
313       comm->AppendBytesToCache(buf, bytes_read, true, status);
314     else if ((bytes_read == 0) && status == eConnectionStatusEndOfFile) {
315       if (comm->GetCloseOnEOF())
316         comm->Disconnect();
317       comm->AppendBytesToCache(buf, bytes_read, true, status);
318     }
319 
320     switch (status) {
321     case eConnectionStatusSuccess:
322       break;
323 
324     case eConnectionStatusEndOfFile:
325       done = true;
326       break;
327     case eConnectionStatusError: // Check GetError() for details
328       if (error.GetType() == eErrorTypePOSIX && error.GetError() == EIO) {
329         // EIO on a pipe is usually caused by remote shutdown
330         comm->Disconnect();
331         done = true;
332       }
333       if (error.Fail())
334         LLDB_LOG(log, "error: {0}, status = {1}", error,
335                  Communication::ConnectionStatusAsCString(status));
336       break;
337     case eConnectionStatusInterrupted: // Synchronization signal from
338                                        // SynchronizeWithReadThread()
339       // The connection returns eConnectionStatusInterrupted only when there is
340       // no input pending to be read, so we can signal that.
341       comm->BroadcastEvent(eBroadcastBitNoMorePendingInput);
342       break;
343     case eConnectionStatusNoConnection:   // No connection
344     case eConnectionStatusLostConnection: // Lost connection while connected to
345                                           // a valid connection
346       done = true;
347       LLVM_FALLTHROUGH;
348     case eConnectionStatusTimedOut: // Request timed out
349       if (error.Fail())
350         LLDB_LOG(log, "error: {0}, status = {1}", error,
351                  Communication::ConnectionStatusAsCString(status));
352       break;
353     }
354   }
355   log = lldb_private::GetLogIfAllCategoriesSet(LIBLLDB_LOG_COMMUNICATION);
356   if (log)
357     log->Printf("%p Communication::ReadThread () thread exiting...", p);
358 
359   comm->m_read_thread_did_exit = true;
360   // Let clients know that this thread is exiting
361   comm->BroadcastEvent(eBroadcastBitNoMorePendingInput);
362   comm->BroadcastEvent(eBroadcastBitReadThreadDidExit);
363   return NULL;
364 }
365 
366 void Communication::SetReadThreadBytesReceivedCallback(
367     ReadThreadBytesReceived callback, void *callback_baton) {
368   m_callback = callback;
369   m_callback_baton = callback_baton;
370 }
371 
372 void Communication::SynchronizeWithReadThread() {
373   // Only one thread can do the synchronization dance at a time.
374   std::lock_guard<std::mutex> guard(m_synchronize_mutex);
375 
376   // First start listening for the synchronization event.
377   ListenerSP listener_sp(
378       Listener::MakeListener("Communication::SyncronizeWithReadThread"));
379   listener_sp->StartListeningForEvents(this, eBroadcastBitNoMorePendingInput);
380 
381   // If the thread is not running, there is no point in synchronizing.
382   if (!m_read_thread_enabled || m_read_thread_did_exit)
383     return;
384 
385   // Notify the read thread.
386   m_connection_sp->InterruptRead();
387 
388   // Wait for the synchronization event.
389   EventSP event_sp;
390   listener_sp->GetEvent(event_sp, llvm::None);
391 }
392 
393 void Communication::SetConnection(Connection *connection) {
394   Disconnect(nullptr);
395   StopReadThread(nullptr);
396   m_connection_sp.reset(connection);
397 }
398 
399 const char *
400 Communication::ConnectionStatusAsCString(lldb::ConnectionStatus status) {
401   switch (status) {
402   case eConnectionStatusSuccess:
403     return "success";
404   case eConnectionStatusError:
405     return "error";
406   case eConnectionStatusTimedOut:
407     return "timed out";
408   case eConnectionStatusNoConnection:
409     return "no connection";
410   case eConnectionStatusLostConnection:
411     return "lost connection";
412   case eConnectionStatusEndOfFile:
413     return "end of file";
414   case eConnectionStatusInterrupted:
415     return "interrupted";
416   }
417 
418   static char unknown_state_string[64];
419   snprintf(unknown_state_string, sizeof(unknown_state_string),
420            "ConnectionStatus = %i", status);
421   return unknown_state_string;
422 }
423