Shift AsyncPipe to NetworkSocket

Summary:
Shift AsyncPipe to be backed by NetworkSocket, as the only things this can be used for are listenable to by libevent. At the moment, the APIs for what are listenable to by libevent are all described in terms of NetworkSocket.
While, by name, this may seem incorrect, in practice, it is correct in function, due to the limitations of how NetworkSocket's are obtainable on Windows, and all existing calls to these functions on posix platforms are already functional by being file descriptor backed.
The plan is to have this, slightly badly named for the location, interface for the moment, and to clean it up in the future.

Reviewed By: yfeldblum

Differential Revision: D14702972

fbshipit-source-id: 785281a7bb9fafc63f5295e33f5bc136b9bc70f1

folly/io/async/AsyncPipe.cpp

@@ -16,6 +16,7 @@
 #include <folly/io/async/AsyncPipe.h>
 
 #include <folly/FileUtil.h>
+#include <folly/detail/FileUtilDetail.h>
 #include <folly/io/async/AsyncSocketException.h>
 
 using folly::IOBuf;
@@ -42,18 +43,28 @@ void AsyncPipeReader::failRead(const AsyncSocketException& ex) {
 
 void AsyncPipeReader::close() {
   unregisterHandler();
-  if (fd_ >= 0) {
+  if (fd_ != NetworkSocket()) {
     changeHandlerFD(NetworkSocket());
 
     if (closeCb_) {
       closeCb_(fd_);
     } else {
-      ::close(fd_);
+      netops::close(fd_);
     }
-    fd_ = -1;
+    fd_ = NetworkSocket();
   }
 }
 
+#if _WIN32
+static int recv_internal(NetworkSocket s, void* buf, size_t count) {
+  auto r = netops::recv(s, buf, count, 0);
+  if (r == -1 && WSAGetLastError() == WSAEWOULDBLOCK) {
+    errno = EAGAIN;
+  }
+  return r;
+}
+#endif
+
 void AsyncPipeReader::handlerReady(uint16_t events) noexcept {
   DestructorGuard dg(this);
   CHECK(events & EventHandler::READ);
@@ -104,7 +115,13 @@ void AsyncPipeReader::handlerReady(uint16_t events) noexcept {
     }
 
     // Perform the read
-    ssize_t bytesRead = folly::readNoInt(fd_, buf, buflen);
+#if _WIN32
+    // On Windows you can't call read on a socket, so call recv instead.
+    ssize_t bytesRead =
+        folly::fileutil_detail::wrapNoInt(recv_internal, fd_, buf, buflen);
+#else
+    ssize_t bytesRead = folly::readNoInt(fd_.toFd(), buf, buflen);
+#endif
 
     if (bytesRead > 0) {
       if (movable) {
@@ -190,15 +207,15 @@ void AsyncPipeWriter::closeNow() {
     failAllWrites(AsyncSocketException(
         AsyncSocketException::NOT_OPEN, "closed with pending writes"));
   }
-  if (fd_ >= 0) {
+  if (fd_ != NetworkSocket()) {
     unregisterHandler();
     changeHandlerFD(NetworkSocket());
     if (closeCb_) {
       closeCb_(fd_);
     } else {
-      close(fd_);
+      netops::close(fd_);
     }
-    fd_ = -1;
+    fd_ = NetworkSocket();
   }
 }
 
@@ -220,6 +237,16 @@ void AsyncPipeWriter::handlerReady(uint16_t events) noexcept {
   handleWrite();
 }
 
+#if _WIN32
+static int send_internal(NetworkSocket s, const void* buf, size_t count) {
+  auto r = netops::send(s, buf, count, 0);
+  if (r == -1 && WSAGetLastError() == WSAEWOULDBLOCK) {
+    errno = EAGAIN;
+  }
+  return r;
+}
+#endif
+
 void AsyncPipeWriter::handleWrite() {
   DestructorGuard dg(this);
   assert(!queue_.empty());
@@ -230,7 +257,13 @@ void AsyncPipeWriter::handleWrite() {
     // someday, support writev.  The logic for partial writes is a bit complex
     const IOBuf* head = curQueue.front();
     CHECK(head->length());
-    ssize_t rc = folly::writeNoInt(fd_, head->data(), head->length());
+#if _WIN32
+    // On Windows you can't call write on a socket.
+    ssize_t rc = folly::fileutil_detail::wrapNoInt(
+        send_internal, fd_, head->data(), head->length());
+#else
+    ssize_t rc = folly::writeNoInt(fd_.toFd(), head->data(), head->length());
+#endif
     if (rc < 0) {
       if (errno == EAGAIN || errno == EWOULDBLOCK) {
         // pipe is full

folly/io/async/AsyncPipe.h

@@ -39,14 +39,20 @@ class AsyncPipeReader : public EventHandler,
       unique_ptr<AsyncPipeReader, folly::DelayedDestruction::Destructor>
           UniquePtr;
 
-  template <typename... Args>
-  static UniquePtr newReader(Args&&... args) {
-    return UniquePtr(new AsyncPipeReader(std::forward<Args>(args)...));
+  static UniquePtr newReader(
+      folly::EventBase* eventBase,
+      NetworkSocket pipeFd) {
+    return UniquePtr(new AsyncPipeReader(eventBase, pipeFd));
   }
 
+  static UniquePtr newReader(folly::EventBase* eventBase, int pipeFd) {
+    return newReader(eventBase, NetworkSocket::fromFd(pipeFd));
+  }
+
+  AsyncPipeReader(folly::EventBase* eventBase, NetworkSocket pipeFd)
+      : EventHandler(eventBase, pipeFd), fd_(pipeFd) {}
   AsyncPipeReader(folly::EventBase* eventBase, int pipeFd)
-      : EventHandler(eventBase, folly::NetworkSocket::fromFd(pipeFd)),
-        fd_(pipeFd) {}
+      : AsyncPipeReader(eventBase, NetworkSocket::fromFd(pipeFd)) {}
 
   /**
    * Set the read callback and automatically install/uninstall the handler
@@ -74,7 +80,7 @@ class AsyncPipeReader : public EventHandler,
   /**
    * Set a special hook to close the socket (otherwise, will call close())
    */
-  void setCloseCallback(std::function<void(int)> closeCb) {
+  void setCloseCallback(std::function<void(NetworkSocket)> closeCb) {
     closeCb_ = closeCb;
   }
 
@@ -85,9 +91,9 @@ class AsyncPipeReader : public EventHandler,
   void failRead(const AsyncSocketException& ex);
   void close();
 
-  int fd_;
+  NetworkSocket fd_;
   AsyncReader::ReadCallback* readCallback_{nullptr};
-  std::function<void(int)> closeCb_;
+  std::function<void(NetworkSocket)> closeCb_;
 };
 
 /**
@@ -101,14 +107,20 @@ class AsyncPipeWriter : public EventHandler,
       unique_ptr<AsyncPipeWriter, folly::DelayedDestruction::Destructor>
           UniquePtr;
 
-  template <typename... Args>
-  static UniquePtr newWriter(Args&&... args) {
-    return UniquePtr(new AsyncPipeWriter(std::forward<Args>(args)...));
+  static UniquePtr newWriter(
+      folly::EventBase* eventBase,
+      NetworkSocket pipeFd) {
+    return UniquePtr(new AsyncPipeWriter(eventBase, pipeFd));
+  }
+
+  static UniquePtr newWriter(folly::EventBase* eventBase, int pipeFd) {
+    return newWriter(eventBase, NetworkSocket::fromFd(pipeFd));
   }
 
+  AsyncPipeWriter(folly::EventBase* eventBase, NetworkSocket pipeFd)
+      : EventHandler(eventBase, pipeFd), fd_(pipeFd) {}
   AsyncPipeWriter(folly::EventBase* eventBase, int pipeFd)
-      : EventHandler(eventBase, folly::NetworkSocket::fromFd(pipeFd)),
-        fd_(pipeFd) {}
+      : AsyncPipeWriter(eventBase, NetworkSocket::fromFd(pipeFd)) {}
 
   /**
    * Asynchronously write the given iobuf to this pipe, and invoke the callback
@@ -121,7 +133,7 @@ class AsyncPipeWriter : public EventHandler,
   /**
    * Set a special hook to close the socket (otherwise, will call close())
    */
-  void setCloseCallback(std::function<void(int)> closeCb) {
+  void setCloseCallback(std::function<void(NetworkSocket)> closeCb) {
     closeCb_ = closeCb;
   }
 
@@ -129,7 +141,7 @@ class AsyncPipeWriter : public EventHandler,
    * Returns true if the pipe is closed
    */
   bool closed() const {
-    return (fd_ < 0 || closeOnEmpty_);
+    return (fd_ == NetworkSocket() || closeOnEmpty_);
   }
 
   /**
@@ -175,10 +187,10 @@ class AsyncPipeWriter : public EventHandler,
   void handleWrite();
   void failAllWrites(const AsyncSocketException& ex);
 
-  int fd_;
+  NetworkSocket fd_;
   std::list<std::pair<folly::IOBufQueue, AsyncWriter::WriteCallback*>> queue_;
   bool closeOnEmpty_{false};
-  std::function<void(int)> closeCb_;
+  std::function<void(NetworkSocket)> closeCb_;
 
   ~AsyncPipeWriter() override {
     closeNow();

folly/io/async/AsyncSSLSocket.cpp

@@ -1008,8 +1008,8 @@ bool AsyncSSLSocket::willBlock(
       // a native handle type to pass to the constructor.
       auto native_handle = NetworkSocket::native_handle_type(ofd);
 
-      auto asyncPipeReader = AsyncPipeReader::newReader(
-          eventBase_, NetworkSocket(native_handle).toFd());
+      auto asyncPipeReader =
+          AsyncPipeReader::newReader(eventBase_, NetworkSocket(native_handle));
       auto asyncPipeReaderPtr = asyncPipeReader.get();
       if (!asyncOperationFinishCallback_) {
         asyncOperationFinishCallback_.reset(

folly/io/async/test/AsyncPipeTest.cpp

@@ -104,8 +104,10 @@ class AsyncPipeTest : public Test {
 
     EXPECT_EQ(::fcntl(pipeFds_[0], F_SETFL, O_NONBLOCK), 0);
     EXPECT_EQ(::fcntl(pipeFds_[1], F_SETFL, O_NONBLOCK), 0);
-    reader_ = folly::AsyncPipeReader::newReader(&eventBase_, pipeFds_[0]);
-    writer_ = folly::AsyncPipeWriter::newWriter(&eventBase_, pipeFds_[1]);
+    reader_ = folly::AsyncPipeReader::newReader(
+        &eventBase_, folly::NetworkSocket::fromFd(pipeFds_[0]));
+    writer_ = folly::AsyncPipeWriter::newWriter(
+        &eventBase_, folly::NetworkSocket::fromFd(pipeFds_[1]));
 
     readCallback_.setMovable(movable);
   }

folly/io/async/test/AsyncSSLSocketTest.cpp

@@ -1402,8 +1402,9 @@ static int customRsaPrivEnc(
   int ret = 0;
   int* retptr = &ret;
 
-  auto asyncPipeWriter =
-      folly::AsyncPipeWriter::newWriter(asyncJobEvb, pipefds[1]);
+  auto hand = folly::NetworkSocket::native_handle_type(pipefds[1]);
+  auto asyncPipeWriter = folly::AsyncPipeWriter::newWriter(
+      asyncJobEvb, folly::NetworkSocket(hand));
 
   asyncJobEvb->runInEventBaseThread([retptr = retptr,
                                      flen = flen,