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