From e1a15f9600e77c094dc368cb5dbe1c2ecd3dcb4f Mon Sep 17 00:00:00 2001 From: Petr Lapukhov Date: Fri, 30 Dec 2016 16:11:41 -0800 Subject: [PATCH] Add more MSG_OOB tests for EventHandler Summary: Sending OOB messages allows for single byte to be sent truly out of band and placed in an OOB buffer. This byte can then be read (by default) independent from normal data. The two test-cases validate that (1) We can receive OOB and normal data on same socket (2) Sending OOB byte does not affect regular data NOTE: sending more than one byte usually means the OOB data interleaved with regular data. Reviewed By: alandau Differential Revision: D4370652 fbshipit-source-id: a573dae7e5afba5a4a410be5371faf4f065c74e9 --- folly/io/async/test/EventHandlerTest.cpp | 253 ++++++++++++++++++----- 1 file changed, 198 insertions(+), 55 deletions(-) diff --git a/folly/io/async/test/EventHandlerTest.cpp b/folly/io/async/test/EventHandlerTest.cpp index 0c3e80c8..2631cfe5 100644 --- a/folly/io/async/test/EventHandlerTest.cpp +++ b/folly/io/async/test/EventHandlerTest.cpp @@ -189,83 +189,226 @@ TEST_F(EventHandlerTest, many_concurrent_consumers) { } #ifdef EV_PRI -TEST(EventHandlerSocketTest, EPOLLPRI) { - std::promise serverReady; - std::thread t([serverReadyFuture = serverReady.get_future()]() mutable { - // client - LOG(INFO) << "Server is ready"; - int sockfd = socket(AF_INET, SOCK_STREAM, 0); +// +// See rfc6093 for extensive discussion on TCP URG semantics. Specificaly, +// it points out that URG mechanism was never intended to be used +// for out-of-band information delivery. However, pretty much every +// implementation interprets the LAST octect or urgent data as the +// OOB byte. +// +class EventHandlerOobTest : public ::testing::Test { + public: + // + // Wait for port number to connect to, then connect and invoke + // clientOps(fd) where fd is the connection file descriptor + // + void runClient(std::function clientOps) { + clientThread = std::thread( + [ serverPortFuture = serverReady.get_future(), clientOps ]() mutable { + int clientFd = socket(AF_INET, SOCK_STREAM, 0); + SCOPE_EXIT { + close(clientFd); + }; + struct hostent* he{nullptr}; + struct sockaddr_in server; + + std::array hostname = {"localhost"}; + he = gethostbyname(hostname.data()); + PCHECK(he); + + memcpy(&server.sin_addr, he->h_addr_list[0], he->h_length); + server.sin_family = AF_INET; + + // block here until port is known + server.sin_port = serverPortFuture.get(); + LOG(INFO) << "Server is ready"; + + PCHECK( + ::connect(clientFd, (struct sockaddr*)&server, sizeof(server)) == + 0); + LOG(INFO) << "Server connection available"; + + clientOps(clientFd); + }); + } + + // + // Bind, get port number, pass it to client, listen/accept and store the + // accepted fd + // + void acceptConn() { + // make the server. + int listenfd = socket(AF_INET, SOCK_STREAM, 0); SCOPE_EXIT { - close(sockfd); + close(listenfd); }; - struct hostent* he; - struct sockaddr_in server; + PCHECK(listenfd != -1) << "unable to open socket"; + + struct sockaddr_in sin; + sin.sin_port = htons(0); + sin.sin_addr.s_addr = INADDR_ANY; + sin.sin_family = AF_INET; + + PCHECK(bind(listenfd, (struct sockaddr*)&sin, sizeof(sin)) >= 0) + << "Can't bind to port"; + listen(listenfd, 5); + + struct sockaddr_in findSockName; + socklen_t sz = sizeof(findSockName); + getsockname(listenfd, (struct sockaddr*)&findSockName, &sz); + serverReady.set_value(findSockName.sin_port); + + struct sockaddr_in cli_addr; + socklen_t clilen = sizeof(cli_addr); + serverFd = accept(listenfd, (struct sockaddr*)&cli_addr, &clilen); + PCHECK(serverFd >= 0) << "can't accept"; + } - const char hostname[] = "localhost"; - he = gethostbyname(hostname); - PCHECK(he); + void SetUp() override {} - memcpy(&server.sin_addr, he->h_addr_list[0], he->h_length); - server.sin_family = AF_INET; - server.sin_port = serverReadyFuture.get(); + void TearDown() override { + clientThread.join(); + close(serverFd); + } - PCHECK(::connect(sockfd, (struct sockaddr*)&server, sizeof(server)) == 0); - LOG(INFO) << "Server connection available"; + EventBase eb; + std::thread clientThread; + std::promise serverReady; + int serverFd{-1}; +}; +// +// Test that sending OOB data is detected by event handler +// +TEST_F(EventHandlerOobTest, EPOLLPRI) { + auto clientOps = [](int fd) { char buffer[] = "banana"; - int n = send(sockfd, buffer, strlen(buffer) + 1, MSG_OOB); + int n = send(fd, buffer, strlen(buffer) + 1, MSG_OOB); + LOG(INFO) << "Client send finished"; PCHECK(n > 0); - }); - SCOPE_EXIT { - t.join(); - }; - // make the server. - int sockfd = socket(AF_INET, SOCK_STREAM, 0); - SCOPE_EXIT { - close(sockfd); - }; - PCHECK(sockfd != -1) << "unable to open socket"; - - struct sockaddr_in sin; - sin.sin_port = htons(0); - sin.sin_addr.s_addr = INADDR_ANY; - sin.sin_family = AF_INET; - - PCHECK(bind(sockfd, (struct sockaddr*)&sin, sizeof(sin)) >= 0) - << "Can't bind to port"; - listen(sockfd, 5); - - struct sockaddr_in findSockName; - socklen_t sz = sizeof(findSockName); - getsockname(sockfd, (struct sockaddr*)&findSockName, &sz); - serverReady.set_value(findSockName.sin_port); - - socklen_t clilen; - struct sockaddr_in cli_addr; - clilen = sizeof(cli_addr); - int newsockfd = accept(sockfd, (struct sockaddr*)&cli_addr, &clilen); - PCHECK(newsockfd >= 0) << "can't accept"; - SCOPE_EXIT { - close(newsockfd); }; - EventBase eb; + runClient(clientOps); + acceptConn(); + struct SockEvent : public EventHandler { SockEvent(EventBase* eb, int fd) : EventHandler(eb, fd), fd_(fd) {} void handlerReady(uint16_t events) noexcept override { EXPECT_TRUE(EventHandler::EventFlags::PRI & events); - char buffer[256]; - int n = read(fd_, buffer, 255); + std::array buffer; + int n = read(fd_, buffer.data(), buffer.size()); + // + // NB: we sent 7 bytes, but only received 6. The last byte + // has been stored in the OOB buffer. + // EXPECT_EQ(6, n); - EXPECT_EQ("banana", std::string(buffer)); + EXPECT_EQ("banana", std::string(buffer.data(), 6)); + // now read the byte stored in OOB buffer + n = recv(fd_, buffer.data(), buffer.size(), MSG_OOB); + EXPECT_EQ(1, n); } private: int fd_; - } sockHandler(&eb, newsockfd); + } sockHandler(&eb, serverFd); + sockHandler.registerHandler(EventHandler::EventFlags::PRI); LOG(INFO) << "Registered Handler"; eb.loop(); } + +// +// Test if we can send an OOB byte and then normal data +// +TEST_F(EventHandlerOobTest, OOB_AND_NORMAL_DATA) { + auto clientOps = [](int sockfd) { + { + // OOB buffer can only hold one byte in most implementations + std::array buffer = {"X"}; + int n = send(sockfd, buffer.data(), 1, MSG_OOB); + PCHECK(n > 0); + } + + { + std::array buffer = {"banana"}; + int n = send(sockfd, buffer.data(), buffer.size(), 0); + PCHECK(n > 0); + } + }; + + runClient(clientOps); + acceptConn(); + + struct SockEvent : public EventHandler { + SockEvent(EventBase* eb, int fd) : EventHandler(eb, fd), eb_(eb), fd_(fd) {} + + void handlerReady(uint16_t events) noexcept override { + std::array buffer; + if (events & EventHandler::EventFlags::PRI) { + int n = recv(fd_, buffer.data(), buffer.size(), MSG_OOB); + EXPECT_EQ(1, n); + EXPECT_EQ("X", std::string(buffer.data(), 1)); + registerHandler(EventHandler::EventFlags::READ); + return; + } + + if (events & EventHandler::EventFlags::READ) { + int n = recv(fd_, buffer.data(), buffer.size(), 0); + EXPECT_EQ(7, n); + EXPECT_EQ("banana", std::string(buffer.data())); + eb_->terminateLoopSoon(); + return; + } + } + + private: + EventBase* eb_; + int fd_; + } sockHandler(&eb, serverFd); + sockHandler.registerHandler( + EventHandler::EventFlags::PRI | EventHandler::EventFlags::READ); + LOG(INFO) << "Registered Handler"; + eb.loopForever(); +} + +// +// Demonstrate that "regular" reads ignore the OOB byte sent to us +// +TEST_F(EventHandlerOobTest, SWALLOW_OOB) { + auto clientOps = [](int sockfd) { + { + std::array buffer = {"X"}; + int n = send(sockfd, buffer.data(), 1, MSG_OOB); + PCHECK(n > 0); + } + + { + std::array buffer = {"banana"}; + int n = send(sockfd, buffer.data(), buffer.size(), 0); + PCHECK(n > 0); + } + }; + + runClient(clientOps); + acceptConn(); + + struct SockEvent : public EventHandler { + SockEvent(EventBase* eb, int fd) : EventHandler(eb, fd), fd_(fd) {} + + void handlerReady(uint16_t events) noexcept override { + std::array buffer; + ASSERT_TRUE(events & EventHandler::EventFlags::READ); + int n = recv(fd_, buffer.data(), buffer.size(), 0); + EXPECT_EQ(7, n); + EXPECT_EQ("banana", std::string(buffer.data())); + } + + private: + int fd_; + } sockHandler(&eb, serverFd); + sockHandler.registerHandler(EventHandler::EventFlags::READ); + LOG(INFO) << "Registered Handler"; + eb.loop(); +} #endif -- 2.34.1