[lldb] [MainLoopPosix] Fix crash upon adding lots of pending callbacks

[mgorny]

Summary

If lots of pending callbacks are added while the main loop has exited
already, the trigger pipe buffer fills in, causing the write to fail
and the related assertion to fail. To avoid this, add a boolean member
indicating whether the callbacks have been triggered already.
If the trigger was done, avoid writing to the pipe until loops proceeds
to run them and resets the variable.

Besides fixing the issue, this also avoids writing to the pipe multiple
times if callbacks are added faster than the loop is able to process
them. Previously, this would lead to the loop performing multiple read
iterations from pipe unnecessarily.

[labath]

I never expected we would have so many callbacks that we'd have to worry about this, but yes, this is one way to fix the problem. Another (slightly simpler, but also less performant) would be to make the write pipe nonblocking, and do not treat EAGAIN as an error.

lldb/source/Host/posix/MainLoopPosix.cpp

408

I /think/ this is not right. The other thread can wake up as soon as the Write call is done, and can proceed to clear the done flag before we are able to set it. If we set it afterwards, then we we suppress all subsequent writes, and the callbacks would never run. If we set the flag before we make the Write call, then it should be ok -- though the flag should probably have a different name then.

[mgorny]

In D135516#3846182, @labath wrote:

I never expected we would have so many callbacks that we'd have to worry about this, but yes, this is one way to fix the problem. Another (slightly simpler, but also less performant) would be to make the write pipe nonblocking, and do not treat EAGAIN as an error.

Well, I've only hit this in a pathological case (sending a lot of small packets without checking whether the loop in the communication thread didn't exit) but I've figured out we want some protection anyway.

The pipe logic uses select helper anyway, so it does return a timeout error if the buffer is full. However, I've figured that it'd be better to avoid polluting the buffer with tons of "interrupts" if only one is really meaningful.

[mgorny]

lldb/source/Host/posix/MainLoopPosix.cpp

408

I think you are correct. Will update that soonish.

[mgorny]

Move and rename the bool.

[labath]

lldb/source/Host/posix/MainLoopPosix.cpp

399–402

Maybe something like if (m_triggering.exchange(true)) return; ?
This version should work as well, but it may save someone from wondering what will happen if two threads execute this concurrently.

[mgorny]

Thanks. I'll do a fresh test run on the updated version and push if it passes.

lldb/source/Host/posix/MainLoopPosix.cpp

399–402

Sure, I'll try that.

lldb/include/lldb/Host/posix/MainLoopPosix.h

 //===-- MainLoopPosix.h -----------------------------------------*- C++ -*-===//
 //
 // Part of the LLVM Project, under the Apache License v2.0 with LLVM Exceptions.
 // See https://llvm.org/LICENSE.txt for license information.
 // SPDX-License-Identifier: Apache-2.0 WITH LLVM-exception
 //
 //===----------------------------------------------------------------------===//
 
 #ifndef LLDB_HOST_POSIX_MAINLOOPPOSIX_H
 #define LLDB_HOST_POSIX_MAINLOOPPOSIX_H
 
 #include "lldb/Host/Config.h"
 #include "lldb/Host/MainLoopBase.h"
 #include "lldb/Host/Pipe.h"
 #include "llvm/ADT/DenseMap.h"
+#include <atomic>
 #include <csignal>
 #include <list>
 #include <vector>
 
 namespace lldb_private {
 
 // Implementation of the MainLoopBase class. It can monitor file descriptors for
 // readability using ppoll, kqueue, or pselect. In addition to the common base,
struct SignalInfo {
     struct sigaction old_action;
     bool was_blocked : 1;
   };
   class RunImpl;
 
   llvm::DenseMap<IOObject::WaitableHandle, Callback> m_read_fds;
   llvm::DenseMap<int, SignalInfo> m_signals;
   Pipe m_trigger_pipe;
+  std::atomic<bool> m_triggering;
 #if HAVE_SYS_EVENT_H
   int m_kqueue;
 #endif
 };
 
 } // namespace lldb_private
 
 #endif // LLDB_HOST_POSIX_MAINLOOPPOSIX_H

lldb/source/Host/posix/MainLoopPosix.cpp

for (const auto &signal : signals) {
     if (loop.m_terminate_request)
       return;
     g_signal_flags[signal] = 0;
     loop.ProcessSignal(signal);
   }
 }
 #endif
 
-MainLoopPosix::MainLoopPosix() {
+MainLoopPosix::MainLoopPosix() : m_triggering(false) {
   Status error = m_trigger_pipe.CreateNew(/*child_process_inherit=*/false);
   assert(error.Success());
   const int trigger_pipe_fd = m_trigger_pipe.GetReadFileDescriptor();
   m_read_fds.insert({trigger_pipe_fd, [trigger_pipe_fd](MainLoopBase &loop) {
                        char c;
                        ssize_t bytes_read = llvm::sys::RetryAfterSignal(
                            -1, ::read, trigger_pipe_fd, &c, 1);
                        assert(bytes_read == 1);
Status MainLoopPosix::Run() {
   while (!m_terminate_request &&
          (m_read_fds.size() > 1 || !m_signals.empty())) {
     error = impl.Poll();
     if (error.Fail())
       return error;
 
     impl.ProcessEvents();
 
+    m_triggering = false;
     ProcessPendingCallbacks();
   }
   return Status();
 }
 
 void MainLoopPosix::ProcessReadObject(IOObject::WaitableHandle handle) {
   auto it = m_read_fds.find(handle);
   if (it != m_read_fds.end())
     it->second(*this); // Do the work
 }
 
 void MainLoopPosix::ProcessSignal(int signo) {
   auto it = m_signals.find(signo);
   if (it != m_signals.end()) {
     // The callback may actually register/unregister signal handlers,
     // so we need to create a copy first.
     llvm::SmallVector<Callback, 4> callbacks_to_run{
         it->second.callbacks.begin(), it->second.callbacks.end()};
     for (auto &x : callbacks_to_run)
       x(*this); // Do the work
   }
 }
 
 void MainLoopPosix::TriggerPendingCallbacks() {
+  if (m_triggering.exchange(true))
+    return;
+
   char c = '.';
   size_t bytes_written;
   Status error = m_trigger_pipe.Write(&c, 1, bytes_written);
   assert(error.Success());
   UNUSED_IF_ASSERT_DISABLED(error);
   assert(bytes_written == 1);
 }

lldb/unittests/Host/MainLoopTest.cpp

std::thread callback2_adder([&]() {
     });
   });
   ASSERT_THAT_ERROR(loop.Run().ToError(), llvm::Succeeded());
   callback2_adder.join();
   ASSERT_TRUE(callback1_called);
   ASSERT_TRUE(callback2_called);
 }
 
+// Regression test for assertion failure if a lot of callbacks end up
+// being queued after loop exits.
+TEST_F(MainLoopTest, PendingCallbackAfterLoopExited) {
+  MainLoop loop;
+  Status error;
+  ASSERT_TRUE(loop.Run().Success());
+  // Try to fill the pipe buffer in.
+  for (int i = 0; i < 65536; ++i)
+    loop.AddPendingCallback([&](MainLoopBase &loop) {});
+}
+
 #ifdef LLVM_ON_UNIX
 TEST_F(MainLoopTest, DetectsEOF) {
 
   PseudoTerminal term;
   ASSERT_THAT_ERROR(term.OpenFirstAvailablePrimary(O_RDWR), llvm::Succeeded());
   ASSERT_THAT_ERROR(term.OpenSecondary(O_RDWR | O_NOCTTY), llvm::Succeeded());
   auto conn = std::make_unique<ConnectionFileDescriptor>(
       term.ReleasePrimaryFileDescriptor(), true);