Replace the singleton "ShadowListener" with a primary and N secondary listeners

[jingham]

Summary

Allow a Broadcaster to have one "Primary" listener, and potentially many secondary Listeners. The primary listener is guaranteed to run the DoOnRemoval first, and only when that is complete will secondary Listeners receive the event.

Prior to Ismail's addition of the notion of a "Secondary" listener, you could only safely observe process events from the Listener you passed in to the Process creation. That was because the act of pulling the process event off the event queue triggered the change of the public state of the process (thus keeping the event stream and the state in sync).

Ismail added a privileged "shadow listener" that could also be informed of the event. That wasn't quite the right model, however, because the real singleton is the primary listener.

That's what this patch implements.

This isn't quite the full solution for the Process Listener. The goal is that the primary Process Listener not only gets to drive the event stream, but that the primary listener gets to react to each event before any other Listener can hear about it. That will allow the process execution logic to run before any other agent gets a chance to change the process state out from under it. For that, I'll need to add a receipt mechanism to the event delivery, and have the forwarding to the pending listeners happen only after the receipt is acknowledged. But that will be a follow on patch.

[mib]

LGTM with some comments :)

lldb/source/Target/Process.cpp

371

nit: this could probably be constexpr

lldb/source/Utility/Broadcaster.cpp

274–276

nit: We could simplify the logic here a little bit

283

Then, as a follow-up to my suggestion, you could check hijacking_listener_sp instead of checking the boolean.

[bulbazord]

lldb/include/lldb/Target/Process.h

348–363

I wonder if it might be better to add a new element to the enum, something like eBroadcastBitAll = eBroadcastBitStageChanged | ... | eBroadcastBitStructuredData,.

If we have to add a new element to this enumeration, I'm not sure everyone will realize that g_all_event_bits needs to be updated in a separate file (even though it's right below the enum definition).

621–624

Why was this moved down?

lldb/source/Utility/Broadcaster.cpp

283

Instead of having the bool flag for is_hijack, you can compare primary_listener_sp against hijacking_listener_sp.

Then the setup above gets less complex, like so:

ListenerSP primary_listener_sp = hijacking_listener_sp;
if (!primary_listener_sp)
  primary_listener_sp = m_primary_listener_sp;

284

I wonder if it might be useful to introduce a variant of GetListeners which takes an event_type and only gives you back the ones you want. That would simplify this loop (and the one in the other branching path). You could also just pass that vector along to the Event instead of filtering and adding them one at a time (so you're not paying for potential vector resizing costs).

290

nit: Checking to see if pair.first is hijacking_listener_sp is redundant. If you've gotten to this point, hijacking_listener_sp must be pointing to nullptr.

291

Why might pair.first be m_primary_listener_sp? I would assume that if a primary listener is set, it wouldn't be among the other listeners in a broadcaster.... But I suppose nothing stops you from doing that.

lldb/test/API/functionalities/interactive_scripted_process/TestInteractiveScriptedProcess.py

28

Remove this comment?

40–46

This is to make sure that the primary listener is getting the run/stop events before the mux_process_listener right?

58

delete?

[JDevlieghere]

lldb/include/lldb/Utility/Event.h

228

Nit: trailing whitespace

lldb/test/API/functionalities/interactive_scripted_process/TestInteractiveScriptedProcess.py

28

Remove?

58

Remove?

[clayborg]

lldb/source/Target/Process.cpp

371

yes, this can be a constexpr and be defined in only in the header file. No need to create storage for it by making it a real static variable

lldb/source/Utility/Broadcaster.cpp

179–182

Does the primary listener need to listen to all event bits?

274–276

I agree this would clean this up nicely.

[jingham]

Respond to review comments.

In the course of implementing Alex's suggestion to add a filter to GetListeners, I noticed that there we were being lax in our dealing with removed listeners. RemoveListener didn't actually do that, it just removed the event bits, and the next pass through GetListeners would actually clear out the entry.

Fixing that required a few other little fixes to get this working again.

[jingham]

I implemented Alex's suggestion for a GetListeners filter, but doing that triggered some bugs due to the fact that we weren't really pruning the m_listeners correctly as listeners come and go. I fixed some bugs there along with implementing this suggestion.

lldb/source/Utility/Broadcaster.cpp

179–182

I don't see a use case for saying "I want to be the one that gets first crack at this broadcaster" but only for some bits. Plus, if we do it that way, we'd really have to have a map of "primary listener -> event mask" and add a way to make yourself the primary listener for only some bits.

We have this ability for HijackListeners, but we never use it, we always just hijack all bits as well...

There's no reason you can't do it that way, but until we see a need for multiple primary listeners, I don't think the extra complexity is warranted.

290

Interestingly enough, that was effectively a check on empty listeners in m_listeners. There shouldn't have been any, but we weren't being very careful about pruning the list...

291

That's really an oversight in GetListeners. That correctly doesn't return the Hijack listener, since that is a hidden agent, but the primary listener only differs from the secondary listeners in how dispatch gets done. I fixed that in this patch.

lldb/test/API/functionalities/interactive_scripted_process/TestInteractiveScriptedProcess.py

40–46

correct.

[bulbazord]

Thanks for all the work to clean up some of those interfaces! I know that was not straightforward. :)

I left a few more comments about minor things I noticed while re-reading this. I'm going to request changes to put this back into your queue.

lldb/include/lldb/Utility/Event.h

233

nit: trailing semicolon

lldb/source/Target/Process.cpp

620–621

Do you need to check for the validity of the primary listener before using it? It can point to nullptr from what I can tell

lldb/source/Utility/Broadcaster.cpp

59

I wonder if we should be reserving at all considering we're using an llvm::SmallVector<$type, 4>. The point of a SmallVector is that on average it should be small and only sometimes should we be exceeding that hardcoded size. This means that if there are 8 listeners but only 2 are relevant, we're taking a SmallVector of size 4 and allocating space for 8 listeners (always triggering a memory allocation) even if it will only ever have 2 listeners in it.

I know that's the existing behavior and you're just preserving it, but something to think about for the future.

179–184

This check is a bit redundant since HasListeners performs the exact same check

198–199

lldb/include/lldb/Target/Process.h

class Process : public std::enable_shared_from_this<Process>,
   friend class ProcessEventData;
   friend class StopInfo;
   friend class Target;
   friend class ThreadList;
 
 public:
   /// Broadcaster event bits definitions.
   enum {
     eBroadcastBitStateChanged = (1 << 0),
     eBroadcastBitInterrupt = (1 << 1),
     eBroadcastBitSTDOUT = (1 << 2),
     eBroadcastBitSTDERR = (1 << 3),
     eBroadcastBitProfileData = (1 << 4),
     eBroadcastBitStructuredData = (1 << 5),
   };
+  // This is all the event bits the public process broadcaster broadcasts.
+  // The process shadow listener signs up for all these bits...
+  static constexpr int g_all_event_bits = eBroadcastBitStateChanged 
+                                        | eBroadcastBitInterrupt
+                                        | eBroadcastBitSTDOUT 
+                                        | eBroadcastBitSTDERR
+                                        | eBroadcastBitProfileData 
+                                        | eBroadcastBitStructuredData;
 
   enum {
     eBroadcastInternalStateControlStop = (1 << 0),
     eBroadcastInternalStateControlPause = (1 << 1),
     eBroadcastInternalStateControlResume = (1 << 2)
   };
 
   typedef Range<lldb::addr_t, lldb::addr_t> LoadRange;
   // We use a read/write lock to allow on or more clients to access the process
public:
   static constexpr llvm::StringRef LaunchSynchronousHijackListenerName =
       "lldb.internal.Process.LaunchSynchronous.hijack";
   static constexpr llvm::StringRef ResumeSynchronousHijackListenerName =
       "lldb.internal.Process.ResumeSynchronous.hijack";
 
   ConstString &GetBroadcasterClass() const override {
     return GetStaticBroadcasterClass();
   }
   
-  void SetShadowListener(lldb::ListenerSP listener_sp) override {
-    Broadcaster::SetShadowListener(listener_sp);
-  }
-
 /// A notification structure that can be used by clients to listen
 /// for changes in a process's lifetime.
 ///
 /// \see RegisterNotificationCallbacks (const Notifications&) @see
 /// UnregisterNotificationCallbacks (const Notifications&)
   typedef struct {
     void *baton;
     void (*initialize)(void *baton, Process *process);
/// UnregisterNotificationCallbacks (const Notifications&)
 
   virtual Status DoLoadCore() {
     Status error;
     error.SetErrorStringWithFormatv(
         "error: {0} does not support loading core files.", GetPluginName());
     return error;
   }
 
+  /// The "ShadowListener" for a process is just an ordinary Listener that 
+  /// listens for all the Process event bits.  It's convenient because you can
+  /// specify it in the LaunchInfo or AttachInfo, so it will get events from
+  /// the very start of the process.
+  void SetShadowListener(lldb::ListenerSP shadow_listener_sp) {
+    if (shadow_listener_sp)
+      AddListener(shadow_listener_sp, g_all_event_bits);
+  }
+
   // FUTURE WORK: GetLoadImageUtilityFunction are the first use we've
   // had of having other plugins cache data in the Process.  This is handy for
   // long-living plugins - like the Platform - which manage interactions whose
   // lifetime is governed by the Process lifetime.  If we find we need to do
   // this more often, we should construct a general solution to the problem.
   // The consensus suggestion was that we have a token based registry in the
   // Process. Some undecided questions are  (1) who manages the tokens.  It's
   // probably best that you add the element  and get back a token that
uint32_t m_extended_thread_stop_id; ///< The natural stop id when
                                       ///extended_thread_list was last updated
   QueueList
       m_queue_list; ///< The list of libdispatch queues at a given stop point
   uint32_t m_queue_list_stop_id; ///< The natural stop id when queue list was
                                  ///last fetched
   std::vector<Notifications> m_notifications; ///< The list of notifications
                                               ///that this process can deliver.
   std::vector<lldb::addr_t> m_image_tokens;
-  lldb::ListenerSP m_listener_sp; ///< Shared pointer to the listener used for
-                                  ///public events.  Can not be empty.
   BreakpointSiteList m_breakpoint_site_list; ///< This is the list of breakpoint
                                              ///locations we intend to insert in
                                              ///the target.
   lldb::DynamicLoaderUP m_dyld_up;
   lldb::JITLoaderListUP m_jit_loaders_up;
   lldb::DynamicCheckerFunctionsUP m_dynamic_checkers_up; ///< The functions used
                                                          /// by the expression
                                                          /// parser to validate

lldb/include/lldb/Utility/Broadcaster.h

public:
   /// with the broadcaster manager and do broadcaster class matching.
   /// FIXME: Probably should make a ManagedBroadcaster subclass with all the
   /// bits needed to work with the BroadcasterManager, so that it is clearer
   /// how to add one.
   virtual ConstString &GetBroadcasterClass() const;
 
   lldb::BroadcasterManagerSP GetManager();
 
-  virtual void SetShadowListener(lldb::ListenerSP listener_sp) {
-    m_broadcaster_sp->m_shadow_listener = listener_sp;
+  void SetPrimaryListener(lldb::ListenerSP listener_sp) {
+    m_broadcaster_sp->SetPrimaryListener(listener_sp);
+  }
+
+  lldb::ListenerSP GetPrimaryListener() {
+    return m_broadcaster_sp->m_primary_listener_sp;
   }
 
 protected:
   /// BroadcasterImpl contains the actual Broadcaster implementation.  The
   /// Broadcaster makes a BroadcasterImpl which lives as long as it does.  The
   /// Listeners & the Events hold a weak pointer to the BroadcasterImpl, so
   /// that they can survive if a Broadcaster they were listening to is
   /// destroyed w/o their being able to unregister from it (which can happen if
const char *GetEventName(uint32_t event_mask) const {
       const auto pos = m_event_names.find(event_mask);
       if (pos != m_event_names.end())
         return pos->second.c_str();
       return nullptr;
     }
 
     bool EventTypeHasListeners(uint32_t event_type);
 
+    void SetPrimaryListener(lldb::ListenerSP listener_sp);
+
     bool RemoveListener(lldb_private::Listener *listener,
                         uint32_t event_mask = UINT32_MAX);
 
     bool RemoveListener(const lldb::ListenerSP &listener_sp,
                         uint32_t event_mask = UINT32_MAX);
 
     bool HijackBroadcaster(const lldb::ListenerSP &listener_sp,
                            uint32_t event_mask = UINT32_MAX);
 
     bool IsHijackedForEvent(uint32_t event_mask);
 
     void RestoreBroadcaster();
 
   protected:
     void PrivateBroadcastEvent(lldb::EventSP &event_sp, bool unique);
 
     const char *GetHijackingListenerName();
 
     typedef llvm::SmallVector<std::pair<lldb::ListenerWP, uint32_t>, 4>
         collection;
     typedef std::map<uint32_t, std::string> event_names_map;
 
     llvm::SmallVector<std::pair<lldb::ListenerSP, uint32_t &>, 4>
-    GetListeners();
+    GetListeners(uint32_t event_mask = UINT32_MAX, bool include_primary = true);
+
+    bool HasListeners(uint32_t event_mask);
 
     /// The broadcaster that this implements.
     Broadcaster &m_broadcaster;
 
     /// Optionally define event names for readability and logging for each
     /// event bit.
     event_names_map m_event_names;
 
+    /// A Broadcaster can have zero, one or many listeners.  A Broadcaster with
+    /// zero listeners is a no-op, with one Listener is trivial.
+    /// In most cases of multiple Listeners,the Broadcaster treats all its
+    /// Listeners as equal, sending each event to all of the Listeners in no
+    /// guaranteed order.
+    /// However, some Broadcasters - in particular the Process broadcaster, can
+    /// designate one Listener to be the "Primary Listener".  In the case of
+    /// the Process Broadcaster, the Listener passed to the Process constructor
+    /// will be the Primary Listener.
+    /// If the broadcaster has a Primary Listener, then the event gets
+    /// sent first to the Primary Listener, and then when the Primary Listener
+    /// pulls the event and the the event's DoOnRemoval finishes running,
+    /// the event is forwarded to all the other Listeners.
+    /// The other wrinkle is that a Broadcaster may be serving a Hijack
+    /// Listener.  If the Hijack Listener is present, events are only sent to
+    /// the Hijack Listener.  We use that, for instance, to absorb all the
+    /// events generated by running an expression so that they don't show up to
+    /// the driver or UI as starts and stops.
+    /// If a Broadcaster has both a Primary and a Hijack Listener, the top-most
+    /// Hijack Listener is treated as the current Primary Listener.
+
     /// A list of Listener / event_mask pairs that are listening to this
     /// broadcaster.
     collection m_listeners;
 
     /// A mutex that protects \a m_listeners.
     std::recursive_mutex m_listeners_mutex;
 
+    /// See the discussion of Broadcasters and Listeners above.
+    lldb::ListenerSP m_primary_listener_sp;
+    // The primary listener listens to all bits:
+    uint32_t m_primary_listener_mask = UINT32_MAX;
+
     /// A simple mechanism to intercept events from a broadcaster
     std::vector<lldb::ListenerSP> m_hijacking_listeners;
 
     /// At some point we may want to have a stack or Listener collections, but
     /// for now this is just for private hijacking.
     std::vector<uint32_t> m_hijacking_masks;
 
-    /// A optional listener that all private events get also broadcasted to,
-    /// on top the hijacked / default listeners.
-    lldb::ListenerSP m_shadow_listener = nullptr;
-
   private:
     BroadcasterImpl(const BroadcasterImpl &) = delete;
     const BroadcasterImpl &operator=(const BroadcasterImpl &) = delete;
   };
 
   typedef std::shared_ptr<BroadcasterImpl> BroadcasterImplSP;
   typedef std::weak_ptr<BroadcasterImpl> BroadcasterImplWP;
 

lldb/include/lldb/Utility/Event.h

private:
   lldb::StructuredDataPluginSP m_plugin_sp;
 
   EventDataStructuredData(const EventDataStructuredData &) = delete;
   const EventDataStructuredData &
   operator=(const EventDataStructuredData &) = delete;
 };
 
 // lldb::Event
-class Event {
+class Event : public std::enable_shared_from_this<Event> {
   friend class Listener;
   friend class EventData;
   friend class Broadcaster::BroadcasterImpl;
 
 public:
   Event(Broadcaster *broadcaster, uint32_t event_type,
         EventData *data = nullptr);
 
Broadcaster::BroadcasterImplSP broadcaster_impl_sp =
         m_broadcaster_wp.lock();
     if (broadcaster_impl_sp)
       return broadcaster_impl_sp->GetBroadcaster() == broadcaster;
     else
       return false;
   }
 
   void Clear() { m_data_sp.reset(); }
 
+  /// This is used by Broadcasters with Primary Listeners to store the other
+  /// Listeners till after the Event's DoOnRemoval has completed.
+  void AddPendingListener(lldb::ListenerSP pending_listener_sp) {
+    m_pending_listeners.push_back(pending_listener_sp);
+  };
+
 private:
   // This is only called by Listener when it pops an event off the queue for
   // the listener.  It calls the Event Data's DoOnRemoval() method, which is
   // virtual and can be overridden by the specific data classes.
 
   void DoOnRemoval();
 
   // Called by Broadcaster::BroadcastEvent prior to letting all the listeners
   // know about it update the contained broadcaster so that events can be
   // popped off one queue and re-broadcast to others.
   void SetBroadcaster(Broadcaster *broadcaster) {
     m_broadcaster_wp = broadcaster->GetBroadcasterImpl();
   }
 
   Broadcaster::BroadcasterImplWP
       m_broadcaster_wp;        // The broadcaster that sent this event
   uint32_t m_type;             // The bit describing this event
   lldb::EventDataSP m_data_sp; // User specific data for this event
+  std::vector<lldb::ListenerSP> m_pending_listeners;
+  std::mutex m_listeners_mutex;
 
   Event(const Event &) = delete;
   const Event &operator=(const Event &) = delete;
   Event() = delete;
 };
 
 } // namespace lldb_private
 
 #endif // LLDB_UTILITY_EVENT_H

lldb/source/Target/Process.cpp

 
 FollowForkMode ProcessProperties::GetFollowForkMode() const {
   const uint32_t idx = ePropertyFollowForkMode;
   return GetPropertyAtIndexAs<FollowForkMode>(
       idx, static_cast<FollowForkMode>(
                g_process_properties[idx].default_uint_value));
 }
 
 ProcessSP Process::FindPlugin(lldb::TargetSP target_sp,
                               llvm::StringRef plugin_name,
                               ListenerSP listener_sp,
                               const FileSpec *crash_file_path,
                               bool can_connect) {
   static uint32_t g_process_unique_id = 0;
 
   ProcessSP process_sp;
   ProcessCreateInstance create_callback = nullptr;
: ProcessProperties(this),
           nullptr, "lldb.process.internal_state_control_broadcaster"),
       m_private_state_listener_sp(
           Listener::MakeListener("lldb.process.internal_state_listener")),
       m_mod_id(), m_process_unique_id(0), m_thread_index_id(0),
       m_thread_id_to_index_id_map(), m_exit_status(-1), m_exit_string(),
       m_exit_status_mutex(), m_thread_mutex(), m_thread_list_real(this),
       m_thread_list(this), m_thread_plans(*this), m_extended_thread_list(this),
       m_extended_thread_stop_id(0), m_queue_list(this), m_queue_list_stop_id(0),
-      m_notifications(), m_image_tokens(), m_listener_sp(listener_sp),
+      m_notifications(), m_image_tokens(),
       m_breakpoint_site_list(), m_dynamic_checkers_up(),
       m_unix_signals_sp(unix_signals_sp), m_abi_sp(), m_process_input_reader(),
       m_stdio_communication("process.stdio"), m_stdio_communication_mutex(),
       m_stdin_forward(false), m_stdout_data(), m_stderr_data(),
       m_profile_data_comm_mutex(), m_profile_data(), m_iohandler_sync(0),
       m_memory_cache(*this), m_allocated_memory_cache(*this),
       m_should_detach(false), m_next_event_action_up(), m_public_run_lock(),
       m_private_run_lock(), m_currently_handling_do_on_removals(false),
Process::Process(lldb::TargetSP target_sp, ListenerSP listener_sp,
 
   m_private_state_control_broadcaster.SetEventName(
       eBroadcastInternalStateControlStop, "control-stop");
   m_private_state_control_broadcaster.SetEventName(
       eBroadcastInternalStateControlPause, "control-pause");
   m_private_state_control_broadcaster.SetEventName(
       eBroadcastInternalStateControlResume, "control-resume");
 
-  m_listener_sp->StartListeningForEvents(
-      this, eBroadcastBitStateChanged | eBroadcastBitInterrupt |
-                eBroadcastBitSTDOUT | eBroadcastBitSTDERR |
-                eBroadcastBitProfileData | eBroadcastBitStructuredData);
+  // The listener passed into process creation is the primary listener:
+  // It always listens for all the event bits for Process:
+  SetPrimaryListener(listener_sp);
 
   m_private_state_listener_sp->StartListeningForEvents(
       &m_private_state_broadcaster,
       eBroadcastBitStateChanged | eBroadcastBitInterrupt);
 
   m_private_state_listener_sp->StartListeningForEvents(
       &m_private_state_control_broadcaster,
       eBroadcastInternalStateControlStop | eBroadcastInternalStateControlPause |

 // the thread that is waiting for new events, so if we needed more than one
 // event for our handling, we would stall.  So instead we do it when we fetch
 // the event off of the queue.
 //
 
 StateType Process::GetNextEvent(EventSP &event_sp) {
   StateType state = eStateInvalid;
 
-  if (m_listener_sp->GetEventForBroadcaster(this, event_sp,
+  if (GetPrimaryListener()->GetEventForBroadcaster(this, event_sp,
                                             std::chrono::seconds(0)) &&
       event_sp)
     state = Process::ProcessEventData::GetStateFromEvent(event_sp.get());
 
   return state;
 }
 
 void Process::SyncIOHandler(uint32_t iohandler_id,
                             const Timeout<std::micro> &timeout) {

 StateType Process::GetStateChangedEvents(EventSP &event_sp,
                                          const Timeout<std::micro> &timeout,
                                          ListenerSP hijack_listener_sp) {
   Log *log = GetLog(LLDBLog::Process);
   LLDB_LOG(log, "timeout = {0}, event_sp)...", timeout);
 
   ListenerSP listener_sp = hijack_listener_sp;
   if (!listener_sp)
-    listener_sp = m_listener_sp;
+    listener_sp = GetPrimaryListener();
 
   StateType state = eStateInvalid;
   if (listener_sp->GetEventForBroadcasterWithType(
           this, eBroadcastBitStateChanged | eBroadcastBitInterrupt, event_sp,
           timeout)) {
     if (event_sp && event_sp->GetType() == eBroadcastBitStateChanged)
       state = Process::ProcessEventData::GetStateFromEvent(event_sp.get());
     else
       LLDB_LOG(log, "got no event or was interrupted.");
   }
 
   LLDB_LOG(log, "timeout = {0}, event_sp) => {1}", timeout, state);
   return state;
 }
 
 Event *Process::PeekAtStateChangedEvents() {
   Log *log = GetLog(LLDBLog::Process);
 
   LLDB_LOGF(log, "Process::%s...", __FUNCTION__);
 
   Event *event_ptr;
-  event_ptr = m_listener_sp->PeekAtNextEventForBroadcasterWithType(
+  event_ptr = GetPrimaryListener()->PeekAtNextEventForBroadcasterWithType(
       this, eBroadcastBitStateChanged);
   if (log) {
     if (event_ptr) {
       LLDB_LOGF(log, "Process::%s (event_ptr) => %s", __FUNCTION__,
                 StateAsCString(ProcessEventData::GetStateFromEvent(event_ptr)));
     } else {
       LLDB_LOGF(log, "Process::%s no events found", __FUNCTION__);
     }
void Process::ProcessEventData::DoOnRemoval(Event *event_ptr) {
 
   if (GetRestarted()) {
     return;
   }
 
   if (!still_should_stop && does_anybody_have_an_opinion) {
     // We've been asked to continue, so do that here.
     SetRestarted(true);
-    // Use the public resume method here, since this is just extending a
-    // public resume.
+    // Use the private resume method here, since we aren't changing the run
+    // lock state.
     process_sp->PrivateResume();
   } else {
     bool hijacked = process_sp->IsHijackedForEvent(eBroadcastBitStateChanged) &&
                     !process_sp->StateChangedIsHijackedForSynchronousResume();
 
     if (!hijacked) {
       // If we didn't restart, run the Stop Hooks here.
       // Don't do that if state changed events aren't hooked up to the

lldb/source/Utility/Broadcaster.cpp

 
 void Broadcaster::CheckInWithManager() {
   if (m_manager_sp) {
     m_manager_sp->SignUpListenersForBroadcaster(*this);
   }
 }
 
 llvm::SmallVector<std::pair<ListenerSP, uint32_t &>, 4>
-Broadcaster::BroadcasterImpl::GetListeners() {
+Broadcaster::BroadcasterImpl::GetListeners(uint32_t event_mask,
+                                           bool include_primary) {
   llvm::SmallVector<std::pair<ListenerSP, uint32_t &>, 4> listeners;
-  listeners.reserve(m_listeners.size());
+  size_t max_count = m_listeners.size();
+  if (include_primary)
+    max_count++;
+  listeners.reserve(max_count);
 
   for (auto it = m_listeners.begin(); it != m_listeners.end();) {
     lldb::ListenerSP curr_listener_sp(it->first.lock());
-    if (curr_listener_sp && it->second) {
+    if (curr_listener_sp) {
+      if (it->second & event_mask)
         listeners.emplace_back(std::move(curr_listener_sp), it->second);
       ++it;
     } else
+      // If our listener_wp didn't resolve, then we should remove this entry.
       it = m_listeners.erase(it);
   }
+  if (include_primary && m_primary_listener_sp)
+    listeners.emplace_back(m_primary_listener_sp, m_primary_listener_mask);
 
   return listeners;
 }
 
+bool Broadcaster::BroadcasterImpl::HasListeners(uint32_t event_mask) {
+  if (m_primary_listener_sp)
+    return true;
+  for (auto it = m_listeners.begin(); it != m_listeners.end(); it++) {
+    // Don't return a listener if the other end of the WP is gone:
+    lldb::ListenerSP curr_listener_sp(it->first.lock());
+    if (curr_listener_sp && (it->second & event_mask))
+      return true;
+  }
+  return false;
+}
+
 void Broadcaster::BroadcasterImpl::Clear() {
   std::lock_guard<std::recursive_mutex> guard(m_listeners_mutex);
 
   // Make sure the listener forgets about this broadcaster. We do this in the
   // broadcaster in case the broadcaster object initiates the removal.
   for (auto &pair : GetListeners())
     pair.first->BroadcasterWillDestruct(&m_broadcaster);
 
   m_listeners.clear();
+  m_primary_listener_sp.reset();
 }
 
 Broadcaster *Broadcaster::BroadcasterImpl::GetBroadcaster() {
   return &m_broadcaster;
 }
 
 bool Broadcaster::BroadcasterImpl::GetEventNames(
     Stream &s, uint32_t event_mask, bool prefix_with_broadcaster_name) const {
if (!listener_sp)
     return 0;
 
   std::lock_guard<std::recursive_mutex> guard(m_listeners_mutex);
 
   // See if we already have this listener, and if so, update its mask
 
   bool handled = false;
 
-  for (auto &pair : GetListeners()) {
+  if (listener_sp == m_primary_listener_sp)
+    // This already handles all bits so just return the mask:
+    return event_mask;
+
+  for (auto &pair : GetListeners(UINT32_MAX, false)) {
     if (pair.first == listener_sp) {
       handled = true;
       pair.second |= event_mask;
       m_broadcaster.AddInitialEventsToListener(listener_sp, event_mask);
       break;
     }
   }
 

 }
 
 bool Broadcaster::BroadcasterImpl::EventTypeHasListeners(uint32_t event_type) {
   std::lock_guard<std::recursive_mutex> guard(m_listeners_mutex);
 
   if (!m_hijacking_listeners.empty() && event_type & m_hijacking_masks.back())
     return true;
 
-  for (auto &pair : GetListeners()) {
-    if (pair.second & event_type)
-      return true;
-  }
-  return false;
+  // The primary listener listens for all event bits:
+  if (m_primary_listener_sp)
+
+  return HasListeners(event_type);
 }
 
 bool Broadcaster::BroadcasterImpl::RemoveListener(
     lldb_private::Listener *listener, uint32_t event_mask) {
   if (!listener)
     return false;
 
-  std::lock_guard<std::recursive_mutex> guard(m_listeners_mutex);
-  for (auto &pair : GetListeners()) {
-    if (pair.first.get() == listener) {
-      pair.second &= ~event_mask;
-      return true;
-    }
+  if (listener == m_primary_listener_sp.get()) {
+    // Primary listeners listen for all the event bits for their broadcaster,
+    // so remove this altogether if asked:
+    m_primary_listener_sp.reset();
+
+  std::lock_guard<std::recursive_mutex> guard(m_listeners_mutex);
+  for (auto it = m_listeners.begin(); it != m_listeners.end();) {
+    lldb::ListenerSP curr_listener_sp(it->first.lock());
+
+    if (!curr_listener_sp) {
+      // The weak pointer for this listener didn't resolve, lets' prune it
+      // as we go.
+      m_listeners.erase(it);
+      continue;
+    }
+
+    if (curr_listener_sp.get() == listener) {
+      it->second &= ~event_mask;
+      // If we removed all the event bits from a listener, remove it from
+      // the list as well.
+      if (!it->second)
+        m_listeners.erase(it);
+      return true;
+    } else
+      it++;
   }
   return false;
 }
 
 bool Broadcaster::BroadcasterImpl::RemoveListener(
     const lldb::ListenerSP &listener_sp, uint32_t event_mask) {
   return RemoveListener(listener_sp.get(), event_mask);
 }
if (log) {
     event_sp->Dump(&event_description);
     LLDB_LOG(log,
              "{0:x} Broadcaster(\"{1}\")::BroadcastEvent (event_sp = {2}, "
              "unique={3}) hijack = {4:x}",
              static_cast<void *>(this), GetBroadcasterName(),
              event_description.GetData(), unique,
              static_cast<void *>(hijacking_listener_sp.get()));
   }
+  ListenerSP primary_listener_sp
+      = hijacking_listener_sp ? hijacking_listener_sp : m_primary_listener_sp;
 
-  if (hijacking_listener_sp) {
-    if (unique && hijacking_listener_sp->PeekAtNextEventForBroadcasterWithType(
+  if (primary_listener_sp) {
+    if (unique && primary_listener_sp->PeekAtNextEventForBroadcasterWithType(
                       &m_broadcaster, event_type))
       return;
-    hijacking_listener_sp->AddEvent(event_sp);
-    if (m_shadow_listener)
-      m_shadow_listener->AddEvent(event_sp);
-  } else {
-    for (auto &pair : GetListeners()) {
-      if (!(pair.second & event_type))
+    // Add the pending listeners but not if the event is hijacked, since that
+    // is given sole access to the event stream it is hijacking.
+    // Make sure to do this before adding the event to the primary or it might
+    // start handling the event before we're done adding all the pending
+    // listeners.
+    if (!hijacking_listener_sp) {
+      for (auto &pair : GetListeners(event_type, false)) {
+        if (unique && pair.first->PeekAtNextEventForBroadcasterWithType(
+                          &m_broadcaster, event_type))
           continue;
+        event_sp->AddPendingListener(pair.first);
+      }
+    }
+    primary_listener_sp->AddEvent(event_sp);
+  } else {
+    for (auto &pair : GetListeners(event_type)) {
       if (unique && pair.first->PeekAtNextEventForBroadcasterWithType(
                         &m_broadcaster, event_type))
         continue;
 
       pair.first->AddEvent(event_sp);
-      if (m_shadow_listener)
-        m_shadow_listener->AddEvent(event_sp);
     }
   }
 }
 
 void Broadcaster::BroadcasterImpl::BroadcastEvent(uint32_t event_type,
                                                   EventData *event_data) {
   auto event_sp = std::make_shared<Event>(event_type, event_data);
   PrivateBroadcastEvent(event_sp, false);
 }
 
 void Broadcaster::BroadcasterImpl::BroadcastEvent(
     uint32_t event_type, const lldb::EventDataSP &event_data_sp) {
   auto event_sp = std::make_shared<Event>(event_type, event_data_sp);
   PrivateBroadcastEvent(event_sp, false);
 }
 
 void Broadcaster::BroadcasterImpl::BroadcastEventIfUnique(
     uint32_t event_type, EventData *event_data) {
   auto event_sp = std::make_shared<Event>(event_type, event_data);
   PrivateBroadcastEvent(event_sp, true);
 }
 
+void Broadcaster::BroadcasterImpl::SetPrimaryListener(lldb::ListenerSP
+                                                      listener_sp) {
+  // This might have already been added as a normal listener, make sure we
+  // don't hold two copies.
+  RemoveListener(listener_sp.get(), UINT32_MAX);
+  m_primary_listener_sp = listener_sp;
+                                                      
+}
+
 bool Broadcaster::BroadcasterImpl::HijackBroadcaster(
     const lldb::ListenerSP &listener_sp, uint32_t event_mask) {
   std::lock_guard<std::recursive_mutex> guard(m_listeners_mutex);
 
   Log *log = GetLog(LLDBLog::Events);
   LLDB_LOG(
       log,
       "{0} Broadcaster(\"{1}\")::HijackBroadcaster (listener(\"{2}\")={3})",

lldb/source/Utility/Event.cpp

 //===-- Event.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/Utility/Event.h"
 
 #include "lldb/Utility/Broadcaster.h"
 #include "lldb/Utility/DataExtractor.h"
 #include "lldb/Utility/Endian.h"
+#include "lldb/Utility/Listener.h"
 #include "lldb/Utility/Stream.h"
 #include "lldb/Utility/StreamString.h"
 #include "lldb/lldb-enumerations.h"
 
 #include "llvm/ADT/StringExtras.h"
 
 #include <algorithm>
 
if (m_data_sp) {
     s->PutChar('{');
     m_data_sp->Dump(s);
     s->PutChar('}');
   } else
     s->Printf("<NULL>");
 }
 
 void Event::DoOnRemoval() {
+  std::lock_guard<std::mutex> guard(m_listeners_mutex);
+
   if (m_data_sp)
     m_data_sp->DoOnRemoval(this);
+  // Now that the event has been handled by the primary event Listener, forward
+  // it to the other Listeners.
+  EventSP me_sp = shared_from_this();
+  for (auto listener_sp : m_pending_listeners)
+    listener_sp->AddEvent(me_sp);
+  m_pending_listeners.clear();
 }
 
 #pragma mark -
 #pragma mark EventData
 
 // EventData functions
 
 EventData::EventData() = default;

lldb/source/Utility/Listener.cpp

if (log != nullptr)
                 static_cast<void *>(event_sp.get()));
 
     if (remove) {
       m_events.erase(pos);
       // Unlock the event queue here.  We've removed this event and are about
       // to return it so it should be okay to get the next event off the queue
       // here - and it might be useful to do that in the "DoOnRemoval".
       lock.unlock();
-      if (!m_is_shadow)
-        event_sp->DoOnRemoval();
     }
     return true;
   }
 
   event_sp.reset();
   return false;
 }
 

lldb/test/API/api/listeners/TestListener.py

def test_clearing_listener(self):
             not event.IsValid(), "We don't get events we aren't listening to."
         )
 
     def test_receiving_breakpoint_added_from_debugger(self):
         """Test that we get breakpoint added events, waiting on event classes on the debugger"""
         self.build()
 
         my_listener = lldb.SBListener("test_listener")
-
         my_listener.StartListeningForEventClass(
             self.dbg,
             lldb.SBTarget.GetBroadcasterClassName(),
             lldb.SBTarget.eBroadcastBitBreakpointChanged,
         )
 
         target = self.createTestTarget()
 

lldb/test/API/functionalities/interactive_scripted_process/TestInteractiveScriptedProcess.py

def setUp(self):
         TestBase.setUp(self)
         # Build and load test program
         self.build()
         self.runCmd("file " + self.getBuildArtifact("a.out"), CURRENT_EXECUTABLE_SET)
         self.main_source_file = lldb.SBFileSpec("main.cpp")
         self.script_module = "interactive_scripted_process"
         self.script_file = self.script_module + ".py"
 
+    # These tests are flakey and sometimes timeout.  They work most of the time
+    # so the basic event flow is right, but somehow the handling is off.
     @skipUnlessDarwin
-    @skipIfDarwin
+    @skipIfDarwin 
     def test_passthrough_launch(self):
         """Test a simple pass-through process launch"""
         self.passthrough_launch()
 
         lldbutil.run_break_set_by_source_regexp(self, "also break here")
         self.assertEqual(self.mux_target.GetNumBreakpoints(), 2)
         error = self.mux_process.Continue()
         self.assertSuccess(error, "Resuming multiplexer scripted process")
         self.assertTrue(self.mux_process.IsValid(), "Got a valid process")
 
         event = lldbutil.fetch_next_event(
+            self, self.dbg.GetListener(), self.mux_process.GetBroadcaster(), timeout=20
+        )
+        self.assertState(lldb.SBProcess.GetStateFromEvent(event), lldb.eStateRunning)
+        event = lldbutil.fetch_next_event(
+            self, self.dbg.GetListener(), self.mux_process.GetBroadcaster()
+        )
+        self.assertState(lldb.SBProcess.GetStateFromEvent(event), lldb.eStateStopped)
+
+        event = lldbutil.fetch_next_event(
             self, self.mux_process_listener, self.mux_process.GetBroadcaster()
         )
         self.assertState(lldb.SBProcess.GetStateFromEvent(event), lldb.eStateRunning)
         event = lldbutil.fetch_next_event(
             self, self.mux_process_listener, self.mux_process.GetBroadcaster()
         )
         self.assertState(lldb.SBProcess.GetStateFromEvent(event), lldb.eStateStopped)
 
     @skipUnlessDarwin
     @skipIfDarwin
     def test_multiplexed_launch(self):
         """Test a multiple interactive scripted process debugging"""
         self.passthrough_launch()
         self.assertEqual(self.dbg.GetNumTargets(), 2)
 
         driving_target = self.mux_process.GetScriptedImplementation().driving_target
         self.assertTrue(driving_target.IsValid(), "Driving target is invalid")
 
def test_multiplexed_launch(self):
             event_target = event_process.GetTarget()
             event_target_idx = self.dbg.GetIndexOfTarget(event_target)
             self.assertFalse(
                 execution_events[event_target_idx][state],
                 "Event already received for this process",
             )
             execution_events[event_target_idx][state] = True
 
+        for _ in range((self.dbg.GetNumTargets() - 1) * 2):
+            fetch_process_event(self, execution_events)
+
+        for target_index, event_states in execution_events.items():
+            for state, is_set in event_states.items():
+                self.assertTrue(is_set, f"Target {target_index} has state {state} set")
+
         event = lldbutil.fetch_next_event(
             self, self.mux_process_listener, self.mux_process.GetBroadcaster()
         )
         self.assertState(lldb.SBProcess.GetStateFromEvent(event), lldb.eStateRunning)
 
         event = lldbutil.fetch_next_event(
             self, self.mux_process_listener, self.mux_process.GetBroadcaster()
         )
         self.assertState(lldb.SBProcess.GetStateFromEvent(event), lldb.eStateStopped)
 
-        for _ in range((self.dbg.GetNumTargets() - 1) * 2):
-            fetch_process_event(self, execution_events)
-
-        for target_index, event_states in execution_events.items():
-            for state, is_set in event_states.items():
-                self.assertTrue(is_set, f"Target {target_index} has state {state} set")
-
     def duplicate_target(self, driving_target):
         exe = driving_target.executable.fullpath
         triple = driving_target.triple
         return self.dbg.CreateTargetWithFileAndTargetTriple(exe, triple)
 
     def get_launch_info(self, class_name, script_dict):
         structured_data = lldb.SBStructuredData()
         structured_data.SetFromJSON(json.dumps(script_dict))
def passthrough_launch(self):
         mux_launch_info.SetShadowListener(self.mux_process_listener)
 
         self.dbg.SetAsync(True)
         error = lldb.SBError()
         self.mux_process = self.mux_target.Launch(mux_launch_info, error)
         self.assertSuccess(error, "Launched multiplexer scripted process")
         self.assertTrue(self.mux_process.IsValid(), "Got a valid process")
 
-        # Check that the mux process started running
+        # Check that the real process started running
         event = lldbutil.fetch_next_event(
-            self, self.mux_process_listener, self.mux_process.GetBroadcaster()
+            self, self.dbg.GetListener(), self.mux_process.GetBroadcaster()
         )
         self.assertState(lldb.SBProcess.GetStateFromEvent(event), lldb.eStateRunning)
-        # Check that the real process started running
+        # Check that the mux process started running
         event = lldbutil.fetch_next_event(
-            self, self.dbg.GetListener(), self.mux_process.GetBroadcaster()
+            self, self.mux_process_listener, self.mux_process.GetBroadcaster()
         )
         self.assertState(lldb.SBProcess.GetStateFromEvent(event), lldb.eStateRunning)
 
         # Check that the real process stopped
         event = lldbutil.fetch_next_event(
             self, self.dbg.GetListener(), self.mux_process.GetBroadcaster()
         )
         self.assertState(lldb.SBProcess.GetStateFromEvent(event), lldb.eStateStopped)

lldb/test/API/python_api/event/TestEvents.py

def test_add_listener_to_broadcaster(self):
 
         # Wait until the 'MyListeningThread' terminates.
         my_thread.join()
 
         # The final judgement. :-)
         self.assertEqual(
             self.state, "stopped", "Both expected state changed events received"
         )
+
+    def wait_for_next_event(self, expected_state, test_shadow = False):
+        """Wait for an event from self.primary & self.shadow listener.
+           If test_shadow is true, we also check that the shadow listener only 
+           receives events AFTER the primary listener does."""
+        # Waiting on the shadow listener shouldn't have events yet because
+        # we haven't fetched them for the primary listener yet:
+        event = lldb.SBEvent()
+
+        if test_shadow:
+            success = self.shadow_listener.WaitForEvent(1, event)
+            self.assertFalse(success, "Shadow listener doesn't pull events")
+
+        # But there should be an event for the primary listener:
+        success = self.primary_listener.WaitForEvent(5, event)
+        self.assertTrue(success, "Primary listener got the event")
+
+        state = lldb.SBProcess.GetStateFromEvent(event)
+        restart = False
+        if state == lldb.eStateStopped:
+            restart = lldb.SBProcess.GetRestartedFromEvent(event)
+
+        if expected_state != None:
+            self.assertEqual(state, expected_state, "Primary thread got the correct event")
+            
+        # And after pulling that one there should be an equivalent event for the shadow
+        # listener:
+        success = self.shadow_listener.WaitForEvent(5, event)
+        self.assertTrue(success, "Shadow listener got event too")
+        self.assertEqual(state, lldb.SBProcess.GetStateFromEvent(event), "It was the same event")
+        self.assertEqual(restart, lldb.SBProcess.GetRestartedFromEvent(event), "It was the same restarted")
+            
+        return state, restart
+
+    def test_shadow_listener(self):
+        self.build()
+        exe = self.getBuildArtifact("a.out")
+
+        # Create a target by the debugger.
+        target = self.dbg.CreateTarget(exe)
+        self.assertTrue(target, VALID_TARGET)
+
+        # Now create a breakpoint on main.c by name 'c'.
+        bkpt1 = target.BreakpointCreateByName("c", "a.out")
+        self.trace("breakpoint:", bkpt1)
+        self.assertTrue(bkpt1.GetNumLocations() == 1, VALID_BREAKPOINT)
+
+        self.primary_listener = lldb.SBListener("my listener")
+        self.shadow_listener = lldb.SBListener("shadow listener")
+
+        self.cur_thread = None
+        
+        error = lldb.SBError()
+        launch_info = target.GetLaunchInfo()
+        launch_info.SetListener(self.primary_listener)
+        launch_info.SetShadowListener(self.shadow_listener)
+
+        self.runCmd("settings set target.process.extra-startup-command QSetLogging:bitmask=LOG_PROCESS|LOG_EXCEPTIONS|LOG_RNB_PACKETS|LOG_STEP;")
+        self.dbg.SetAsync(True)
+
+        self.process = target.Launch(launch_info, error)
+        self.assertSuccess(error, "Process launched successfully")
+
+        # Keep fetching events from the primary to trigger the do on removal and
+        # then from the shadow listener, and make sure they match:
+
+        # Events in the launch sequence might be platform dependent, so don't
+        # expect any particular event till we get the stopped:
+        state = lldb.eStateInvalid
+        while state != lldb.eStateStopped:
+            state, restart = self.wait_for_next_event(None, False)
+
+        # Okay, we're now at a good stop, so try a next:
+        self.cur_thread = self.process.threads[0]
+
+        # Make sure we're at our expected breakpoint:
+        self.assertTrue(self.cur_thread.IsValid(), "Got a zeroth thread")
+        self.assertEqual(self.cur_thread.stop_reason, lldb.eStopReasonBreakpoint)
+        self.assertEqual(self.cur_thread.GetStopReasonDataCount(), 2, "Only one breakpoint/loc here")
+        self.assertEqual(bkpt1.GetID(), self.cur_thread.GetStopReasonDataAtIndex(0), "Hit the right breakpoint")
+        # Disable the first breakpoint so it doesn't get in the way...
+        bkpt1.SetEnabled(False)
+
+        self.cur_thread.StepOver()
+        # We'll run the test for "shadow listener blocked by primary listener
+        # for the first couple rounds, then we'll skip the 1 second pause...
+        self.wait_for_next_event(lldb.eStateRunning, True)
+        self.wait_for_next_event(lldb.eStateStopped, True)
+
+        # Next try an auto-continue breakpoint and make sure the shadow listener got
+        # the resumed info as well.  Note that I'm not explicitly counting
+        # running events here.  At the point when I wrote this lldb sometimes
+        # emits two running events in a row.  Apparently the code to coalesce running
+        # events isn't working.  But that's not what this test is testing, we're really
+        # testing that the primary & shadow listeners hear the same thing and in the
+        # right order.
+
+        main_spec = lldb.SBFileSpec("main.c")
+        bkpt2 = target.BreakpointCreateBySourceRegex("b.2. returns %d", main_spec)
+        self.assertTrue(bkpt2.GetNumLocations() > 0, "BP2 worked")
+        bkpt2.SetAutoContinue(True)
+
+        bkpt3 = target.BreakpointCreateBySourceRegex("a.3. returns %d", main_spec)
+        self.assertTrue(bkpt3.GetNumLocations() > 0, "BP3 worked")
+
+        state = lldb.eStateStopped
+        restarted = False
+
+        # Put in a counter to make sure we don't spin forever if there is some
+        # error in the logic.
+        counter = 0
+        while state != lldb.eStateExited:
+            counter += 1
+            self.assertLess(counter, 50, "Took more than 50 events to hit two breakpoints.")
+            if state == lldb.eStateStopped and not restarted:
+                self.process.Continue()
+            state, restarted  = self.wait_for_next_event(None, False)
+