Skip to content

F
folly
Commit 6ee198b0 authored by Subodh Iyengar's avatar Subodh Iyengar Committed by Facebook Github Bot
Browse files
Options

Clang format AsyncUDPSocket 

Summary:
Clang format files related to AsyncUDPSocket.
This helps keep changes consistent.

Reviewed By: yfeldblum, lnicco

Differential Revision: D7632865

fbshipit-source-id: 0761f0cfd06417334e680ca0880280d65480fe7e
parent ba208d71
Hide whitespace changes
Inline Side-by-side
Showing with 94 additions and 124 deletions
folly/io/async/AsyncUDPSocket.cpp
View file @ 6ee198b0
......@@ -51,9 +51,10 @@ AsyncUDPSocket::~AsyncUDPSocket() {
void AsyncUDPSocket::bind(const folly::SocketAddress& address) {
int socket = fsp::socket(address.getFamily(), SOCK_DGRAM, IPPROTO_UDP);
if (socket == -1) {
throw AsyncSocketException(AsyncSocketException::NOT_OPEN,
"error creating async udp socket",
errno);
throw AsyncSocketException(
AsyncSocketException::NOT_OPEN,
"error creating async udp socket",
errno);
}
auto g = folly::makeGuard([&] { ::close(socket); });
......@@ -61,37 +62,33 @@ void AsyncUDPSocket::bind(const folly::SocketAddress& address) {
// put the socket in non-blocking mode
int ret = fcntl(socket, F_SETFL, O_NONBLOCK);
if (ret != 0) {
throw AsyncSocketException(AsyncSocketException::NOT_OPEN,
"failed to put socket in non-blocking mode",
errno);
throw AsyncSocketException(
AsyncSocketException::NOT_OPEN,
"failed to put socket in non-blocking mode",
errno);
}
if (reuseAddr_) {
// put the socket in reuse mode
int value = 1;
if (setsockopt(socket,
SOL_SOCKET,
SO_REUSEADDR,
&value,
sizeof(value)) != 0) {
throw AsyncSocketException(AsyncSocketException::NOT_OPEN,
"failed to put socket in reuse mode",
errno);
if (setsockopt(socket, SOL_SOCKET, SO_REUSEADDR, &value, sizeof(value)) !=
0) {
throw AsyncSocketException(
AsyncSocketException::NOT_OPEN,
"failed to put socket in reuse mode",
errno);
}
}
if (reusePort_) {
// put the socket in port reuse mode
int value = 1;
if (setsockopt(socket,
SOL_SOCKET,
SO_REUSEPORT,
&value,
sizeof(value)) != 0) {
throw AsyncSocketException(AsyncSocketException::NOT_OPEN,
"failed to put socket in reuse_port mode",
errno);
if (setsockopt(socket, SOL_SOCKET, SO_REUSEPORT, &value, sizeof(value)) !=
0) {
throw AsyncSocketException(
AsyncSocketException::NOT_OPEN,
"failed to put socket in reuse_port mode",
errno);
}
}
......@@ -100,50 +97,38 @@ void AsyncUDPSocket::bind(const folly::SocketAddress& address) {
// Set busy_poll time in microseconds on the socket.
// It sets how long socket will be in busy_poll mode when no event occurs.
int value = busyPollUs_;
if (setsockopt(socket,
SOL_SOCKET,
SO_BUSY_POLL,
&value,
sizeof(value)) != 0) {
throw AsyncSocketException(AsyncSocketException::NOT_OPEN,
"failed to set SO_BUSY_POLL on the socket",
errno);
if (setsockopt(socket, SOL_SOCKET, SO_BUSY_POLL, &value, sizeof(value)) !=
0) {
throw AsyncSocketException(
AsyncSocketException::NOT_OPEN,
"failed to set SO_BUSY_POLL on the socket",
errno);
}
#else /* SO_BUSY_POLL is not supported*/
throw AsyncSocketException(AsyncSocketException::NOT_OPEN,
"SO_BUSY_POLL is not supported",
errno);
throw AsyncSocketException(
AsyncSocketException::NOT_OPEN, "SO_BUSY_POLL is not supported", errno);
#endif
}
if (rcvBuf_ > 0) {
// Set the size of the buffer for the received messages in rx_queues.
int value = rcvBuf_;
if (setsockopt(socket,
SOL_SOCKET,
SO_RCVBUF,
&value,
sizeof(value)) != 0) {
throw AsyncSocketException(AsyncSocketException::NOT_OPEN,
"failed to set SO_RCVBUF on the socket",
errno);
if (setsockopt(socket, SOL_SOCKET, SO_RCVBUF, &value, sizeof(value)) != 0) {
throw AsyncSocketException(
AsyncSocketException::NOT_OPEN,
"failed to set SO_RCVBUF on the socket",
errno);
}
}
if (sndBuf_ > 0) {
// Set the size of the buffer for the sent messages in tx_queues.
int value = rcvBuf_;
if (setsockopt(socket,
SOL_SOCKET,
SO_SNDBUF,
&value,
sizeof(value)) != 0) {
throw AsyncSocketException(AsyncSocketException::NOT_OPEN,
"failed to set SO_SNDBUF on the socket",
errno);
if (setsockopt(socket, SOL_SOCKET, SO_SNDBUF, &value, sizeof(value)) != 0) {
throw AsyncSocketException(
AsyncSocketException::NOT_OPEN,
"failed to set SO_SNDBUF on the socket",
errno);
}
}
......@@ -152,9 +137,7 @@ void AsyncUDPSocket::bind(const folly::SocketAddress& address) {
int flag = 1;
if (setsockopt(socket, IPPROTO_IPV6, IPV6_V6ONLY, &flag, sizeof(flag))) {
throw AsyncSocketException(
AsyncSocketException::NOT_OPEN,
"Failed to set IPV6_V6ONLY",
errno);
AsyncSocketException::NOT_OPEN, "Failed to set IPV6_V6ONLY", errno);
}
}
......@@ -223,14 +206,15 @@ void AsyncUDPSocket::setFD(int fd, FDOwnership ownership) {
localAddress_.setFromLocalAddress(fd_);
}
ssize_t AsyncUDPSocket::write(const folly::SocketAddress& address,
const std::unique_ptr<folly::IOBuf>& buf) {
ssize_t AsyncUDPSocket::write(
const folly::SocketAddress& address,
const std::unique_ptr<folly::IOBuf>& buf) {
// UDP's typical MTU size is 1500, so high number of buffers
// really do not make sense. Optimze for buffer chains with
// buffers less than 16, which is the highest I can think of
// for a real use case.
iovec vec[16];
size_t iovec_len = buf->fillIov(vec, sizeof(vec)/sizeof(vec[0]));
size_t iovec_len = buf->fillIov(vec, sizeof(vec) / sizeof(vec[0]));
if (UNLIKELY(iovec_len == 0)) {
buf->coalesce();
vec[0].iov_base = const_cast<uint8_t*>(buf->data());
......@@ -241,8 +225,10 @@ ssize_t AsyncUDPSocket::write(const folly::SocketAddress& address,
return writev(address, vec, iovec_len);
}
ssize_t AsyncUDPSocket::writev(const folly::SocketAddress& address,
const struct iovec* vec, size_t iovec_len) {
ssize_t AsyncUDPSocket::writev(
const folly::SocketAddress& address,
const struct iovec* vec,
size_t iovec_len) {
CHECK_NE(-1, fd_) << "Socket not yet bound";
sockaddr_storage addrStorage;
......@@ -266,8 +252,8 @@ void AsyncUDPSocket::resumeRead(ReadCallback* cob) {
readCallback_ = CHECK_NOTNULL(cob);
if (!updateRegistration()) {
AsyncSocketException ex(AsyncSocketException::NOT_OPEN,
"failed to register for accept events");
AsyncSocketException ex(
AsyncSocketException::NOT_OPEN, "failed to register for accept events");
readCallback_ = nullptr;
cob->onReadError(ex);
......@@ -318,7 +304,6 @@ void AsyncUDPSocket::handleRead() noexcept {
AsyncSocketException::BAD_ARGS,
"AsyncUDPSocket::getReadBuffer() returned empty buffer");
auto cob = readCallback_;
readCallback_ = nullptr;
......@@ -353,9 +338,8 @@ void AsyncUDPSocket::handleRead() noexcept {
return;
}
AsyncSocketException ex(AsyncSocketException::INTERNAL_ERROR,
"::recvfrom() failed",
errno);
AsyncSocketException ex(
AsyncSocketException::INTERNAL_ERROR, "::recvfrom() failed", errno);
// In case of UDP we can continue reading from the socket
// even if the current request fails. We notify the user
......
folly/io/async/AsyncUDPSocket.h
View file @ 6ee198b0
......@@ -33,10 +33,7 @@ namespace folly {
*/
class AsyncUDPSocket : public EventHandler {
public:
enum class FDOwnership {
OWNS,
SHARED
};
enum class FDOwnership { OWNS, SHARED };
class ReadCallback {
public:
......@@ -48,16 +45,17 @@ class AsyncUDPSocket : public EventHandler {
* and if there were more bytes in datagram, we will end up
* dropping them.
*/
virtual void getReadBuffer(void** buf, size_t* len) noexcept = 0;
virtual void getReadBuffer(void** buf, size_t* len) noexcept = 0;
/**
* Invoked when a new datagraom is available on the socket. `len`
* is the number of bytes read and `truncated` is true if we had
* to drop few bytes because of running out of buffer space.
*/
virtual void onDataAvailable(const folly::SocketAddress& client,
size_t len,
bool truncated) noexcept = 0;
virtual void onDataAvailable(
const folly::SocketAddress& client,
size_t len,
bool truncated) noexcept = 0;
/**
* Invoked when there is an error reading from the socket.
......@@ -66,8 +64,7 @@ class AsyncUDPSocket : public EventHandler {
* But you have to re-register readCallback yourself after
* onReadError.
*/
virtual void onReadError(const AsyncSocketException& ex)
noexcept = 0;
virtual void onReadError(const AsyncSocketException& ex) noexcept = 0;
/**
* Invoked when socket is closed and a read callback is registered.
......@@ -112,14 +109,17 @@ class AsyncUDPSocket : public EventHandler {
* Send the data in buffer to destination. Returns the return code from
* ::sendmsg.
*/
virtual ssize_t write(const folly::SocketAddress& address,
const std::unique_ptr<folly::IOBuf>& buf);
virtual ssize_t write(
const folly::SocketAddress& address,
const std::unique_ptr<folly::IOBuf>& buf);
/**
* Send data in iovec to destination. Returns the return code from sendmsg.
*/
virtual ssize_t writev(const folly::SocketAddress& address,
const struct iovec* vec, size_t veclen);
virtual ssize_t writev(
const folly::SocketAddress& address,
const struct iovec* vec,
size_t veclen);
/**
* Start reading datagrams
......
folly/io/async/test/AsyncUDPSocketTest.cpp
View file @ 6ee198b0
......@@ -26,36 +26,34 @@
#include <folly/portability/GMock.h>
#include <folly/portability/GTest.h>
using folly::AsyncUDPSocket;
using folly::AsyncUDPServerSocket;
using folly::AsyncTimeout;
using folly::AsyncUDPServerSocket;
using folly::AsyncUDPSocket;
using folly::EventBase;
using folly::SocketAddress;
using folly::IOBuf;
using folly::SocketAddress;
using namespace testing;
class UDPAcceptor
: public AsyncUDPServerSocket::Callback {
class UDPAcceptor : public AsyncUDPServerSocket::Callback {
public:
UDPAcceptor(EventBase* evb, int n): evb_(evb), n_(n) {
}
UDPAcceptor(EventBase* evb, int n) : evb_(evb), n_(n) {}
void onListenStarted() noexcept override {}
void onListenStopped() noexcept override {}
void onDataAvailable(std::shared_ptr<folly::AsyncUDPSocket> /* socket */,
const folly::SocketAddress& client,
std::unique_ptr<folly::IOBuf> data,
bool truncated) noexcept override {
void onDataAvailable(
std::shared_ptr<folly::AsyncUDPSocket> /* socket */,
const folly::SocketAddress& client,
std::unique_ptr<folly::IOBuf> data,
bool truncated) noexcept override {
lastClient_ = client;
lastMsg_ = data->moveToFbString().toStdString();
auto len = data->computeChainDataLength();
VLOG(4) << "Worker " << n_ << " read " << len << " bytes "
<< "(trun:" << truncated << ") from " << client.describe()
<< " - " << lastMsg_;
<< "(trun:" << truncated << ") from " << client.describe() << " - "
<< lastMsg_;
sendPong();
}
......@@ -81,8 +79,7 @@ class UDPAcceptor
class UDPServer {
public:
UDPServer(EventBase* evb, folly::SocketAddress addr, int n)
: evb_(evb), addr_(addr), evbs_(n) {
}
: evb_(evb), addr_(addr), evbs_(n) {}
void start() {
CHECK(evb_->isInEventBaseThread());
......@@ -101,12 +98,10 @@ class UDPServer {
// Add numWorkers thread
int i = 0;
for (auto& evb: evbs_) {
for (auto& evb : evbs_) {
acceptors_.emplace_back(&evb, i);
std::thread t([&] () {
evb.loopForever();
});
std::thread t([&]() { evb.loopForever(); });
evb.waitUntilRunning();
......@@ -127,11 +122,11 @@ class UDPServer {
socket_->close();
socket_.reset();
for (auto& evb: evbs_) {
for (auto& evb : evbs_) {
evb.terminateLoopSoon();
}
for (auto& t: threads_) {
for (auto& t : threads_) {
t.join();
}
}
......@@ -146,14 +141,9 @@ class UDPServer {
std::vector<UDPAcceptor> acceptors_;
};
class UDPClient
: private AsyncUDPSocket::ReadCallback,
private AsyncTimeout {
class UDPClient : private AsyncUDPSocket::ReadCallback, private AsyncTimeout {
public:
explicit UDPClient(EventBase* evb)
: AsyncTimeout(evb),
evb_(evb) {
}
explicit UDPClient(EventBase* evb) : AsyncTimeout(evb), evb_(evb) {}
void start(const folly::SocketAddress& server, int n) {
CHECK(evb_->isInEventBaseThread());
......@@ -193,8 +183,7 @@ class UDPClient
--n_;
scheduleTimeout(5);
socket_->write(
server_,
folly::IOBuf::copyBuffer(folly::to<std::string>("PING ", n_)));
server_, folly::IOBuf::copyBuffer(folly::to<std::string>("PING ", n_)));
}
void getReadBuffer(void** buf, size_t* len) noexcept override {
......@@ -202,11 +191,12 @@ class UDPClient
*len = 1024;
}
void onDataAvailable(const folly::SocketAddress& client,
size_t len,
bool truncated) noexcept override {
void onDataAvailable(
const folly::SocketAddress& client,
size_t len,
bool truncated) noexcept override {
VLOG(4) << "Read " << len << " bytes (trun:" << truncated << ") from "
<< client.describe() << " - " << std::string(buf_, len);
<< client.describe() << " - " << std::string(buf_, len);
VLOG(4) << n_ << " left";
++pongRecvd_;
......@@ -251,9 +241,7 @@ TEST(AsyncSocketTest, PingPong) {
UDPServer server(&sevb, folly::SocketAddress("127.0.0.1", 0), 4);
// Start event loop in a separate thread
auto serverThread = std::thread([&sevb] () {
sevb.loopForever();
});
auto serverThread = std::thread([&sevb]() { sevb.loopForever(); });
// Wait for event loop to start
sevb.waitUntilRunning();
......@@ -265,15 +253,13 @@ TEST(AsyncSocketTest, PingPong) {
UDPClient client(&cevb);
// Start event loop in a separate thread
auto clientThread = std::thread([&cevb] () {
cevb.loopForever();
});
auto clientThread = std::thread([&cevb]() { cevb.loopForever(); });
// Wait for event loop to start
cevb.waitUntilRunning();
// Send ping
cevb.runInEventBaseThread([&] () { client.start(server.address(), 1000); });
cevb.runInEventBaseThread([&]() { client.start(server.address(), 1000); });
// Wait for client to finish
clientThread.join();
......@@ -283,7 +269,7 @@ TEST(AsyncSocketTest, PingPong) {
CHECK_GT(client.pongRecvd(), 0);
// Shutdown server
sevb.runInEventBaseThread([&] () {
sevb.runInEventBaseThread([&]() {
server.shutdown();
sevb.terminateLoopSoon();
});
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment