Relanding #2: Fixing crash that can occur if signal is modified while firing.

The crash occurs if a slot causes the very next slot in iteration order
to be disconnected.

Relanding after fixing a race condition that this CL revealed. Previously
the race resulted in an invalidated iterator, but now it will result in the
iterator being modified, so TSan catches it.

BUG=webrtc:7527

Review-Url: https://codereview.webrtc.org/2846593005
Cr-Commit-Position: refs/heads/master@{#18124}

webrtc/base/asyncinvoker.cc

@@ -20,7 +20,6 @@ AsyncInvoker::AsyncInvoker() : invocation_complete_(false, false) {}
 
 AsyncInvoker::~AsyncInvoker() {
   destroying_ = true;
-  SignalInvokerDestroyed();
   // Messages for this need to be cleared *before* our destructor is complete.
   MessageQueueManager::Clear(this);
   // And we need to wait for any invocations that are still in progress on
@@ -126,8 +125,9 @@ NotifyingAsyncClosureBase::NotifyingAsyncClosureBase(
       calling_thread_(calling_thread) {
   calling_thread->SignalQueueDestroyed.connect(
       this, &NotifyingAsyncClosureBase::CancelCallback);
-  invoker->SignalInvokerDestroyed.connect(
-      this, &NotifyingAsyncClosureBase::CancelCallback);
+  // Note: We don't need to listen for the invoker being destroyed, because it
+  // will wait for this closure to be destroyed (and pending_invocations_ to go
+  // to 0) before its destructor completes.
 }
 
 NotifyingAsyncClosureBase::~NotifyingAsyncClosureBase() {

webrtc/base/asyncinvoker.h

@@ -144,9 +144,6 @@ class AsyncInvoker : public MessageHandler {
   // behavior is desired, call Flush() before destroying this object.
   void Flush(Thread* thread, uint32_t id = MQID_ANY);
 
-  // Signaled when this object is destructed.
-  sigslot::signal0<> SignalInvokerDestroyed;
-
  private:
   void OnMessage(Message* msg) override;
   void DoInvoke(const Location& posted_from,

webrtc/base/sigslot.h

@@ -394,7 +394,8 @@ namespace sigslot {
 	protected:
 		typedef std::list< _opaque_connection > connections_list;
 
-		_signal_base() : _signal_base_interface(&_signal_base::do_slot_disconnect, &_signal_base::do_slot_duplicate)
+		_signal_base() : _signal_base_interface(&_signal_base::do_slot_disconnect, &_signal_base::do_slot_duplicate),
+    m_current_iterator(m_connected_slots.end())
 		{
 		}
 
@@ -407,7 +408,8 @@ namespace sigslot {
 		_signal_base& operator= (_signal_base const& that);
 
 	public:
-		_signal_base(const _signal_base& o) : _signal_base_interface(&_signal_base::do_slot_disconnect, &_signal_base::do_slot_duplicate) {
+		_signal_base(const _signal_base& o) : _signal_base_interface(&_signal_base::do_slot_disconnect, &_signal_base::do_slot_duplicate),
+   m_current_iterator(m_connected_slots.end()) {
 			lock_block<mt_policy> lock(this);
       for (const auto& connection : o.m_connected_slots)
 			{
@@ -432,6 +434,10 @@ namespace sigslot {
 				m_connected_slots.pop_front();
 				pdest->signal_disconnect(static_cast< _signal_base_interface* >(this));
 			}
+      // If disconnect_all is called while the signal is firing, advance the
+      // current slot iterator to the end to avoid an invalidated iterator from
+      // being dereferenced.
+      m_current_iterator = m_connected_slots.end();
 		}
 
 #if !defined(NDEBUG)
@@ -460,7 +466,13 @@ namespace sigslot {
 			{
 				if(it->getdest() == pclass)
 				{
-					m_connected_slots.erase(it);
+          // If we're currently using this iterator because the signal is
+          // firing, advance it to avoid it being invalidated.
+          if (m_current_iterator == it) {
+            m_current_iterator = m_connected_slots.erase(it);
+          } else {
+            m_connected_slots.erase(it);
+          }
 					pclass->signal_disconnect(static_cast< _signal_base_interface* >(this));
 					return;
 				}
@@ -484,8 +496,14 @@ namespace sigslot {
 
 				if(it->getdest() == pslot)
 				{
-					self->m_connected_slots.erase(it);
-				}
+          // If we're currently using this iterator because the signal is
+          // firing, advance it to avoid it being invalidated.
+          if (self->m_current_iterator == it) {
+            self->m_current_iterator = self->m_connected_slots.erase(it);
+          } else {
+            self->m_connected_slots.erase(it);
+          }
+        }
 
 				it = itNext;
 			}
@@ -511,7 +529,11 @@ namespace sigslot {
 
 	protected:
 		connections_list m_connected_slots;
-	};
+
+                // Used to handle a slot being disconnected while a signal is
+                // firing (iterating m_connected_slots).
+                connections_list::iterator m_current_iterator;
+        };
 
 	template<class mt_policy = SIGSLOT_DEFAULT_MT_POLICY>
 	class has_slots : public has_slots_interface, public mt_policy
@@ -605,16 +627,15 @@ namespace sigslot {
 		void emit(Args... args)
 		{
 			lock_block<mt_policy> lock(this);
-			typename connections_list::const_iterator it = this->m_connected_slots.begin();
-			typename connections_list::const_iterator itEnd = this->m_connected_slots.end();
-
-			while(it != itEnd)
-			{
-				_opaque_connection const& conn = *it;
-				++it;
-
-				conn.emit<Args...>(args...);
-			}
+                        this->m_current_iterator =
+                            this->m_connected_slots.begin();
+                        while (this->m_current_iterator !=
+                               this->m_connected_slots.end()) {
+                          _opaque_connection const& conn =
+                              *this->m_current_iterator;
+                          ++(this->m_current_iterator);
+                          conn.emit<Args...>(args...);
+                        }
 		}
 
 		void operator()(Args... args)

webrtc/base/sigslot_unittest.cc

@@ -272,3 +272,86 @@ TEST(SigslotTest, CopyConnectedSlot) {
   signal();
   EXPECT_EQ(1, copied_receiver.signal_count());
 }
+
+// Just used for the test below.
+class Disconnector : public sigslot::has_slots<> {
+ public:
+  Disconnector(SigslotReceiver<>* receiver1, SigslotReceiver<>* receiver2)
+      : receiver1_(receiver1), receiver2_(receiver2) {}
+
+  void Connect(sigslot::signal<>* signal) {
+    signal_ = signal;
+    signal->connect(this, &Disconnector::Disconnect);
+  }
+
+ private:
+  void Disconnect() {
+    receiver1_->Disconnect();
+    receiver2_->Disconnect();
+    signal_->disconnect(this);
+  }
+
+  sigslot::signal<>* signal_;
+  SigslotReceiver<>* receiver1_;
+  SigslotReceiver<>* receiver2_;
+};
+
+// Test that things work as expected if a signal is disconnected from a slot
+// while it's firing.
+TEST(SigslotTest, DisconnectFromSignalWhileFiring) {
+  sigslot::signal<> signal;
+  SigslotReceiver<> receiver1;
+  SigslotReceiver<> receiver2;
+  SigslotReceiver<> receiver3;
+  Disconnector disconnector(&receiver1, &receiver2);
+
+  // From this ordering, receiver1 should receive the signal, then the
+  // disconnector will be invoked, causing receiver2 to be disconnected before
+  // it receives the signal. And receiver3 should also receive the signal,
+  // since it was never disconnected.
+  receiver1.Connect(&signal);
+  disconnector.Connect(&signal);
+  receiver2.Connect(&signal);
+  receiver3.Connect(&signal);
+  signal();
+
+  EXPECT_EQ(1, receiver1.signal_count());
+  EXPECT_EQ(0, receiver2.signal_count());
+  EXPECT_EQ(1, receiver3.signal_count());
+}
+
+// Uses disconnect_all instead of disconnect.
+class Disconnector2 : public sigslot::has_slots<> {
+ public:
+  void Connect(sigslot::signal<>* signal) {
+    signal_ = signal;
+    signal->connect(this, &Disconnector2::Disconnect);
+  }
+
+ private:
+  void Disconnect() {
+    signal_->disconnect_all();
+  }
+
+  sigslot::signal<>* signal_;
+};
+
+// Test that things work as expected if a signal is disconnected from a slot
+// while it's firing using disconnect_all.
+TEST(SigslotTest, CallDisconnectAllWhileSignalFiring) {
+  sigslot::signal<> signal;
+  SigslotReceiver<> receiver1;
+  SigslotReceiver<> receiver2;
+  Disconnector2 disconnector;
+
+  // From this ordering, receiver1 should receive the signal, then the
+  // disconnector will be invoked, causing receiver2 to be disconnected before
+  // it receives the signal.
+  receiver1.Connect(&signal);
+  disconnector.Connect(&signal);
+  receiver2.Connect(&signal);
+  signal();
+
+  EXPECT_EQ(1, receiver1.signal_count());
+  EXPECT_EQ(0, receiver2.signal_count());
+}