2 * Copyright 2017 Facebook, Inc.
4 * Licensed under the Apache License, Version 2.0 (the "License");
5 * you may not use this file except in compliance with the License.
6 * You may obtain a copy of the License at
8 * http://www.apache.org/licenses/LICENSE-2.0
10 * Unless required by applicable law or agreed to in writing, software
11 * distributed under the License is distributed on an "AS IS" BASIS,
12 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13 * See the License for the specific language governing permissions and
14 * limitations under the License.
21 #include <folly/MPMCQueue.h>
22 #include <folly/ScopeGuard.h>
23 #include <folly/io/async/EventBase.h>
24 #include <folly/io/async/EventHandler.h>
25 #include <folly/portability/GMock.h>
26 #include <folly/portability/GTest.h>
27 #include <folly/portability/Sockets.h>
28 #include <sys/eventfd.h>
31 using namespace folly;
32 using namespace testing;
34 void runInThreadsAndWait(
35 size_t nthreads, function<void(size_t)> cb) {
36 vector<thread> threads(nthreads);
37 for (size_t i = 0; i < nthreads; ++i) {
38 threads[i] = thread(cb, i);
40 for (size_t i = 0; i < nthreads; ++i) {
45 void runInThreadsAndWait(vector<function<void()>> cbs) {
46 runInThreadsAndWait(cbs.size(), [&](size_t k) { cbs[k](); });
49 class EventHandlerMock : public EventHandler {
51 EventHandlerMock(EventBase* eb, int fd) : EventHandler(eb, fd) {}
52 // gmock can't mock noexcept methods, so we need an intermediary
53 MOCK_METHOD1(_handlerReady, void(uint16_t));
54 void handlerReady(uint16_t events) noexcept override {
55 _handlerReady(events);
59 class EventHandlerTest : public Test {
63 void SetUp() override {
64 efd = eventfd(0, EFD_SEMAPHORE);
65 ASSERT_THAT(efd, Gt(0));
68 void TearDown() override {
75 void efd_write(uint64_t val) {
76 write(efd, &val, sizeof(val));
81 read(efd, &val, sizeof(val));
86 TEST_F(EventHandlerTest, simple) {
87 const size_t writes = 4;
88 size_t readsRemaining = writes;
91 EventHandlerMock eh(&eb, efd);
92 eh.registerHandler(EventHandler::READ | EventHandler::PERSIST);
93 EXPECT_CALL(eh, _handlerReady(_))
95 .WillRepeatedly(Invoke([&](uint16_t /* events */) {
97 if (--readsRemaining == 0) {
98 eh.unregisterHandler();
104 EXPECT_EQ(0, readsRemaining);
107 TEST_F(EventHandlerTest, many_concurrent_producers) {
108 const size_t writes = 200;
109 const size_t nproducers = 20;
110 size_t readsRemaining = writes;
112 runInThreadsAndWait({
115 EventHandlerMock eh(&eb, efd);
116 eh.registerHandler(EventHandler::READ | EventHandler::PERSIST);
117 EXPECT_CALL(eh, _handlerReady(_))
119 .WillRepeatedly(Invoke([&](uint16_t /* events */) {
121 if (--readsRemaining == 0) {
122 eh.unregisterHandler();
128 runInThreadsAndWait(nproducers,
129 [&](size_t /* k */) {
130 for (size_t i = 0; i < writes / nproducers; ++i) {
131 this_thread::sleep_for(chrono::milliseconds(1));
138 EXPECT_EQ(0, readsRemaining);
141 TEST_F(EventHandlerTest, many_concurrent_consumers) {
142 const size_t writes = 200;
143 const size_t nproducers = 8;
144 const size_t nconsumers = 20;
145 atomic<size_t> writesRemaining(writes);
146 atomic<size_t> readsRemaining(writes);
148 MPMCQueue<nullptr_t> queue(writes / 10);
150 runInThreadsAndWait({
154 [&](size_t /* k */) {
155 size_t thReadsRemaining = writes / nconsumers;
157 EventHandlerMock eh(&eb, efd);
158 eh.registerHandler(EventHandler::READ | EventHandler::PERSIST);
159 EXPECT_CALL(eh, _handlerReady(_))
160 .WillRepeatedly(Invoke([&](uint16_t /* events */) {
162 if (!queue.readIfNotEmpty(val)) {
167 if (--thReadsRemaining == 0) {
168 eh.unregisterHandler();
175 runInThreadsAndWait(nproducers,
176 [&](size_t /* k */) {
177 for (size_t i = 0; i < writes / nproducers; ++i) {
178 this_thread::sleep_for(chrono::milliseconds(1));
179 queue.blockingWrite(nullptr);
187 EXPECT_EQ(0, writesRemaining);
188 EXPECT_EQ(0, readsRemaining);
193 // See rfc6093 for extensive discussion on TCP URG semantics. Specificaly,
194 // it points out that URG mechanism was never intended to be used
195 // for out-of-band information delivery. However, pretty much every
196 // implementation interprets the LAST octect or urgent data as the
199 class EventHandlerOobTest : public ::testing::Test {
202 // Wait for port number to connect to, then connect and invoke
203 // clientOps(fd) where fd is the connection file descriptor
205 void runClient(std::function<void(int fd)> clientOps) {
206 clientThread = std::thread(
207 [ serverPortFuture = serverReady.get_future(), clientOps ]() mutable {
208 int clientFd = socket(AF_INET, SOCK_STREAM, 0);
212 struct hostent* he{nullptr};
213 struct sockaddr_in server;
215 std::array<const char, 10> hostname = {"localhost"};
216 he = gethostbyname(hostname.data());
219 memcpy(&server.sin_addr, he->h_addr_list[0], he->h_length);
220 server.sin_family = AF_INET;
222 // block here until port is known
223 server.sin_port = serverPortFuture.get();
224 LOG(INFO) << "Server is ready";
227 ::connect(clientFd, (struct sockaddr*)&server, sizeof(server)) ==
229 LOG(INFO) << "Server connection available";
236 // Bind, get port number, pass it to client, listen/accept and store the
241 int listenfd = socket(AF_INET, SOCK_STREAM, 0);
245 PCHECK(listenfd != -1) << "unable to open socket";
247 struct sockaddr_in sin;
248 sin.sin_port = htons(0);
249 sin.sin_addr.s_addr = INADDR_ANY;
250 sin.sin_family = AF_INET;
252 PCHECK(bind(listenfd, (struct sockaddr*)&sin, sizeof(sin)) >= 0)
253 << "Can't bind to port";
256 struct sockaddr_in findSockName;
257 socklen_t sz = sizeof(findSockName);
258 getsockname(listenfd, (struct sockaddr*)&findSockName, &sz);
259 serverReady.set_value(findSockName.sin_port);
261 struct sockaddr_in cli_addr;
262 socklen_t clilen = sizeof(cli_addr);
263 serverFd = accept(listenfd, (struct sockaddr*)&cli_addr, &clilen);
264 PCHECK(serverFd >= 0) << "can't accept";
267 void SetUp() override {}
269 void TearDown() override {
275 std::thread clientThread;
276 std::promise<decltype(sockaddr_in::sin_port)> serverReady;
281 // Test that sending OOB data is detected by event handler
283 TEST_F(EventHandlerOobTest, EPOLLPRI) {
284 auto clientOps = [](int fd) {
285 char buffer[] = "banana";
286 int n = send(fd, buffer, strlen(buffer) + 1, MSG_OOB);
287 LOG(INFO) << "Client send finished";
291 runClient(clientOps);
294 struct SockEvent : public EventHandler {
295 SockEvent(EventBase* eb, int fd) : EventHandler(eb, fd), fd_(fd) {}
297 void handlerReady(uint16_t events) noexcept override {
298 EXPECT_TRUE(EventHandler::EventFlags::PRI & events);
299 std::array<char, 255> buffer;
300 int n = read(fd_, buffer.data(), buffer.size());
302 // NB: we sent 7 bytes, but only received 6. The last byte
303 // has been stored in the OOB buffer.
306 EXPECT_EQ("banana", std::string(buffer.data(), 6));
307 // now read the byte stored in OOB buffer
308 n = recv(fd_, buffer.data(), buffer.size(), MSG_OOB);
314 } sockHandler(&eb, serverFd);
316 sockHandler.registerHandler(EventHandler::EventFlags::PRI);
317 LOG(INFO) << "Registered Handler";
322 // Test if we can send an OOB byte and then normal data
324 TEST_F(EventHandlerOobTest, OOB_AND_NORMAL_DATA) {
325 auto clientOps = [](int sockfd) {
327 // OOB buffer can only hold one byte in most implementations
328 std::array<char, 2> buffer = {"X"};
329 int n = send(sockfd, buffer.data(), 1, MSG_OOB);
334 std::array<char, 7> buffer = {"banana"};
335 int n = send(sockfd, buffer.data(), buffer.size(), 0);
340 runClient(clientOps);
343 struct SockEvent : public EventHandler {
344 SockEvent(EventBase* eb, int fd) : EventHandler(eb, fd), eb_(eb), fd_(fd) {}
346 void handlerReady(uint16_t events) noexcept override {
347 std::array<char, 255> buffer;
348 if (events & EventHandler::EventFlags::PRI) {
349 int n = recv(fd_, buffer.data(), buffer.size(), MSG_OOB);
351 EXPECT_EQ("X", std::string(buffer.data(), 1));
352 registerHandler(EventHandler::EventFlags::READ);
356 if (events & EventHandler::EventFlags::READ) {
357 int n = recv(fd_, buffer.data(), buffer.size(), 0);
359 EXPECT_EQ("banana", std::string(buffer.data()));
360 eb_->terminateLoopSoon();
368 } sockHandler(&eb, serverFd);
369 sockHandler.registerHandler(
370 EventHandler::EventFlags::PRI | EventHandler::EventFlags::READ);
371 LOG(INFO) << "Registered Handler";
376 // Demonstrate that "regular" reads ignore the OOB byte sent to us
378 TEST_F(EventHandlerOobTest, SWALLOW_OOB) {
379 auto clientOps = [](int sockfd) {
381 std::array<char, 2> buffer = {"X"};
382 int n = send(sockfd, buffer.data(), 1, MSG_OOB);
387 std::array<char, 7> buffer = {"banana"};
388 int n = send(sockfd, buffer.data(), buffer.size(), 0);
393 runClient(clientOps);
396 struct SockEvent : public EventHandler {
397 SockEvent(EventBase* eb, int fd) : EventHandler(eb, fd), fd_(fd) {}
399 void handlerReady(uint16_t events) noexcept override {
400 std::array<char, 255> buffer;
401 ASSERT_TRUE(events & EventHandler::EventFlags::READ);
402 int n = recv(fd_, buffer.data(), buffer.size(), 0);
404 EXPECT_EQ("banana", std::string(buffer.data()));
409 } sockHandler(&eb, serverFd);
410 sockHandler.registerHandler(EventHandler::EventFlags::READ);
411 LOG(INFO) << "Registered Handler";