Support epoll in PhysicalSocketServer.

webrtc/base/physicalsocketserver.cc

 /*
  *  Copyright 2004 The WebRTC Project Authors. All rights reserved.
  *
  *  Use of this source code is governed by a BSD-style license
  *  that can be found in the LICENSE file in the root of the source
  *  tree. An additional intellectual property rights grant can be found
  *  in the file PATENTS.  All contributing project authors may
  *  be found in the AUTHORS file in the root of the source tree.
  */
 #include "webrtc/base/physicalsocketserver.h"
@@ skipping 19 matching lines @@
 
 #if defined(WEBRTC_WIN)
 #define WIN32_LEAN_AND_MEAN
 #include <windows.h>
 #include <winsock2.h>
 #include <ws2tcpip.h>
 #undef SetPort
 #endif
 
 #include <algorithm>
+#include <limits>
 #include <map>
 
 #include "webrtc/base/arraysize.h"
 #include "webrtc/base/basictypes.h"
 #include "webrtc/base/byteorder.h"
 #include "webrtc/base/checks.h"
 #include "webrtc/base/logging.h"
 #include "webrtc/base/networkmonitor.h"
 #include "webrtc/base/nullsocketserver.h"
 #include "webrtc/base/timeutils.h"
@@ skipping 24 matching lines @@
 
 int64_t GetSocketRecvTimestamp(int socket) {
   return -1;
 }
 #endif
 
 #if defined(WEBRTC_WIN)
 typedef char* SockOptArg;
 #endif
 
+#if defined(WEBRTC_POSIX) && defined(WEBRTC_LINUX)
+// EPOLLRDHUP is only defined starting with Linux 2.6.17.
+#if !defined(EPOLLRDHUP)
+#define EPOLLRDHUP 0x2000
+#endif
+#endif
+
 namespace rtc {
 
 std::unique_ptr<SocketServer> SocketServer::CreateDefault() {
 #if defined(__native_client__)
   return std::unique_ptr<SocketServer>(new rtc::NullSocketServer);
 #else
   return std::unique_ptr<SocketServer>(new rtc::PhysicalSocketServer);
 #endif
 }
 
@@ skipping 33 matching lines @@
     resolver_(nullptr) {
 #if defined(WEBRTC_WIN)
   // EnsureWinsockInit() ensures that winsock is initialized. The default
   // version of this function doesn't do anything because winsock is
   // initialized by constructor of a static object. If neccessary libjingle
   // users can link it with a different version of this function by replacing
   // win32socketinit.cc. See win32socketinit.cc for more details.
   EnsureWinsockInit();
 #endif
   if (s_ != INVALID_SOCKET) {
-    enabled_events_ = DE_READ | DE_WRITE;
+    SetEnabledEvents(DE_READ | DE_WRITE);

[Taylor Brandstetter, 2017/05/17 07:20:16]

nit: The "SetEnabledEvents" etc. change seems like it could be in an independent
CL

[joachim, 2017/05/17 21:17:02]

Moved to https://codereview.webrtc.org/2893723002/

 
     int type = SOCK_STREAM;
     socklen_t len = sizeof(type);
     const int res =
         getsockopt(s_, SOL_SOCKET, SO_TYPE, (SockOptArg)&type, &len);
     RTC_DCHECK_EQ(0, res);
     udp_ = (SOCK_DGRAM == type);
   }
 }
 
 PhysicalSocket::~PhysicalSocket() {
   Close();
 }
 
 bool PhysicalSocket::Create(int family, int type) {
   Close();
   s_ = ::socket(family, type, 0);
   udp_ = (SOCK_DGRAM == type);
   UpdateLastError();
-  if (udp_)
-    enabled_events_ = DE_READ | DE_WRITE;
+  if (udp_) {
+    SetEnabledEvents(DE_READ | DE_WRITE);
+  }
   return s_ != INVALID_SOCKET;
 }
 
 SocketAddress PhysicalSocket::GetLocalAddress() const {
   sockaddr_storage addr_storage = {0};
   socklen_t addrlen = sizeof(addr_storage);
   sockaddr* addr = reinterpret_cast<sockaddr*>(&addr_storage);
   int result = ::getsockname(s_, addr, &addrlen);
   SocketAddress address;
   if (result >= 0) {
@@ skipping 91 matching lines @@
 int PhysicalSocket::DoConnect(const SocketAddress& connect_addr) {
   if ((s_ == INVALID_SOCKET) &&
       !Create(connect_addr.family(), SOCK_STREAM)) {
     return SOCKET_ERROR;
   }
   sockaddr_storage addr_storage;
   size_t len = connect_addr.ToSockAddrStorage(&addr_storage);
   sockaddr* addr = reinterpret_cast<sockaddr*>(&addr_storage);
   int err = ::connect(s_, addr, static_cast<int>(len));
   UpdateLastError();
+  uint8_t events = DE_READ | DE_WRITE;
   if (err == 0) {
     state_ = CS_CONNECTED;
   } else if (IsBlockingError(GetError())) {
     state_ = CS_CONNECTING;
-    enabled_events_ |= DE_CONNECT;
+    events |= DE_CONNECT;
   } else {
     return SOCKET_ERROR;
   }
 
-  enabled_events_ |= DE_READ | DE_WRITE;
+  EnableEvents(events);
   return 0;
 }
 
 int PhysicalSocket::GetError() const {
   CritScope cs(&crit_);
   return error_;
 }
 
 void PhysicalSocket::SetError(int error) {
   CritScope cs(&crit_);
@@ skipping 45 matching lines @@
 #else
       0
 #endif
       );
   UpdateLastError();
   MaybeRemapSendError();
   // We have seen minidumps where this may be false.
   RTC_DCHECK(sent <= static_cast<int>(cb));
   if ((sent > 0 && sent < static_cast<int>(cb)) ||
       (sent < 0 && IsBlockingError(GetError()))) {
-    enabled_events_ |= DE_WRITE;
+    EnableEvents(DE_WRITE);
   }
   return sent;
 }
 
 int PhysicalSocket::SendTo(const void* buffer,
                            size_t length,
                            const SocketAddress& addr) {
   sockaddr_storage saddr;
   size_t len = addr.ToSockAddrStorage(&saddr);
   int sent = DoSendTo(
       s_, static_cast<const char *>(buffer), static_cast<int>(length),
 #if defined(WEBRTC_LINUX) && !defined(WEBRTC_ANDROID)
       // Suppress SIGPIPE. See above for explanation.
       MSG_NOSIGNAL,
 #else
       0,
 #endif
       reinterpret_cast<sockaddr*>(&saddr), static_cast<int>(len));
   UpdateLastError();
   MaybeRemapSendError();
   // We have seen minidumps where this may be false.
   RTC_DCHECK(sent <= static_cast<int>(length));
   if ((sent > 0 && sent < static_cast<int>(length)) ||
       (sent < 0 && IsBlockingError(GetError()))) {
-    enabled_events_ |= DE_WRITE;
+    EnableEvents(DE_WRITE);
   }
   return sent;
 }
 
 int PhysicalSocket::Recv(void* buffer, size_t length, int64_t* timestamp) {
   int received = ::recv(s_, static_cast<char*>(buffer),
                         static_cast<int>(length), 0);
   if ((received == 0) && (length != 0)) {
     // Note: on graceful shutdown, recv can return 0.  In this case, we
     // pretend it is blocking, and then signal close, so that simplifying
     // assumptions can be made about Recv.
     LOG(LS_WARNING) << "EOF from socket; deferring close event";
     // Must turn this back on so that the select() loop will notice the close
     // event.
-    enabled_events_ |= DE_READ;
+    EnableEvents(DE_READ);
     SetError(EWOULDBLOCK);
     return SOCKET_ERROR;
   }
   if (timestamp) {
     *timestamp = GetSocketRecvTimestamp(s_);
   }
   UpdateLastError();
   int error = GetError();
   bool success = (received >= 0) || IsBlockingError(error);
   if (udp_ || success) {
-    enabled_events_ |= DE_READ;
+    EnableEvents(DE_READ);
   }
   if (!success) {
     LOG_F(LS_VERBOSE) << "Error = " << error;
   }
   return received;
 }
 
 int PhysicalSocket::RecvFrom(void* buffer,
                              size_t length,
                              SocketAddress* out_addr,
                              int64_t* timestamp) {
   sockaddr_storage addr_storage;
   socklen_t addr_len = sizeof(addr_storage);
   sockaddr* addr = reinterpret_cast<sockaddr*>(&addr_storage);
   int received = ::recvfrom(s_, static_cast<char*>(buffer),
                             static_cast<int>(length), 0, addr, &addr_len);
   if (timestamp) {
     *timestamp = GetSocketRecvTimestamp(s_);
   }
   UpdateLastError();
   if ((received >= 0) && (out_addr != nullptr))
     SocketAddressFromSockAddrStorage(addr_storage, out_addr);
   int error = GetError();
   bool success = (received >= 0) || IsBlockingError(error);
   if (udp_ || success) {
-    enabled_events_ |= DE_READ;
+    EnableEvents(DE_READ);
   }
   if (!success) {
     LOG_F(LS_VERBOSE) << "Error = " << error;
   }
   return received;
 }
 
 int PhysicalSocket::Listen(int backlog) {
   int err = ::listen(s_, backlog);
   UpdateLastError();
   if (err == 0) {
     state_ = CS_CONNECTING;
-    enabled_events_ |= DE_ACCEPT;
+    EnableEvents(DE_ACCEPT);
 #if !defined(NDEBUG)
     dbg_addr_ = "Listening @ ";
     dbg_addr_.append(GetLocalAddress().ToString());
 #endif
   }
   return err;
 }
 
 AsyncSocket* PhysicalSocket::Accept(SocketAddress* out_addr) {
   // Always re-subscribe DE_ACCEPT to make sure new incoming connections will
   // trigger an event even if DoAccept returns an error here.
-  enabled_events_ |= DE_ACCEPT;
+  EnableEvents(DE_ACCEPT);
   sockaddr_storage addr_storage;
   socklen_t addr_len = sizeof(addr_storage);
   sockaddr* addr = reinterpret_cast<sockaddr*>(&addr_storage);
   SOCKET s = DoAccept(s_, addr, &addr_len);
   UpdateLastError();
   if (s == INVALID_SOCKET)
     return nullptr;
   if (out_addr != nullptr)
     SocketAddressFromSockAddrStorage(addr_storage, out_addr);
   return ss_->WrapSocket(s);
 }
 
 int PhysicalSocket::Close() {
   if (s_ == INVALID_SOCKET)
     return 0;
   int err = ::closesocket(s_);
   UpdateLastError();
   s_ = INVALID_SOCKET;
   state_ = CS_CLOSED;
-  enabled_events_ = 0;
+  SetEnabledEvents(0);
   if (resolver_) {
     resolver_->Destroy(false);
     resolver_ = nullptr;
   }
   return err;
 }
 
 int PhysicalSocket::EstimateMTU(uint16_t* mtu) {
   SocketAddress addr = GetRemoteAddress();
   if (addr.IsAnyIP()) {
@@ skipping 105 matching lines @@
   // Reference/ManPages/man2/sendto.2.html
   // ENOBUFS - The output queue for a network interface is full.
   // This generally indicates that the interface has stopped sending,
   // but may be caused by transient congestion.
   if (GetError() == ENOBUFS) {
     SetError(EWOULDBLOCK);
   }
 #endif
 }
 
+void PhysicalSocket::SetEnabledEvents(uint8_t events) {
+  enabled_events_ = events;
+}
+
+void PhysicalSocket::EnableEvents(uint8_t events) {
+  enabled_events_ |= events;
+}
+
+void PhysicalSocket::DisableEvents(uint8_t events) {
+  enabled_events_ &= ~events;
+}
+
 int PhysicalSocket::TranslateOption(Option opt, int* slevel, int* sopt) {
   switch (opt) {
     case OPT_DONTFRAGMENT:
 #if defined(WEBRTC_WIN)
       *slevel = IPPROTO_IP;
       *sopt = IP_DONTFRAGMENT;
       break;
 #elif defined(WEBRTC_MAC) || defined(BSD) || defined(__native_client__)
       LOG(LS_WARNING) << "Socket::OPT_DONTFRAGMENT not supported.";
       return -1;
@@ skipping 178 matching lines @@
 
 #if defined(WEBRTC_WIN)
 
 void SocketDispatcher::OnEvent(uint32_t ff, int err) {
   int cache_id = id_;
   // Make sure we deliver connect/accept first. Otherwise, consumers may see
   // something like a READ followed by a CONNECT, which would be odd.
   if (((ff & DE_CONNECT) != 0) && (id_ == cache_id)) {
     if (ff != DE_CONNECT)
       LOG(LS_VERBOSE) << "Signalled with DE_CONNECT: " << ff;
-    enabled_events_ &= ~DE_CONNECT;
+    DisableEvents(DE_CONNECT);
 #if !defined(NDEBUG)
     dbg_addr_ = "Connected @ ";
     dbg_addr_.append(GetRemoteAddress().ToString());
 #endif
     SignalConnectEvent(this);
   }
   if (((ff & DE_ACCEPT) != 0) && (id_ == cache_id)) {
-    enabled_events_ &= ~DE_ACCEPT;
+    DisableEvents(DE_ACCEPT);
     SignalReadEvent(this);
   }
   if ((ff & DE_READ) != 0) {
-    enabled_events_ &= ~DE_READ;
+    DisableEvents(DE_READ);
     SignalReadEvent(this);
   }
   if (((ff & DE_WRITE) != 0) && (id_ == cache_id)) {
-    enabled_events_ &= ~DE_WRITE;
+    DisableEvents(DE_WRITE);
     SignalWriteEvent(this);
   }
   if (((ff & DE_CLOSE) != 0) && (id_ == cache_id)) {
     signal_close_ = true;
     signal_err_ = err;
   }
 }
 
 #elif defined(WEBRTC_POSIX)
 
 void SocketDispatcher::OnEvent(uint32_t ff, int err) {
+#if defined(WEBRTC_LINUX)

[Taylor Brandstetter, 2017/05/17 07:20:16]

nit: It may be cleaner to do "#ifdef WEBRTC_LINUX #define USE_EPOLL" at the top
of this file, then these could be "#ifdef USE_EPOLL" which is more readable.

[joachim, 2017/05/17 21:17:02]

Done (named it "WEBRTC_USE_EPOLL").

+  // Remember currently enabled events so we can combine multiple changes
+  // into one update call later.
+  PushEnabledEvents();
+#endif
   // Make sure we deliver connect/accept first. Otherwise, consumers may see
   // something like a READ followed by a CONNECT, which would be odd.
   if ((ff & DE_CONNECT) != 0) {
-    enabled_events_ &= ~DE_CONNECT;
+    DisableEvents(DE_CONNECT);
     SignalConnectEvent(this);
   }
   if ((ff & DE_ACCEPT) != 0) {
-    enabled_events_ &= ~DE_ACCEPT;
+    DisableEvents(DE_ACCEPT);
     SignalReadEvent(this);
   }
   if ((ff & DE_READ) != 0) {
-    enabled_events_ &= ~DE_READ;
+    DisableEvents(DE_READ);
     SignalReadEvent(this);
   }
   if ((ff & DE_WRITE) != 0) {
-    enabled_events_ &= ~DE_WRITE;
+    DisableEvents(DE_WRITE);
     SignalWriteEvent(this);
   }
   if ((ff & DE_CLOSE) != 0) {
     // The socket is now dead to us, so stop checking it.
-    enabled_events_ = 0;
+    SetEnabledEvents(0);
     SignalCloseEvent(this, err);
   }
+#if defined(WEBRTC_LINUX)
+  PopEnabledEvents();
+#endif
 }
 
+#if defined(WEBRTC_LINUX)
+
+void SocketDispatcher::PushEnabledEvents() {
+  enabled_events_stack_.push_back(enabled_events_);

[Taylor Brandstetter, 2017/05/17 07:20:16]

Will the stack ever have a size > 1? If so, can you leave a comment explaining
how? If not, I think this code could be simplified somewhat.

[joachim, 2017/05/17 21:17:02]

Currently it will only be one element. I'll change it and rename to
"SaveEnabledEvents" / "RestoreEnabledEvents".

+}
+
+void SocketDispatcher::PopEnabledEvents() {
+  RTC_DCHECK(!enabled_events_stack_.empty());
+  uint8_t old_events = enabled_events_stack_.back();
+  enabled_events_stack_.pop_back();
+  MaybeUpdateDispatcher(old_events);
+}
+
+void SocketDispatcher::MaybeUpdateDispatcher(uint8_t old_events) {
+  if (enabled_events_ != old_events && enabled_events_stack_.empty()) {
+    ss_->Update(this);
+  }
+}
+
+void SocketDispatcher::SetEnabledEvents(uint8_t events) {
+  uint8_t old_events = enabled_events_;
+  PhysicalSocket::SetEnabledEvents(events);
+  MaybeUpdateDispatcher(old_events);
+}
+
+void SocketDispatcher::EnableEvents(uint8_t events) {
+  uint8_t old_events = enabled_events_;
+  PhysicalSocket::EnableEvents(events);
+  MaybeUpdateDispatcher(old_events);
+}
+
+void SocketDispatcher::DisableEvents(uint8_t events) {
+  uint8_t old_events = enabled_events_;
+  PhysicalSocket::DisableEvents(events);
+  MaybeUpdateDispatcher(old_events);
+}
+
+#endif // WEBRTC_LINUX
+
 #endif // WEBRTC_POSIX
 
 int SocketDispatcher::Close() {
   if (s_ == INVALID_SOCKET)
     return 0;
 
 #if defined(WEBRTC_WIN)
   id_ = 0;
   signal_close_ = false;
 #endif
@@ skipping 310 matching lines @@
     if (pf_)
       *pf_ = false;
   }
 
  private:
   bool *pf_;
 };
 
 PhysicalSocketServer::PhysicalSocketServer()
     : fWait_(false) {
+#if defined(WEBRTC_POSIX) && defined(WEBRTC_LINUX)
+  // Since Linux 2.6.8, the size argument is ignored, but must be greater than
+  // zero. Before that the size served as hint to the kernel for the amount of
+  // space to initially allocate in internal data structures.
+  epoll_fd_ = epoll_create(FD_SETSIZE);
+  if (epoll_fd_ == -1) {
+    LOG_E(LS_WARNING, EN, errno) << "epoll_create";
+    epoll_fd_ = INVALID_SOCKET;

[Taylor Brandstetter, 2017/05/17 07:20:16]

Could you leave a comment mentioning that the code will fall back to select() if
epoll_create() failed?

[joachim, 2017/05/17 21:17:02]

Done.

+  }
+#endif
   signal_wakeup_ = new Signaler(this, &fWait_);
 #if defined(WEBRTC_WIN)
   socket_ev_ = WSACreateEvent();
 #endif
 }
 
 PhysicalSocketServer::~PhysicalSocketServer() {
 #if defined(WEBRTC_WIN)
   WSACloseEvent(socket_ev_);
 #endif
 #if defined(WEBRTC_POSIX)
   signal_dispatcher_.reset();
 #endif
   delete signal_wakeup_;
+#if defined(WEBRTC_POSIX) && defined(WEBRTC_LINUX)
+  if (epoll_fd_ != INVALID_SOCKET) {
+    close(epoll_fd_);
+  }
+#endif
   RTC_DCHECK(dispatchers_.empty());
 }
 
 void PhysicalSocketServer::WakeUp() {
   signal_wakeup_->Signal();
 }
 
 Socket* PhysicalSocketServer::CreateSocket(int type) {
   return CreateSocket(AF_INET, type);
 }
@@ skipping 27 matching lines @@
   if (dispatcher->Initialize()) {
     return dispatcher;
   } else {
     delete dispatcher;
     return nullptr;
   }
 }
 
 void PhysicalSocketServer::Add(Dispatcher *pdispatcher) {
   CritScope cs(&crit_);
+#if defined(WEBRTC_WIN)
   // Prevent duplicates. This can cause dead dispatchers to stick around.
   DispatcherList::iterator pos = std::find(dispatchers_.begin(),
                                            dispatchers_.end(),
                                            pdispatcher);
   if (pos != dispatchers_.end())
     return;
   dispatchers_.push_back(pdispatcher);
+#else

[Taylor Brandstetter, 2017/05/17 07:20:16]

Is there a reason the set can't be used on Windows?

[joachim, 2017/05/17 21:17:02]

The Windows code has some special handling to invalidate iterators used in
"Wait" below when removing dispatchers from the PSS. For this it needs the
position of the removed dispatcher, so we can't use a set there.

I didn't take a deeper look and in any case if something is changed there, I
think it should be done in a separate CL.

+  dispatchers_.insert(pdispatcher);
+#if defined(WEBRTC_POSIX) && defined(WEBRTC_LINUX)
+  if (epoll_fd_ != INVALID_SOCKET) {
+    AddEpoll(pdispatcher);
+  }
+#endif  // WEBRTC_POSIX && WEBRTC_LINUX
+#endif  // WEBRTC_WIN
 }
 
 void PhysicalSocketServer::Remove(Dispatcher *pdispatcher) {
   CritScope cs(&crit_);
+#if defined(WEBRTC_WIN)
   DispatcherList::iterator pos = std::find(dispatchers_.begin(),
                                            dispatchers_.end(),
                                            pdispatcher);
   // We silently ignore duplicate calls to Add, so we should silently ignore
   // the (expected) symmetric calls to Remove. Note that this may still hide
   // a real issue, so we at least log a warning about it.
   if (pos == dispatchers_.end()) {
     LOG(LS_WARNING) << "PhysicalSocketServer asked to remove a unknown "
                     << "dispatcher, potentially from a duplicate call to Add.";
     return;
   }
   size_t index = pos - dispatchers_.begin();
   dispatchers_.erase(pos);
   for (IteratorList::iterator it = iterators_.begin(); it != iterators_.end();
        ++it) {
     if (index < **it) {
       --**it;
     }
   }
+#else
+  if (!dispatchers_.erase(pdispatcher)) {
+    LOG(LS_WARNING) << "PhysicalSocketServer asked to remove a unknown "
+                    << "dispatcher, potentially from a duplicate call to Add.";
+    return;
+  }
+#if defined(WEBRTC_POSIX) && defined(WEBRTC_LINUX)
+  if (epoll_fd_ != INVALID_SOCKET) {
+    RemoveEpoll(pdispatcher);
+  }
+#endif  // WEBRTC_POSIX && WEBRTC_LINUX
+#endif  // WEBRTC_WIN
+}
+
+void PhysicalSocketServer::Update(Dispatcher *pdispatcher) {

[Taylor Brandstetter, 2017/05/17 07:20:16]

It seems like the code would be more readable if there was an "#ifdef" at the
call site, which called a method named "UpdateEpoll" directly.

[joachim, 2017/05/17 21:17:02]

The call site for this is the SocketDispatcher. I wanted to hide the "epoll"
implementation detail of the PSS and keep the public method names consistent.

Already available are "Add" and "Remove" with "Update" being added in this CL. I
you really think it would be more readable to name this "UpdateEpoll" and call
that from the SocketDispatcher, I can change it.

[Taylor Brandstetter, 2017/05/17 23:19:02]

I see now; in that case this is fine.

+#if defined(WEBRTC_POSIX) && defined(WEBRTC_LINUX)
+  if (epoll_fd_ == INVALID_SOCKET) {
+    return;
+  }
+
+  CritScope cs(&crit_);
+  if (dispatchers_.find(pdispatcher) == dispatchers_.end()) {
+    return;
+  }
+
+  UpdateEpoll(pdispatcher);
+#endif
 }
 
 #if defined(WEBRTC_POSIX)
+
 bool PhysicalSocketServer::Wait(int cmsWait, bool process_io) {
+#if defined(WEBRTC_LINUX)
+  // We don't keep a dedicated epoll descriptor containing only non-IO (i.e.
+  // signaling) dispatchers, so the default "select" loop will be used in that
+  // case.
+  if (epoll_fd_ != INVALID_SOCKET && process_io) {
+    return WaitEpoll(cmsWait);
+  }
+#endif
+  return WaitSelect(cmsWait, process_io);
+}
+
+bool PhysicalSocketServer::WaitSelect(int cmsWait, bool process_io) {
   // Calculate timing information
 
   struct timeval* ptvWait = nullptr;
   struct timeval tvWait;
   struct timeval tvStop;
   if (cmsWait != kForever) {
     // Calculate wait timeval
     tvWait.tv_sec = cmsWait / 1000;
     tvWait.tv_usec = (cmsWait % 1000) * 1000;
     ptvWait = &tvWait;
@@ skipping 22 matching lines @@
   __msan_unpoison(&fdsRead, sizeof(fdsRead));
   __msan_unpoison(&fdsWrite, sizeof(fdsWrite));
 #endif
 
   fWait_ = true;
 
   while (fWait_) {
     int fdmax = -1;
     {
       CritScope cr(&crit_);
-      for (size_t i = 0; i < dispatchers_.size(); ++i) {
+      for (Dispatcher* pdispatcher : dispatchers_) {
         // Query dispatchers for read and write wait state
-        Dispatcher *pdispatcher = dispatchers_[i];
         RTC_DCHECK(pdispatcher);
         if (!process_io && (pdispatcher != signal_wakeup_))
           continue;
         int fd = pdispatcher->GetDescriptor();
+        // "select"ing a file descriptor that is equal to or larger than
+        // FD_SETSIZE will result in undefined behavior.
+        RTC_DCHECK_LT(fd, FD_SETSIZE);
         if (fd > fdmax)
           fdmax = fd;
 
         uint32_t ff = pdispatcher->GetRequestedEvents();
         if (ff & (DE_READ | DE_ACCEPT))
           FD_SET(fd, &fdsRead);
         if (ff & (DE_WRITE | DE_CONNECT))
           FD_SET(fd, &fdsWrite);
       }
     }
@@ skipping 13 matching lines @@
       // Else ignore the error and keep going. If this EINTR was for one of the
       // signals managed by this PhysicalSocketServer, the
       // PosixSignalDeliveryDispatcher will be in the signaled state in the next
       // iteration.
     } else if (n == 0) {
       // If timeout, return success
       return true;
     } else {
       // We have signaled descriptors
       CritScope cr(&crit_);
-      for (size_t i = 0; i < dispatchers_.size(); ++i) {
-        Dispatcher *pdispatcher = dispatchers_[i];
+      for (Dispatcher *pdispatcher : dispatchers_) {
         int fd = pdispatcher->GetDescriptor();
         uint32_t ff = 0;
         int errcode = 0;
 
         // Reap any error code, which can be signaled through reads or writes.
         // TODO(pthatcher): Should we set errcode if getsockopt fails?
         if (FD_ISSET(fd, &fdsRead) || FD_ISSET(fd, &fdsWrite)) {
           socklen_t len = sizeof(errcode);
           ::getsockopt(fd, SOL_SOCKET, SO_ERROR, &errcode, &len);
         }
@@ skipping 53 matching lines @@
           ptvWait->tv_usec += 1000000;
           ptvWait->tv_sec -= 1;
         }
       }
     }
   }
 
   return true;
 }
 
+#if defined(WEBRTC_LINUX)
+
+// Initial number of events to process with one call to "epoll_wait".
+static const size_t kInitialEpollEvents = 128;
+
+// Maximum number of events to process with one call to "epoll_wait".
+static const size_t kMaxEpollEvents = 8192;
+
+static int GetEpollEvents(Dispatcher* pdispatcher) {
+  int events = 0;
+  uint32_t ff = pdispatcher->GetRequestedEvents();
+  if (ff & (DE_READ | DE_ACCEPT)) {
+    events |= EPOLLIN;
+  }
+  if (ff & (DE_WRITE | DE_CONNECT)) {
+    events |= EPOLLOUT;
+  }
+  return events;
+}
+
+void PhysicalSocketServer::AddEpoll(Dispatcher* pdispatcher) {
+  RTC_DCHECK(epoll_fd_ != INVALID_SOCKET);
+  int fd = pdispatcher->GetDescriptor();
+  RTC_DCHECK(fd != INVALID_SOCKET);
+  if (fd == INVALID_SOCKET) {
+    return;
+  }
+
+  struct epoll_event event = {0};
+  event.events = GetEpollEvents(pdispatcher);
+  event.data.ptr = pdispatcher;
+  int err = epoll_ctl(epoll_fd_, EPOLL_CTL_ADD, fd, &event);
+  RTC_DCHECK_EQ(err, 0);
+  if (err == -1) {
+    LOG_E(LS_ERROR, EN, errno) << "epoll_ctl EPOLL_CTL_ADD";
+  }
+}
+
+void PhysicalSocketServer::RemoveEpoll(Dispatcher* pdispatcher) {
+  RTC_DCHECK(epoll_fd_ != INVALID_SOCKET);
+  int fd = pdispatcher->GetDescriptor();
+  RTC_DCHECK(fd != INVALID_SOCKET);
+  if (fd == INVALID_SOCKET) {
+    return;
+  }
+
+  struct epoll_event event = {0};
+  int err = epoll_ctl(epoll_fd_, EPOLL_CTL_DEL, fd, &event);
+  RTC_DCHECK(err == 0 || errno == ENOENT);
+  if (err == -1) {
+    if (errno == ENOENT) {
+      // Socket has already been closed.
+      LOG_E(LS_VERBOSE, EN, errno) << "epoll_ctl EPOLL_CTL_DEL";
+    } else {
+      LOG_E(LS_ERROR, EN, errno) << "epoll_ctl EPOLL_CTL_DEL";
+    }
+  }
+}
+
+void PhysicalSocketServer::UpdateEpoll(Dispatcher* pdispatcher) {
+  RTC_DCHECK(epoll_fd_ != INVALID_SOCKET);
+  int fd = pdispatcher->GetDescriptor();
+  RTC_DCHECK(fd != INVALID_SOCKET);
+  if (fd == INVALID_SOCKET) {
+    return;
+  }
+
+  struct epoll_event event = {0};
+  event.events = GetEpollEvents(pdispatcher);
+  event.data.ptr = pdispatcher;
+  int err = epoll_ctl(epoll_fd_, EPOLL_CTL_MOD, fd, &event);
+  RTC_DCHECK_EQ(err, 0);
+  if (err == -1) {
+    LOG_E(LS_ERROR, EN, errno) << "epoll_ctl EPOLL_CTL_MOD";
+  }
+}
+
+bool PhysicalSocketServer::WaitEpoll(int cmsWait) {
+  RTC_DCHECK(epoll_fd_ != INVALID_SOCKET);
+  int64_t tvWait = -1;
+  int64_t tvStop = -1;
+  if (cmsWait != kForever) {
+    tvWait = cmsWait;
+    tvStop = TimeAfter(cmsWait);
+  }
+
+  if (epoll_events_.empty()) {
+    // The initial space to receive events is created only if epoll is used.
+    epoll_events_.resize(kInitialEpollEvents);
+  }
+
+  fWait_ = true;
+
+  while (fWait_) {
+    // Wait then call handlers as appropriate
+    // < 0 means error
+    // 0 means timeout
+    // > 0 means count of descriptors ready
+    int n = epoll_wait(epoll_fd_, &epoll_events_[0],
+        static_cast<int>(epoll_events_.size()), static_cast<int>(tvWait));
+    if (n < 0) {
+      if (errno != EINTR) {
+        LOG_E(LS_ERROR, EN, errno) << "epoll";
+        return false;
+      }
+      // Else ignore the error and keep going. If this EINTR was for one of the
+      // signals managed by this PhysicalSocketServer, the
+      // PosixSignalDeliveryDispatcher will be in the signaled state in the next
+      // iteration.
+    } else if (n == 0) {
+      // If timeout, return success
+      return true;
+    } else {
+      // We have signaled descriptors
+      CritScope cr(&crit_);
+      for (int i = 0; i < n; ++i) {
+        const epoll_event& event = epoll_events_[i];
+        Dispatcher* pdispatcher = static_cast<Dispatcher*>(event.data.ptr);
+        if (dispatchers_.find(pdispatcher) == dispatchers_.end()) {
+          // The dispatcher for this socket no longer exists.
+          continue;
+        }
+
+        int fd = pdispatcher->GetDescriptor();
+        uint32_t ff = 0;
+        int errcode = 0;
+
+        // Reap any error code.
+        // TODO(pthatcher): Should we set errcode if getsockopt fails?
+        if (event.events & (EPOLLRDHUP | EPOLLERR | EPOLLHUP)) {
+          socklen_t len = sizeof(errcode);
+          ::getsockopt(fd, SOL_SOCKET, SO_ERROR, &errcode, &len);
+        }
+
+        // Check readable descriptors. If we're waiting on an accept, signal
+        // that. Otherwise we're waiting for data, check to see if we're
+        // readable or really closed.
+        // TODO(pthatcher): Only peek at TCP descriptors.
+        if ((event.events & (EPOLLIN | EPOLLPRI))) {
+          if (pdispatcher->GetRequestedEvents() & DE_ACCEPT) {
+            ff |= DE_ACCEPT;
+          } else if (errcode || pdispatcher->IsDescriptorClosed()) {
+            ff |= DE_CLOSE;
+          } else {
+            ff |= DE_READ;
+          }
+        }
+
+        // Check writable descriptors. If we're waiting on a connect, detect
+        // success versus failure by the reaped error code.
+        if ((event.events & EPOLLOUT)) {
+          if (pdispatcher->GetRequestedEvents() & DE_CONNECT) {
+            if (!errcode) {
+              ff |= DE_CONNECT;
+            } else {
+              ff |= DE_CLOSE;
+            }
+          } else {
+            ff |= DE_WRITE;
+          }
+        }
+
+        // Tell the descriptor about the event.
+        if (ff != 0) {
+          pdispatcher->OnPreEvent(ff);
+          pdispatcher->OnEvent(ff, errcode);
+        }
+      }
+    }
+
+    if (static_cast<size_t>(n) == epoll_events_.size() &&
+        epoll_events_.size() < kMaxEpollEvents) {
+      // We used the complete space to receive events, increase size for future
+      // iterations.
+      epoll_events_.resize(std::max(epoll_events_.size() * 2, kMaxEpollEvents));
+    }
+
+    if (cmsWait != kForever) {
+      tvWait = TimeDiff(tvStop, TimeMillis());
+      if (tvWait < 0) {
+        // Return success on timeout.
+        return true;
+      }
+    }
+  }
+
+  return true;
+}
+
+#endif  // WEBRTC_LINUX
+
 static void GlobalSignalHandler(int signum) {
   PosixSignalHandler::Instance()->OnPosixSignalReceived(signum);
 }
 
 bool PhysicalSocketServer::SetPosixSignalHandler(int signum,
                                                  void (*handler)(int)) {
   // If handler is SIG_IGN or SIG_DFL then clear our user-level handler,
   // otherwise set one.
   if (handler == SIG_IGN || handler == SIG_DFL) {
     if (!InstallSignal(signum, handler)) {
@@ skipping 201 matching lines @@
        break;
     }
   }
 
   // Done
   return true;
 }
 #endif  // WEBRTC_WIN
 
 }  // namespace rtc