folly/io/async/AsyncSocket.h

   1 /*
   2  * Copyright 2016 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 #pragma once
  18
  19 #include <folly/Optional.h>
  20 #include <folly/SocketAddress.h>
  21 #include <folly/detail/SocketFastOpen.h>
  22 #include <folly/io/IOBuf.h>
  23 #include <folly/io/ShutdownSocketSet.h>
  24 #include <folly/io/async/AsyncSocketException.h>
  25 #include <folly/io/async/AsyncTimeout.h>
  26 #include <folly/io/async/AsyncTransport.h>
  27 #include <folly/io/async/DelayedDestruction.h>
  28 #include <folly/io/async/EventHandler.h>
  29 #include <folly/portability/Sockets.h>
  30
  31 #include <sys/types.h>
  32
  33 #include <chrono>
  34 #include <memory>
  35 #include <map>
  36
  37 namespace folly {
  38
  39 /**
  40  * A class for performing asynchronous I/O on a socket.
  41  *
  42  * AsyncSocket allows users to asynchronously wait for data on a socket, and
  43  * to asynchronously send data.
  44  *
  45  * The APIs for reading and writing are intentionally asymmetric.  Waiting for
  46  * data to read is a persistent API: a callback is installed, and is notified
  47  * whenever new data is available.  It continues to be notified of new events
  48  * until it is uninstalled.
  49  *
  50  * AsyncSocket does not provide read timeout functionality, because it
  51  * typically cannot determine when the timeout should be active.  Generally, a
  52  * timeout should only be enabled when processing is blocked waiting on data
  53  * from the remote endpoint.  For server sockets, the timeout should not be
  54  * active if the server is currently processing one or more outstanding
  55  * requests for this socket.  For client sockets, the timeout should not be
  56  * active if there are no requests pending on the socket.  Additionally, if a
  57  * client has multiple pending requests, it will ususally want a separate
  58  * timeout for each request, rather than a single read timeout.
  59  *
  60  * The write API is fairly intuitive: a user can request to send a block of
  61  * data, and a callback will be informed once the entire block has been
  62  * transferred to the kernel, or on error.  AsyncSocket does provide a send
  63  * timeout, since most callers want to give up if the remote end stops
  64  * responding and no further progress can be made sending the data.
  65  */
  66
  67 #ifdef _MSC_VER
  68 // We do a dynamic_cast on this, in
  69 // AsyncTransportWrapper::getUnderlyingTransport so be safe and
  70 // force displacements for it. See:
  71 // https://msdn.microsoft.com/en-us/library/7sf3txa8.aspx
  72 #pragma vtordisp(push, 2)
  73 #endif
  74 class AsyncSocket : virtual public AsyncTransportWrapper {
  75  public:
  76   typedef std::unique_ptr<AsyncSocket, Destructor> UniquePtr;
  77
  78   class ConnectCallback {
  79    public:
  80     virtual ~ConnectCallback() = default;
  81
  82     /**
  83      * connectSuccess() will be invoked when the connection has been
  84      * successfully established.
  85      */
  86     virtual void connectSuccess() noexcept = 0;
  87
  88     /**
  89      * connectErr() will be invoked if the connection attempt fails.
  90      *
  91      * @param ex        An exception describing the error that occurred.
  92      */
  93     virtual void connectErr(const AsyncSocketException& ex)
  94       noexcept = 0;
  95   };
  96
  97   explicit AsyncSocket();
  98   /**
  99    * Create a new unconnected AsyncSocket.
 100    *
 101    * connect() must later be called on this socket to establish a connection.
 102    */
 103   explicit AsyncSocket(EventBase* evb);
 104
 105   void setShutdownSocketSet(ShutdownSocketSet* ss);
 106
 107   /**
 108    * Create a new AsyncSocket and begin the connection process.
 109    *
 110    * @param evb             EventBase that will manage this socket.
 111    * @param address         The address to connect to.
 112    * @param connectTimeout  Optional timeout in milliseconds for the connection
 113    *                        attempt.
 114    */
 115   AsyncSocket(EventBase* evb,
 116                const folly::SocketAddress& address,
 117                uint32_t connectTimeout = 0);
 118
 119   /**
 120    * Create a new AsyncSocket and begin the connection process.
 121    *
 122    * @param evb             EventBase that will manage this socket.
 123    * @param ip              IP address to connect to (dotted-quad).
 124    * @param port            Destination port in host byte order.
 125    * @param connectTimeout  Optional timeout in milliseconds for the connection
 126    *                        attempt.
 127    */
 128   AsyncSocket(EventBase* evb,
 129                const std::string& ip,
 130                uint16_t port,
 131                uint32_t connectTimeout = 0);
 132
 133   /**
 134    * Create a AsyncSocket from an already connected socket file descriptor.
 135    *
 136    * Note that while AsyncSocket enables TCP_NODELAY for sockets it creates
 137    * when connecting, it does not change the socket options when given an
 138    * existing file descriptor.  If callers want TCP_NODELAY enabled when using
 139    * this version of the constructor, they need to explicitly call
 140    * setNoDelay(true) after the constructor returns.
 141    *
 142    * @param evb EventBase that will manage this socket.
 143    * @param fd  File descriptor to take over (should be a connected socket).
 144    */
 145   AsyncSocket(EventBase* evb, int fd);
 146
 147   /**
 148    * Helper function to create a shared_ptr<AsyncSocket>.
 149    *
 150    * This passes in the correct destructor object, since AsyncSocket's
 151    * destructor is protected and cannot be invoked directly.
 152    */
 153   static std::shared_ptr<AsyncSocket> newSocket(EventBase* evb) {
 154     return std::shared_ptr<AsyncSocket>(new AsyncSocket(evb),
 155                                            Destructor());
 156   }
 157
 158   /**
 159    * Helper function to create a shared_ptr<AsyncSocket>.
 160    */
 161   static std::shared_ptr<AsyncSocket> newSocket(
 162       EventBase* evb,
 163       const folly::SocketAddress& address,
 164       uint32_t connectTimeout = 0) {
 165     return std::shared_ptr<AsyncSocket>(
 166         new AsyncSocket(evb, address, connectTimeout),
 167         Destructor());
 168   }
 169
 170   /**
 171    * Helper function to create a shared_ptr<AsyncSocket>.
 172    */
 173   static std::shared_ptr<AsyncSocket> newSocket(
 174       EventBase* evb,
 175       const std::string& ip,
 176       uint16_t port,
 177       uint32_t connectTimeout = 0) {
 178     return std::shared_ptr<AsyncSocket>(
 179         new AsyncSocket(evb, ip, port, connectTimeout),
 180         Destructor());
 181   }
 182
 183   /**
 184    * Helper function to create a shared_ptr<AsyncSocket>.
 185    */
 186   static std::shared_ptr<AsyncSocket> newSocket(EventBase* evb, int fd) {
 187     return std::shared_ptr<AsyncSocket>(new AsyncSocket(evb, fd),
 188                                            Destructor());
 189   }
 190
 191   /**
 192    * Destroy the socket.
 193    *
 194    * AsyncSocket::destroy() must be called to destroy the socket.
 195    * The normal destructor is private, and should not be invoked directly.
 196    * This prevents callers from deleting a AsyncSocket while it is invoking a
 197    * callback.
 198    */
 199   virtual void destroy() override;
 200
 201   /**
 202    * Get the EventBase used by this socket.
 203    */
 204   EventBase* getEventBase() const override {
 205     return eventBase_;
 206   }
 207
 208   /**
 209    * Get the file descriptor used by the AsyncSocket.
 210    */
 211   virtual int getFd() const {
 212     return fd_;
 213   }
 214
 215   /**
 216    * Extract the file descriptor from the AsyncSocket.
 217    *
 218    * This will immediately cause any installed callbacks to be invoked with an
 219    * error.  The AsyncSocket may no longer be used after the file descriptor
 220    * has been extracted.
 221    *
 222    * Returns the file descriptor.  The caller assumes ownership of the
 223    * descriptor, and it will not be closed when the AsyncSocket is destroyed.
 224    */
 225   virtual int detachFd();
 226
 227   /**
 228    * Uniquely identifies a handle to a socket option value. Each
 229    * combination of level and option name corresponds to one socket
 230    * option value.
 231    */
 232   class OptionKey {
 233    public:
 234     bool operator<(const OptionKey& other) const {
 235       if (level == other.level) {
 236         return optname < other.optname;
 237       }
 238       return level < other.level;
 239     }
 240     int apply(int fd, int val) const {
 241       return setsockopt(fd, level, optname, &val, sizeof(val));
 242     }
 243     int level;
 244     int optname;
 245   };
 246
 247   // Maps from a socket option key to its value
 248   typedef std::map<OptionKey, int> OptionMap;
 249
 250   static const OptionMap emptyOptionMap;
 251   static const folly::SocketAddress& anyAddress();
 252
 253   /**
 254    * Initiate a connection.
 255    *
 256    * @param callback  The callback to inform when the connection attempt
 257    *                  completes.
 258    * @param address   The address to connect to.
 259    * @param timeout   A timeout value, in milliseconds.  If the connection
 260    *                  does not succeed within this period,
 261    *                  callback->connectError() will be invoked.
 262    */
 263   virtual void connect(
 264       ConnectCallback* callback,
 265       const folly::SocketAddress& address,
 266       int timeout = 0,
 267       const OptionMap& options = emptyOptionMap,
 268       const folly::SocketAddress& bindAddr = anyAddress()) noexcept;
 269
 270   void connect(
 271       ConnectCallback* callback,
 272       const std::string& ip,
 273       uint16_t port,
 274       int timeout = 0,
 275       const OptionMap& options = emptyOptionMap) noexcept;
 276
 277   /**
 278    * If a connect request is in-flight, cancels it and closes the socket
 279    * immediately. Otherwise, this is a no-op.
 280    *
 281    * This does not invoke any connection related callbacks. Call this to
 282    * prevent any connect callback while cleaning up, etc.
 283    */
 284   void cancelConnect();
 285
 286   /**
 287    * Set the send timeout.
 288    *
 289    * If write requests do not make any progress for more than the specified
 290    * number of milliseconds, fail all pending writes and close the socket.
 291    *
 292    * If write requests are currently pending when setSendTimeout() is called,
 293    * the timeout interval is immediately restarted using the new value.
 294    *
 295    * (See the comments for AsyncSocket for an explanation of why AsyncSocket
 296    * provides setSendTimeout() but not setRecvTimeout().)
 297    *
 298    * @param milliseconds  The timeout duration, in milliseconds.  If 0, no
 299    *                      timeout will be used.
 300    */
 301   void setSendTimeout(uint32_t milliseconds) override;
 302
 303   /**
 304    * Get the send timeout.
 305    *
 306    * @return Returns the current send timeout, in milliseconds.  A return value
 307    *         of 0 indicates that no timeout is set.
 308    */
 309   uint32_t getSendTimeout() const override {
 310     return sendTimeout_;
 311   }
 312
 313   /**
 314    * Set the maximum number of reads to execute from the underlying
 315    * socket each time the EventBase detects that new ingress data is
 316    * available. The default is unlimited, but callers can use this method
 317    * to limit the amount of data read from the socket per event loop
 318    * iteration.
 319    *
 320    * @param maxReads  Maximum number of reads per data-available event;
 321    *                  a value of zero means unlimited.
 322    */
 323   void setMaxReadsPerEvent(uint16_t maxReads) {
 324     maxReadsPerEvent_ = maxReads;
 325   }
 326
 327   /**
 328    * Get the maximum number of reads this object will execute from
 329    * the underlying socket each time the EventBase detects that new
 330    * ingress data is available.
 331    *
 332    * @returns Maximum number of reads per data-available event; a value
 333    *          of zero means unlimited.
 334    */
 335   uint16_t getMaxReadsPerEvent() const {
 336     return maxReadsPerEvent_;
 337   }
 338
 339   // Read and write methods
 340   void setReadCB(ReadCallback* callback) override;
 341   ReadCallback* getReadCallback() const override;
 342
 343   void write(WriteCallback* callback, const void* buf, size_t bytes,
 344              WriteFlags flags = WriteFlags::NONE) override;
 345   void writev(WriteCallback* callback, const iovec* vec, size_t count,
 346               WriteFlags flags = WriteFlags::NONE) override;
 347   void writeChain(WriteCallback* callback,
 348                   std::unique_ptr<folly::IOBuf>&& buf,
 349                   WriteFlags flags = WriteFlags::NONE) override;
 350
 351   class WriteRequest;
 352   virtual void writeRequest(WriteRequest* req);
 353   void writeRequestReady() {
 354     handleWrite();
 355   }
 356
 357   // Methods inherited from AsyncTransport
 358   void close() override;
 359   void closeNow() override;
 360   void closeWithReset() override;
 361   void shutdownWrite() override;
 362   void shutdownWriteNow() override;
 363
 364   bool readable() const override;
 365   bool isPending() const override;
 366   virtual bool hangup() const;
 367   bool good() const override;
 368   bool error() const override;
 369   void attachEventBase(EventBase* eventBase) override;
 370   void detachEventBase() override;
 371   bool isDetachable() const override;
 372
 373   void getLocalAddress(
 374     folly::SocketAddress* address) const override;
 375   void getPeerAddress(
 376     folly::SocketAddress* address) const override;
 377
 378   bool isEorTrackingEnabled() const override { return false; }
 379
 380   void setEorTracking(bool /*track*/) override {}
 381
 382   bool connecting() const override {
 383     return (state_ == StateEnum::CONNECTING);
 384   }
 385
 386   virtual bool isClosedByPeer() const {
 387     return (state_ == StateEnum::CLOSED &&
 388             (readErr_ == READ_EOF || readErr_ == READ_ERROR));
 389   }
 390
 391   virtual bool isClosedBySelf() const {
 392     return (state_ == StateEnum::CLOSED &&
 393             (readErr_ != READ_EOF && readErr_ != READ_ERROR));
 394   }
 395
 396   size_t getAppBytesWritten() const override {
 397     return appBytesWritten_;
 398   }
 399
 400   size_t getRawBytesWritten() const override {
 401     return getAppBytesWritten();
 402   }
 403
 404   size_t getAppBytesReceived() const override {
 405     return appBytesReceived_;
 406   }
 407
 408   size_t getRawBytesReceived() const override {
 409     return getAppBytesReceived();
 410   }
 411
 412   std::chrono::nanoseconds getConnectTime() const {
 413     return connectEndTime_ - connectStartTime_;
 414   }
 415
 416   std::chrono::milliseconds getConnectTimeout() const {
 417     return connectTimeout_;
 418   }
 419
 420   bool getTFOAttempted() const {
 421     return tfoAttempted_;
 422   }
 423
 424   /**
 425    * Returns whether or not the attempt to use TFO
 426    * finished successfully. This does not necessarily
 427    * mean TFO worked, just that trying to use TFO
 428    * succeeded.
 429    */
 430   bool getTFOFinished() const {
 431     return tfoFinished_;
 432   }
 433
 434   // Methods controlling socket options
 435
 436   /**
 437    * Force writes to be transmitted immediately.
 438    *
 439    * This controls the TCP_NODELAY socket option.  When enabled, TCP segments
 440    * are sent as soon as possible, even if it is not a full frame of data.
 441    * When disabled, the data may be buffered briefly to try and wait for a full
 442    * frame of data.
 443    *
 444    * By default, TCP_NODELAY is enabled for AsyncSocket objects.
 445    *
 446    * This method will fail if the socket is not currently open.
 447    *
 448    * @return Returns 0 if the TCP_NODELAY flag was successfully updated,
 449    *         or a non-zero errno value on error.
 450    */
 451   int setNoDelay(bool noDelay);
 452
 453
 454   /**
 455    * Set the FD_CLOEXEC flag so that the socket will be closed if the program
 456    * later forks and execs.
 457    */
 458   void setCloseOnExec();
 459
 460   /*
 461    * Set the Flavor of Congestion Control to be used for this Socket
 462    * Please check '/lib/modules/<kernel>/kernel/net/ipv4' for tcp_*.ko
 463    * first to make sure the module is available for plugging in
 464    * Alternatively you can choose from net.ipv4.tcp_allowed_congestion_control
 465    */
 466   int setCongestionFlavor(const std::string &cname);
 467
 468   /*
 469    * Forces ACKs to be sent immediately
 470    *
 471    * @return Returns 0 if the TCP_QUICKACK flag was successfully updated,
 472    *         or a non-zero errno value on error.
 473    */
 474   int setQuickAck(bool quickack);
 475
 476   /**
 477    * Set the send bufsize
 478    */
 479   int setSendBufSize(size_t bufsize);
 480
 481   /**
 482    * Set the recv bufsize
 483    */
 484   int setRecvBufSize(size_t bufsize);
 485
 486   /**
 487    * Sets a specific tcp personality
 488    * Available only on kernels 3.2 and greater
 489    */
 490   #define SO_SET_NAMESPACE        41
 491   int setTCPProfile(int profd);
 492
 493   /**
 494    * Generic API for reading a socket option.
 495    *
 496    * @param level     same as the "level" parameter in getsockopt().
 497    * @param optname   same as the "optname" parameter in getsockopt().
 498    * @param optval    pointer to the variable in which the option value should
 499    *                  be returned.
 500    * @param optlen    value-result argument, initially containing the size of
 501    *                  the buffer pointed to by optval, and modified on return
 502    *                  to indicate the actual size of the value returned.
 503    * @return          same as the return value of getsockopt().
 504    */
 505   template <typename T>
 506   int getSockOpt(int level, int optname, T* optval, socklen_t* optlen) {
 507     return getsockopt(fd_, level, optname, (void*) optval, optlen);
 508   }
 509
 510   /**
 511    * Generic API for setting a socket option.
 512    *
 513    * @param level     same as the "level" parameter in getsockopt().
 514    * @param optname   same as the "optname" parameter in getsockopt().
 515    * @param optval    the option value to set.
 516    * @return          same as the return value of setsockopt().
 517    */
 518   template <typename T>
 519   int setSockOpt(int  level,  int  optname,  const T *optval) {
 520     return setsockopt(fd_, level, optname, optval, sizeof(T));
 521   }
 522
 523   virtual void setPeek(bool peek) {
 524     peek_ = peek;
 525   }
 526
 527   /**
 528    * Enables TFO behavior on the AsyncSocket if FOLLY_ALLOW_TFO
 529    * is set.
 530    */
 531   void enableTFO() {
 532     // No-op if folly does not allow tfo
 533 #if FOLLY_ALLOW_TFO
 534     tfoEnabled_ = true;
 535 #endif
 536   }
 537
 538   enum class StateEnum : uint8_t {
 539     UNINIT,
 540     CONNECTING,
 541     ESTABLISHED,
 542     CLOSED,
 543     ERROR,
 544     FAST_OPEN,
 545   };
 546
 547   void setBufferCallback(BufferCallback* cb);
 548
 549   /**
 550    * writeReturn is the total number of bytes written, or WRITE_ERROR on error.
 551    * If no data has been written, 0 is returned.
 552    * exception is a more specific exception that cause a write error.
 553    * Not all writes have exceptions associated with them thus writeReturn
 554    * should be checked to determine whether the operation resulted in an error.
 555    */
 556   struct WriteResult {
 557     explicit WriteResult(ssize_t ret) : writeReturn(ret) {}
 558
 559     WriteResult(ssize_t ret, std::unique_ptr<const AsyncSocketException> e)
 560         : writeReturn(ret), exception(std::move(e)) {}
 561
 562     ssize_t writeReturn;
 563     std::unique_ptr<const AsyncSocketException> exception;
 564   };
 565
 566   /**
 567    * readReturn is the number of bytes read, or READ_EOF on EOF, or
 568    * READ_ERROR on error, or READ_BLOCKING if the operation will
 569    * block.
 570    * exception is a more specific exception that may have caused a read error.
 571    * Not all read errors have exceptions associated with them thus readReturn
 572    * should be checked to determine whether the operation resulted in an error.
 573    */
 574   struct ReadResult {
 575     explicit ReadResult(ssize_t ret) : readReturn(ret) {}
 576
 577     ReadResult(ssize_t ret, std::unique_ptr<const AsyncSocketException> e)
 578         : readReturn(ret), exception(std::move(e)) {}
 579
 580     ssize_t readReturn;
 581     std::unique_ptr<const AsyncSocketException> exception;
 582   };
 583
 584   /**
 585    * A WriteRequest object tracks information about a pending write operation.
 586    */
 587   class WriteRequest {
 588    public:
 589     WriteRequest(AsyncSocket* socket, WriteCallback* callback) :
 590       socket_(socket), callback_(callback) {}
 591
 592     virtual void start() {};
 593
 594     virtual void destroy() = 0;
 595
 596     virtual WriteResult performWrite() = 0;
 597
 598     virtual void consume() = 0;
 599
 600     virtual bool isComplete() = 0;
 601
 602     WriteRequest* getNext() const {
 603       return next_;
 604     }
 605
 606     WriteCallback* getCallback() const {
 607       return callback_;
 608     }
 609
 610     uint32_t getTotalBytesWritten() const {
 611       return totalBytesWritten_;
 612     }
 613
 614     void append(WriteRequest* next) {
 615       assert(next_ == nullptr);
 616       next_ = next;
 617     }
 618
 619     void fail(const char* fn, const AsyncSocketException& ex) {
 620       socket_->failWrite(fn, ex);
 621     }
 622
 623     void bytesWritten(size_t count) {
 624       totalBytesWritten_ += count;
 625       socket_->appBytesWritten_ += count;
 626     }
 627
 628    protected:
 629     // protected destructor, to ensure callers use destroy()
 630     virtual ~WriteRequest() {}
 631
 632     AsyncSocket* socket_;         ///< parent socket
 633     WriteRequest* next_{nullptr};          ///< pointer to next WriteRequest
 634     WriteCallback* callback_;     ///< completion callback
 635     uint32_t totalBytesWritten_{0};  ///< total bytes written
 636   };
 637
 638  protected:
 639   enum ReadResultEnum {
 640     READ_EOF = 0,
 641     READ_ERROR = -1,
 642     READ_BLOCKING = -2,
 643     READ_NO_ERROR = -3,
 644   };
 645
 646   enum WriteResultEnum {
 647     WRITE_ERROR = -1,
 648   };
 649
 650   /**
 651    * Protected destructor.
 652    *
 653    * Users of AsyncSocket must never delete it directly.  Instead, invoke
 654    * destroy() instead.  (See the documentation in DelayedDestruction.h for
 655    * more details.)
 656    */
 657   ~AsyncSocket();
 658
 659   friend std::ostream& operator << (std::ostream& os, const StateEnum& state);
 660
 661   enum ShutdownFlags {
 662     /// shutdownWrite() called, but we are still waiting on writes to drain
 663     SHUT_WRITE_PENDING = 0x01,
 664     /// writes have been completely shut down
 665     SHUT_WRITE = 0x02,
 666     /**
 667      * Reads have been shutdown.
 668      *
 669      * At the moment we don't distinguish between remote read shutdown
 670      * (received EOF from the remote end) and local read shutdown.  We can
 671      * only receive EOF when a read callback is set, and we immediately inform
 672      * it of the EOF.  Therefore there doesn't seem to be any reason to have a
 673      * separate state of "received EOF but the local side may still want to
 674      * read".
 675      *
 676      * We also don't currently provide any API for only shutting down the read
 677      * side of a socket.  (This is a no-op as far as TCP is concerned, anyway.)
 678      */
 679     SHUT_READ = 0x04,
 680   };
 681
 682   class BytesWriteRequest;
 683
 684   class WriteTimeout : public AsyncTimeout {
 685    public:
 686     WriteTimeout(AsyncSocket* socket, EventBase* eventBase)
 687       : AsyncTimeout(eventBase)
 688       , socket_(socket) {}
 689
 690     virtual void timeoutExpired() noexcept {
 691       socket_->timeoutExpired();
 692     }
 693
 694    private:
 695     AsyncSocket* socket_;
 696   };
 697
 698   class IoHandler : public EventHandler {
 699    public:
 700     IoHandler(AsyncSocket* socket, EventBase* eventBase)
 701       : EventHandler(eventBase, -1)
 702       , socket_(socket) {}
 703     IoHandler(AsyncSocket* socket, EventBase* eventBase, int fd)
 704       : EventHandler(eventBase, fd)
 705       , socket_(socket) {}
 706
 707     virtual void handlerReady(uint16_t events) noexcept {
 708       socket_->ioReady(events);
 709     }
 710
 711    private:
 712     AsyncSocket* socket_;
 713   };
 714
 715   void init();
 716
 717   class ImmediateReadCB : public folly::EventBase::LoopCallback {
 718    public:
 719     explicit ImmediateReadCB(AsyncSocket* socket) : socket_(socket) {}
 720     void runLoopCallback() noexcept override {
 721       DestructorGuard dg(socket_);
 722       socket_->checkForImmediateRead();
 723     }
 724    private:
 725     AsyncSocket* socket_;
 726   };
 727
 728   /**
 729    * Schedule checkForImmediateRead to be executed in the next loop
 730    * iteration.
 731    */
 732   void scheduleImmediateRead() noexcept {
 733     if (good()) {
 734       eventBase_->runInLoop(&immediateReadHandler_);
 735     }
 736   }
 737
 738   /**
 739    * Schedule handleInitalReadWrite to run in the next iteration.
 740    */
 741   void scheduleInitialReadWrite() noexcept {
 742     if (good()) {
 743       DestructorGuard dg(this);
 744       eventBase_->runInLoop([this, dg] {
 745         if (good()) {
 746           handleInitialReadWrite();
 747         }
 748       });
 749     }
 750   }
 751
 752   // event notification methods
 753   void ioReady(uint16_t events) noexcept;
 754   virtual void checkForImmediateRead() noexcept;
 755   virtual void handleInitialReadWrite() noexcept;
 756   virtual void prepareReadBuffer(void** buf, size_t* buflen) noexcept;
 757   virtual void handleRead() noexcept;
 758   virtual void handleWrite() noexcept;
 759   virtual void handleConnect() noexcept;
 760   void timeoutExpired() noexcept;
 761
 762   /**
 763    * Attempt to read from the socket.
 764    *
 765    * @param buf      The buffer to read data into.
 766    * @param buflen   The length of the buffer.
 767    *
 768    * @return Returns a read result. See read result for details.
 769    */
 770   virtual ReadResult performRead(void** buf, size_t* buflen, size_t* offset);
 771
 772   /**
 773    * Populate an iovec array from an IOBuf and attempt to write it.
 774    *
 775    * @param callback Write completion/error callback.
 776    * @param vec      Target iovec array; caller retains ownership.
 777    * @param count    Number of IOBufs to write, beginning at start of buf.
 778    * @param buf      Chain of iovecs.
 779    * @param flags    set of flags for the underlying write calls, like cork
 780    */
 781   void writeChainImpl(WriteCallback* callback, iovec* vec,
 782                       size_t count, std::unique_ptr<folly::IOBuf>&& buf,
 783                       WriteFlags flags);
 784
 785   /**
 786    * Write as much data as possible to the socket without blocking,
 787    * and queue up any leftover data to send when the socket can
 788    * handle writes again.
 789    *
 790    * @param callback The callback to invoke when the write is completed.
 791    * @param vec      Array of buffers to write; this method will make a
 792    *                 copy of the vector (but not the buffers themselves)
 793    *                 if the write has to be completed asynchronously.
 794    * @param count    Number of elements in vec.
 795    * @param buf      The IOBuf that manages the buffers referenced by
 796    *                 vec, or a pointer to nullptr if the buffers are not
 797    *                 associated with an IOBuf.  Note that ownership of
 798    *                 the IOBuf is transferred here; upon completion of
 799    *                 the write, the AsyncSocket deletes the IOBuf.
 800    * @param flags    Set of write flags.
 801    */
 802   void writeImpl(WriteCallback* callback, const iovec* vec, size_t count,
 803                  std::unique_ptr<folly::IOBuf>&& buf,
 804                  WriteFlags flags = WriteFlags::NONE);
 805
 806   /**
 807    * Attempt to write to the socket.
 808    *
 809    * @param vec             The iovec array pointing to the buffers to write.
 810    * @param count           The length of the iovec array.
 811    * @param flags           Set of write flags.
 812    * @param countWritten    On return, the value pointed to by this parameter
 813    *                          will contain the number of iovec entries that were
 814    *                          fully written.
 815    * @param partialWritten  On return, the value pointed to by this parameter
 816    *                          will contain the number of bytes written in the
 817    *                          partially written iovec entry.
 818    *
 819    * @return Returns a WriteResult. See WriteResult for more details.
 820    */
 821   virtual WriteResult performWrite(
 822       const iovec* vec,
 823       uint32_t count,
 824       WriteFlags flags,
 825       uint32_t* countWritten,
 826       uint32_t* partialWritten);
 827
 828   /**
 829    * Sends the message over the socket using sendmsg
 830    *
 831    * @param msg       Message to send
 832    * @param msg_flags Flags to pass to sendmsg
 833    */
 834   AsyncSocket::WriteResult
 835   sendSocketMessage(int fd, struct msghdr* msg, int msg_flags);
 836
 837   virtual ssize_t tfoSendMsg(int fd, struct msghdr* msg, int msg_flags);
 838
 839   int socketConnect(const struct sockaddr* addr, socklen_t len);
 840
 841   virtual void scheduleConnectTimeout();
 842   void registerForConnectEvents();
 843
 844   bool updateEventRegistration();
 845
 846   /**
 847    * Update event registration.
 848    *
 849    * @param enable Flags of events to enable. Set it to 0 if no events
 850    * need to be enabled in this call.
 851    * @param disable Flags of events
 852    * to disable. Set it to 0 if no events need to be disabled in this
 853    * call.
 854    *
 855    * @return true iff the update is successful.
 856    */
 857   bool updateEventRegistration(uint16_t enable, uint16_t disable);
 858
 859   // Actually close the file descriptor and set it to -1 so we don't
 860   // accidentally close it again.
 861   void doClose();
 862
 863   // error handling methods
 864   void startFail();
 865   void finishFail();
 866   void fail(const char* fn, const AsyncSocketException& ex);
 867   void failConnect(const char* fn, const AsyncSocketException& ex);
 868   void failRead(const char* fn, const AsyncSocketException& ex);
 869   void failWrite(const char* fn, WriteCallback* callback, size_t bytesWritten,
 870                  const AsyncSocketException& ex);
 871   void failWrite(const char* fn, const AsyncSocketException& ex);
 872   void failAllWrites(const AsyncSocketException& ex);
 873   virtual void invokeConnectErr(const AsyncSocketException& ex);
 874   virtual void invokeConnectSuccess();
 875   void invalidState(ConnectCallback* callback);
 876   void invalidState(ReadCallback* callback);
 877   void invalidState(WriteCallback* callback);
 878
 879   std::string withAddr(const std::string& s);
 880
 881   StateEnum state_;                     ///< StateEnum describing current state
 882   uint8_t shutdownFlags_;               ///< Shutdown state (ShutdownFlags)
 883   uint16_t eventFlags_;                 ///< EventBase::HandlerFlags settings
 884   int fd_;                              ///< The socket file descriptor
 885   mutable folly::SocketAddress addr_;    ///< The address we tried to connect to
 886   mutable folly::SocketAddress localAddr_;
 887                                         ///< The address we are connecting from
 888   uint32_t sendTimeout_;                ///< The send timeout, in milliseconds
 889   uint16_t maxReadsPerEvent_;           ///< Max reads per event loop iteration
 890   EventBase* eventBase_;                ///< The EventBase
 891   WriteTimeout writeTimeout_;           ///< A timeout for connect and write
 892   IoHandler ioHandler_;                 ///< A EventHandler to monitor the fd
 893   ImmediateReadCB immediateReadHandler_; ///< LoopCallback for checking read
 894
 895   ConnectCallback* connectCallback_;    ///< ConnectCallback
 896   ReadCallback* readCallback_;          ///< ReadCallback
 897   WriteRequest* writeReqHead_;          ///< Chain of WriteRequests
 898   WriteRequest* writeReqTail_;          ///< End of WriteRequest chain
 899   ShutdownSocketSet* shutdownSocketSet_;
 900   size_t appBytesReceived_;             ///< Num of bytes received from socket
 901   size_t appBytesWritten_;              ///< Num of bytes written to socket
 902   bool isBufferMovable_{false};
 903
 904   bool peek_{false}; // Peek bytes.
 905
 906   int8_t readErr_{READ_NO_ERROR};      ///< The read error encountered, if any.
 907
 908   std::chrono::steady_clock::time_point connectStartTime_;
 909   std::chrono::steady_clock::time_point connectEndTime_;
 910
 911   std::chrono::milliseconds connectTimeout_{0};
 912
 913   BufferCallback* bufferCallback_{nullptr};
 914   bool tfoEnabled_{false};
 915   bool tfoAttempted_{false};
 916   bool tfoFinished_{false};
 917 };
 918 #ifdef _MSC_VER
 919 #pragma vtordisp(pop)
 920 #endif
 921
 922 } // folly