Commit ccc65a2e authored by Pranjal Raihan's avatar Pranjal Raihan Committed by Facebook GitHub Bot

Allow AsyncServerSocket::setQueueTimeout to accept an Observer

Summary: `AsyncServerSocket::getQueueTimeout` also returns an `Observer` now.

Reviewed By: andriigrynenko

Differential Revision: D25146630

fbshipit-source-id: 106ffc744eb6a3c09625897e72484054e0330247
parent 2e32138d
...@@ -1021,8 +1021,9 @@ void AsyncServerSocket::dispatchSocket( ...@@ -1021,8 +1021,9 @@ void AsyncServerSocket::dispatchSocket(
msg.type = MessageType::MSG_NEW_CONN; msg.type = MessageType::MSG_NEW_CONN;
msg.address = std::move(address); msg.address = std::move(address);
msg.fd = socket; msg.fd = socket;
if (queueTimeout_.count() != 0) { auto queueTimeout = *queueTimeout_;
msg.deadline = std::chrono::steady_clock::now() + queueTimeout_; if (queueTimeout.count() != 0) {
msg.deadline = std::chrono::steady_clock::now() + queueTimeout;
} }
// Loop until we find a free queue to write to // Loop until we find a free queue to write to
......
...@@ -18,6 +18,7 @@ ...@@ -18,6 +18,7 @@
#include <folly/SocketAddress.h> #include <folly/SocketAddress.h>
#include <folly/String.h> #include <folly/String.h>
#include <folly/experimental/observer/Observer.h>
#include <folly/io/ShutdownSocketSet.h> #include <folly/io/ShutdownSocketSet.h>
#include <folly/io/async/AsyncSocketBase.h> #include <folly/io/async/AsyncSocketBase.h>
#include <folly/io/async/AsyncTimeout.h> #include <folly/io/async/AsyncTimeout.h>
...@@ -210,6 +211,7 @@ class AsyncServerSocket : public DelayedDestruction, public AsyncSocketBase { ...@@ -210,6 +211,7 @@ class AsyncServerSocket : public DelayedDestruction, public AsyncSocketBase {
static const uint32_t kDefaultMaxAcceptAtOnce = 30; static const uint32_t kDefaultMaxAcceptAtOnce = 30;
static const uint32_t kDefaultCallbackAcceptAtOnce = 5; static const uint32_t kDefaultCallbackAcceptAtOnce = 5;
static const uint32_t kDefaultMaxMessagesInQueue = 1024; static const uint32_t kDefaultMaxMessagesInQueue = 1024;
/** /**
* Create a new AsyncServerSocket with the specified EventBase. * Create a new AsyncServerSocket with the specified EventBase.
* *
...@@ -545,7 +547,10 @@ class AsyncServerSocket : public DelayedDestruction, public AsyncSocketBase { ...@@ -545,7 +547,10 @@ class AsyncServerSocket : public DelayedDestruction, public AsyncSocketBase {
* Get the duration after which new connection messages will be dropped from * Get the duration after which new connection messages will be dropped from
* the NotificationQueue if it has not started processing yet. * the NotificationQueue if it has not started processing yet.
*/ */
std::chrono::nanoseconds getQueueTimeout() const { return queueTimeout_; } const folly::observer::AtomicObserver<std::chrono::nanoseconds>&
getQueueTimeout() const {
return queueTimeout_;
}
/** /**
* Set the duration after which new connection messages will be dropped from * Set the duration after which new connection messages will be dropped from
...@@ -557,9 +562,13 @@ class AsyncServerSocket : public DelayedDestruction, public AsyncSocketBase { ...@@ -557,9 +562,13 @@ class AsyncServerSocket : public DelayedDestruction, public AsyncSocketBase {
* *
* The default value (of 0) means that messages will never expire. * The default value (of 0) means that messages will never expire.
*/ */
void setQueueTimeout(std::chrono::nanoseconds duration) { void setQueueTimeout(
folly::observer::Observer<std::chrono::nanoseconds> duration) {
queueTimeout_ = duration; queueTimeout_ = duration;
} }
void setQueueTimeout(std::chrono::nanoseconds duration) {
setQueueTimeout(folly::observer::makeStaticObserver(duration));
}
/** /**
* Get the maximum number of unprocessed messages which a NotificationQueue * Get the maximum number of unprocessed messages which a NotificationQueue
...@@ -898,7 +907,8 @@ class AsyncServerSocket : public DelayedDestruction, public AsyncSocketBase { ...@@ -898,7 +907,8 @@ class AsyncServerSocket : public DelayedDestruction, public AsyncSocketBase {
ConnectionEventCallback* connectionEventCallback_{nullptr}; ConnectionEventCallback* connectionEventCallback_{nullptr};
bool tosReflect_{false}; bool tosReflect_{false};
bool zeroCopyVal_{false}; bool zeroCopyVal_{false};
std::chrono::nanoseconds queueTimeout_{0}; folly::observer::AtomicObserver<std::chrono::nanoseconds> queueTimeout_{
folly::observer::makeStaticObserver(std::chrono::nanoseconds::zero())};
}; };
} // namespace folly } // namespace folly
...@@ -4203,7 +4203,7 @@ TEST(AsyncSocketTest, getBufInUse) { ...@@ -4203,7 +4203,7 @@ TEST(AsyncSocketTest, getBufInUse) {
} }
#endif #endif
TEST(AsyncSocketTest, ConnectionExpiry) { TEST(AsyncSocketTest, QueueTimeout) {
// Create a new AsyncServerSocket // Create a new AsyncServerSocket
EventBase eventBase; EventBase eventBase;
std::shared_ptr<AsyncServerSocket> serverSocket( std::shared_ptr<AsyncServerSocket> serverSocket(
...@@ -4213,8 +4213,8 @@ TEST(AsyncSocketTest, ConnectionExpiry) { ...@@ -4213,8 +4213,8 @@ TEST(AsyncSocketTest, ConnectionExpiry) {
folly::SocketAddress serverAddress; folly::SocketAddress serverAddress;
serverSocket->getAddress(&serverAddress); serverSocket->getAddress(&serverAddress);
constexpr auto kConnectionExpiryDuration = milliseconds(10); constexpr auto kConnectionTimeout = milliseconds(10);
serverSocket->setQueueTimeout(kConnectionExpiryDuration); serverSocket->setQueueTimeout(kConnectionTimeout);
TestAcceptCallback acceptCb; TestAcceptCallback acceptCb;
acceptCb.setConnectionAcceptedFn( acceptCb.setConnectionAcceptedFn(
...@@ -4225,7 +4225,7 @@ TEST(AsyncSocketTest, ConnectionExpiry) { ...@@ -4225,7 +4225,7 @@ TEST(AsyncSocketTest, ConnectionExpiry) {
// Allow plenty of time for the AsyncSocketServer's event loop to run. // Allow plenty of time for the AsyncSocketServer's event loop to run.
// This should leave no doubt that the acceptor thread has enough time // This should leave no doubt that the acceptor thread has enough time
// to dequeue. If the dequeue succeeds, then our expiry code is broken. // to dequeue. If the dequeue succeeds, then our expiry code is broken.
constexpr auto kEventLoopTime = kConnectionExpiryDuration * 5; constexpr auto kEventLoopTime = kConnectionTimeout * 5;
eventBase.runInEventBaseThread([&]() { eventBase.runInEventBaseThread([&]() {
eventBase.tryRunAfterDelay( eventBase.tryRunAfterDelay(
[&]() { serverSocket->removeAcceptCallback(&acceptCb, nullptr); }, [&]() { serverSocket->removeAcceptCallback(&acceptCb, nullptr); },
...@@ -4233,7 +4233,7 @@ TEST(AsyncSocketTest, ConnectionExpiry) { ...@@ -4233,7 +4233,7 @@ TEST(AsyncSocketTest, ConnectionExpiry) {
}); });
// After the first message is enqueued, sleep long enough so that the // After the first message is enqueued, sleep long enough so that the
// second message expires before it has a chance to dequeue. // second message expires before it has a chance to dequeue.
std::this_thread::sleep_for(kConnectionExpiryDuration); std::this_thread::sleep_for(kConnectionTimeout);
}); });
ScopedEventBaseThread acceptThread("ioworker_test"); ScopedEventBaseThread acceptThread("ioworker_test");
......
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