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/Host/HostThread.h"
13 #include "lldb/Host/ThreadLauncher.h"
14 #include "lldb/Utility/Connection.h"
15 #include "lldb/Utility/ConstString.h"
16 #include "lldb/Utility/Event.h"
17 #include "lldb/Utility/Listener.h"
18 #include "lldb/Utility/Log.h"
19 #include "lldb/Utility/Logging.h"
20 #include "lldb/Utility/Status.h"
21
22 #include "llvm/ADT/None.h"
23 #include "llvm/ADT/Optional.h"
24 #include "llvm/Support/Compiler.h"
25
26 #include <algorithm>
27 #include <chrono>
28 #include <cstring>
29 #include <memory>
30
31 #include <errno.h>
32 #include <inttypes.h>
33 #include <stdio.h>
34
35 using namespace lldb;
36 using namespace lldb_private;
37
GetStaticBroadcasterClass()38 ConstString &Communication::GetStaticBroadcasterClass() {
39 static ConstString class_name("lldb.communication");
40 return class_name;
41 }
42
Communication(const char * name)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
~Communication()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
Clear()72 void Communication::Clear() {
73 SetReadThreadBytesReceivedCallback(nullptr, nullptr);
74 Disconnect(nullptr);
75 StopReadThread(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_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
Disconnect(Status * error_ptr)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
IsConnected() const115 bool Communication::IsConnected() const {
116 lldb::ConnectionSP connection_sp(m_connection_sp);
117 return (connection_sp ? connection_sp->IsConnected() : false);
118 }
119
HasConnection() const120 bool Communication::HasConnection() const {
121 return m_connection_sp.get() != nullptr;
122 }
123
Read(void * dst,size_t dst_len,const Timeout<std::micro> & timeout,ConnectionStatus & status,Status * error_ptr)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
Write(const void * src,size_t src_len,ConnectionStatus & status,Status * error_ptr)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
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_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
StopReadThread(Status * error_ptr)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
JoinReadThread(Status * error_ptr)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
GetCachedBytes(void * dst,size_t dst_len)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
AppendBytesToCache(const uint8_t * bytes,size_t len,bool broadcast,ConnectionStatus status)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
ReadFromConnection(void * dst,size_t dst_len,const Timeout<std::micro> & timeout,ConnectionStatus & status,Status * error_ptr)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
ReadThreadIsRunning()294 bool Communication::ReadThreadIsRunning() { return m_read_thread_enabled; }
295
ReadThread(lldb::thread_arg_t p)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
SetReadThreadBytesReceivedCallback(ReadThreadBytesReceived callback,void * callback_baton)366 void Communication::SetReadThreadBytesReceivedCallback(
367 ReadThreadBytesReceived callback, void *callback_baton) {
368 m_callback = callback;
369 m_callback_baton = callback_baton;
370 }
371
SynchronizeWithReadThread()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
SetConnection(Connection * connection)393 void Communication::SetConnection(Connection *connection) {
394 Disconnect(nullptr);
395 StopReadThread(nullptr);
396 m_connection_sp.reset(connection);
397 }
398
399 const char *
ConnectionStatusAsCString(lldb::ConnectionStatus status)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