mollyim/webrtc
1
0
Fork 0
mirror of https://github.com/mollyim/webrtc.git synced 2025-05-12 21:30:45 +01:00
Code Issues Projects Releases Packages Wiki Activity Actions
webrtc/rtc_base/physical_socket_server.cc
mmorrison 25eeda1872 Fix socket not getting registered for epoll events 
When epoll is enabled in the PhysicalSocketServer, a socket may
not get registered for its epoll events. If an AsyncSocket is
closed and re-created during one of its signal callbacks, its
old epoll events and new epolls events bitmasks may be the same,
even though the fd has changed. This causes the epoll implementation
to not register the new fd for any events.

Fix this by resetting the saved events bitmask when the socket is
closed. This ensures the new fd, if any, is registered if needed.

Bug: webrtc:11497
Change-Id: Idea499e09aefdf292430d1a774a046f963603b95
Reviewed-on: https://webrtc-review.googlesource.com/c/src/+/173103
Reviewed-by: Taylor <deadbeef@webrtc.org>
Reviewed-by: Karl Wiberg <kwiberg@webrtc.org>
Commit-Queue: Karl Wiberg <kwiberg@webrtc.org>
Cr-Commit-Position: refs/heads/master@{#31039}
2020-04-09 10:17:47 +00:00

1977 lines
58 KiB
C++
Raw Blame History

/*
* 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 "rtc_base/physical_socket_server.h"
#if defined(_MSC_VER) && _MSC_VER < 1300
#pragma warning(disable : 4786)
#endif
#ifdef MEMORY_SANITIZER
#include <sanitizer/msan_interface.h>
#endif
#if defined(WEBRTC_POSIX)
#include <fcntl.h>
#include <string.h>
#if defined(WEBRTC_USE_EPOLL)
// "poll" will be used to wait for the signal dispatcher.
#include <poll.h>
#endif
#include <signal.h>
#include <sys/ioctl.h>
#include <sys/select.h>
#include <sys/time.h>
#include <unistd.h>
#endif
#if defined(WEBRTC_WIN)
#include <windows.h>
#include <winsock2.h>
#include <ws2tcpip.h>
#undef SetPort
#endif
#include <errno.h>
#include <algorithm>
#include <map>
#include "rtc_base/arraysize.h"
#include "rtc_base/byte_order.h"
#include "rtc_base/checks.h"
#include "rtc_base/logging.h"
#include "rtc_base/network_monitor.h"
#include "rtc_base/null_socket_server.h"
#include "rtc_base/time_utils.h"
#if defined(WEBRTC_LINUX)
#include <linux/sockios.h>
#endif
#if defined(WEBRTC_WIN)
#define LAST_SYSTEM_ERROR (::GetLastError())
#elif defined(__native_client__) && __native_client__
#define LAST_SYSTEM_ERROR (0)
#elif defined(WEBRTC_POSIX)
#define LAST_SYSTEM_ERROR (errno)
#endif // WEBRTC_WIN
#if defined(WEBRTC_POSIX)
#include <netinet/tcp.h> // for TCP_NODELAY
#define IP_MTU 14 // Until this is integrated from linux/in.h to netinet/in.h
typedef void* SockOptArg;
#endif // WEBRTC_POSIX
#if defined(WEBRTC_POSIX) && !defined(WEBRTC_MAC) && !defined(__native_client__)
int64_t GetSocketRecvTimestamp(int socket) {
struct timeval tv_ioctl;
int ret = ioctl(socket, SIOCGSTAMP, &tv_ioctl);
if (ret != 0)
return -1;
int64_t timestamp =
rtc::kNumMicrosecsPerSec * static_cast<int64_t>(tv_ioctl.tv_sec) +
static_cast<int64_t>(tv_ioctl.tv_usec);
return timestamp;
}
#else
int64_t GetSocketRecvTimestamp(int socket) {
return -1;
}
#endif
#if defined(WEBRTC_WIN)
typedef char* SockOptArg;
#endif
#if defined(WEBRTC_USE_EPOLL)
// POLLRDHUP / EPOLLRDHUP are only defined starting with Linux 2.6.17.
#if !defined(POLLRDHUP)
#define POLLRDHUP 0x2000
#endif
#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
}
PhysicalSocket::PhysicalSocket(PhysicalSocketServer* ss, SOCKET s)
: ss_(ss),
s_(s),
error_(0),
state_((s == INVALID_SOCKET) ? CS_CLOSED : CS_CONNECTED),
resolver_(nullptr) {
if (s_ != INVALID_SOCKET) {
SetEnabledEvents(DE_READ | DE_WRITE);
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);
family_ = family;
UpdateLastError();
if (udp_) {
SetEnabledEvents(DE_READ | DE_WRITE);
}
return s_ != INVALID_SOCKET;
}
SocketAddress PhysicalSocket::GetLocalAddress() const {
sockaddr_storage addr_storage = {};
socklen_t addrlen = sizeof(addr_storage);
sockaddr* addr = reinterpret_cast<sockaddr*>(&addr_storage);
int result = ::getsockname(s_, addr, &addrlen);
SocketAddress address;
if (result >= 0) {
SocketAddressFromSockAddrStorage(addr_storage, &address);
} else {
RTC_LOG(LS_WARNING) << "GetLocalAddress: unable to get local addr, socket="
<< s_;
}
return address;
}
SocketAddress PhysicalSocket::GetRemoteAddress() const {
sockaddr_storage addr_storage = {};
socklen_t addrlen = sizeof(addr_storage);
sockaddr* addr = reinterpret_cast<sockaddr*>(&addr_storage);
int result = ::getpeername(s_, addr, &addrlen);
SocketAddress address;
if (result >= 0) {
SocketAddressFromSockAddrStorage(addr_storage, &address);
} else {
RTC_LOG(LS_WARNING)
<< "GetRemoteAddress: unable to get remote addr, socket=" << s_;
}
return address;
}
int PhysicalSocket::Bind(const SocketAddress& bind_addr) {
SocketAddress copied_bind_addr = bind_addr;
// If a network binder is available, use it to bind a socket to an interface
// instead of bind(), since this is more reliable on an OS with a weak host
// model.
if (ss_->network_binder() && !bind_addr.IsAnyIP()) {
NetworkBindingResult result =
ss_->network_binder()->BindSocketToNetwork(s_, bind_addr.ipaddr());
if (result == NetworkBindingResult::SUCCESS) {
// Since the network binder handled binding the socket to the desired
// network interface, we don't need to (and shouldn't) include an IP in
// the bind() call; bind() just needs to assign a port.
copied_bind_addr.SetIP(GetAnyIP(copied_bind_addr.ipaddr().family()));
} else if (result == NetworkBindingResult::NOT_IMPLEMENTED) {
RTC_LOG(LS_INFO) << "Can't bind socket to network because "
"network binding is not implemented for this OS.";
} else {
if (bind_addr.IsLoopbackIP()) {
// If we couldn't bind to a loopback IP (which should only happen in
// test scenarios), continue on. This may be expected behavior.
RTC_LOG(LS_VERBOSE) << "Binding socket to loopback address"
<< " failed; result: " << static_cast<int>(result);
} else {
RTC_LOG(LS_WARNING) << "Binding socket to network address"
<< " failed; result: " << static_cast<int>(result);
// If a network binding was attempted and failed, we should stop here
// and not try to use the socket. Otherwise, we may end up sending
// packets with an invalid source address.
// See: https://bugs.chromium.org/p/webrtc/issues/detail?id=7026
return -1;
}
}
}
sockaddr_storage addr_storage;
size_t len = copied_bind_addr.ToSockAddrStorage(&addr_storage);
sockaddr* addr = reinterpret_cast<sockaddr*>(&addr_storage);
int err = ::bind(s_, addr, static_cast<int>(len));
UpdateLastError();
#if !defined(NDEBUG)
if (0 == err) {
dbg_addr_ = "Bound @ ";
dbg_addr_.append(GetLocalAddress().ToString());
}
#endif
return err;
}
int PhysicalSocket::Connect(const SocketAddress& addr) {
// TODO(pthatcher): Implicit creation is required to reconnect...
// ...but should we make it more explicit?
if (state_ != CS_CLOSED) {
SetError(EALREADY);
return SOCKET_ERROR;
}
if (addr.IsUnresolvedIP()) {
RTC_LOG(LS_VERBOSE) << "Resolving addr in PhysicalSocket::Connect";
resolver_ = new AsyncResolver();
resolver_->SignalDone.connect(this, &PhysicalSocket::OnResolveResult);
resolver_->Start(addr);
state_ = CS_CONNECTING;
return 0;
}
return DoConnect(addr);
}
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;
events |= DE_CONNECT;
} else {
return SOCKET_ERROR;
}
EnableEvents(events);
return 0;
}
int PhysicalSocket::GetError() const {
CritScope cs(&crit_);
return error_;
}
void PhysicalSocket::SetError(int error) {
CritScope cs(&crit_);
error_ = error;
}
AsyncSocket::ConnState PhysicalSocket::GetState() const {
return state_;
}
int PhysicalSocket::GetOption(Option opt, int* value) {
int slevel;
int sopt;
if (TranslateOption(opt, &slevel, &sopt) == -1)
return -1;
socklen_t optlen = sizeof(*value);
int ret = ::getsockopt(s_, slevel, sopt, (SockOptArg)value, &optlen);
if (ret == -1) {
return -1;
}
if (opt == OPT_DONTFRAGMENT) {
#if defined(WEBRTC_LINUX) && !defined(WEBRTC_ANDROID)
*value = (*value != IP_PMTUDISC_DONT) ? 1 : 0;
#endif
} else if (opt == OPT_DSCP) {
#if defined(WEBRTC_POSIX)
// unshift DSCP value to get six most significant bits of IP DiffServ field
*value >>= 2;
#endif
}
return ret;
}
int PhysicalSocket::SetOption(Option opt, int value) {
int slevel;
int sopt;
if (TranslateOption(opt, &slevel, &sopt) == -1)
return -1;
if (opt == OPT_DONTFRAGMENT) {
#if defined(WEBRTC_LINUX) && !defined(WEBRTC_ANDROID)
value = (value) ? IP_PMTUDISC_DO : IP_PMTUDISC_DONT;
#endif
} else if (opt == OPT_DSCP) {
#if defined(WEBRTC_POSIX)
// shift DSCP value to fit six most significant bits of IP DiffServ field
value <<= 2;
#endif
}
#if defined(WEBRTC_POSIX)
if (sopt == IPV6_TCLASS) {
// Set the IPv4 option in all cases to support dual-stack sockets.
::setsockopt(s_, IPPROTO_IP, IP_TOS, (SockOptArg)&value, sizeof(value));
}
#endif
return ::setsockopt(s_, slevel, sopt, (SockOptArg)&value, sizeof(value));
}
int PhysicalSocket::Send(const void* pv, size_t cb) {
int sent = DoSend(
s_, reinterpret_cast<const char*>(pv), static_cast<int>(cb),
#if defined(WEBRTC_LINUX) && !defined(WEBRTC_ANDROID)
// Suppress SIGPIPE. Without this, attempting to send on a socket whose
// other end is closed will result in a SIGPIPE signal being raised to
// our process, which by default will terminate the process, which we
// don't want. By specifying this flag, we'll just get the error EPIPE
// instead and can handle the error gracefully.
MSG_NOSIGNAL
#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()))) {
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()))) {
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.
RTC_LOG(LS_WARNING) << "EOF from socket; deferring close event";
// Must turn this back on so that the select() loop will notice the close
// event.
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) {
EnableEvents(DE_READ);
}
if (!success) {
RTC_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) {
EnableEvents(DE_READ);
}
if (!success) {
RTC_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;
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.
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;
SetEnabledEvents(0);
if (resolver_) {
resolver_->Destroy(false);
resolver_ = nullptr;
}
return err;
}
SOCKET PhysicalSocket::DoAccept(SOCKET socket,
sockaddr* addr,
socklen_t* addrlen) {
return ::accept(socket, addr, addrlen);
}
int PhysicalSocket::DoSend(SOCKET socket, const char* buf, int len, int flags) {
return ::send(socket, buf, len, flags);
}
int PhysicalSocket::DoSendTo(SOCKET socket,
const char* buf,
int len,
int flags,
const struct sockaddr* dest_addr,
socklen_t addrlen) {
return ::sendto(socket, buf, len, flags, dest_addr, addrlen);
}
void PhysicalSocket::OnResolveResult(AsyncResolverInterface* resolver) {
if (resolver != resolver_) {
return;
}
int error = resolver_->GetError();
if (error == 0) {
error = DoConnect(resolver_->address());
} else {
Close();
}
if (error) {
SetError(error);
SignalCloseEvent(this, error);
}
}
void PhysicalSocket::UpdateLastError() {
SetError(LAST_SYSTEM_ERROR);
}
void PhysicalSocket::MaybeRemapSendError() {
#if defined(WEBRTC_MAC)
// https://developer.apple.com/library/mac/documentation/Darwin/
// 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__)
RTC_LOG(LS_WARNING) << "Socket::OPT_DONTFRAGMENT not supported.";
return -1;
#elif defined(WEBRTC_POSIX)
*slevel = IPPROTO_IP;
*sopt = IP_MTU_DISCOVER;
break;
#endif
case OPT_RCVBUF:
*slevel = SOL_SOCKET;
*sopt = SO_RCVBUF;
break;
case OPT_SNDBUF:
*slevel = SOL_SOCKET;
*sopt = SO_SNDBUF;
break;
case OPT_NODELAY:
*slevel = IPPROTO_TCP;
*sopt = TCP_NODELAY;
break;
case OPT_DSCP:
#if defined(WEBRTC_POSIX)
if (family_ == AF_INET6) {
*slevel = IPPROTO_IPV6;
*sopt = IPV6_TCLASS;
} else {
*slevel = IPPROTO_IP;
*sopt = IP_TOS;
}
break;
#else
RTC_LOG(LS_WARNING) << "Socket::OPT_DSCP not supported.";
return -1;
#endif
case OPT_RTP_SENDTIME_EXTN_ID:
return -1; // No logging is necessary as this not a OS socket option.
default:
RTC_NOTREACHED();
return -1;
}
return 0;
}
SocketDispatcher::SocketDispatcher(PhysicalSocketServer* ss)
#if defined(WEBRTC_WIN)
: PhysicalSocket(ss),
id_(0),
signal_close_(false)
#else
: PhysicalSocket(ss)
#endif
{
}
SocketDispatcher::SocketDispatcher(SOCKET s, PhysicalSocketServer* ss)
#if defined(WEBRTC_WIN)
: PhysicalSocket(ss, s),
id_(0),
signal_close_(false)
#else
: PhysicalSocket(ss, s)
#endif
{
}
SocketDispatcher::~SocketDispatcher() {
Close();
}
bool SocketDispatcher::Initialize() {
RTC_DCHECK(s_ != INVALID_SOCKET);
// Must be a non-blocking
#if defined(WEBRTC_WIN)
u_long argp = 1;
ioctlsocket(s_, FIONBIO, &argp);
#elif defined(WEBRTC_POSIX)
fcntl(s_, F_SETFL, fcntl(s_, F_GETFL, 0) | O_NONBLOCK);
#endif
#if defined(WEBRTC_IOS)
// iOS may kill sockets when the app is moved to the background
// (specifically, if the app doesn't use the "voip" UIBackgroundMode). When
// we attempt to write to such a socket, SIGPIPE will be raised, which by
// default will terminate the process, which we don't want. By specifying
// this socket option, SIGPIPE will be disabled for the socket.
int value = 1;
::setsockopt(s_, SOL_SOCKET, SO_NOSIGPIPE, &value, sizeof(value));
#endif
ss_->Add(this);
return true;
}
bool SocketDispatcher::Create(int type) {
return Create(AF_INET, type);
}
bool SocketDispatcher::Create(int family, int type) {
// Change the socket to be non-blocking.
if (!PhysicalSocket::Create(family, type))
return false;
if (!Initialize())
return false;
#if defined(WEBRTC_WIN)
do {
id_ = ++next_id_;
} while (id_ == 0);
#endif
return true;
}
#if defined(WEBRTC_WIN)
WSAEVENT SocketDispatcher::GetWSAEvent() {
return WSA_INVALID_EVENT;
}
SOCKET SocketDispatcher::GetSocket() {
return s_;
}
bool SocketDispatcher::CheckSignalClose() {
if (!signal_close_)
return false;
char ch;
if (recv(s_, &ch, 1, MSG_PEEK) > 0)
return false;
state_ = CS_CLOSED;
signal_close_ = false;
SignalCloseEvent(this, signal_err_);
return true;
}
int SocketDispatcher::next_id_ = 0;
#elif defined(WEBRTC_POSIX)
int SocketDispatcher::GetDescriptor() {
return s_;
}
bool SocketDispatcher::IsDescriptorClosed() {
if (udp_) {
// The MSG_PEEK trick doesn't work for UDP, since (at least in some
// circumstances) it requires reading an entire UDP packet, which would be
// bad for performance here. So, just check whether |s_| has been closed,
// which should be sufficient.
return s_ == INVALID_SOCKET;
}
// We don't have a reliable way of distinguishing end-of-stream
// from readability. So test on each readable call. Is this
// inefficient? Probably.
char ch;
ssize_t res = ::recv(s_, &ch, 1, MSG_PEEK);
if (res > 0) {
// Data available, so not closed.
return false;
} else if (res == 0) {
// EOF, so closed.
return true;
} else { // error
switch (errno) {
// Returned if we've already closed s_.
case EBADF:
// Returned during ungraceful peer shutdown.
case ECONNRESET:
return true;
// The normal blocking error; don't log anything.
case EWOULDBLOCK:
// Interrupted system call.
case EINTR:
return false;
default:
// Assume that all other errors are just blocking errors, meaning the
// connection is still good but we just can't read from it right now.
// This should only happen when connecting (and at most once), because
// in all other cases this function is only called if the file
// descriptor is already known to be in the readable state. However,
// it's not necessary a problem if we spuriously interpret a
// "connection lost"-type error as a blocking error, because typically
// the next recv() will get EOF, so we'll still eventually notice that
// the socket is closed.
RTC_LOG_ERR(LS_WARNING) << "Assuming benign blocking error";
return false;
}
}
}
#endif // WEBRTC_POSIX
uint32_t SocketDispatcher::GetRequestedEvents() {
return enabled_events();
}
void SocketDispatcher::OnPreEvent(uint32_t ff) {
if ((ff & DE_CONNECT) != 0)
state_ = CS_CONNECTED;
#if defined(WEBRTC_WIN)
// We set CS_CLOSED from CheckSignalClose.
#elif defined(WEBRTC_POSIX)
if ((ff & DE_CLOSE) != 0)
state_ = CS_CLOSED;
#endif
}
#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)
RTC_LOG(LS_VERBOSE) << "Signalled with DE_CONNECT: " << ff;
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)) {
DisableEvents(DE_ACCEPT);
SignalReadEvent(this);
}
if ((ff & DE_READ) != 0) {
DisableEvents(DE_READ);
SignalReadEvent(this);
}
if (((ff & DE_WRITE) != 0) && (id_ == cache_id)) {
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_USE_EPOLL)
// Remember currently enabled events so we can combine multiple changes
// into one update call later.
// The signal handlers might re-enable events disabled here, so we can't
// keep a list of events to disable at the end of the method. This list
// would not be updated with the events enabled by the signal handlers.
StartBatchedEventUpdates();
#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)
FinishBatchedEventUpdates();
#endif
}
#endif // WEBRTC_POSIX
#if defined(WEBRTC_USE_EPOLL)
static int GetEpollEvents(uint32_t ff) {
int events = 0;
if (ff & (DE_READ | DE_ACCEPT)) {
events |= EPOLLIN;
}
if (ff & (DE_WRITE | DE_CONNECT)) {
events |= EPOLLOUT;
}
return events;
}
void SocketDispatcher::StartBatchedEventUpdates() {
RTC_DCHECK_EQ(saved_enabled_events_, -1);
saved_enabled_events_ = enabled_events();
}
void SocketDispatcher::FinishBatchedEventUpdates() {
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 (GetEpollEvents(enabled_events()) != GetEpollEvents(old_events) &&
saved_enabled_events_ == -1) {
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
#if defined(WEBRTC_USE_EPOLL)
// If we're batching events, the socket can be closed and reopened
// during the batch. Set saved_enabled_events_ to 0 here so the new
// socket, if any, has the correct old events bitfield
if (saved_enabled_events_ != -1) {
saved_enabled_events_ = 0;
}
#endif
ss_->Remove(this);
return PhysicalSocket::Close();
}
#if defined(WEBRTC_POSIX)
class EventDispatcher : public Dispatcher {
public:
EventDispatcher(PhysicalSocketServer* ss) : ss_(ss), fSignaled_(false) {
if (pipe(afd_) < 0)
RTC_LOG(LERROR) << "pipe failed";
ss_->Add(this);
}
~EventDispatcher() override {
ss_->Remove(this);
close(afd_[0]);
close(afd_[1]);
}
virtual void Signal() {
CritScope cs(&crit_);
if (!fSignaled_) {
const uint8_t b[1] = {0};
const ssize_t res = write(afd_[1], b, sizeof(b));
RTC_DCHECK_EQ(1, res);
fSignaled_ = true;
}
}
uint32_t GetRequestedEvents() override { return DE_READ; }
void OnPreEvent(uint32_t ff) override {
// It is not possible to perfectly emulate an auto-resetting event with
// pipes. This simulates it by resetting before the event is handled.
CritScope cs(&crit_);
if (fSignaled_) {
uint8_t b[4]; // Allow for reading more than 1 byte, but expect 1.
const ssize_t res = read(afd_[0], b, sizeof(b));
RTC_DCHECK_EQ(1, res);
fSignaled_ = false;
}
}
void OnEvent(uint32_t ff, int err) override { RTC_NOTREACHED(); }
int GetDescriptor() override { return afd_[0]; }
bool IsDescriptorClosed() override { return false; }
private:
PhysicalSocketServer* ss_;
int afd_[2];
bool fSignaled_;
CriticalSection crit_;
};
// These two classes use the self-pipe trick to deliver POSIX signals to our
// select loop. This is the only safe, reliable, cross-platform way to do
// non-trivial things with a POSIX signal in an event-driven program (until
// proper pselect() implementations become ubiquitous).
class PosixSignalHandler {
public:
// POSIX only specifies 32 signals, but in principle the system might have
// more and the programmer might choose to use them, so we size our array
// for 128.
static const int kNumPosixSignals = 128;
// There is just a single global instance. (Signal handlers do not get any
// sort of user-defined void * parameter, so they can't access anything that
// isn't global.)
static PosixSignalHandler* Instance() {
static PosixSignalHandler* const instance = new PosixSignalHandler();
return instance;
}
// Returns true if the given signal number is set.
bool IsSignalSet(int signum) const {
RTC_DCHECK(signum < static_cast<int>(arraysize(received_signal_)));
if (signum < static_cast<int>(arraysize(received_signal_))) {
return received_signal_[signum];
} else {
return false;
}
}
// Clears the given signal number.
void ClearSignal(int signum) {
RTC_DCHECK(signum < static_cast<int>(arraysize(received_signal_)));
if (signum < static_cast<int>(arraysize(received_signal_))) {
received_signal_[signum] = false;
}
}
// Returns the file descriptor to monitor for signal events.
int GetDescriptor() const { return afd_[0]; }
// This is called directly from our real signal handler, so it must be
// signal-handler-safe. That means it cannot assume anything about the
// user-level state of the process, since the handler could be executed at any
// time on any thread.
void OnPosixSignalReceived(int signum) {
if (signum >= static_cast<int>(arraysize(received_signal_))) {
// We don't have space in our array for this.
return;
}
// Set a flag saying we've seen this signal.
received_signal_[signum] = true;
// Notify application code that we got a signal.
const uint8_t b[1] = {0};
if (-1 == write(afd_[1], b, sizeof(b))) {
// Nothing we can do here. If there's an error somehow then there's
// nothing we can safely do from a signal handler.
// No, we can't even safely log it.
// But, we still have to check the return value here. Otherwise,
// GCC 4.4.1 complains ignoring return value. Even (void) doesn't help.
return;
}
}
private:
PosixSignalHandler() {
if (pipe(afd_) < 0) {
RTC_LOG_ERR(LS_ERROR) << "pipe failed";
return;
}
if (fcntl(afd_[0], F_SETFL, O_NONBLOCK) < 0) {
RTC_LOG_ERR(LS_WARNING) << "fcntl #1 failed";
}
if (fcntl(afd_[1], F_SETFL, O_NONBLOCK) < 0) {
RTC_LOG_ERR(LS_WARNING) << "fcntl #2 failed";
}
memset(const_cast<void*>(static_cast<volatile void*>(received_signal_)), 0,
sizeof(received_signal_));
}
~PosixSignalHandler() {
int fd1 = afd_[0];
int fd2 = afd_[1];
// We clobber the stored file descriptor numbers here or else in principle
// a signal that happens to be delivered during application termination
// could erroneously write a zero byte to an unrelated file handle in
// OnPosixSignalReceived() if some other file happens to be opened later
// during shutdown and happens to be given the same file descriptor number
// as our pipe had. Unfortunately even with this precaution there is still a
// race where that could occur if said signal happens to be handled
// concurrently with this code and happens to have already read the value of
// afd_[1] from memory before we clobber it, but that's unlikely.
afd_[0] = -1;
afd_[1] = -1;
close(fd1);
close(fd2);
}
int afd_[2];
// These are boolean flags that will be set in our signal handler and read
// and cleared from Wait(). There is a race involved in this, but it is
// benign. The signal handler sets the flag before signaling the pipe, so
// we'll never end up blocking in select() while a flag is still true.
// However, if two of the same signal arrive close to each other then it's
// possible that the second time the handler may set the flag while it's still
// true, meaning that signal will be missed. But the first occurrence of it
// will still be handled, so this isn't a problem.
// Volatile is not necessary here for correctness, but this data _is_ volatile
// so I've marked it as such.
volatile uint8_t received_signal_[kNumPosixSignals];
};
class PosixSignalDispatcher : public Dispatcher {
public:
PosixSignalDispatcher(PhysicalSocketServer* owner) : owner_(owner) {
owner_->Add(this);
}
~PosixSignalDispatcher() override { owner_->Remove(this); }
uint32_t GetRequestedEvents() override { return DE_READ; }
void OnPreEvent(uint32_t ff) override {
// Events might get grouped if signals come very fast, so we read out up to
// 16 bytes to make sure we keep the pipe empty.
uint8_t b[16];
ssize_t ret = read(GetDescriptor(), b, sizeof(b));
if (ret < 0) {
RTC_LOG_ERR(LS_WARNING) << "Error in read()";
} else if (ret == 0) {
RTC_LOG(LS_WARNING) << "Should have read at least one byte";
}
}
void OnEvent(uint32_t ff, int err) override {
for (int signum = 0; signum < PosixSignalHandler::kNumPosixSignals;
++signum) {
if (PosixSignalHandler::Instance()->IsSignalSet(signum)) {
PosixSignalHandler::Instance()->ClearSignal(signum);
HandlerMap::iterator i = handlers_.find(signum);
if (i == handlers_.end()) {
// This can happen if a signal is delivered to our process at around
// the same time as we unset our handler for it. It is not an error
// condition, but it's unusual enough to be worth logging.
RTC_LOG(LS_INFO) << "Received signal with no handler: " << signum;
} else {
// Otherwise, execute our handler.
(*i->second)(signum);
}
}
}
}
int GetDescriptor() override {
return PosixSignalHandler::Instance()->GetDescriptor();
}
bool IsDescriptorClosed() override { return false; }
void SetHandler(int signum, void (*handler)(int)) {
handlers_[signum] = handler;
}
void ClearHandler(int signum) { handlers_.erase(signum); }
bool HasHandlers() { return !handlers_.empty(); }
private:
typedef std::map<int, void (*)(int)> HandlerMap;
HandlerMap handlers_;
// Our owner.
PhysicalSocketServer* owner_;
};
#endif // WEBRTC_POSIX
#if defined(WEBRTC_WIN)
static uint32_t FlagsToEvents(uint32_t events) {
uint32_t ffFD = FD_CLOSE;
if (events & DE_READ)
ffFD |= FD_READ;
if (events & DE_WRITE)
ffFD |= FD_WRITE;
if (events & DE_CONNECT)
ffFD |= FD_CONNECT;
if (events & DE_ACCEPT)
ffFD |= FD_ACCEPT;
return ffFD;
}
class EventDispatcher : public Dispatcher {
public:
EventDispatcher(PhysicalSocketServer* ss) : ss_(ss) {
hev_ = WSACreateEvent();
if (hev_) {
ss_->Add(this);
}
}
~EventDispatcher() override {
if (hev_ != nullptr) {
ss_->Remove(this);
WSACloseEvent(hev_);
hev_ = nullptr;
}
}
virtual void Signal() {
if (hev_ != nullptr)
WSASetEvent(hev_);
}
uint32_t GetRequestedEvents() override { return 0; }
void OnPreEvent(uint32_t ff) override { WSAResetEvent(hev_); }
void OnEvent(uint32_t ff, int err) override {}
WSAEVENT GetWSAEvent() override { return hev_; }
SOCKET GetSocket() override { return INVALID_SOCKET; }
bool CheckSignalClose() override { return false; }
private:
PhysicalSocketServer* ss_;
WSAEVENT hev_;
};
#endif // WEBRTC_WIN
// Sets the value of a boolean value to false when signaled.
class Signaler : public EventDispatcher {
public:
Signaler(PhysicalSocketServer* ss, bool* pf) : EventDispatcher(ss), pf_(pf) {}
~Signaler() override {}
void OnEvent(uint32_t ff, int err) override {
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.
RTC_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 family, int type) {
PhysicalSocket* socket = new PhysicalSocket(this);
if (socket->Create(family, type)) {
return socket;
} else {
delete socket;
return nullptr;
}
}
AsyncSocket* PhysicalSocketServer::CreateAsyncSocket(int family, int type) {
SocketDispatcher* dispatcher = new SocketDispatcher(this);
if (dispatcher->Create(family, type)) {
return dispatcher;
} else {
delete dispatcher;
return nullptr;
}
}
AsyncSocket* PhysicalSocketServer::WrapSocket(SOCKET s) {
SocketDispatcher* dispatcher = new SocketDispatcher(s, this);
if (dispatcher->Initialize()) {
return dispatcher;
} else {
delete dispatcher;
return nullptr;
}
}
void PhysicalSocketServer::Add(Dispatcher* pdispatcher) {
CritScope cs(&crit_);
if (processing_dispatchers_) {
// A dispatcher is being added while a "Wait" call is processing the
// list of socket events.
// Defer adding to "dispatchers_" set until processing is done to avoid
// invalidating the iterator in "Wait".
pending_remove_dispatchers_.erase(pdispatcher);
pending_add_dispatchers_.insert(pdispatcher);
} else {
dispatchers_.insert(pdispatcher);
}
#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 (processing_dispatchers_) {
// A dispatcher is being removed while a "Wait" call is processing the
// list of socket events.
// Defer removal from "dispatchers_" set until processing is done to avoid
// invalidating the iterator in "Wait".
if (!pending_add_dispatchers_.erase(pdispatcher) &&
dispatchers_.find(pdispatcher) == dispatchers_.end()) {
RTC_LOG(LS_WARNING) << "PhysicalSocketServer asked to remove a unknown "
"dispatcher, potentially from a duplicate call to "
"Add.";
return;
}
pending_remove_dispatchers_.insert(pdispatcher);
} else if (!dispatchers_.erase(pdispatcher)) {
RTC_LOG(LS_WARNING)
<< "PhysicalSocketServer asked to remove a unknown "
"dispatcher, potentially from a duplicate call to Add.";
return;
}
#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
}
void PhysicalSocketServer::AddRemovePendingDispatchers() {
if (!pending_add_dispatchers_.empty()) {
for (Dispatcher* pdispatcher : pending_add_dispatchers_) {
dispatchers_.insert(pdispatcher);
}
pending_add_dispatchers_.clear();
}
if (!pending_remove_dispatchers_.empty()) {
for (Dispatcher* pdispatcher : pending_remove_dispatchers_) {
dispatchers_.erase(pdispatcher);
}
pending_remove_dispatchers_.clear();
}
}
#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 the non-IO
// (i.e. signaling) dispatcher, so "poll" will be used instead of the default
// "select" to support sockets larger than FD_SETSIZE.
if (!process_io) {
return WaitPoll(cmsWait, signal_wakeup_);
} else if (epoll_fd_ != INVALID_SOCKET) {
return WaitEpoll(cmsWait);
}
#endif
return WaitSelect(cmsWait, process_io);
}
static void ProcessEvents(Dispatcher* dispatcher,
bool readable,
bool writable,
bool check_error) {
int errcode = 0;
// TODO(pthatcher): Should we set errcode if getsockopt fails?
if (check_error) {
socklen_t len = sizeof(errcode);
::getsockopt(dispatcher->GetDescriptor(), SOL_SOCKET, SO_ERROR, &errcode,
&len);
}
uint32_t ff = 0;
// 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 (readable) {
if (dispatcher->GetRequestedEvents() & DE_ACCEPT) {
ff |= DE_ACCEPT;
} else if (errcode || dispatcher->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 (writable) {
if (dispatcher->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) {
dispatcher->OnPreEvent(ff);
dispatcher->OnEvent(ff, errcode);
}
}
bool PhysicalSocketServer::WaitSelect(int cmsWait, bool process_io) {
// Calculate timing information
struct timeval* ptvWait = nullptr;
struct timeval tvWait;
int64_t stop_us;
if (cmsWait != kForever) {
// Calculate wait timeval
tvWait.tv_sec = cmsWait / 1000;
tvWait.tv_usec = (cmsWait % 1000) * 1000;
ptvWait = &tvWait;
// Calculate when to return
stop_us = rtc::TimeMicros() + cmsWait * 1000;
}
// Zero all fd_sets. Don't need to do this inside the loop since
// select() zeros the descriptors not signaled
fd_set fdsRead;
FD_ZERO(&fdsRead);
fd_set fdsWrite;
FD_ZERO(&fdsWrite);
// Explicitly unpoison these FDs on MemorySanitizer which doesn't handle the
// inline assembly in FD_ZERO.
// http://crbug.com/344505
#ifdef MEMORY_SANITIZER
__msan_unpoison(&fdsRead, sizeof(fdsRead));
__msan_unpoison(&fdsWrite, sizeof(fdsWrite));
#endif
fWait_ = true;
while (fWait_) {
int fdmax = -1;
{
CritScope cr(&crit_);
// TODO(jbauch): Support re-entrant waiting.
RTC_DCHECK(!processing_dispatchers_);
for (Dispatcher* pdispatcher : dispatchers_) {
// Query dispatchers for read and write wait state
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);
}
}
// Wait then call handlers as appropriate
// < 0 means error
// 0 means timeout
// > 0 means count of descriptors ready
int n = select(fdmax + 1, &fdsRead, &fdsWrite, nullptr, ptvWait);
// If error, return error.
if (n < 0) {
if (errno != EINTR) {
RTC_LOG_E(LS_ERROR, EN, errno) << "select";
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_);
processing_dispatchers_ = true;
for (Dispatcher* pdispatcher : dispatchers_) {
int fd = pdispatcher->GetDescriptor();
bool readable = FD_ISSET(fd, &fdsRead);
if (readable) {
FD_CLR(fd, &fdsRead);
}
bool writable = FD_ISSET(fd, &fdsWrite);
if (writable) {
FD_CLR(fd, &fdsWrite);
}
// The error code can be signaled through reads or writes.
ProcessEvents(pdispatcher, readable, writable, readable || writable);
}
processing_dispatchers_ = false;
// Process deferred dispatchers that have been added/removed while the
// events were handled above.
AddRemovePendingDispatchers();
}
// Recalc the time remaining to wait. Doing it here means it doesn't get
// calced twice the first time through the loop
if (ptvWait) {
ptvWait->tv_sec = 0;
ptvWait->tv_usec = 0;
int64_t time_left_us = stop_us - rtc::TimeMicros();
if (time_left_us > 0) {
ptvWait->tv_sec = time_left_us / rtc::kNumMicrosecsPerSec;
ptvWait->tv_usec = time_left_us % rtc::kNumMicrosecsPerSec;
}
}
}
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;
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->GetRequestedEvents());
event.data.ptr = pdispatcher;
int err = epoll_ctl(epoll_fd_, EPOLL_CTL_ADD, fd, &event);
RTC_DCHECK_EQ(err, 0);
if (err == -1) {
RTC_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.
RTC_LOG_E(LS_VERBOSE, EN, errno) << "epoll_ctl EPOLL_CTL_DEL";
} else {
RTC_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->GetRequestedEvents());
event.data.ptr = pdispatcher;
int err = epoll_ctl(epoll_fd_, EPOLL_CTL_MOD, fd, &event);
RTC_DCHECK_EQ(err, 0);
if (err == -1) {
RTC_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) {
RTC_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;
}
bool readable = (event.events & (EPOLLIN | EPOLLPRI));
bool writable = (event.events & EPOLLOUT);
bool check_error = (event.events & (EPOLLRDHUP | EPOLLERR | EPOLLHUP));
ProcessEvents(pdispatcher, readable, writable, check_error);
}
}
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;
}
bool PhysicalSocketServer::WaitPoll(int cmsWait, Dispatcher* dispatcher) {
RTC_DCHECK(dispatcher);
int64_t tvWait = -1;
int64_t tvStop = -1;
if (cmsWait != kForever) {
tvWait = cmsWait;
tvStop = TimeAfter(cmsWait);
}
fWait_ = true;
struct pollfd fds = {0};
int fd = dispatcher->GetDescriptor();
fds.fd = fd;
while (fWait_) {
uint32_t ff = dispatcher->GetRequestedEvents();
fds.events = 0;
if (ff & (DE_READ | DE_ACCEPT)) {
fds.events |= POLLIN;
}
if (ff & (DE_WRITE | DE_CONNECT)) {
fds.events |= POLLOUT;
}
fds.revents = 0;
// Wait then call handlers as appropriate
// < 0 means error
// 0 means timeout
// > 0 means count of descriptors ready
int n = poll(&fds, 1, static_cast<int>(tvWait));
if (n < 0) {
if (errno != EINTR) {
RTC_LOG_E(LS_ERROR, EN, errno) << "poll";
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 (should only be the passed dispatcher).
RTC_DCHECK_EQ(n, 1);
RTC_DCHECK_EQ(fds.fd, fd);
bool readable = (fds.revents & (POLLIN | POLLPRI));
bool writable = (fds.revents & POLLOUT);
bool check_error = (fds.revents & (POLLRDHUP | POLLERR | POLLHUP));
ProcessEvents(dispatcher, readable, writable, check_error);
}
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)) {
return false;
}
if (signal_dispatcher_) {
signal_dispatcher_->ClearHandler(signum);
if (!signal_dispatcher_->HasHandlers()) {
signal_dispatcher_.reset();
}
}
} else {
if (!signal_dispatcher_) {
signal_dispatcher_.reset(new PosixSignalDispatcher(this));
}
signal_dispatcher_->SetHandler(signum, handler);
if (!InstallSignal(signum, &GlobalSignalHandler)) {
return false;
}
}
return true;
}
Dispatcher* PhysicalSocketServer::signal_dispatcher() {
return signal_dispatcher_.get();
}
bool PhysicalSocketServer::InstallSignal(int signum, void (*handler)(int)) {
struct sigaction act;
// It doesn't really matter what we set this mask to.
if (sigemptyset(&act.sa_mask) != 0) {
RTC_LOG_ERR(LS_ERROR) << "Couldn't set mask";
return false;
}
act.sa_handler = handler;
#if !defined(__native_client__)
// Use SA_RESTART so that our syscalls don't get EINTR, since we don't need it
// and it's a nuisance. Though some syscalls still return EINTR and there's no
// real standard for which ones. :(
act.sa_flags = SA_RESTART;
#else
act.sa_flags = 0;
#endif
if (sigaction(signum, &act, nullptr) != 0) {
RTC_LOG_ERR(LS_ERROR) << "Couldn't set sigaction";
return false;
}
return true;
}
#endif // WEBRTC_POSIX
#if defined(WEBRTC_WIN)
bool PhysicalSocketServer::Wait(int cmsWait, bool process_io) {
int64_t cmsTotal = cmsWait;
int64_t cmsElapsed = 0;
int64_t msStart = Time();
fWait_ = true;
while (fWait_) {
std::vector<WSAEVENT> events;
std::vector<Dispatcher*> event_owners;
events.push_back(socket_ev_);
{
CritScope cr(&crit_);
// TODO(jbauch): Support re-entrant waiting.
RTC_DCHECK(!processing_dispatchers_);
// Calling "CheckSignalClose" might remove a closed dispatcher from the
// set. This must be deferred to prevent invalidating the iterator.
processing_dispatchers_ = true;
for (Dispatcher* disp : dispatchers_) {
if (!process_io && (disp != signal_wakeup_))
continue;
SOCKET s = disp->GetSocket();
if (disp->CheckSignalClose()) {
// We just signalled close, don't poll this socket
} else if (s != INVALID_SOCKET) {
WSAEventSelect(s, events[0],
FlagsToEvents(disp->GetRequestedEvents()));
} else {
events.push_back(disp->GetWSAEvent());
event_owners.push_back(disp);
}
}
processing_dispatchers_ = false;
// Process deferred dispatchers that have been added/removed while the
// events were handled above.
AddRemovePendingDispatchers();
}
// Which is shorter, the delay wait or the asked wait?
int64_t cmsNext;
if (cmsWait == kForever) {
cmsNext = cmsWait;
} else {
cmsNext = std::max<int64_t>(0, cmsTotal - cmsElapsed);
}
// Wait for one of the events to signal
DWORD dw =
WSAWaitForMultipleEvents(static_cast<DWORD>(events.size()), &events[0],
false, static_cast<DWORD>(cmsNext), false);
if (dw == WSA_WAIT_FAILED) {
// Failed?
// TODO(pthatcher): need a better strategy than this!
WSAGetLastError();
RTC_NOTREACHED();
return false;
} else if (dw == WSA_WAIT_TIMEOUT) {
// Timeout?
return true;
} else {
// Figure out which one it is and call it
CritScope cr(&crit_);
int index = dw - WSA_WAIT_EVENT_0;
if (index > 0) {
--index; // The first event is the socket event
Dispatcher* disp = event_owners[index];
// The dispatcher could have been removed while waiting for events.
if (dispatchers_.find(disp) != dispatchers_.end()) {
disp->OnPreEvent(0);
disp->OnEvent(0, 0);
}
} else if (process_io) {
processing_dispatchers_ = true;
for (Dispatcher* disp : dispatchers_) {
SOCKET s = disp->GetSocket();
if (s == INVALID_SOCKET)
continue;
WSANETWORKEVENTS wsaEvents;
int err = WSAEnumNetworkEvents(s, events[0], &wsaEvents);
if (err == 0) {
{
if ((wsaEvents.lNetworkEvents & FD_READ) &&
wsaEvents.iErrorCode[FD_READ_BIT] != 0) {
RTC_LOG(WARNING)
<< "PhysicalSocketServer got FD_READ_BIT error "
<< wsaEvents.iErrorCode[FD_READ_BIT];
}
if ((wsaEvents.lNetworkEvents & FD_WRITE) &&
wsaEvents.iErrorCode[FD_WRITE_BIT] != 0) {
RTC_LOG(WARNING)
<< "PhysicalSocketServer got FD_WRITE_BIT error "
<< wsaEvents.iErrorCode[FD_WRITE_BIT];
}
if ((wsaEvents.lNetworkEvents & FD_CONNECT) &&
wsaEvents.iErrorCode[FD_CONNECT_BIT] != 0) {
RTC_LOG(WARNING)
<< "PhysicalSocketServer got FD_CONNECT_BIT error "
<< wsaEvents.iErrorCode[FD_CONNECT_BIT];
}
if ((wsaEvents.lNetworkEvents & FD_ACCEPT) &&
wsaEvents.iErrorCode[FD_ACCEPT_BIT] != 0) {
RTC_LOG(WARNING)
<< "PhysicalSocketServer got FD_ACCEPT_BIT error "
<< wsaEvents.iErrorCode[FD_ACCEPT_BIT];
}
if ((wsaEvents.lNetworkEvents & FD_CLOSE) &&
wsaEvents.iErrorCode[FD_CLOSE_BIT] != 0) {
RTC_LOG(WARNING)
<< "PhysicalSocketServer got FD_CLOSE_BIT error "
<< wsaEvents.iErrorCode[FD_CLOSE_BIT];
}
}
uint32_t ff = 0;
int errcode = 0;
if (wsaEvents.lNetworkEvents & FD_READ)
ff |= DE_READ;
if (wsaEvents.lNetworkEvents & FD_WRITE)
ff |= DE_WRITE;
if (wsaEvents.lNetworkEvents & FD_CONNECT) {
if (wsaEvents.iErrorCode[FD_CONNECT_BIT] == 0) {
ff |= DE_CONNECT;
} else {
ff |= DE_CLOSE;
errcode = wsaEvents.iErrorCode[FD_CONNECT_BIT];
}
}
if (wsaEvents.lNetworkEvents & FD_ACCEPT)
ff |= DE_ACCEPT;
if (wsaEvents.lNetworkEvents & FD_CLOSE) {
ff |= DE_CLOSE;
errcode = wsaEvents.iErrorCode[FD_CLOSE_BIT];
}
if (ff != 0) {
disp->OnPreEvent(ff);
disp->OnEvent(ff, errcode);
}
}
}
processing_dispatchers_ = false;
// Process deferred dispatchers that have been added/removed while the
// events were handled above.
AddRemovePendingDispatchers();
}
// Reset the network event until new activity occurs
WSAResetEvent(socket_ev_);
}
// Break?
if (!fWait_)
break;
cmsElapsed = TimeSince(msStart);
if ((cmsWait != kForever) && (cmsElapsed >= cmsWait)) {
break;
}
}
// Done
return true;
}
#endif // WEBRTC_WIN
} // namespace rtc
Reference in a new issue View git blame Copy permalink