1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224
|
// Copyright 2012 The Chromium Authors
// Use of this source code is governed by a BSD-style license that can be
// found in the LICENSE file.
#include "base/sync_socket.h"
#include <errno.h>
#include <fcntl.h>
#include <limits.h>
#include <poll.h>
#include <stddef.h>
#include <stdio.h>
#include <sys/ioctl.h>
#include <sys/socket.h>
#include <sys/types.h>
#include "base/check_op.h"
#include "base/containers/span.h"
#include "base/files/file_util.h"
#include "base/numerics/safe_conversions.h"
#include "base/threading/scoped_blocking_call.h"
#include "build/build_config.h"
#if BUILDFLAG(IS_SOLARIS)
#include <sys/filio.h>
#endif
namespace base {
namespace {
// To avoid users sending negative message lengths to Send/Receive
// we clamp message lengths, which are size_t, to no more than INT_MAX.
const size_t kMaxMessageLength = static_cast<size_t>(INT_MAX);
// Writes |length| of |buffer| into |handle|. Returns the number of bytes
// written or zero on error. |length| must be greater than 0.
size_t SendHelper(SyncSocket::Handle handle,
const void* buffer,
size_t length) {
DCHECK_GT(length, 0u);
DCHECK_LE(length, kMaxMessageLength);
DCHECK_NE(handle, SyncSocket::kInvalidHandle);
return WriteFileDescriptor(
handle, make_span(static_cast<const uint8_t*>(buffer), length))
? length
: 0;
}
} // namespace
// static
bool SyncSocket::CreatePair(SyncSocket* socket_a, SyncSocket* socket_b) {
DCHECK_NE(socket_a, socket_b);
DCHECK(!socket_a->IsValid());
DCHECK(!socket_b->IsValid());
#if BUILDFLAG(IS_APPLE)
int nosigpipe = 1;
#endif // BUILDFLAG(IS_APPLE)
ScopedHandle handles[2];
{
Handle raw_handles[2] = {kInvalidHandle, kInvalidHandle};
if (socketpair(AF_UNIX, SOCK_STREAM, 0, raw_handles) != 0) {
return false;
}
handles[0].reset(raw_handles[0]);
handles[1].reset(raw_handles[1]);
}
#if BUILDFLAG(IS_APPLE)
// On OSX an attempt to read or write to a closed socket may generate a
// SIGPIPE rather than returning -1. setsockopt will shut this off.
if (0 != setsockopt(handles[0].get(), SOL_SOCKET, SO_NOSIGPIPE, &nosigpipe,
sizeof(nosigpipe)) ||
0 != setsockopt(handles[1].get(), SOL_SOCKET, SO_NOSIGPIPE, &nosigpipe,
sizeof(nosigpipe))) {
return false;
}
#endif
// Copy the handles out for successful return.
socket_a->handle_ = std::move(handles[0]);
socket_b->handle_ = std::move(handles[1]);
return true;
}
void SyncSocket::Close() {
handle_.reset();
}
size_t SyncSocket::Send(const void* buffer, size_t length) {
ScopedBlockingCall scoped_blocking_call(FROM_HERE, BlockingType::MAY_BLOCK);
return SendHelper(handle(), buffer, length);
}
size_t SyncSocket::Receive(void* buffer, size_t length) {
ScopedBlockingCall scoped_blocking_call(FROM_HERE, BlockingType::MAY_BLOCK);
DCHECK_GT(length, 0u);
DCHECK_LE(length, kMaxMessageLength);
DCHECK(IsValid());
char* charbuffer = static_cast<char*>(buffer);
if (ReadFromFD(handle(), charbuffer, length))
return length;
return 0;
}
size_t SyncSocket::ReceiveWithTimeout(void* buffer,
size_t length,
TimeDelta timeout) {
ScopedBlockingCall scoped_blocking_call(FROM_HERE, BlockingType::MAY_BLOCK);
DCHECK_GT(length, 0u);
DCHECK_LE(length, kMaxMessageLength);
DCHECK(IsValid());
// Only timeouts greater than zero and less than one second are allowed.
DCHECK_GT(timeout.InMicroseconds(), 0);
DCHECK_LT(timeout.InMicroseconds(), Seconds(1).InMicroseconds());
// Track the start time so we can reduce the timeout as data is read.
TimeTicks start_time = TimeTicks::Now();
const TimeTicks finish_time = start_time + timeout;
struct pollfd pollfd;
pollfd.fd = handle();
pollfd.events = POLLIN;
pollfd.revents = 0;
size_t bytes_read_total = 0;
while (bytes_read_total < length) {
const TimeDelta this_timeout = finish_time - TimeTicks::Now();
const int timeout_ms =
static_cast<int>(this_timeout.InMillisecondsRoundedUp());
if (timeout_ms <= 0)
break;
const int poll_result = poll(&pollfd, 1, timeout_ms);
// Handle EINTR manually since we need to update the timeout value.
if (poll_result == -1 && errno == EINTR)
continue;
// Return if other type of error or a timeout.
if (poll_result <= 0)
return bytes_read_total;
// poll() only tells us that data is ready for reading, not how much. We
// must Peek() for the amount ready for reading to avoid blocking.
// At hang up (POLLHUP), the write end has been closed and there might still
// be data to be read.
// No special handling is needed for error (POLLERR); we can let any of the
// following operations fail and handle it there.
DCHECK(pollfd.revents & (POLLIN | POLLHUP | POLLERR)) << pollfd.revents;
const size_t bytes_to_read = std::min(Peek(), length - bytes_read_total);
// There may be zero bytes to read if the socket at the other end closed.
if (!bytes_to_read)
return bytes_read_total;
const size_t bytes_received =
Receive(static_cast<char*>(buffer) + bytes_read_total, bytes_to_read);
bytes_read_total += bytes_received;
if (bytes_received != bytes_to_read)
return bytes_read_total;
}
return bytes_read_total;
}
size_t SyncSocket::Peek() {
DCHECK(IsValid());
int number_chars = 0;
if (ioctl(handle_.get(), FIONREAD, &number_chars) == -1) {
// If there is an error in ioctl, signal that the channel would block.
return 0;
}
return checked_cast<size_t>(number_chars);
}
bool SyncSocket::IsValid() const {
return handle_.is_valid();
}
SyncSocket::Handle SyncSocket::handle() const {
return handle_.get();
}
SyncSocket::Handle SyncSocket::Release() {
return handle_.release();
}
bool CancelableSyncSocket::Shutdown() {
DCHECK(IsValid());
return HANDLE_EINTR(shutdown(handle(), SHUT_RDWR)) >= 0;
}
size_t CancelableSyncSocket::Send(const void* buffer, size_t length) {
DCHECK_GT(length, 0u);
DCHECK_LE(length, kMaxMessageLength);
DCHECK(IsValid());
const int flags = fcntl(handle(), F_GETFL);
if (flags != -1 && (flags & O_NONBLOCK) == 0) {
// Set the socket to non-blocking mode for sending if its original mode
// is blocking.
fcntl(handle(), F_SETFL, flags | O_NONBLOCK);
}
const size_t len = SendHelper(handle(), buffer, length);
if (flags != -1 && (flags & O_NONBLOCK) == 0) {
// Restore the original flags.
fcntl(handle(), F_SETFL, flags);
}
return len;
}
// static
bool CancelableSyncSocket::CreatePair(CancelableSyncSocket* socket_a,
CancelableSyncSocket* socket_b) {
return SyncSocket::CreatePair(socket_a, socket_b);
}
} // namespace base
|