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/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 <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(lldb_private::GetLogIfAnyCategoriesSet(LIBLLDB_LOG_OBJECT | 51 LIBLLDB_LOG_COMMUNICATION), 52 "{0} Communication::Communication (name = {1})", 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_LOG(lldb_private::GetLogIfAnyCategoriesSet(LIBLLDB_LOG_OBJECT | 66 LIBLLDB_LOG_COMMUNICATION), 67 "{0} Communication::~Communication (name = {1})", this, 68 GetBroadcasterName().AsCString()); 69 Clear(); 70 } 71 72 void Communication::Clear() { 73 SetReadThreadBytesReceivedCallback(nullptr, nullptr); 74 StopReadThread(nullptr); 75 Disconnect(nullptr); 76 } 77 78 ConnectionStatus Communication::Connect(const char *url, Status *error_ptr) { 79 Clear(); 80 81 LLDB_LOG(lldb_private::GetLogIfAllCategoriesSet(LIBLLDB_LOG_COMMUNICATION), 82 "{0} Communication::Connect (url = {1})", 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_LOG(lldb_private::GetLogIfAllCategoriesSet(LIBLLDB_LOG_COMMUNICATION), 94 "{0} Communication::Disconnect ()", this); 95 96 assert((!m_read_thread_enabled || m_read_thread_did_exit) && 97 "Disconnecting while the read thread is running is racy!"); 98 lldb::ConnectionSP connection_sp(m_connection_sp); 99 if (connection_sp) { 100 ConnectionStatus status = connection_sp->Disconnect(error_ptr); 101 // We currently don't protect connection_sp with any mutex for multi- 102 // threaded environments. So lets not nuke our connection class without 103 // putting some multi-threaded protections in. We also probably don't want 104 // to pay for the overhead it might cause if every time we access the 105 // connection we have to take a lock. 106 // 107 // This unique pointer will cleanup after itself when this object goes 108 // away, so there is no need to currently have it destroy itself 109 // immediately upon disconnect. 110 // connection_sp.reset(); 111 return status; 112 } 113 return eConnectionStatusNoConnection; 114 } 115 116 bool Communication::IsConnected() const { 117 lldb::ConnectionSP connection_sp(m_connection_sp); 118 return (connection_sp ? connection_sp->IsConnected() : false); 119 } 120 121 bool Communication::HasConnection() const { 122 return m_connection_sp.get() != nullptr; 123 } 124 125 size_t Communication::Read(void *dst, size_t dst_len, 126 const Timeout<std::micro> &timeout, 127 ConnectionStatus &status, Status *error_ptr) { 128 Log *log = GetLogIfAllCategoriesSet(LIBLLDB_LOG_COMMUNICATION); 129 LLDB_LOG( 130 log, 131 "this = {0}, dst = {1}, dst_len = {2}, timeout = {3}, connection = {4}", 132 this, dst, dst_len, timeout, m_connection_sp.get()); 133 134 if (m_read_thread_enabled) { 135 // We have a dedicated read thread that is getting data for us 136 size_t cached_bytes = GetCachedBytes(dst, dst_len); 137 if (cached_bytes > 0 || (timeout && timeout->count() == 0)) { 138 status = eConnectionStatusSuccess; 139 return cached_bytes; 140 } 141 142 if (!m_connection_sp) { 143 if (error_ptr) 144 error_ptr->SetErrorString("Invalid connection."); 145 status = eConnectionStatusNoConnection; 146 return 0; 147 } 148 149 ListenerSP listener_sp(Listener::MakeListener("Communication::Read")); 150 listener_sp->StartListeningForEvents( 151 this, eBroadcastBitReadThreadGotBytes | eBroadcastBitReadThreadDidExit); 152 EventSP event_sp; 153 while (listener_sp->GetEvent(event_sp, timeout)) { 154 const uint32_t event_type = event_sp->GetType(); 155 if (event_type & eBroadcastBitReadThreadGotBytes) { 156 return GetCachedBytes(dst, dst_len); 157 } 158 159 if (event_type & eBroadcastBitReadThreadDidExit) { 160 if (GetCloseOnEOF()) 161 Disconnect(nullptr); 162 break; 163 } 164 } 165 return 0; 166 } 167 168 // We aren't using a read thread, just read the data synchronously in this 169 // thread. 170 return ReadFromConnection(dst, dst_len, timeout, status, error_ptr); 171 } 172 173 size_t Communication::Write(const void *src, size_t src_len, 174 ConnectionStatus &status, Status *error_ptr) { 175 lldb::ConnectionSP connection_sp(m_connection_sp); 176 177 std::lock_guard<std::mutex> guard(m_write_mutex); 178 LLDB_LOG(lldb_private::GetLogIfAllCategoriesSet(LIBLLDB_LOG_COMMUNICATION), 179 "{0} Communication::Write (src = {1}, src_len = {2}" 180 ") connection = {3}", 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 size_t Communication::WriteAll(const void *src, size_t src_len, 193 ConnectionStatus &status, Status *error_ptr) { 194 size_t total_written = 0; 195 do 196 total_written += Write(static_cast<const char *>(src) + total_written, 197 src_len - total_written, status, error_ptr); 198 while (status == eConnectionStatusSuccess && total_written < src_len); 199 return total_written; 200 } 201 202 bool Communication::StartReadThread(Status *error_ptr) { 203 if (error_ptr) 204 error_ptr->Clear(); 205 206 if (m_read_thread.IsJoinable()) 207 return true; 208 209 LLDB_LOG(lldb_private::GetLogIfAllCategoriesSet(LIBLLDB_LOG_COMMUNICATION), 210 "{0} Communication::StartReadThread ()", this); 211 212 const std::string thread_name = 213 llvm::formatv("<lldb.comm.{0}>", GetBroadcasterName()); 214 215 m_read_thread_enabled = true; 216 m_read_thread_did_exit = false; 217 auto maybe_thread = ThreadLauncher::LaunchThread( 218 thread_name, Communication::ReadThread, this); 219 if (maybe_thread) { 220 m_read_thread = *maybe_thread; 221 } else { 222 if (error_ptr) 223 *error_ptr = Status(maybe_thread.takeError()); 224 else { 225 LLDB_LOG(lldb_private::GetLogIfAllCategoriesSet(LIBLLDB_LOG_HOST), 226 "failed to launch host thread: {}", 227 llvm::toString(maybe_thread.takeError())); 228 } 229 } 230 231 if (!m_read_thread.IsJoinable()) 232 m_read_thread_enabled = false; 233 234 return m_read_thread_enabled; 235 } 236 237 bool Communication::StopReadThread(Status *error_ptr) { 238 if (!m_read_thread.IsJoinable()) 239 return true; 240 241 LLDB_LOG(lldb_private::GetLogIfAllCategoriesSet(LIBLLDB_LOG_COMMUNICATION), 242 "{0} Communication::StopReadThread ()", this); 243 244 m_read_thread_enabled = false; 245 246 BroadcastEvent(eBroadcastBitReadThreadShouldExit, nullptr); 247 248 // error = m_read_thread.Cancel(); 249 250 Status error = m_read_thread.Join(nullptr); 251 return error.Success(); 252 } 253 254 bool Communication::JoinReadThread(Status *error_ptr) { 255 if (!m_read_thread.IsJoinable()) 256 return true; 257 258 Status error = m_read_thread.Join(nullptr); 259 return error.Success(); 260 } 261 262 size_t Communication::GetCachedBytes(void *dst, size_t dst_len) { 263 std::lock_guard<std::recursive_mutex> guard(m_bytes_mutex); 264 if (!m_bytes.empty()) { 265 // If DST is nullptr and we have a thread, then return the number of bytes 266 // that are available so the caller can call again 267 if (dst == nullptr) 268 return m_bytes.size(); 269 270 const size_t len = std::min<size_t>(dst_len, m_bytes.size()); 271 272 ::memcpy(dst, m_bytes.c_str(), len); 273 m_bytes.erase(m_bytes.begin(), m_bytes.begin() + len); 274 275 return len; 276 } 277 return 0; 278 } 279 280 void Communication::AppendBytesToCache(const uint8_t *bytes, size_t len, 281 bool broadcast, 282 ConnectionStatus status) { 283 LLDB_LOG(lldb_private::GetLogIfAllCategoriesSet(LIBLLDB_LOG_COMMUNICATION), 284 "{0} Communication::AppendBytesToCache (src = {1}, src_len = {2}, " 285 "broadcast = {3})", 286 this, bytes, (uint64_t)len, broadcast); 287 if ((bytes == nullptr || len == 0) && 288 (status != lldb::eConnectionStatusEndOfFile)) 289 return; 290 if (m_callback) { 291 // If the user registered a callback, then call it and do not broadcast 292 m_callback(m_callback_baton, bytes, len); 293 } else if (bytes != nullptr && len > 0) { 294 std::lock_guard<std::recursive_mutex> guard(m_bytes_mutex); 295 m_bytes.append((const char *)bytes, len); 296 if (broadcast) 297 BroadcastEventIfUnique(eBroadcastBitReadThreadGotBytes); 298 } 299 } 300 301 size_t Communication::ReadFromConnection(void *dst, size_t dst_len, 302 const Timeout<std::micro> &timeout, 303 ConnectionStatus &status, 304 Status *error_ptr) { 305 lldb::ConnectionSP connection_sp(m_connection_sp); 306 if (connection_sp) 307 return connection_sp->Read(dst, dst_len, timeout, status, error_ptr); 308 309 if (error_ptr) 310 error_ptr->SetErrorString("Invalid connection."); 311 status = eConnectionStatusNoConnection; 312 return 0; 313 } 314 315 bool Communication::ReadThreadIsRunning() { return m_read_thread_enabled; } 316 317 lldb::thread_result_t Communication::ReadThread(lldb::thread_arg_t p) { 318 Communication *comm = (Communication *)p; 319 320 Log *log(lldb_private::GetLogIfAllCategoriesSet(LIBLLDB_LOG_COMMUNICATION)); 321 322 LLDB_LOGF(log, "%p Communication::ReadThread () thread starting...", p); 323 324 uint8_t buf[1024]; 325 326 Status error; 327 ConnectionStatus status = eConnectionStatusSuccess; 328 bool done = false; 329 bool disconnect = false; 330 while (!done && comm->m_read_thread_enabled) { 331 size_t bytes_read = comm->ReadFromConnection( 332 buf, sizeof(buf), std::chrono::seconds(5), status, &error); 333 if (bytes_read > 0 || status == eConnectionStatusEndOfFile) 334 comm->AppendBytesToCache(buf, bytes_read, true, status); 335 336 switch (status) { 337 case eConnectionStatusSuccess: 338 break; 339 340 case eConnectionStatusEndOfFile: 341 done = true; 342 disconnect = comm->GetCloseOnEOF(); 343 break; 344 case eConnectionStatusError: // Check GetError() for details 345 if (error.GetType() == eErrorTypePOSIX && error.GetError() == EIO) { 346 // EIO on a pipe is usually caused by remote shutdown 347 disconnect = comm->GetCloseOnEOF(); 348 done = true; 349 } 350 if (error.Fail()) 351 LLDB_LOG(log, "error: {0}, status = {1}", error, 352 Communication::ConnectionStatusAsString(status)); 353 break; 354 case eConnectionStatusInterrupted: // Synchronization signal from 355 // SynchronizeWithReadThread() 356 // The connection returns eConnectionStatusInterrupted only when there is 357 // no input pending to be read, so we can signal that. 358 comm->BroadcastEvent(eBroadcastBitNoMorePendingInput); 359 break; 360 case eConnectionStatusNoConnection: // No connection 361 case eConnectionStatusLostConnection: // Lost connection while connected to 362 // a valid connection 363 done = true; 364 LLVM_FALLTHROUGH; 365 case eConnectionStatusTimedOut: // Request timed out 366 if (error.Fail()) 367 LLDB_LOG(log, "error: {0}, status = {1}", error, 368 Communication::ConnectionStatusAsString(status)); 369 break; 370 } 371 } 372 log = lldb_private::GetLogIfAllCategoriesSet(LIBLLDB_LOG_COMMUNICATION); 373 if (log) 374 LLDB_LOGF(log, "%p Communication::ReadThread () thread exiting...", p); 375 376 // Handle threads wishing to synchronize with us. 377 { 378 // Prevent new ones from showing up. 379 comm->m_read_thread_did_exit = true; 380 381 // Unblock any existing thread waiting for the synchronization signal. 382 comm->BroadcastEvent(eBroadcastBitNoMorePendingInput); 383 384 // Wait for the thread to finish... 385 std::lock_guard<std::mutex> guard(comm->m_synchronize_mutex); 386 // ... and disconnect. 387 if (disconnect) 388 comm->Disconnect(); 389 } 390 391 // Let clients know that this thread is exiting 392 comm->BroadcastEvent(eBroadcastBitReadThreadDidExit); 393 return {}; 394 } 395 396 void Communication::SetReadThreadBytesReceivedCallback( 397 ReadThreadBytesReceived callback, void *callback_baton) { 398 m_callback = callback; 399 m_callback_baton = callback_baton; 400 } 401 402 void Communication::SynchronizeWithReadThread() { 403 // Only one thread can do the synchronization dance at a time. 404 std::lock_guard<std::mutex> guard(m_synchronize_mutex); 405 406 // First start listening for the synchronization event. 407 ListenerSP listener_sp( 408 Listener::MakeListener("Communication::SyncronizeWithReadThread")); 409 listener_sp->StartListeningForEvents(this, eBroadcastBitNoMorePendingInput); 410 411 // If the thread is not running, there is no point in synchronizing. 412 if (!m_read_thread_enabled || m_read_thread_did_exit) 413 return; 414 415 // Notify the read thread. 416 m_connection_sp->InterruptRead(); 417 418 // Wait for the synchronization event. 419 EventSP event_sp; 420 listener_sp->GetEvent(event_sp, llvm::None); 421 } 422 423 void Communication::SetConnection(std::unique_ptr<Connection> connection) { 424 Disconnect(nullptr); 425 StopReadThread(nullptr); 426 m_connection_sp = std::move(connection); 427 } 428 429 std::string 430 Communication::ConnectionStatusAsString(lldb::ConnectionStatus status) { 431 switch (status) { 432 case eConnectionStatusSuccess: 433 return "success"; 434 case eConnectionStatusError: 435 return "error"; 436 case eConnectionStatusTimedOut: 437 return "timed out"; 438 case eConnectionStatusNoConnection: 439 return "no connection"; 440 case eConnectionStatusLostConnection: 441 return "lost connection"; 442 case eConnectionStatusEndOfFile: 443 return "end of file"; 444 case eConnectionStatusInterrupted: 445 return "interrupted"; 446 } 447 448 return "@" + std::to_string(status); 449 } 450