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 62 matching lines @@
 
 int64_t GetSocketRecvTimestamp(int socket) {
   return -1;
 }
 #endif
 
 #if defined(WEBRTC_WIN)
 typedef char* SockOptArg;
 #endif
 
+#if defined(WEBRTC_USE_EPOLL)
+// 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 665 matching lines @@
   }
   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_USE_EPOLL)
+  // Remember currently enabled events so we can combine multiple changes
+  // into one update call later.
+  SaveEnabledEvents();
+#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) {
     DisableEvents(DE_CONNECT);
     SignalConnectEvent(this);
   }
   if ((ff & DE_ACCEPT) != 0) {
     DisableEvents(DE_ACCEPT);
     SignalReadEvent(this);
   }
   if ((ff & DE_READ) != 0) {
     DisableEvents(DE_READ);
     SignalReadEvent(this);
   }
   if ((ff & DE_WRITE) != 0) {
     DisableEvents(DE_WRITE);
     SignalWriteEvent(this);
   }
   if ((ff & DE_CLOSE) != 0) {
     // The socket is now dead to us, so stop checking it.
     SetEnabledEvents(0);
     SignalCloseEvent(this, err);
   }
+#if defined(WEBRTC_USE_EPOLL)
+  RestoreEnabledEvents();

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

This name isn't very intuitive; it sounds like it would do "enabled_events_ =
saved_enabled_events_". Better names may be "StartBatchedEventUpdates" and
"FinishBatchedEventUpdates", or something along those lines.

Though in my opinion, this is a little overkill in the first place. This method
could just do:

uint8_t events_to_disable = 0;
if ((ff & DE_CONNECT) != 0) {
  events_to_disable |= DE_CONNECT;
  SignalConnectEvent(this);
}
/* ... */
if ((ff & DE_CLOSE) != 0) {
  events_to_disable = 0xff;
  SignalCloseEvent(this, err);
}
DisableEvents(events_to_disable);

That would avoid needing an extra member variable and two methods.

[joachim, 2017/05/18 12:41:49]

Keeping a bitmask of events to disable won't work, because the signal handlers
might re-enable some events. These don't get updated in "events_to_disable" and
would wrongly be disabled at the end.

[Taylor Brandstetter, 2017/05/18 21:59:56]

Ah, that's what I was missing. Can you mention it in a comment?

[joachim, 2017/05/19 20:46:23]

Done.

+#endif
 }
 
 #endif // WEBRTC_POSIX
 
+#if defined(WEBRTC_USE_EPOLL)
+
+void SocketDispatcher::SaveEnabledEvents() {
+  RTC_DCHECK_EQ(saved_enabled_events_, -1);
+  saved_enabled_events_ = enabled_events();
+}
+
+void SocketDispatcher::RestoreEnabledEvents() {
+  RTC_DCHECK_NE(saved_enabled_events_, -1);
+  uint8_t old_events = static_cast<uint8_t>(saved_enabled_events_);
+  saved_enabled_events_ = -1;
+  MaybeUpdateDispatcher(old_events);
+}
+
+void SocketDispatcher::MaybeUpdateDispatcher(uint8_t old_events) {
+  if (enabled_events() != old_events && saved_enabled_events_ == -1) {

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

Since DE_READ and DE_ACCEPT both map to EPOLLIN: suppose the enabled events go
from "DE_ACCEPT" to "DE_ACCEPT | DE_READ". This would result in epoll_ctl being
called a second time with the same arguments, right?

Should this function use a condition like this instead, to avoid redundant
system calls?

  if (EpollEvents(enabled_events()) != EpollEvents(old_events) ...)

[joachim, 2017/05/18 12:41:49]

Right, thanks for spotting.

+    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_USE_EPOLL
+
 int SocketDispatcher::Close() {
   if (s_ == INVALID_SOCKET)
     return 0;
 
 #if defined(WEBRTC_WIN)
   id_ = 0;
   signal_close_ = false;
 #endif
   ss_->Remove(this);
   return PhysicalSocket::Close();
@@ skipping 308 matching lines @@
     if (pf_)
       *pf_ = false;
   }
 
  private:
   bool *pf_;
 };
 
 PhysicalSocketServer::PhysicalSocketServer()
     : fWait_(false) {
+#if defined(WEBRTC_USE_EPOLL)
+  // 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) {
+    // Not an error, will fall back to "select" below.
+    LOG_E(LS_WARNING, EN, errno) << "epoll_create";
+    epoll_fd_ = INVALID_SOCKET;
+  }
+#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_USE_EPOLL)
+  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
+  dispatchers_.insert(pdispatcher);
+#endif  // WEBRTC_WIN
+#if defined(WEBRTC_USE_EPOLL)
+  if (epoll_fd_ != INVALID_SOCKET) {
+    AddEpoll(pdispatcher);
+  }
+#endif  // WEBRTC_USE_EPOLL
 }
 
 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;
+  }
+#endif  // WEBRTC_WIN
+#if defined(WEBRTC_USE_EPOLL)
+  if (epoll_fd_ != INVALID_SOCKET) {
+    RemoveEpoll(pdispatcher);
+  }
+#endif  // WEBRTC_USE_EPOLL
+}
+
+void PhysicalSocketServer::Update(Dispatcher *pdispatcher) {
+#if defined(WEBRTC_USE_EPOLL)
+  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_USE_EPOLL)
+  // 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);
+}

[zhaoyanfeng, 2017/05/18 07:46:17]

If process_io == false, still use select?
That seems crash when fd goes above 1024.

[joachim, 2017/05/18 12:41:49]

The wakeup dispatcher is created early during construction of the PSS, so its fd
should be below 1024. But you're right, to be on the safe side I should keep a
separate epoll descriptor for it.

[zhaoyanfeng, 2017/05/18 14:08:37]

I have tested it in our server but failed, our server contains many modules,
WebRTC is one of them, so the fd maybe above 1024 when the wakeup dispatcher is
created.
We'd better use epoll here completely.
Thanks for fixing.
 

[joachim, 2017/05/18 19:31:53]

I changed it now to use "poll" to wait for the signaling dispatcher instead of
maintaining a separate epoll dispatcher.

+
+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_USE_EPOLL)
+
+// 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_USE_EPOLL
+
 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