folly/io/async/test/EventHandlerTest.cpp

   1 /*
   2  * Copyright 2017 Facebook, Inc.
   3  *
   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
   7  *
   8  *   http://www.apache.org/licenses/LICENSE-2.0
   9  *
  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.
  15  */
  16
  17 #include <bitset>
  18 #include <future>
  19 #include <thread>
  20
  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>
  29
  30 using namespace std;
  31 using namespace folly;
  32 using namespace testing;
  33
  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);
  39   }
  40   for (size_t i = 0; i < nthreads; ++i) {
  41     threads[i].join();
  42   }
  43 }
  44
  45 void runInThreadsAndWait(vector<function<void()>> cbs) {
  46   runInThreadsAndWait(cbs.size(), [&](size_t k) { cbs[k](); });
  47 }
  48
  49 class EventHandlerMock : public EventHandler {
  50 public:
  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);
  56   }
  57 };
  58
  59 class EventHandlerTest : public Test {
  60 public:
  61   int efd = 0;
  62
  63   void SetUp() override {
  64     efd = eventfd(0, EFD_SEMAPHORE);
  65     ASSERT_THAT(efd, Gt(0));
  66   }
  67
  68   void TearDown() override {
  69     if (efd > 0) {
  70       close(efd);
  71     }
  72     efd = 0;
  73   }
  74
  75   void efd_write(uint64_t val) {
  76     write(efd, &val, sizeof(val));
  77   }
  78
  79   uint64_t efd_read() {
  80     uint64_t val = 0;
  81     read(efd, &val, sizeof(val));
  82     return val;
  83   }
  84 };
  85
  86 TEST_F(EventHandlerTest, simple) {
  87   const size_t writes = 4;
  88   size_t readsRemaining = writes;
  89
  90   EventBase eb;
  91   EventHandlerMock eh(&eb, efd);
  92   eh.registerHandler(EventHandler::READ | EventHandler::PERSIST);
  93   EXPECT_CALL(eh, _handlerReady(_))
  94       .Times(writes)
  95       .WillRepeatedly(Invoke([&](uint16_t /* events */) {
  96         efd_read();
  97         if (--readsRemaining == 0) {
  98           eh.unregisterHandler();
  99         }
 100       }));
 101   efd_write(writes);
 102   eb.loop();
 103
 104   EXPECT_EQ(0, readsRemaining);
 105 }
 106
 107 TEST_F(EventHandlerTest, many_concurrent_producers) {
 108   const size_t writes = 200;
 109   const size_t nproducers = 20;
 110   size_t readsRemaining = writes;
 111
 112   runInThreadsAndWait({
 113       [&] {
 114         EventBase eb;
 115         EventHandlerMock eh(&eb, efd);
 116         eh.registerHandler(EventHandler::READ | EventHandler::PERSIST);
 117         EXPECT_CALL(eh, _handlerReady(_))
 118             .Times(writes)
 119             .WillRepeatedly(Invoke([&](uint16_t /* events */) {
 120               efd_read();
 121               if (--readsRemaining == 0) {
 122                 eh.unregisterHandler();
 123               }
 124             }));
 125         eb.loop();
 126       },
 127       [&] {
 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));
 132                                 efd_write(1);
 133                               }
 134                             });
 135       },
 136   });
 137
 138   EXPECT_EQ(0, readsRemaining);
 139 }
 140
 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);
 147
 148   MPMCQueue<nullptr_t> queue(writes / 10);
 149
 150   runInThreadsAndWait({
 151       [&] {
 152         runInThreadsAndWait(
 153             nconsumers,
 154             [&](size_t /* k */) {
 155               size_t thReadsRemaining = writes / nconsumers;
 156               EventBase eb;
 157               EventHandlerMock eh(&eb, efd);
 158               eh.registerHandler(EventHandler::READ | EventHandler::PERSIST);
 159               EXPECT_CALL(eh, _handlerReady(_))
 160                   .WillRepeatedly(Invoke([&](uint16_t /* events */) {
 161                     nullptr_t val;
 162                     if (!queue.readIfNotEmpty(val)) {
 163                       return;
 164                     }
 165                     efd_read();
 166                     --readsRemaining;
 167                     if (--thReadsRemaining == 0) {
 168                       eh.unregisterHandler();
 169                     }
 170                   }));
 171               eb.loop();
 172             });
 173       },
 174       [&] {
 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);
 180                                 efd_write(1);
 181                                 --writesRemaining;
 182                               }
 183                             });
 184       },
 185   });
 186
 187   EXPECT_EQ(0, writesRemaining);
 188   EXPECT_EQ(0, readsRemaining);
 189 }
 190
 191 #ifdef EV_PRI
 192 //
 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
 197 // OOB byte.
 198 //
 199 class EventHandlerOobTest : public ::testing::Test {
 200  public:
 201   //
 202   // Wait for port number to connect to, then connect and invoke
 203   // clientOps(fd) where fd is the connection file descriptor
 204   //
 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);
 209           SCOPE_EXIT {
 210             close(clientFd);
 211           };
 212           struct hostent* he{nullptr};
 213           struct sockaddr_in server;
 214
 215           std::array<const char, 10> hostname = {"localhost"};
 216           he = gethostbyname(hostname.data());
 217           PCHECK(he);
 218
 219           memcpy(&server.sin_addr, he->h_addr_list[0], he->h_length);
 220           server.sin_family = AF_INET;
 221
 222           // block here until port is known
 223           server.sin_port = serverPortFuture.get();
 224           LOG(INFO) << "Server is ready";
 225
 226           PCHECK(
 227               ::connect(clientFd, (struct sockaddr*)&server, sizeof(server)) ==
 228               0);
 229           LOG(INFO) << "Server connection available";
 230
 231           clientOps(clientFd);
 232         });
 233   }
 234
 235   //
 236   // Bind, get port number, pass it to client, listen/accept and store the
 237   // accepted fd
 238   //
 239   void acceptConn() {
 240     // make the server.
 241     int listenfd = socket(AF_INET, SOCK_STREAM, 0);
 242     SCOPE_EXIT {
 243       close(listenfd);
 244     };
 245     PCHECK(listenfd != -1) << "unable to open socket";
 246
 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;
 251
 252     PCHECK(bind(listenfd, (struct sockaddr*)&sin, sizeof(sin)) >= 0)
 253         << "Can't bind to port";
 254     listen(listenfd, 5);
 255
 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);
 260
 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";
 265   }
 266
 267   void SetUp() override {}
 268
 269   void TearDown() override {
 270     clientThread.join();
 271     close(serverFd);
 272   }
 273
 274   EventBase eb;
 275   std::thread clientThread;
 276   std::promise<decltype(sockaddr_in::sin_port)> serverReady;
 277   int serverFd{-1};
 278 };
 279
 280 //
 281 // Test that sending OOB data is detected by event handler
 282 //
 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";
 288     PCHECK(n > 0);
 289   };
 290
 291   runClient(clientOps);
 292   acceptConn();
 293
 294   struct SockEvent : public EventHandler {
 295     SockEvent(EventBase* eb, int fd) : EventHandler(eb, fd), fd_(fd) {}
 296
 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());
 301       //
 302       // NB: we sent 7 bytes, but only received 6. The last byte
 303       // has been stored in the OOB buffer.
 304       //
 305       EXPECT_EQ(6, n);
 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);
 309       EXPECT_EQ(1, n);
 310     }
 311
 312    private:
 313     int fd_;
 314   } sockHandler(&eb, serverFd);
 315
 316   sockHandler.registerHandler(EventHandler::EventFlags::PRI);
 317   LOG(INFO) << "Registered Handler";
 318   eb.loop();
 319 }
 320
 321 //
 322 // Test if we can send an OOB byte and then normal data
 323 //
 324 TEST_F(EventHandlerOobTest, OOB_AND_NORMAL_DATA) {
 325   auto clientOps = [](int sockfd) {
 326     {
 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);
 330       PCHECK(n > 0);
 331     }
 332
 333     {
 334       std::array<char, 7> buffer = {"banana"};
 335       int n = send(sockfd, buffer.data(), buffer.size(), 0);
 336       PCHECK(n > 0);
 337     }
 338   };
 339
 340   runClient(clientOps);
 341   acceptConn();
 342
 343   struct SockEvent : public EventHandler {
 344     SockEvent(EventBase* eb, int fd) : EventHandler(eb, fd), eb_(eb), fd_(fd) {}
 345
 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);
 350         EXPECT_EQ(1, n);
 351         EXPECT_EQ("X", std::string(buffer.data(), 1));
 352         registerHandler(EventHandler::EventFlags::READ);
 353         return;
 354       }
 355
 356       if (events & EventHandler::EventFlags::READ) {
 357         int n = recv(fd_, buffer.data(), buffer.size(), 0);
 358         EXPECT_EQ(7, n);
 359         EXPECT_EQ("banana", std::string(buffer.data()));
 360         eb_->terminateLoopSoon();
 361         return;
 362       }
 363     }
 364
 365    private:
 366     EventBase* eb_;
 367     int fd_;
 368   } sockHandler(&eb, serverFd);
 369   sockHandler.registerHandler(
 370       EventHandler::EventFlags::PRI | EventHandler::EventFlags::READ);
 371   LOG(INFO) << "Registered Handler";
 372   eb.loopForever();
 373 }
 374
 375 //
 376 // Demonstrate that "regular" reads ignore the OOB byte sent to us
 377 //
 378 TEST_F(EventHandlerOobTest, SWALLOW_OOB) {
 379   auto clientOps = [](int sockfd) {
 380     {
 381       std::array<char, 2> buffer = {"X"};
 382       int n = send(sockfd, buffer.data(), 1, MSG_OOB);
 383       PCHECK(n > 0);
 384     }
 385
 386     {
 387       std::array<char, 7> buffer = {"banana"};
 388       int n = send(sockfd, buffer.data(), buffer.size(), 0);
 389       PCHECK(n > 0);
 390     }
 391   };
 392
 393   runClient(clientOps);
 394   acceptConn();
 395
 396   struct SockEvent : public EventHandler {
 397     SockEvent(EventBase* eb, int fd) : EventHandler(eb, fd), fd_(fd) {}
 398
 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);
 403       EXPECT_EQ(7, n);
 404       EXPECT_EQ("banana", std::string(buffer.data()));
 405     }
 406
 407    private:
 408     int fd_;
 409   } sockHandler(&eb, serverFd);
 410   sockHandler.registerHandler(EventHandler::EventFlags::READ);
 411   LOG(INFO) << "Registered Handler";
 412   eb.loop();
 413 }
 414 #endif