[lldb] Fix ThreadedCommunication races

[labath]

Summary

The Read function could end up blocking if data (or EOF) arrived just as
it was about to start waiting for the events. This was only discovered
now, because we did not have unit tests for this functionality before.
We need to check for data *after* we start listening for incoming
events. There were no changes to the read thread code needed, as we
already use this pattern in SynchronizeWithReadThread, so I just updated
the comments to make it clear that it is used for reading as well.

[labath]

I've only seen one flake on linux so far, but the windows bot seems more susceptible to this problem.

[mgorny]

lldb/source/Core/ThreadedCommunication.cpp

113–119

Can you think of any reason not to move listener setup before the first GetCachedBytes() call instead of duplicating it?

[labath]

lldb/source/Core/ThreadedCommunication.cpp

113–119

The only possible reason is "efficiency" (avoiding the creation of the listener and all that goes with it). But I'm definitely not convinced that this actually matters.

[mgorny]

Thanks for noticing and fixing this.

lldb/source/Core/ThreadedCommunication.cpp

113–119

Ok, I'm not bent on it either way.

lldb/source/Core/ThreadedCommunication.cpp

if (m_read_thread_enabled) {
 
     if (!m_connection_sp) {
       if (error_ptr)
         error_ptr->SetErrorString("Invalid connection.");
       status = eConnectionStatusNoConnection;
       return 0;
     }
 
+    // No data yet, we have to start listening.
     ListenerSP listener_sp(
         Listener::MakeListener("ThreadedCommunication::Read"));
     listener_sp->StartListeningForEvents(
         this, eBroadcastBitReadThreadGotBytes | eBroadcastBitReadThreadDidExit);
+
+    // Re-check for data, as it might have arrived while we were setting up our
+    // listener.
+    cached_bytes = GetCachedBytes(dst, dst_len);
+    if (cached_bytes > 0) {
+      status = eConnectionStatusSuccess;
+      return cached_bytes;
+    }
+
     EventSP event_sp;
-    while (listener_sp->GetEvent(event_sp, timeout)) {
-      const uint32_t event_type = event_sp->GetType();
-      if (event_type & eBroadcastBitReadThreadGotBytes) {
-        return GetCachedBytes(dst, dst_len);
-      }
+    // Explicitly check for the thread exit, for the same reason.
+    if (m_read_thread_did_exit) {
+      // We've missed the event, lets just conjure one up.
+      event_sp = std::make_shared<Event>(eBroadcastBitReadThreadDidExit);
+    } else {
+      if (!listener_sp->GetEvent(event_sp, timeout)) {
+        if (error_ptr)
+          error_ptr->SetErrorString("Timed out.");
+        status = eConnectionStatusTimedOut;
+        return 0;
+      }
+    }
 
-      if (event_type & eBroadcastBitReadThreadDidExit) {
-        // If the thread exited of its own accord, it either means it
-        // hit an end-of-file condition or an error.
-        status = m_pass_status;
-        if (error_ptr)
-          *error_ptr = std::move(m_pass_error);
 
-        if (GetCloseOnEOF())
-          Disconnect(nullptr);
-        return 0;
-      }
-    }
-
-    if (error_ptr)
-      error_ptr->SetErrorString("Timed out.");
-    status = eConnectionStatusTimedOut;
-    return 0;
+    llvm_unreachable("Got unexpected event type!");
   }
 
   // We aren't using a read thread, just read the data synchronously in this
   // thread.
   return Communication::Read(dst, dst_len, timeout, status, error_ptr);
 }
 
 bool ThreadedCommunication::StartReadThread(Status *error_ptr) {
case eConnectionStatusTimedOut: // Request timed out
                  ThreadedCommunication::ConnectionStatusAsString(status));
       break;
     }
   }
   m_pass_status = status;
   m_pass_error = std::move(error);
   LLDB_LOG(log, "Communication({0}) thread exiting...", this);
 
-  // Handle threads wishing to synchronize with us.
-  {
-    // Prevent new ones from showing up.
-    m_read_thread_did_exit = true;
+  // Start shutting down. We need to do this in a very specific order to ensure
+  // we don't race with threads wanting to read/synchronize with us.
+
+  // First, we signal our intent to exit. This ensures no new thread start
+  // waiting on events from us.
 
-    // Unblock any existing thread waiting for the synchronization signal.
-    BroadcastEvent(eBroadcastBitNoMorePendingInput);
 
-    // Wait for the thread to finish...
+  {
+    // Wait for the synchronization thread to finish...
     std::lock_guard<std::mutex> guard(m_synchronize_mutex);
     // ... and disconnect.
     if (disconnect)
       Disconnect();
   }
 
-  // Let clients know that this thread is exiting
+  // Finally, unblock any readers waiting for us to exit.
   BroadcastEvent(eBroadcastBitReadThreadDidExit);
   return {};
 }
 
 void ThreadedCommunication::SetReadThreadBytesReceivedCallback(
     ReadThreadBytesReceived callback, void *callback_baton) {
   m_callback = callback;
   m_callback_baton = callback_baton;