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