Fix ABA problem when iterating epoll events.

Original patch contributed by andrey.semashev@gmail.com.

In PhysicalSocketServer::WaitEpoll(), the loop verifies that the
signalled dispatcher is in dispatchers_ set. It does so by looking up
the dispatcher pointer in the set. This is vulnerable to the ABA
problem because one dispatcher may be removed and destroyed and another
created and added with the same address before epoll reports an event
for the old dispatcher. The same issue exists for other Wait
implementations, if a dispatcher is removed and a new one added with
the same socket handle is the old.

This is avoided by using a 64-bit key for looking up the dispatcher
in the set. The key is set from a running counter which gets incremented
when a dispatcher is added to the set, so even if the same dispatcher
pointer is added, removed and added again, the key value will be
different.

This changes the storage of dispatchers_ from a set to a flat_hash_map,
which uses a bit more memory but has faster lookup (O(1) as opposed to
O(log n)).

Bug: webrtc:11124
Change-Id: I6d206e1a367b58ba971edca9b48af7664384b797
Reviewed-on: https://webrtc-review.googlesource.com/c/src/+/181027
Commit-Queue: Taylor <deadbeef@webrtc.org>
Reviewed-by: Karl Wiberg <kwiberg@webrtc.org>
Cr-Commit-Position: refs/heads/master@{#32019}

rtc_base/socket_unittest.cc

@@ -149,6 +149,15 @@ void SocketTest::TestCloseInClosedCallbackIPv6() {
   CloseInClosedCallbackInternal(kIPv6Loopback);
 }
 
+void SocketTest::TestDeleteInReadCallbackIPv4() {
+  DeleteInReadCallbackInternal(kIPv4Loopback);
+}
+
+void SocketTest::TestDeleteInReadCallbackIPv6() {
+  MAYBE_SKIP_IPV6;
+  DeleteInReadCallbackInternal(kIPv6Loopback);
+}
+
 void SocketTest::TestSocketServerWaitIPv4() {
   SocketServerWaitInternal(kIPv4Loopback);
 }
@@ -651,6 +660,42 @@ void SocketTest::CloseInClosedCallbackInternal(const IPAddress& loopback) {
   EXPECT_TRUE(Socket::CS_CLOSED == client->GetState());
 }
 
+// Helper class specifically for the test below.
+class SocketDeleter : public sigslot::has_slots<> {
+ public:
+  explicit SocketDeleter(std::unique_ptr<AsyncSocket> socket)
+      : socket_(std::move(socket)) {}
+
+  void Delete(AsyncSocket* other) { socket_.reset(); }
+
+  bool deleted() const { return socket_ == nullptr; }
+
+ private:
+  std::unique_ptr<AsyncSocket> socket_;
+};
+
+// Tested deleting a socket within another socket's read callback. A previous
+// iteration of the select loop failed in this situation, if both sockets
+// became readable at the same time.
+void SocketTest::DeleteInReadCallbackInternal(const IPAddress& loopback) {
+  std::unique_ptr<AsyncSocket> socket1(
+      ss_->CreateAsyncSocket(loopback.family(), SOCK_DGRAM));
+  std::unique_ptr<AsyncSocket> socket2(
+      ss_->CreateAsyncSocket(loopback.family(), SOCK_DGRAM));
+  EXPECT_EQ(0, socket1->Bind(SocketAddress(loopback, 0)));
+  EXPECT_EQ(0, socket2->Bind(SocketAddress(loopback, 0)));
+  EXPECT_EQ(3, socket1->SendTo("foo", 3, socket1->GetLocalAddress()));
+  EXPECT_EQ(3, socket2->SendTo("bar", 3, socket1->GetLocalAddress()));
+  // Sleep a while to ensure sends are both completed at the same time.
+  Thread::SleepMs(1000);
+
+  // Configure the helper class to delete socket 2 when socket 1 has a read
+  // event.
+  SocketDeleter deleter(std::move(socket2));
+  socket1->SignalReadEvent.connect(&deleter, &SocketDeleter::Delete);
+  EXPECT_TRUE_WAIT(deleter.deleted(), kTimeout);
+}
+
 class Sleeper : public MessageHandler {
  public:
   void OnMessage(Message* msg) override { Thread::Current()->SleepMs(500); }