[WIP][lldb] Add "event" capability to the MainLoop class

[mgorny]

To be honest, I'm not sure if "event" is the best name for it. I get it's a WSA name but it's a bit non-obvious to outsiders (I mean, technically everything is an event). Not that I have a better name in mind.

lldb/include/lldb/Host/MainLoop.h

66 ↗

(On Diff #449941)

Could you include a short comment explaining how it's used? I suppose something like "will be executed upon being explicitly signaled via NotifyEvent() from any thread".

lldb/source/Host/common/MainLoop.cpp

121 ↗

(On Diff #449941)

I think the reliance on < vs <= here could be confusing. Perhaps add a note that we popped the "event event" above, or maybe add a last_event const-var prior to popping it.

133 ↗

(On Diff #449941)

Let's make sure something didn't corrupt the state.

[mgorny]

BTW do you need me to implement the POSIX counterpart to this?

[labath]

In D131160#3701839, @mgorny wrote:

To be honest, I'm not sure if "event" is the best name for it. I get it's a WSA name but it's a bit non-obvious to outsiders (I mean, technically everything is an event). Not that I have a better name in mind.

Yeah, I know... I checked the thesaurus for "event", and I didn't find anything reasonable. And "signal" is obviously a bad idea..

Another possibility (one that just occurred to me right now) would be to use a completely different paradigm to achieve the effect. For example, we could extend the PendingCallback mechanism to enable adding callbacks from different threads (and have them wake the loop if necessary). That would be a more generic mechanism, and it would avoid the need to introduce a new concept (and having to name it).

In D131160#3701907, @mgorny wrote:

BTW do you need me to implement the POSIX counterpart to this?

That would be appreciated, but I think we first need to figure out the API, and how this code is going to be structured (I'm going to try handling the second part now)...

[mgorny]

Rebase on top of D131159 and partially implement the Linux counterpart.

Note that I have only covered sys/event.h for now and that I haven't tested the updated Windows code.

[mgorny]

lldb/source/Host/posix/MainLoopPosix.cpp

74–82

I'm wondering whether I should be repeating this all over the place or perhaps just adding m_event_pipe to m_read_fds.

108–112

Also I'm wondering about adding a ProcessEvents() wrapper that would take care of this.

[mgorny]

Ok, I've just realized that the pipe writing solution isn't going to work when the notification happens from the same thread (as the unittest does right now). @labath, do you happen to have an idea how to solve that reasonably cleanly?

[mgorny]

In D131160#3726642, @mgorny wrote:

Ok, I've just realized that the pipe writing solution isn't going to work when the notification happens from the same thread (as the unittest does right now). @labath, do you happen to have an idea how to solve that reasonably cleanly?

Ah, nevermind, for some reason I thought pipe writes are going to block until read happens. Perhaps we don't need to worry about that.

[mgorny]

Add m_event_pipe handler to m_read_fds to avoid reinventing the wheel three times.

[mgorny]

Add a test that notifying actually works on top of pending read.

[labath]

Thanks for taking care of the posix part.

I'm sort of warming up to the idea of reusing pending callback machinery for this. Besides avoiding the awkward naming problem, it also avoid another somewhat strange situation where our EventHandle offer additional actions on the event (which is different from all our other handle types, which only support deletion). All it would take is to add some locking around the PendingCallback manipulations, and write to the pipe (or trigger the event) whenever a new callback is registered, so that the Run function can invoke it.

It's a slightly different programming paradigm, and I suppose it also depends on which mechanism would be more natural for our InterruptRead use case (have you tried implementing it using this?), but I think the two mechanisms should be equivalent (as in, one could implement one using the other -- fairly easily).

What do you think about that?

In D131160#3726698, @mgorny wrote:

In D131160#3726642, @mgorny wrote:

Ok, I've just realized that the pipe writing solution isn't going to work when the notification happens from the same thread (as the unittest does right now). @labath, do you happen to have an idea how to solve that reasonably cleanly?

Ah, nevermind, for some reason I thought pipe writes are going to block until read happens. Perhaps we don't need to worry about that.

A pipe has a fixed (on linux, it's configurable) capacity, and it should be sufficiently large that we don't have to worry about it. If we wanted to be super safe, we could make the writes non-blocking. It should be safe to ignore EAGAINs in this situation, as that means the pipe already contains some unprocessed bytes, and all of the pending notifications will be processed once the reading thread "catches up".

lldb/source/Host/posix/MainLoopPosix.cpp

242

I like this.

[mgorny]

In D131160#3732668, @labath wrote:

Thanks for taking care of the posix part.

I'm sort of warming up to the idea of reusing pending callback machinery for this. Besides avoiding the awkward naming problem, it also avoid another somewhat strange situation where our EventHandle offer additional actions on the event (which is different from all our other handle types, which only support deletion). All it would take is to add some locking around the PendingCallback manipulations, and write to the pipe (or trigger the event) whenever a new callback is registered, so that the Run function can invoke it.

To be honest, I don't have a strong opinion, so I leave the decision to you. I was also thinking about this but decided to avoid changing the paradigm while rebasing/finishing the existing code, especially that I can't test it on Windows. But I can do it if you like.

It's a slightly different programming paradigm, and I suppose it also depends on which mechanism would be more natural for our InterruptRead use case (have you tried implementing it using this?), but I think the two mechanisms should be equivalent (as in, one could implement one using the other -- fairly easily).

Actually, I've started experimenting on rewriting the read thread option in Communication using this (with the goal of using the thread both for reads and writes).

lldb/source/Host/posix/MainLoopPosix.cpp

242

Thanks. I definitely like it more than doing it three times too ;-).

[mgorny]

Ok, so I'm going to give it a shot today. Could you clarify one thing though?

All it would take is to add some locking around the PendingCallback manipulations

Are you thinking of locking directly in MainLoop classes, or in the caller? It'd feel a bit inconsistent if MainLoops provided locking around pending callback manipulations and not other methods. However, I suppose making everything thread-safe wouldn't hurt either.

[labath]

In D131160#3735024, @mgorny wrote:

Ok, so I'm going to give it a shot today. Could you clarify one thing though?

All it would take is to add some locking around the PendingCallback manipulations

Are you thinking of locking directly in MainLoop classes, or in the caller? It'd feel a bit inconsistent if MainLoops provided locking around pending callback manipulations and not other methods. However, I suppose making everything thread-safe wouldn't hurt either.

I'm not sure this can be done from the outside, as the callback adittions need to synchronize with the accesses that happen inside the Run function. It is somewhat inconsistent, but for example python has special call_soon/call_soon_threadsafe versions for code which needs to schedule callbacks from other threads. We could do something like that. Or we could also just lock everything...

[mgorny]

Rebase. Replace "events" with triggering pending callbacks.

[mgorny]

@labath, what do you think of this?

Also I'm wondering about error handling. Should we introduce fallible constructor for this?

[labath]

What's the reasoning behind TriggerPendingCallbacks? I was assuming that the addition of a callback would cause it to run automatically...

[labath]

In D131160#3735372, @mgorny wrote:

Also I'm wondering about error handling. Should we introduce fallible constructor for this?

I'm not particularly worried about this. I might even make the Pipe constructor non-fallible, as the only way that can fail is if we exhaust our file descriptor limit -- and I very much doubt our ability to recover from that.

[mgorny]

In that case, is there something more I should do about this patch or are you going to take over from here?

[labath]

In D131160#3751959, @mgorny wrote:

In that case, is there something more I should do about this patch or are you going to take over from here?

It wasn't clear to me whether you intend to finish this patch (and I was waiting on your response to this). That said, if you don't have any strong opinions there, then I can try to finish this.

[mgorny]

In D131160#3738805, @labath wrote:

What's the reasoning behind TriggerPendingCallbacks? I was assuming that the addition of a callback would cause it to run automatically...

To be honest, I didn't think about it much. The original idea is that we can either have "non-important" callbacks that will be run at some point in the future, and "important" callbacks that should be run ASAP. However, I'm not really married to this idea, so I can make them triggered immediately if you prefer that.

In D131160#3761163, @labath wrote:

In D131160#3751959, @mgorny wrote:

In that case, is there something more I should do about this patch or are you going to take over from here?

It wasn't clear to me whether you intend to finish this patch (and I was waiting on your response to this). That said, if you don't have any strong opinions there, then I can try to finish this.

Ah, sorry, missed that comment. I can change that logic if you prefer. Otherwise, I think the next thing that needs to be here is testing/fixing the Windows code, and I'd be grateful if you could do that.

[labath]

Back to me.

[labath]

• Make TriggerPendingCallbacks protected
• fix the windows version

[labath]

In D131160#3763936, @mgorny wrote:

In D131160#3738805, @labath wrote:

What's the reasoning behind TriggerPendingCallbacks? I was assuming that the addition of a callback would cause it to run automatically...

To be honest, I didn't think about it much. The original idea is that we can either have "non-important" callbacks that will be run at some point in the future, and "important" callbacks that should be run ASAP. However, I'm not really married to this idea, so I can make them triggered immediately if you prefer that.

I don't think we need to make that distinction.

In D131160#3761163, @labath wrote:

In D131160#3751959, @mgorny wrote:

In that case, is there something more I should do about this patch or are you going to take over from here?

It wasn't clear to me whether you intend to finish this patch (and I was waiting on your response to this). That said, if you don't have any strong opinions there, then I can try to finish this.

Ah, sorry, missed that comment. I can change that logic if you prefer. Otherwise, I think the next thing that needs to be here is testing/fixing the Windows code, and I'd be grateful if you could do that.

I've fixed the windows version, and I've removed TriggerPendingCallbacks (from the public interface).

I think this should be good to go now.

[mgorny]

LGTM. I've just rebased and tested my patches on top of this, and they seem to work.

[mgorny]

I think this may have broken tests on Darwin.

[labath]

Are you sure about that? https://green.lab.llvm.org/green/view/LLDB/job/lldb-cmake/46639 is just running the commit now. I'd actually suspect the breakage is caused by D133181.

[mgorny]

In D131160#3771693, @labath wrote:

Are you sure about that? https://green.lab.llvm.org/green/view/LLDB/job/lldb-cmake/46639 is just running the commit now. I'd actually suspect the breakage is caused by D133181.

Ah, sorry, I was just guessing since it looked relevant.

lldb/include/lldb/Host/MainLoopBase.h

 //===-- MainLoopBase.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_MAINLOOPBASE_H
 #define LLDB_HOST_MAINLOOPBASE_H
 
 #include "lldb/Utility/IOObject.h"
 #include "lldb/Utility/Status.h"
 #include "llvm/ADT/DenseMap.h"
 #include "llvm/Support/ErrorHandling.h"
+
+#include <atomic>
 #include <functional>
+#include <list>
 
 namespace lldb_private {
 
 // The purpose of this class is to enable multiplexed processing of data from
 // different sources without resorting to multi-threading. Clients can register
 // IOObjects, which will be monitored for readability, and when they become
 // ready, the specified callback will be invoked. Monitoring for writability is
 // not supported, but can be easily added if needed.
 //
 // The RegisterReadObject function return a handle, which controls the duration
 // of the monitoring. When this handle is destroyed, the callback is
 // deregistered.
 //
 // Since this class is primarily intended to be used for single-threaded
 // processing, it does not attempt to perform any internal synchronisation and
 // any concurrent accesses must be protected  externally. However, it is
 // perfectly legitimate to have more than one instance of this class running on
 // separate threads, or even a single thread.
 class MainLoopBase {
 private:
   class ReadHandle;
+  class EventHandle;
 
 public:
   MainLoopBase() : m_terminate_request(false) {}
   virtual ~MainLoopBase() { assert(m_read_fds.size() == 0); }
 
   typedef std::unique_ptr<ReadHandle> ReadHandleUP;
+  typedef std::unique_ptr<EventHandle> EventHandleUP;
 
   typedef std::function<void(MainLoopBase &)> Callback;
 
   virtual ReadHandleUP RegisterReadObject(const lldb::IOObjectSP &object_sp,
                                           const Callback &callback,
                                           Status &error);
 
+  virtual EventHandleUP RegisterEvent(const Callback &callback);
+
   // Add a pending callback that will be executed once after all the pending
   // events are processed. The callback will be executed even if termination
   // was requested.
   virtual void AddPendingCallback(const Callback &callback) {
     m_pending_callbacks.push_back(callback);
   }
 
   // Waits for registered events and invoke the proper callbacks. Returns when
   // all callbacks deregister themselves or when someone requests termination.
   virtual Status Run() { llvm_unreachable("Not implemented"); }
 
   // This should only be performed from a callback. Do not attempt to terminate
   // the processing from another thread.
   virtual void RequestTermination() { m_terminate_request = true; }
 
 protected:
+  struct EventInfo {
+    EventInfo(Callback callback)
+        : notified(false), callback(std::move(callback)) {}
+    std::atomic<bool> notified;
+    Callback callback;
+  };
+
   ReadHandleUP CreateReadHandle(const lldb::IOObjectSP &object_sp) {
     return ReadHandleUP(new ReadHandle(*this, object_sp->GetWaitableHandle()));
   }
 
   virtual void UnregisterReadObject(IOObject::WaitableHandle handle);
 
+  EventHandleUP CreateEventHandle(const Callback &callback) {
+    return EventHandleUP(
+        new EventHandle(*this, m_events.emplace(m_events.end(), callback)));
+  }
+
+  virtual void UnregisterEvent(std::list<EventInfo>::iterator info_it);
+
   void ProcessReadObject(IOObject::WaitableHandle handle);
+  void ProcessEvents();
   void ProcessPendingCallbacks();
 
+  virtual void NotifyEvent(std::list<EventInfo>::iterator info_it) = 0;
+
   llvm::DenseMap<IOObject::WaitableHandle, Callback> m_read_fds;
+  std::list<EventInfo> m_events;
   std::vector<Callback> m_pending_callbacks;
   bool m_terminate_request : 1;
 
 private:
   class ReadHandle {
   public:
     ~ReadHandle() { m_mainloop.UnregisterReadObject(m_handle); }
 
   private:
     ReadHandle(MainLoopBase &mainloop, IOObject::WaitableHandle handle)
         : m_mainloop(mainloop), m_handle(handle) {}
 
     MainLoopBase &m_mainloop;
     IOObject::WaitableHandle m_handle;
 
     friend class MainLoopBase;
     ReadHandle(const ReadHandle &) = delete;
     const ReadHandle &operator=(const ReadHandle &) = delete;
   };
 
+  class EventHandle {
+  public:
+    ~EventHandle() { m_mainloop.UnregisterEvent(m_info_it); }
+
+    void Notify() { m_mainloop.NotifyEvent(m_info_it); }
+
+  private:
+    EventHandle(MainLoopBase &mainloop, std::list<EventInfo>::iterator info_it)
+        : m_mainloop(mainloop), m_info_it(info_it) {}
+
+    MainLoopBase &m_mainloop;
+    std::list<EventInfo>::iterator m_info_it;
+
+    friend class MainLoopBase;
+    EventHandle(const EventHandle &) = delete;
+    const EventHandle &operator=(const EventHandle &) = delete;
+  };
+
   MainLoopBase(const MainLoopBase &) = delete;
   const MainLoopBase &operator=(const MainLoopBase &) = delete;
 };
 
 } // namespace lldb_private
 
 #endif // LLDB_HOST_MAINLOOPBASE_H

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 <csignal>
 #include <list>
 #include <vector>
 
 namespace lldb_private {
 
 // Implementation of the MainLoopBase class. It can monitor file descriptors for
public:
   // a normal function and perform things which would not be safe in a signal
   // handler. However, since the callback is not invoked synchronously, you
   // cannot use this mechanism to handle SIGSEGV and the like.
   SignalHandleUP RegisterSignal(int signo, const Callback &callback,
                                 Status &error);
 
   Status Run() override;
 
+  void NotifyEvent(std::list<EventInfo>::iterator info_it) override;
+
 protected:
   void UnregisterSignal(int signo, std::list<Callback>::iterator callback_it);
 
 private:
   void ProcessSignal(int signo);
 
   class SignalHandle {
   public:
private:
   struct SignalInfo {
     std::list<Callback> callbacks;
     struct sigaction old_action;
     bool was_blocked : 1;
   };
   class RunImpl;
 
   llvm::DenseMap<int, SignalInfo> m_signals;
+  Pipe m_event_pipe;
 #if HAVE_SYS_EVENT_H
   int m_kqueue;
 #endif
 };
 
 } // namespace lldb_private
 
 #endif // LLDB_HOST_POSIX_MAINLOOPPOSIX_H

lldb/include/lldb/Host/windows/MainLoopWindows.h

 
 namespace lldb_private {
 
 // Windows-specific implementation of the MainLoopBase class. It can monitor
 // socket descriptors for readability using WSAEventSelect. Non-socket file
 // descriptors are not supported.
 class MainLoopWindows : public MainLoopBase {
 public:
+  MainLoopWindows();
+  ~MainLoopWindows() override;
+
   ReadHandleUP RegisterReadObject(const lldb::IOObjectSP &object_sp,
                                   const Callback &callback,
                                   Status &error) override;
 
   Status Run() override;
 
+  void NotifyEvent(std::list<EventInfo>::iterator info_it) override;
+
 private:
   llvm::Expected<size_t> Poll();
+  void *m_event_event;
 };
 
 } // namespace lldb_private
 
 #endif // LLDB_HOST_WINDOWS_MAINLOOPWINDOWS_H

lldb/source/Host/common/MainLoopBase.cpp

if (!inserted) {
     error.SetErrorStringWithFormat("File descriptor %d already monitored.",
                                    object_sp->GetWaitableHandle());
     return nullptr;
   }
 
   return CreateReadHandle(object_sp);
 }
 
+MainLoopBase::EventHandleUP
+MainLoopBase::RegisterEvent(const Callback &callback) {
+  return CreateEventHandle(callback);
+}
+
 void MainLoopBase::UnregisterReadObject(IOObject::WaitableHandle handle) {
   bool erased = m_read_fds.erase(handle);
   UNUSED_IF_ASSERT_DISABLED(erased);
   assert(erased);
 }
 
+void MainLoopBase::UnregisterEvent(std::list<EventInfo>::iterator info_it) {
+  m_events.erase(info_it);
+}
+
 void MainLoopBase::ProcessReadObject(IOObject::WaitableHandle handle) {
   auto it = m_read_fds.find(handle);
   if (it != m_read_fds.end())
     it->second(*this); // Do the work
 }
 
+void MainLoopBase::ProcessEvents() {
+  // Create a copy as some events may become unregistered by the callbacks.
+  llvm::SmallVector<Callback, 4> callbacks_to_run;
+  for (auto &info : m_events) {
+    if (info.notified.exchange(false, std::memory_order_acq_rel) == true)
+      callbacks_to_run.push_back(info.callback);
+  }
+  for (auto &cb : callbacks_to_run)
+    cb(*this); // Do the work
+}
+
 void MainLoopBase::ProcessPendingCallbacks() {
   for (const Callback &callback : m_pending_callbacks)
     callback(*this);
   m_pending_callbacks.clear();
 }

lldb/source/Host/posix/MainLoopPosix.cpp

 //===-- MainLoopPosix.cpp -------------------------------------------------===//
 //
 // 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
 //
 //===----------------------------------------------------------------------===//
 
 #include "lldb/Host/posix/MainLoopPosix.h"
+
 #include "lldb/Host/Config.h"
 #include "lldb/Host/PosixApi.h"
 #include "lldb/Utility/Status.h"
 #include "llvm/Config/llvm-config.h"
 #include <algorithm>
 #include <cassert>
 #include <cerrno>
 #include <csignal>

 #endif
 
   sigset_t get_sigmask();
 #endif
 };
 
 #if HAVE_SYS_EVENT_H
 MainLoopPosix::RunImpl::RunImpl(MainLoopPosix &loop) : loop(loop) {
-  in_events.reserve(loop.m_read_fds.size());
+  in_events.reserve(loop.m_read_fds.size() + 1);
 }
 
 Status MainLoopPosix::RunImpl::Poll() {
-  in_events.resize(loop.m_read_fds.size());
+  in_events.resize(loop.m_read_fds.size() + 1);
   unsigned i = 0;
   for (auto &fd : loop.m_read_fds)
     EV_SET(&in_events[i++], fd.first, EVFILT_READ, EV_ADD, 0, 0, 0);
+  EV_SET(&in_events[i++], m_event_pipe.GetReadFileDescriptor());
 
   num_events = kevent(loop.m_kqueue, in_events.data(), in_events.size(),
                       out_events, llvm::array_lengthof(out_events), nullptr);
 
   if (num_events < 0) {
     if (errno == EINTR) {
       // in case of EINTR, let the main loop run one iteration
       // we need to zero num_events to avoid assertions failing
       num_events = 0;
     } else
       return Status(errno, eErrorTypePOSIX);
   }
   return Status();
 }
 
 void MainLoopPosix::RunImpl::ProcessEvents() {
+  const int event_pipe_fd = m_event_pipe.GetReadFileDescriptor();
+
   assert(num_events >= 0);
   for (int i = 0; i < num_events; ++i) {
     if (loop.m_terminate_request)
       return;
     switch (out_events[i].filter) {
     case EVFILT_READ:
+      if (out_events[i].ident == event_pipe_fd) {
+        char c;
+        ssize_t bytes_read =
+            llvm::sys::RetryAfterSignal(-1, ::read, event_pipe_fd, &c, 1);
+        assert(bytes_read == 1);
+        UNUSED_IF_ASSERT_DISABLED(bytes_read);
+        loop.ProcessEvents();
+      } else
         loop.ProcessReadObject(out_events[i].ident);
       break;
     case EVFILT_SIGNAL:
       loop.ProcessSignal(out_events[i].ident);
       break;
     default:
       llvm_unreachable("Unknown event");
     }
   }
 }
 #else
 MainLoopPosix::RunImpl::RunImpl(MainLoopPosix &loop) : loop(loop) {
 #ifndef __ANDROID__
-  read_fds.reserve(loop.m_read_fds.size());
+  read_fds.reserve(loop.m_read_fds.size() + 1);
 #endif
 }
 
 sigset_t MainLoopPosix::RunImpl::get_sigmask() {
   sigset_t sigmask;
   int ret = pthread_sigmask(SIG_SETMASK, nullptr, &sigmask);
   assert(ret == 0);
   (void)ret;
 
   for (const auto &sig : loop.m_signals)
     sigdelset(&sigmask, sig.first);
   return sigmask;
 }
 
 #ifdef __ANDROID__
 Status MainLoopPosix::RunImpl::Poll() {
   // ppoll(2) is not supported on older all android versions. Also, older
   // versions android (API <= 19) implemented pselect in a non-atomic way, as a
   // combination of pthread_sigmask and select. This is not sufficient for us,
   // as we rely on the atomicity to correctly implement signal polling, so we
   // call the underlying syscall ourselves.
 
   FD_ZERO(&read_fd_set);
   int nfds = 0;
+  // TODO: "event"
   for (const auto &fd : loop.m_read_fds) {
     FD_SET(fd.first, &read_fd_set);
     nfds = std::max(nfds, fd.first + 1);
   }
 
   union {
     sigset_t set;
     uint64_t pad;
Status MainLoopPosix::RunImpl::Poll() {
   return Status();
 }
 #else
 Status MainLoopPosix::RunImpl::Poll() {
   read_fds.clear();
 
   sigset_t sigmask = get_sigmask();
 
+  // TODO: "event"
   for (const auto &fd : loop.m_read_fds) {
     struct pollfd pfd;
     pfd.fd = fd.first;
     pfd.events = POLLIN;
     pfd.revents = 0;
     read_fds.push_back(pfd);
   }
 
if (loop.m_terminate_request)
       return;
     g_signal_flags[signal] = 0;
     loop.ProcessSignal(signal);
   }
 }
 #endif
 
 MainLoopPosix::MainLoopPosix() {
+  Status error = m_event_pipe.CreateNew(/*child_process_inherit=*/ false);
+  assert(error.Success());
 #if HAVE_SYS_EVENT_H
   m_kqueue = kqueue();
   assert(m_kqueue >= 0);
 #endif
 }
 
 MainLoopPosix::~MainLoopPosix() {
 #if HAVE_SYS_EVENT_H
   close(m_kqueue);
 #endif
+  m_event_pipe.Close();
   assert(m_signals.size() == 0);
 }
 
 // We shall block the signal, then install the signal handler. The signal will
 // be unblocked in the Run() function to check for signal delivery.
 MainLoopPosix::SignalHandleUP
 MainLoopPosix::RegisterSignal(int signo, const Callback &callback,
                               Status &error) {

 
 Status MainLoopPosix::Run() {
   m_terminate_request = false;
 
   Status error;
   RunImpl impl(*this);
 
   // run until termination or until we run out of things to listen to
-  while (!m_terminate_request && (!m_read_fds.empty() || !m_signals.empty())) {
+  while (!m_terminate_request &&
+         (!m_read_fds.empty() || !m_signals.empty() || !m_events.empty())) {
 
     error = impl.Poll();
     if (error.Fail())
       return error;
 
     impl.ProcessEvents();
 
     ProcessPendingCallbacks();
   }
   return Status();
 }
 
 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::NotifyEvent(std::list<EventInfo>::iterator info_it) {
+  // This must be set before we notify the event.
+  info_it->notified.store(true, std::memory_order_release);
+  char c = '.';
+  size_t bytes_written;
+  Status error = m_event_pipe.Write(&c, 1, bytes_written);
+  assert(error.Success());
+  UNUSED_IF_ASSERT_DISABLED(error);
+  assert(bytes_written == 1);
+}

lldb/source/Host/windows/MainLoopWindows.cpp

 #include <csignal>
 #include <ctime>
 #include <vector>
 #include <winsock2.h>
 
 using namespace lldb;
 using namespace lldb_private;
 
+MainLoopWindows::MainLoopWindows() {
+  m_event_event = WSACreateEvent();
+  assert(m_event_event != WSA_INVALID_EVENT);
+}
+
+MainLoopWindows::~MainLoopWindows() {
+  BOOL result = WSACloseEvent(m_event_event);
+  assert(result == TRUE);
+}
+
 llvm::Expected<size_t> MainLoopWindows::Poll() {
-  std::vector<WSAEVENT> read_events;
-  read_events.reserve(m_read_fds.size());
+  std::vector<WSAEVENT> events;
+  events.reserve(m_read_fds.size() + 1);
   for (auto &fd : m_read_fds) {
     WSAEVENT event = WSACreateEvent();
     assert(event != WSA_INVALID_EVENT);
 
     int result =
         WSAEventSelect(fd.first, event, FD_READ | FD_ACCEPT | FD_CLOSE);
     assert(result == 0);
 
-    read_events.push_back(event);
+    events.push_back(event);
   }
+  events.push_back(loop.m_event_event);
 
-  DWORD result = WSAWaitForMultipleEvents(
-      read_events.size(), read_events.data(), FALSE, WSA_INFINITE, FALSE);
+  DWORD result = WSAWaitForMultipleEvents(events.size(), events.data(), FALSE,
+                                          WSA_INFINITE, FALSE);
 
   for (auto &fd : m_read_fds) {
     int result = WSAEventSelect(fd.first, WSA_INVALID_EVENT, 0);
     assert(result == 0);
   }
 
-  for (auto &event : read_events) {
+  events.pop_back();
+  for (auto &event : events) {
     BOOL result = WSACloseEvent(event);
     assert(result == TRUE);
   }
 
-  if (result >= WSA_WAIT_EVENT_0 &&
-      result < WSA_WAIT_EVENT_0 + read_events.size())
+  if (result >= WSA_WAIT_EVENT_0 && result <= WSA_WAIT_EVENT_0 + events.size())
     return result - WSA_WAIT_EVENT_0;
 
   return llvm::createStringError(llvm::inconvertibleErrorCode(),
                                  "WSAWaitForMultipleEvents failed");
 }
 
 MainLoopWindows::ReadHandleUP
 MainLoopWindows::RegisterReadObject(const IOObjectSP &object_sp,
                                     const Callback &callback, Status &error) {
   if (object_sp->GetFdType() != IOObject::eFDTypeSocket) {
     error.SetErrorString(
         "MainLoopWindows: non-socket types unsupported on Windows");
     return nullptr;
   }
   return MainLoopBase::RegisterReadObject(object_sp, callback, error);
 }
 
 Status MainLoopWindows::Run() {
   m_terminate_request = false;
 
   Status error;
 
   // run until termination or until we run out of things to listen to
-  while (!m_terminate_request && !m_read_fds.empty()) {
+  while (!m_terminate_request && (!m_read_fds.empty() || !m_events.empty())) {
 
     llvm::Expected<size_t> signaled_event = Poll();
     if (!signaled_event)
       return Status(signaled_event.takeError());
 
+    if (*signaled_event < m_read_fds.size()) {
       auto &fd_info = *std::next(m_read_fds.begin(), *signaled_event);
-
       ProcessReadObject(fd_info.first);
+    } else {
+      assert(*signaled_event == m_read_fds.size());
+      // Must be called before events are processed.
+      WSAResetEvent(m_event_event);
+      ProcessEvents();
+    }
+
     ProcessPendingCallbacks();
   }
   return Status();
 }
+
+void MainLoopWindows::NotifyEvent(std::list<EventInfo>::iterator info_it) {
+  // This must be set before we notify the event.
+  info_it->notified.store(true, std::memory_order_release);
+  WSASetEvent(m_event_event);
+}

lldb/unittests/Host/MainLoopTest.cpp

auto handle = loop.RegisterReadObject(
       error);
   ASSERT_TRUE(error.Success());
   ASSERT_TRUE(handle);
   ASSERT_TRUE(loop.Run().Success());
   // 2 iterations of read callback + 1 call of pending callback.
   ASSERT_EQ(3u, callback_count);
 }
 
+TEST_F(MainLoopTest, Event) {
+  MainLoop loop;
+  bool event_called = false;
+  auto handle = loop.RegisterEvent([&](MainLoopBase &loop) {
+    event_called = true;
+    loop.RequestTermination();
+  });
+  handle->Notify();
+  ASSERT_THAT_ERROR(loop.Run().ToError(), llvm::Succeeded());
+  ASSERT_TRUE(event_called);
+}
+
 #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);