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   /**
 435    * Returns whether or not TFO "worked", or, succeeded
 436    * in actually being used.
 437    */
 438   bool getTFOSucceeded() const {
 439     return tfoSucceeded_;
 440   }
 441
 442   // Methods controlling socket options
 443
 444   /**
 445    * Force writes to be transmitted immediately.
 446    *
 447    * This controls the TCP_NODELAY socket option.  When enabled, TCP segments
 448    * are sent as soon as possible, even if it is not a full frame of data.
 449    * When disabled, the data may be buffered briefly to try and wait for a full
 450    * frame of data.
 451    *
 452    * By default, TCP_NODELAY is enabled for AsyncSocket objects.
 453    *
 454    * This method will fail if the socket is not currently open.
 455    *
 456    * @return Returns 0 if the TCP_NODELAY flag was successfully updated,
 457    *         or a non-zero errno value on error.
 458    */
 459   int setNoDelay(bool noDelay);
 460
 461
 462   /**
 463    * Set the FD_CLOEXEC flag so that the socket will be closed if the program
 464    * later forks and execs.
 465    */
 466   void setCloseOnExec();
 467
 468   /*
 469    * Set the Flavor of Congestion Control to be used for this Socket
 470    * Please check '/lib/modules/<kernel>/kernel/net/ipv4' for tcp_*.ko
 471    * first to make sure the module is available for plugging in
 472    * Alternatively you can choose from net.ipv4.tcp_allowed_congestion_control
 473    */
 474   int setCongestionFlavor(const std::string &cname);
 475
 476   /*
 477    * Forces ACKs to be sent immediately
 478    *
 479    * @return Returns 0 if the TCP_QUICKACK flag was successfully updated,
 480    *         or a non-zero errno value on error.
 481    */
 482   int setQuickAck(bool quickack);
 483
 484   /**
 485    * Set the send bufsize
 486    */
 487   int setSendBufSize(size_t bufsize);
 488
 489   /**
 490    * Set the recv bufsize
 491    */
 492   int setRecvBufSize(size_t bufsize);
 493
 494   /**
 495    * Sets a specific tcp personality
 496    * Available only on kernels 3.2 and greater
 497    */
 498   #define SO_SET_NAMESPACE        41
 499   int setTCPProfile(int profd);
 500
 501   /**
 502    * Generic API for reading a socket option.
 503    *
 504    * @param level     same as the "level" parameter in getsockopt().
 505    * @param optname   same as the "optname" parameter in getsockopt().
 506    * @param optval    pointer to the variable in which the option value should
 507    *                  be returned.
 508    * @param optlen    value-result argument, initially containing the size of
 509    *                  the buffer pointed to by optval, and modified on return
 510    *                  to indicate the actual size of the value returned.
 511    * @return          same as the return value of getsockopt().
 512    */
 513   template <typename T>
 514   int getSockOpt(int level, int optname, T* optval, socklen_t* optlen) {
 515     return getsockopt(fd_, level, optname, (void*) optval, optlen);
 516   }
 517
 518   /**
 519    * Generic API for setting a socket option.
 520    *
 521    * @param level     same as the "level" parameter in getsockopt().
 522    * @param optname   same as the "optname" parameter in getsockopt().
 523    * @param optval    the option value to set.
 524    * @return          same as the return value of setsockopt().
 525    */
 526   template <typename T>
 527   int setSockOpt(int  level,  int  optname,  const T *optval) {
 528     return setsockopt(fd_, level, optname, optval, sizeof(T));
 529   }
 530
 531   virtual void setPeek(bool peek) {
 532     peek_ = peek;
 533   }
 534
 535   /**
 536    * Enables TFO behavior on the AsyncSocket if FOLLY_ALLOW_TFO
 537    * is set.
 538    */
 539   void enableTFO() {
 540     // No-op if folly does not allow tfo
 541 #if FOLLY_ALLOW_TFO
 542     tfoEnabled_ = true;
 543 #endif
 544   }
 545
 546   enum class StateEnum : uint8_t {
 547     UNINIT,
 548     CONNECTING,
 549     ESTABLISHED,
 550     CLOSED,
 551     ERROR,
 552     FAST_OPEN,
 553   };
 554
 555   void setBufferCallback(BufferCallback* cb);
 556
 557   /**
 558    * writeReturn is the total number of bytes written, or WRITE_ERROR on error.
 559    * If no data has been written, 0 is returned.
 560    * exception is a more specific exception that cause a write error.
 561    * Not all writes have exceptions associated with them thus writeReturn
 562    * should be checked to determine whether the operation resulted in an error.
 563    */
 564   struct WriteResult {
 565     explicit WriteResult(ssize_t ret) : writeReturn(ret) {}
 566
 567     WriteResult(ssize_t ret, std::unique_ptr<const AsyncSocketException> e)
 568         : writeReturn(ret), exception(std::move(e)) {}
 569
 570     ssize_t writeReturn;
 571     std::unique_ptr<const AsyncSocketException> exception;
 572   };
 573
 574   /**
 575    * readReturn is the number of bytes read, or READ_EOF on EOF, or
 576    * READ_ERROR on error, or READ_BLOCKING if the operation will
 577    * block.
 578    * exception is a more specific exception that may have caused a read error.
 579    * Not all read errors have exceptions associated with them thus readReturn
 580    * should be checked to determine whether the operation resulted in an error.
 581    */
 582   struct ReadResult {
 583     explicit ReadResult(ssize_t ret) : readReturn(ret) {}
 584
 585     ReadResult(ssize_t ret, std::unique_ptr<const AsyncSocketException> e)
 586         : readReturn(ret), exception(std::move(e)) {}
 587
 588     ssize_t readReturn;
 589     std::unique_ptr<const AsyncSocketException> exception;
 590   };
 591
 592   /**
 593    * A WriteRequest object tracks information about a pending write operation.
 594    */
 595   class WriteRequest {
 596    public:
 597     WriteRequest(AsyncSocket* socket, WriteCallback* callback) :
 598       socket_(socket), callback_(callback) {}
 599
 600     virtual void start() {};
 601
 602     virtual void destroy() = 0;
 603
 604     virtual WriteResult performWrite() = 0;
 605
 606     virtual void consume() = 0;
 607
 608     virtual bool isComplete() = 0;
 609
 610     WriteRequest* getNext() const {
 611       return next_;
 612     }
 613
 614     WriteCallback* getCallback() const {
 615       return callback_;
 616     }
 617
 618     uint32_t getTotalBytesWritten() const {
 619       return totalBytesWritten_;
 620     }
 621
 622     void append(WriteRequest* next) {
 623       assert(next_ == nullptr);
 624       next_ = next;
 625     }
 626
 627     void fail(const char* fn, const AsyncSocketException& ex) {
 628       socket_->failWrite(fn, ex);
 629     }
 630
 631     void bytesWritten(size_t count) {
 632       totalBytesWritten_ += count;
 633       socket_->appBytesWritten_ += count;
 634     }
 635
 636    protected:
 637     // protected destructor, to ensure callers use destroy()
 638     virtual ~WriteRequest() {}
 639
 640     AsyncSocket* socket_;         ///< parent socket
 641     WriteRequest* next_{nullptr};          ///< pointer to next WriteRequest
 642     WriteCallback* callback_;     ///< completion callback
 643     uint32_t totalBytesWritten_{0};  ///< total bytes written
 644   };
 645
 646  protected:
 647   enum ReadResultEnum {
 648     READ_EOF = 0,
 649     READ_ERROR = -1,
 650     READ_BLOCKING = -2,
 651     READ_NO_ERROR = -3,
 652   };
 653
 654   enum WriteResultEnum {
 655     WRITE_ERROR = -1,
 656   };
 657
 658   /**
 659    * Protected destructor.
 660    *
 661    * Users of AsyncSocket must never delete it directly.  Instead, invoke
 662    * destroy() instead.  (See the documentation in DelayedDestruction.h for
 663    * more details.)
 664    */
 665   ~AsyncSocket();
 666
 667   friend std::ostream& operator << (std::ostream& os, const StateEnum& state);
 668
 669   enum ShutdownFlags {
 670     /// shutdownWrite() called, but we are still waiting on writes to drain
 671     SHUT_WRITE_PENDING = 0x01,
 672     /// writes have been completely shut down
 673     SHUT_WRITE = 0x02,
 674     /**
 675      * Reads have been shutdown.
 676      *
 677      * At the moment we don't distinguish between remote read shutdown
 678      * (received EOF from the remote end) and local read shutdown.  We can
 679      * only receive EOF when a read callback is set, and we immediately inform
 680      * it of the EOF.  Therefore there doesn't seem to be any reason to have a
 681      * separate state of "received EOF but the local side may still want to
 682      * read".
 683      *
 684      * We also don't currently provide any API for only shutting down the read
 685      * side of a socket.  (This is a no-op as far as TCP is concerned, anyway.)
 686      */
 687     SHUT_READ = 0x04,
 688   };
 689
 690   class BytesWriteRequest;
 691
 692   class WriteTimeout : public AsyncTimeout {
 693    public:
 694     WriteTimeout(AsyncSocket* socket, EventBase* eventBase)
 695       : AsyncTimeout(eventBase)
 696       , socket_(socket) {}
 697
 698     virtual void timeoutExpired() noexcept {
 699       socket_->timeoutExpired();
 700     }
 701
 702    private:
 703     AsyncSocket* socket_;
 704   };
 705
 706   class IoHandler : public EventHandler {
 707    public:
 708     IoHandler(AsyncSocket* socket, EventBase* eventBase)
 709       : EventHandler(eventBase, -1)
 710       , socket_(socket) {}
 711     IoHandler(AsyncSocket* socket, EventBase* eventBase, int fd)
 712       : EventHandler(eventBase, fd)
 713       , socket_(socket) {}
 714
 715     virtual void handlerReady(uint16_t events) noexcept {
 716       socket_->ioReady(events);
 717     }
 718
 719    private:
 720     AsyncSocket* socket_;
 721   };
 722
 723   void init();
 724
 725   class ImmediateReadCB : public folly::EventBase::LoopCallback {
 726    public:
 727     explicit ImmediateReadCB(AsyncSocket* socket) : socket_(socket) {}
 728     void runLoopCallback() noexcept override {
 729       DestructorGuard dg(socket_);
 730       socket_->checkForImmediateRead();
 731     }
 732    private:
 733     AsyncSocket* socket_;
 734   };
 735
 736   /**
 737    * Schedule checkForImmediateRead to be executed in the next loop
 738    * iteration.
 739    */
 740   void scheduleImmediateRead() noexcept {
 741     if (good()) {
 742       eventBase_->runInLoop(&immediateReadHandler_);
 743     }
 744   }
 745
 746   // event notification methods
 747   void ioReady(uint16_t events) noexcept;
 748   virtual void checkForImmediateRead() noexcept;
 749   virtual void handleInitialReadWrite() noexcept;
 750   virtual void prepareReadBuffer(void** buf, size_t* buflen) noexcept;
 751   virtual void handleRead() noexcept;
 752   virtual void handleWrite() noexcept;
 753   virtual void handleConnect() noexcept;
 754   void timeoutExpired() noexcept;
 755
 756   /**
 757    * Attempt to read from the socket.
 758    *
 759    * @param buf      The buffer to read data into.
 760    * @param buflen   The length of the buffer.
 761    *
 762    * @return Returns a read result. See read result for details.
 763    */
 764   virtual ReadResult performRead(void** buf, size_t* buflen, size_t* offset);
 765
 766   /**
 767    * Populate an iovec array from an IOBuf and attempt to write it.
 768    *
 769    * @param callback Write completion/error callback.
 770    * @param vec      Target iovec array; caller retains ownership.
 771    * @param count    Number of IOBufs to write, beginning at start of buf.
 772    * @param buf      Chain of iovecs.
 773    * @param flags    set of flags for the underlying write calls, like cork
 774    */
 775   void writeChainImpl(WriteCallback* callback, iovec* vec,
 776                       size_t count, std::unique_ptr<folly::IOBuf>&& buf,
 777                       WriteFlags flags);
 778
 779   /**
 780    * Write as much data as possible to the socket without blocking,
 781    * and queue up any leftover data to send when the socket can
 782    * handle writes again.
 783    *
 784    * @param callback The callback to invoke when the write is completed.
 785    * @param vec      Array of buffers to write; this method will make a
 786    *                 copy of the vector (but not the buffers themselves)
 787    *                 if the write has to be completed asynchronously.
 788    * @param count    Number of elements in vec.
 789    * @param buf      The IOBuf that manages the buffers referenced by
 790    *                 vec, or a pointer to nullptr if the buffers are not
 791    *                 associated with an IOBuf.  Note that ownership of
 792    *                 the IOBuf is transferred here; upon completion of
 793    *                 the write, the AsyncSocket deletes the IOBuf.
 794    * @param flags    Set of write flags.
 795    */
 796   void writeImpl(WriteCallback* callback, const iovec* vec, size_t count,
 797                  std::unique_ptr<folly::IOBuf>&& buf,
 798                  WriteFlags flags = WriteFlags::NONE);
 799
 800   /**
 801    * Attempt to write to the socket.
 802    *
 803    * @param vec             The iovec array pointing to the buffers to write.
 804    * @param count           The length of the iovec array.
 805    * @param flags           Set of write flags.
 806    * @param countWritten    On return, the value pointed to by this parameter
 807    *                          will contain the number of iovec entries that were
 808    *                          fully written.
 809    * @param partialWritten  On return, the value pointed to by this parameter
 810    *                          will contain the number of bytes written in the
 811    *                          partially written iovec entry.
 812    *
 813    * @return Returns a WriteResult. See WriteResult for more details.
 814    */
 815   virtual WriteResult performWrite(
 816       const iovec* vec,
 817       uint32_t count,
 818       WriteFlags flags,
 819       uint32_t* countWritten,
 820       uint32_t* partialWritten);
 821
 822   /**
 823    * Sends the message over the socket using sendmsg
 824    *
 825    * @param msg       Message to send
 826    * @param msg_flags Flags to pass to sendmsg
 827    */
 828   AsyncSocket::WriteResult
 829   sendSocketMessage(int fd, struct msghdr* msg, int msg_flags);
 830
 831   virtual ssize_t tfoSendMsg(int fd, struct msghdr* msg, int msg_flags);
 832
 833   int socketConnect(const struct sockaddr* addr, socklen_t len);
 834
 835   void scheduleConnectTimeoutAndRegisterForEvents();
 836
 837   bool updateEventRegistration();
 838
 839   /**
 840    * Update event registration.
 841    *
 842    * @param enable Flags of events to enable. Set it to 0 if no events
 843    * need to be enabled in this call.
 844    * @param disable Flags of events
 845    * to disable. Set it to 0 if no events need to be disabled in this
 846    * call.
 847    *
 848    * @return true iff the update is successful.
 849    */
 850   bool updateEventRegistration(uint16_t enable, uint16_t disable);
 851
 852   // Actually close the file descriptor and set it to -1 so we don't
 853   // accidentally close it again.
 854   void doClose();
 855
 856   // error handling methods
 857   void startFail();
 858   void finishFail();
 859   void fail(const char* fn, const AsyncSocketException& ex);
 860   void failConnect(const char* fn, const AsyncSocketException& ex);
 861   void failRead(const char* fn, const AsyncSocketException& ex);
 862   void failWrite(const char* fn, WriteCallback* callback, size_t bytesWritten,
 863                  const AsyncSocketException& ex);
 864   void failWrite(const char* fn, const AsyncSocketException& ex);
 865   void failAllWrites(const AsyncSocketException& ex);
 866   void invokeConnectErr(const AsyncSocketException& ex);
 867   virtual void invokeConnectSuccess();
 868   void invalidState(ConnectCallback* callback);
 869   void invalidState(ReadCallback* callback);
 870   void invalidState(WriteCallback* callback);
 871
 872   std::string withAddr(const std::string& s);
 873
 874   StateEnum state_;                     ///< StateEnum describing current state
 875   uint8_t shutdownFlags_;               ///< Shutdown state (ShutdownFlags)
 876   uint16_t eventFlags_;                 ///< EventBase::HandlerFlags settings
 877   int fd_;                              ///< The socket file descriptor
 878   mutable folly::SocketAddress addr_;    ///< The address we tried to connect to
 879   mutable folly::SocketAddress localAddr_;
 880                                         ///< The address we are connecting from
 881   uint32_t sendTimeout_;                ///< The send timeout, in milliseconds
 882   uint16_t maxReadsPerEvent_;           ///< Max reads per event loop iteration
 883   EventBase* eventBase_;                ///< The EventBase
 884   WriteTimeout writeTimeout_;           ///< A timeout for connect and write
 885   IoHandler ioHandler_;                 ///< A EventHandler to monitor the fd
 886   ImmediateReadCB immediateReadHandler_; ///< LoopCallback for checking read
 887
 888   ConnectCallback* connectCallback_;    ///< ConnectCallback
 889   ReadCallback* readCallback_;          ///< ReadCallback
 890   WriteRequest* writeReqHead_;          ///< Chain of WriteRequests
 891   WriteRequest* writeReqTail_;          ///< End of WriteRequest chain
 892   ShutdownSocketSet* shutdownSocketSet_;
 893   size_t appBytesReceived_;             ///< Num of bytes received from socket
 894   size_t appBytesWritten_;              ///< Num of bytes written to socket
 895   bool isBufferMovable_{false};
 896
 897   bool peek_{false}; // Peek bytes.
 898
 899   int8_t readErr_{READ_NO_ERROR};      ///< The read error encountered, if any.
 900
 901   std::chrono::steady_clock::time_point connectStartTime_;
 902   std::chrono::steady_clock::time_point connectEndTime_;
 903
 904   std::chrono::milliseconds connectTimeout_{0};
 905
 906   BufferCallback* bufferCallback_{nullptr};
 907   bool tfoEnabled_{false};
 908   bool tfoAttempted_{false};
 909   bool tfoFinished_{false};
 910   bool tfoSucceeded_{false};
 911 };
 912 #ifdef _MSC_VER
 913 #pragma vtordisp(pop)
 914 #endif
 915
 916 } // folly