[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

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

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

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 ↗

(On Diff #452714)

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 ↗

(On Diff #452714)

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

230 ↗

(On Diff #453071)

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

230 ↗

(On Diff #453071)

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/MainLoop.h

 // 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 (with some limitations on signal
 // monitoring).
 // TODO: Add locking if this class is to be used in a multi-threaded context.
 class MainLoop : public MainLoopBase {
 private:
   class SignalHandle;
+  class EventHandle;
+  struct EventInfo;
 
 public:
   typedef std::unique_ptr<SignalHandle> SignalHandleUP;
+  typedef std::unique_ptr<EventHandle> EventHandleUP;
 
   MainLoop();
   ~MainLoop() override;
 
   ReadHandleUP RegisterReadObject(const lldb::IOObjectSP &object_sp,
                                   const Callback &callback,
                                   Status &error) override;
 
   // Listening for signals from multiple MainLoop instances is perfectly safe
   // as long as they don't try to listen for the same signal. The callback
   // function is invoked when the control returns to the Run() function, not
   // when the hander is executed. This mean that you can treat the callback as
   // 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);
 
+  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.
   void AddPendingCallback(const Callback &callback) override;
 
   Status Run() override;
 
   // This should only be performed from a callback. Do not attempt to terminate
   // the processing from another thread.
   // TODO: Add synchronization if we want to be terminated from another thread.
   void RequestTermination() override { m_terminate_request = true; }
 
 protected:
   void UnregisterReadObject(IOObject::WaitableHandle handle) override;
 
   void UnregisterSignal(int signo, std::list<Callback>::iterator callback_it);
 
+  void UnregisterEvent(std::list<EventInfo>::iterator info_it);
+
 private:
   void ProcessReadObject(IOObject::WaitableHandle handle);
   void ProcessSignal(int signo);
+  void ProcessEvents();
+
+  void NotifyEvent(std::list<EventInfo>::iterator info_it);
 
   class SignalHandle {
   public:
     ~SignalHandle() { m_mainloop.UnregisterSignal(m_signo, m_callback_it); }
 
   private:
     SignalHandle(MainLoop &mainloop, int signo,
                  std::list<Callback>::iterator callback_it)
         : m_mainloop(mainloop), m_signo(signo), m_callback_it(callback_it) {}
 
     MainLoop &m_mainloop;
     int m_signo;
     std::list<Callback>::iterator m_callback_it;
 
     friend class MainLoop;
     SignalHandle(const SignalHandle &) = delete;
     const SignalHandle &operator=(const SignalHandle &) = delete;
   };
 
+  class EventHandle {
+  public:
+    EventHandle(MainLoop &mainloop, std::list<EventInfo>::iterator info_it)
+        : m_mainloop(mainloop), m_info_it(info_it) {}
+    ~EventHandle() { m_mainloop.UnregisterEvent(m_info_it); }
+
+    void Notify() { m_mainloop.NotifyEvent(m_info_it); }
+
+  private:
+    MainLoop &m_mainloop;
+    std::list<EventInfo>::iterator m_info_it;
+  };
+
   struct SignalInfo {
     std::list<Callback> callbacks;
 #ifndef SIGNAL_POLLING_UNSUPPORTED
     struct sigaction old_action;
 #endif
     bool was_blocked : 1;
   };
+  struct EventInfo {
+    EventInfo(Callback callback)
+        : notified(false), callback(std::move(callback)) {}
+    std::atomic<bool> notified;
+    Callback callback;
+  };
   class RunImpl;
 
   llvm::DenseMap<IOObject::WaitableHandle, Callback> m_read_fds;
   llvm::DenseMap<int, SignalInfo> m_signals;
+  std::list<EventInfo> m_events;
   std::vector<Callback> m_pending_callbacks;
 #if HAVE_SYS_EVENT_H
   int m_kqueue;
 #endif
+#ifdef _WIN32
+  void *m_event_event;
+#else
+  Pipe m_event_pipe;
+#endif
   bool m_terminate_request : 1;
 };
 
 } // namespace lldb_private
 
 #endif // LLDB_HOST_MAINLOOP_H

lldb/source/Host/common/MainLoop.cpp

 
 MainLoop::RunImpl::RunImpl(MainLoop &loop) : loop(loop) {
 }
 
 MainLoop::RunImpl::~RunImpl() {
 }
 
 Status MainLoop::RunImpl::Poll() {
-  std::vector<WSAEVENT> read_events;
-  read_events.reserve(loop.m_read_fds.size());
+  std::vector<WSAEVENT> events;
+  events.reserve(loop.m_read_fds.size() + 1);
   for(auto &fd : loop.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 : loop.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()) {
     signaled_event = result - WSA_WAIT_EVENT_0;
     return Status();
   }
   return Status("WSAWaitForMultipleEvents failed");
 }
 
 void MainLoop::RunImpl::ProcessEvents() {
   assert(signaled_event);
+  if (*signaled_event < loop.m_read_fds.size()) {
     auto &fd_info = *std::next(loop.m_read_fds.begin(), *signaled_event);
-
     loop.ProcessReadObject(fd_info.first);
+  } else {
+    // Must be called before events are processed.
+    WSAResetEvent(loop.m_event_event);
+    loop.ProcessEvents();
+  }
 }
 #else
 class MainLoop::RunImpl {
 public:
   RunImpl(MainLoop &loop);
   ~RunImpl() = default;
 
   Status Poll();
if (loop.m_terminate_request)
       return;
     g_signal_flags[signal] = 0;
     loop.ProcessSignal(signal);
   }
 }
 #endif
 #endif
 
+void MainLoop::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);
+#ifdef _WIN32
+  WSASetEvent(m_event_event);
+#else
+  ...
+#endif
+}
+
 MainLoop::MainLoop() : m_terminate_request(false) {
 #if HAVE_SYS_EVENT_H
   m_kqueue = kqueue();
   assert(m_kqueue >= 0);
 #endif
+#ifdef _WIN32
+  m_event_event = WSACreateEvent();
+  assert(m_event_event != WSA_INVALID_EVENT);
+#else
+  ...
+#endif
 }
 MainLoop::~MainLoop() {
 #if HAVE_SYS_EVENT_H
   close(m_kqueue);
 #endif
   assert(m_read_fds.size() == 0);
   assert(m_signals.size() == 0);
 }
#endif
   info.was_blocked = sigismember(&old_set, signo);
   auto insert_ret = m_signals.insert({signo, info});
 
   return SignalHandleUP(new SignalHandle(
       *this, signo, insert_ret.first->second.callbacks.begin()));
 #endif
 }
 
+MainLoop::EventHandleUP MainLoop::RegisterEvent(const Callback &callback) {
+  return EventHandleUP(new EventHandle(*this, m_events.emplace(m_events.end(), callback)));
+}
+
 void MainLoop::AddPendingCallback(const Callback &callback) {
   m_pending_callbacks.push_back(callback);
 }
 
 void MainLoop::UnregisterReadObject(IOObject::WaitableHandle handle) {
   bool erased = m_read_fds.erase(handle);
   UNUSED_IF_ASSERT_DISABLED(erased);
   assert(erased);
 }
 
+void MainLoop::UnregisterEvent(std::list<EventInfo>::iterator info_it) {
+  m_events.erase(info_it);
+}
+
 void MainLoop::UnregisterSignal(int signo,
                                 std::list<Callback>::iterator callback_it) {
 #if SIGNAL_POLLING_UNSUPPORTED
   Status("Signal polling is not supported on this platform.");
 #else
   auto it = m_signals.find(signo);
   assert(it != m_signals.end());
 

 
 Status MainLoop::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();
 
     for (const Callback &callback : m_pending_callbacks)
void MainLoop::ProcessSignal(int signo) {
   }
 }
 
 void MainLoop::ProcessReadObject(IOObject::WaitableHandle handle) {
   auto it = m_read_fds.find(handle);
   if (it != m_read_fds.end())
     it->second(*this); // Do the work
 }
+
+void MainLoop::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
+}

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);