folly/io/async/AsyncSocket.h

   1 /*
   2  * Copyright 2014-present 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 #pragma once
  17
  18 #include <folly/Optional.h>
  19 #include <folly/SocketAddress.h>
  20 #include <folly/detail/SocketFastOpen.h>
  21 #include <folly/io/IOBuf.h>
  22 #include <folly/io/ShutdownSocketSet.h>
  23 #include <folly/io/async/AsyncSocketException.h>
  24 #include <folly/io/async/AsyncTimeout.h>
  25 #include <folly/io/async/AsyncTransport.h>
  26 #include <folly/io/async/DelayedDestruction.h>
  27 #include <folly/io/async/EventHandler.h>
  28 #include <folly/portability/Sockets.h>
  29
  30 #include <sys/types.h>
  31
  32 #include <chrono>
  33 #include <map>
  34 #include <memory>
  35
  36 namespace folly {
  37
  38 /**
  39  * A class for performing asynchronous I/O on a socket.
  40  *
  41  * AsyncSocket allows users to asynchronously wait for data on a socket, and
  42  * to asynchronously send data.
  43  *
  44  * The APIs for reading and writing are intentionally asymmetric.  Waiting for
  45  * data to read is a persistent API: a callback is installed, and is notified
  46  * whenever new data is available.  It continues to be notified of new events
  47  * until it is uninstalled.
  48  *
  49  * AsyncSocket does not provide read timeout functionality, because it
  50  * typically cannot determine when the timeout should be active.  Generally, a
  51  * timeout should only be enabled when processing is blocked waiting on data
  52  * from the remote endpoint.  For server sockets, the timeout should not be
  53  * active if the server is currently processing one or more outstanding
  54  * requests for this socket.  For client sockets, the timeout should not be
  55  * active if there are no requests pending on the socket.  Additionally, if a
  56  * client has multiple pending requests, it will ususally want a separate
  57  * timeout for each request, rather than a single read timeout.
  58  *
  59  * The write API is fairly intuitive: a user can request to send a block of
  60  * data, and a callback will be informed once the entire block has been
  61  * transferred to the kernel, or on error.  AsyncSocket does provide a send
  62  * timeout, since most callers want to give up if the remote end stops
  63  * responding and no further progress can be made sending the data.
  64  */
  65
  66 #if defined __linux__ && !defined SO_NO_TRANSPARENT_TLS
  67 #define SO_NO_TRANSPARENT_TLS 200
  68 #endif
  69
  70 #if defined __linux__ && !defined SO_NO_TSOCKS
  71 #define SO_NO_TSOCKS 201
  72 #endif
  73
  74 #ifdef _MSC_VER
  75 // We do a dynamic_cast on this, in
  76 // AsyncTransportWrapper::getUnderlyingTransport so be safe and
  77 // force displacements for it. See:
  78 // https://msdn.microsoft.com/en-us/library/7sf3txa8.aspx
  79 #pragma vtordisp(push, 2)
  80 #endif
  81 class AsyncSocket : virtual public AsyncTransportWrapper {
  82  public:
  83   typedef std::unique_ptr<AsyncSocket, Destructor> UniquePtr;
  84
  85   class ConnectCallback {
  86    public:
  87     virtual ~ConnectCallback() = default;
  88
  89     /**
  90      * connectSuccess() will be invoked when the connection has been
  91      * successfully established.
  92      */
  93     virtual void connectSuccess() noexcept = 0;
  94
  95     /**
  96      * connectErr() will be invoked if the connection attempt fails.
  97      *
  98      * @param ex        An exception describing the error that occurred.
  99      */
 100     virtual void connectErr(const AsyncSocketException& ex)
 101       noexcept = 0;
 102   };
 103
 104   class EvbChangeCallback {
 105    public:
 106     virtual ~EvbChangeCallback() = default;
 107
 108     // Called when the socket has been attached to a new EVB
 109     // and is called from within that EVB thread
 110     virtual void evbAttached(AsyncSocket* socket) = 0;
 111
 112     // Called when the socket is detached from an EVB and
 113     // is called from the EVB thread being detached
 114     virtual void evbDetached(AsyncSocket* socket) = 0;
 115   };
 116
 117   /**
 118    * This interface is implemented only for platforms supporting
 119    * per-socket error queues.
 120    */
 121   class ErrMessageCallback {
 122    public:
 123     virtual ~ErrMessageCallback() = default;
 124
 125     /**
 126      * errMessage() will be invoked when kernel puts a message to
 127      * the error queue associated with the socket.
 128      *
 129      * @param cmsg      Reference to cmsghdr structure describing
 130      *                  a message read from error queue associated
 131      *                  with the socket.
 132      */
 133     virtual void
 134     errMessage(const cmsghdr& cmsg) noexcept = 0;
 135
 136     /**
 137      * errMessageError() will be invoked if an error occurs reading a message
 138      * from the socket error stream.
 139      *
 140      * @param ex        An exception describing the error that occurred.
 141      */
 142     virtual void errMessageError(const AsyncSocketException& ex) noexcept = 0;
 143   };
 144
 145   class SendMsgParamsCallback {
 146    public:
 147     virtual ~SendMsgParamsCallback() = default;
 148
 149     /**
 150      * getFlags() will be invoked to retrieve the desired flags to be passed
 151      * to ::sendmsg() system call. This method was intentionally declared
 152      * non-virtual, so there is no way to override it. Instead feel free to
 153      * override getFlagsImpl(flags, defaultFlags) method instead, and enjoy
 154      * the convenience of defaultFlags passed there.
 155      *
 156      * @param flags     Write flags requested for the given write operation
 157      */
 158     int getFlags(folly::WriteFlags flags, bool zeroCopyEnabled) noexcept {
 159       return getFlagsImpl(flags, getDefaultFlags(flags, zeroCopyEnabled));
 160     }
 161
 162     /**
 163      * getAncillaryData() will be invoked to initialize ancillary data
 164      * buffer referred by "msg_control" field of msghdr structure passed to
 165      * ::sendmsg() system call. The function assumes that the size of buffer
 166      * is not smaller than the value returned by getAncillaryDataSize() method
 167      * for the same combination of flags.
 168      *
 169      * @param flags     Write flags requested for the given write operation
 170      * @param data      Pointer to ancillary data buffer to initialize.
 171      */
 172     virtual void getAncillaryData(
 173       folly::WriteFlags /*flags*/,
 174       void* /*data*/) noexcept {}
 175
 176     /**
 177      * getAncillaryDataSize() will be invoked to retrieve the size of
 178      * ancillary data buffer which should be passed to ::sendmsg() system call
 179      *
 180      * @param flags     Write flags requested for the given write operation
 181      */
 182     virtual uint32_t getAncillaryDataSize(folly::WriteFlags /*flags*/)
 183         noexcept {
 184       return 0;
 185     }
 186
 187     static const size_t maxAncillaryDataSize{0x5000};
 188
 189    private:
 190     /**
 191      * getFlagsImpl() will be invoked by getFlags(folly::WriteFlags flags)
 192      * method to retrieve the flags to be passed to ::sendmsg() system call.
 193      * SendMsgParamsCallback::getFlags() is calling this method, and returns
 194      * its results directly to the caller in AsyncSocket.
 195      * Classes inheriting from SendMsgParamsCallback are welcome to override
 196      * this method to force SendMsgParamsCallback to return its own set
 197      * of flags.
 198      *
 199      * @param flags        Write flags requested for the given write operation
 200      * @param defaultflags A set of message flags returned by getDefaultFlags()
 201      *                     method for the given "flags" mask.
 202      */
 203     virtual int getFlagsImpl(folly::WriteFlags /*flags*/, int defaultFlags) {
 204       return defaultFlags;
 205     }
 206
 207     /**
 208      * getDefaultFlags() will be invoked by  getFlags(folly::WriteFlags flags)
 209      * to retrieve the default set of flags, and pass them to getFlagsImpl(...)
 210      *
 211      * @param flags     Write flags requested for the given write operation
 212      */
 213     int getDefaultFlags(folly::WriteFlags flags, bool zeroCopyEnabled) noexcept;
 214   };
 215
 216   explicit AsyncSocket();
 217   /**
 218    * Create a new unconnected AsyncSocket.
 219    *
 220    * connect() must later be called on this socket to establish a connection.
 221    */
 222   explicit AsyncSocket(EventBase* evb);
 223
 224   void setShutdownSocketSet(const std::weak_ptr<ShutdownSocketSet>& wSS);
 225
 226   /**
 227    * Create a new AsyncSocket and begin the connection process.
 228    *
 229    * @param evb             EventBase that will manage this socket.
 230    * @param address         The address to connect to.
 231    * @param connectTimeout  Optional timeout in milliseconds for the connection
 232    *                        attempt.
 233    */
 234   AsyncSocket(EventBase* evb,
 235                const folly::SocketAddress& address,
 236                uint32_t connectTimeout = 0);
 237
 238   /**
 239    * Create a new AsyncSocket and begin the connection process.
 240    *
 241    * @param evb             EventBase that will manage this socket.
 242    * @param ip              IP address to connect to (dotted-quad).
 243    * @param port            Destination port in host byte order.
 244    * @param connectTimeout  Optional timeout in milliseconds for the connection
 245    *                        attempt.
 246    */
 247   AsyncSocket(EventBase* evb,
 248                const std::string& ip,
 249                uint16_t port,
 250                uint32_t connectTimeout = 0);
 251
 252   /**
 253    * Create a AsyncSocket from an already connected socket file descriptor.
 254    *
 255    * Note that while AsyncSocket enables TCP_NODELAY for sockets it creates
 256    * when connecting, it does not change the socket options when given an
 257    * existing file descriptor.  If callers want TCP_NODELAY enabled when using
 258    * this version of the constructor, they need to explicitly call
 259    * setNoDelay(true) after the constructor returns.
 260    *
 261    * @param evb EventBase that will manage this socket.
 262    * @param fd  File descriptor to take over (should be a connected socket).
 263    * @param zeroCopyBufId Zerocopy buf id to start with.
 264    */
 265   AsyncSocket(EventBase* evb, int fd, uint32_t zeroCopyBufId = 0);
 266
 267   /**
 268    * Create an AsyncSocket from a different, already connected AsyncSocket.
 269    *
 270    * Similar to AsyncSocket(evb, fd) when fd was previously owned by an
 271    * AsyncSocket.
 272    */
 273   explicit AsyncSocket(AsyncSocket::UniquePtr);
 274
 275   /**
 276    * Helper function to create a shared_ptr<AsyncSocket>.
 277    *
 278    * This passes in the correct destructor object, since AsyncSocket's
 279    * destructor is protected and cannot be invoked directly.
 280    */
 281   static std::shared_ptr<AsyncSocket> newSocket(EventBase* evb) {
 282     return std::shared_ptr<AsyncSocket>(new AsyncSocket(evb),
 283                                            Destructor());
 284   }
 285
 286   /**
 287    * Helper function to create a shared_ptr<AsyncSocket>.
 288    */
 289   static std::shared_ptr<AsyncSocket> newSocket(
 290       EventBase* evb,
 291       const folly::SocketAddress& address,
 292       uint32_t connectTimeout = 0) {
 293     return std::shared_ptr<AsyncSocket>(
 294         new AsyncSocket(evb, address, connectTimeout),
 295         Destructor());
 296   }
 297
 298   /**
 299    * Helper function to create a shared_ptr<AsyncSocket>.
 300    */
 301   static std::shared_ptr<AsyncSocket> newSocket(
 302       EventBase* evb,
 303       const std::string& ip,
 304       uint16_t port,
 305       uint32_t connectTimeout = 0) {
 306     return std::shared_ptr<AsyncSocket>(
 307         new AsyncSocket(evb, ip, port, connectTimeout),
 308         Destructor());
 309   }
 310
 311   /**
 312    * Helper function to create a shared_ptr<AsyncSocket>.
 313    */
 314   static std::shared_ptr<AsyncSocket> newSocket(EventBase* evb, int fd) {
 315     return std::shared_ptr<AsyncSocket>(new AsyncSocket(evb, fd),
 316                                            Destructor());
 317   }
 318
 319   /**
 320    * Destroy the socket.
 321    *
 322    * AsyncSocket::destroy() must be called to destroy the socket.
 323    * The normal destructor is private, and should not be invoked directly.
 324    * This prevents callers from deleting a AsyncSocket while it is invoking a
 325    * callback.
 326    */
 327   void destroy() override;
 328
 329   /**
 330    * Get the EventBase used by this socket.
 331    */
 332   EventBase* getEventBase() const override {
 333     return eventBase_;
 334   }
 335
 336   /**
 337    * Get the file descriptor used by the AsyncSocket.
 338    */
 339   virtual int getFd() const {
 340     return fd_;
 341   }
 342
 343   /**
 344    * Extract the file descriptor from the AsyncSocket.
 345    *
 346    * This will immediately cause any installed callbacks to be invoked with an
 347    * error.  The AsyncSocket may no longer be used after the file descriptor
 348    * has been extracted.
 349    *
 350    * This method should be used with care as the resulting fd is not guaranteed
 351    * to perfectly reflect the state of the AsyncSocket (security state,
 352    * pre-received data, etc.).
 353    *
 354    * Returns the file descriptor.  The caller assumes ownership of the
 355    * descriptor, and it will not be closed when the AsyncSocket is destroyed.
 356    */
 357   virtual int detachFd();
 358
 359   /**
 360    * Uniquely identifies a handle to a socket option value. Each
 361    * combination of level and option name corresponds to one socket
 362    * option value.
 363    */
 364   class OptionKey {
 365    public:
 366     bool operator<(const OptionKey& other) const {
 367       if (level == other.level) {
 368         return optname < other.optname;
 369       }
 370       return level < other.level;
 371     }
 372     int apply(int fd, int val) const {
 373       return setsockopt(fd, level, optname, &val, sizeof(val));
 374     }
 375     int level;
 376     int optname;
 377   };
 378
 379   // Maps from a socket option key to its value
 380   typedef std::map<OptionKey, int> OptionMap;
 381
 382   static const OptionMap emptyOptionMap;
 383   static const folly::SocketAddress& anyAddress();
 384
 385   /**
 386    * Initiate a connection.
 387    *
 388    * @param callback  The callback to inform when the connection attempt
 389    *                  completes.
 390    * @param address   The address to connect to.
 391    * @param timeout   A timeout value, in milliseconds.  If the connection
 392    *                  does not succeed within this period,
 393    *                  callback->connectError() will be invoked.
 394    */
 395   virtual void connect(
 396       ConnectCallback* callback,
 397       const folly::SocketAddress& address,
 398       int timeout = 0,
 399       const OptionMap& options = emptyOptionMap,
 400       const folly::SocketAddress& bindAddr = anyAddress()) noexcept;
 401
 402   void connect(
 403       ConnectCallback* callback,
 404       const std::string& ip,
 405       uint16_t port,
 406       int timeout = 0,
 407       const OptionMap& options = emptyOptionMap) noexcept;
 408
 409   /**
 410    * If a connect request is in-flight, cancels it and closes the socket
 411    * immediately. Otherwise, this is a no-op.
 412    *
 413    * This does not invoke any connection related callbacks. Call this to
 414    * prevent any connect callback while cleaning up, etc.
 415    */
 416   void cancelConnect();
 417
 418   /**
 419    * Set the send timeout.
 420    *
 421    * If write requests do not make any progress for more than the specified
 422    * number of milliseconds, fail all pending writes and close the socket.
 423    *
 424    * If write requests are currently pending when setSendTimeout() is called,
 425    * the timeout interval is immediately restarted using the new value.
 426    *
 427    * (See the comments for AsyncSocket for an explanation of why AsyncSocket
 428    * provides setSendTimeout() but not setRecvTimeout().)
 429    *
 430    * @param milliseconds  The timeout duration, in milliseconds.  If 0, no
 431    *                      timeout will be used.
 432    */
 433   void setSendTimeout(uint32_t milliseconds) override;
 434
 435   /**
 436    * Get the send timeout.
 437    *
 438    * @return Returns the current send timeout, in milliseconds.  A return value
 439    *         of 0 indicates that no timeout is set.
 440    */
 441   uint32_t getSendTimeout() const override {
 442     return sendTimeout_;
 443   }
 444
 445   /**
 446    * Set the maximum number of reads to execute from the underlying
 447    * socket each time the EventBase detects that new ingress data is
 448    * available. The default is unlimited, but callers can use this method
 449    * to limit the amount of data read from the socket per event loop
 450    * iteration.
 451    *
 452    * @param maxReads  Maximum number of reads per data-available event;
 453    *                  a value of zero means unlimited.
 454    */
 455   void setMaxReadsPerEvent(uint16_t maxReads) {
 456     maxReadsPerEvent_ = maxReads;
 457   }
 458
 459   /**
 460    * Get the maximum number of reads this object will execute from
 461    * the underlying socket each time the EventBase detects that new
 462    * ingress data is available.
 463    *
 464    * @returns Maximum number of reads per data-available event; a value
 465    *          of zero means unlimited.
 466    */
 467   uint16_t getMaxReadsPerEvent() const {
 468     return maxReadsPerEvent_;
 469   }
 470
 471   /**
 472    * Set a pointer to ErrMessageCallback implementation which will be
 473    * receiving notifications for messages posted to the error queue
 474    * associated with the socket.
 475    * ErrMessageCallback is implemented only for platforms with
 476    * per-socket error message queus support (recvmsg() system call must
 477    * )
 478    *
 479    */
 480   virtual void setErrMessageCB(ErrMessageCallback* callback);
 481
 482   /**
 483    * Get a pointer to ErrMessageCallback implementation currently
 484    * registered with this socket.
 485    *
 486    */
 487   virtual ErrMessageCallback* getErrMessageCallback() const;
 488
 489   /**
 490    * Set a pointer to SendMsgParamsCallback implementation which
 491    * will be used to form ::sendmsg() system call parameters
 492    *
 493    */
 494   virtual void setSendMsgParamCB(SendMsgParamsCallback* callback);
 495
 496   /**
 497    * Get a pointer to SendMsgParamsCallback implementation currently
 498    * registered with this socket.
 499    *
 500    */
 501   virtual SendMsgParamsCallback* getSendMsgParamsCB() const;
 502
 503   // Read and write methods
 504   void setReadCB(ReadCallback* callback) override;
 505   ReadCallback* getReadCallback() const override;
 506
 507   bool setZeroCopy(bool enable);
 508   bool getZeroCopy() const {
 509     return zeroCopyEnabled_;
 510   }
 511
 512   uint32_t getZeroCopyBufId() const {
 513     return zeroCopyBufId_;
 514   }
 515
 516   void write(WriteCallback* callback, const void* buf, size_t bytes,
 517              WriteFlags flags = WriteFlags::NONE) override;
 518   void writev(WriteCallback* callback, const iovec* vec, size_t count,
 519               WriteFlags flags = WriteFlags::NONE) override;
 520   void writeChain(WriteCallback* callback,
 521                   std::unique_ptr<folly::IOBuf>&& buf,
 522                   WriteFlags flags = WriteFlags::NONE) override;
 523
 524   class WriteRequest;
 525   virtual void writeRequest(WriteRequest* req);
 526   void writeRequestReady() {
 527     handleWrite();
 528   }
 529
 530   // Methods inherited from AsyncTransport
 531   void close() override;
 532   void closeNow() override;
 533   void closeWithReset() override;
 534   void shutdownWrite() override;
 535   void shutdownWriteNow() override;
 536
 537   bool readable() const override;
 538   bool writable() const override;
 539   bool isPending() const override;
 540   virtual bool hangup() const;
 541   bool good() const override;
 542   bool error() const override;
 543   void attachEventBase(EventBase* eventBase) override;
 544   void detachEventBase() override;
 545   bool isDetachable() const override;
 546
 547   void getLocalAddress(
 548     folly::SocketAddress* address) const override;
 549   void getPeerAddress(
 550     folly::SocketAddress* address) const override;
 551
 552   bool isEorTrackingEnabled() const override {
 553     return trackEor_;
 554   }
 555
 556   void setEorTracking(bool track) override {
 557     trackEor_ = track;
 558   }
 559
 560   bool connecting() const override {
 561     return (state_ == StateEnum::CONNECTING);
 562   }
 563
 564   virtual bool isClosedByPeer() const {
 565     return (state_ == StateEnum::CLOSED &&
 566             (readErr_ == READ_EOF || readErr_ == READ_ERROR));
 567   }
 568
 569   virtual bool isClosedBySelf() const {
 570     return (state_ == StateEnum::CLOSED &&
 571             (readErr_ != READ_EOF && readErr_ != READ_ERROR));
 572   }
 573
 574   size_t getAppBytesWritten() const override {
 575     return appBytesWritten_;
 576   }
 577
 578   size_t getRawBytesWritten() const override {
 579     return getAppBytesWritten();
 580   }
 581
 582   size_t getAppBytesReceived() const override {
 583     return appBytesReceived_;
 584   }
 585
 586   size_t getRawBytesReceived() const override {
 587     return getAppBytesReceived();
 588   }
 589
 590   std::chrono::nanoseconds getConnectTime() const {
 591     return connectEndTime_ - connectStartTime_;
 592   }
 593
 594   std::chrono::milliseconds getConnectTimeout() const {
 595     return connectTimeout_;
 596   }
 597
 598   bool getTFOAttempted() const {
 599     return tfoAttempted_;
 600   }
 601
 602   /**
 603    * Returns whether or not the attempt to use TFO
 604    * finished successfully. This does not necessarily
 605    * mean TFO worked, just that trying to use TFO
 606    * succeeded.
 607    */
 608   bool getTFOFinished() const {
 609     return tfoFinished_;
 610   }
 611
 612   /**
 613    * Returns whether or not TFO attempt succeded on this
 614    * connection.
 615    * For servers this is pretty straightforward API and can
 616    * be invoked right after the connection is accepted. This API
 617    * will perform one syscall.
 618    * This API is a bit tricky to use for clients, since clients
 619    * only know this for sure after the SYN-ACK is returned. So it's
 620    * appropriate to call this only after the first application
 621    * data is read from the socket when the caller knows that
 622    * the SYN has been ACKed by the server.
 623    */
 624   bool getTFOSucceded() const;
 625
 626   // Methods controlling socket options
 627
 628   /**
 629    * Force writes to be transmitted immediately.
 630    *
 631    * This controls the TCP_NODELAY socket option.  When enabled, TCP segments
 632    * are sent as soon as possible, even if it is not a full frame of data.
 633    * When disabled, the data may be buffered briefly to try and wait for a full
 634    * frame of data.
 635    *
 636    * By default, TCP_NODELAY is enabled for AsyncSocket objects.
 637    *
 638    * This method will fail if the socket is not currently open.
 639    *
 640    * @return Returns 0 if the TCP_NODELAY flag was successfully updated,
 641    *         or a non-zero errno value on error.
 642    */
 643   int setNoDelay(bool noDelay);
 644
 645
 646   /**
 647    * Set the FD_CLOEXEC flag so that the socket will be closed if the program
 648    * later forks and execs.
 649    */
 650   void setCloseOnExec();
 651
 652   /*
 653    * Set the Flavor of Congestion Control to be used for this Socket
 654    * Please check '/lib/modules/<kernel>/kernel/net/ipv4' for tcp_*.ko
 655    * first to make sure the module is available for plugging in
 656    * Alternatively you can choose from net.ipv4.tcp_allowed_congestion_control
 657    */
 658   int setCongestionFlavor(const std::string &cname);
 659
 660   /*
 661    * Forces ACKs to be sent immediately
 662    *
 663    * @return Returns 0 if the TCP_QUICKACK flag was successfully updated,
 664    *         or a non-zero errno value on error.
 665    */
 666   int setQuickAck(bool quickack);
 667
 668   /**
 669    * Set the send bufsize
 670    */
 671   int setSendBufSize(size_t bufsize);
 672
 673   /**
 674    * Set the recv bufsize
 675    */
 676   int setRecvBufSize(size_t bufsize);
 677
 678   /**
 679    * Sets a specific tcp personality
 680    * Available only on kernels 3.2 and greater
 681    */
 682   #define SO_SET_NAMESPACE        41
 683   int setTCPProfile(int profd);
 684
 685   /**
 686    * Generic API for reading a socket option.
 687    *
 688    * @param level     same as the "level" parameter in getsockopt().
 689    * @param optname   same as the "optname" parameter in getsockopt().
 690    * @param optval    pointer to the variable in which the option value should
 691    *                  be returned.
 692    * @param optlen    value-result argument, initially containing the size of
 693    *                  the buffer pointed to by optval, and modified on return
 694    *                  to indicate the actual size of the value returned.
 695    * @return          same as the return value of getsockopt().
 696    */
 697   template <typename T>
 698   int getSockOpt(int level, int optname, T* optval, socklen_t* optlen) {
 699     return getsockopt(fd_, level, optname, (void*) optval, optlen);
 700   }
 701
 702   /**
 703    * Generic API for setting a socket option.
 704    *
 705    * @param level     same as the "level" parameter in getsockopt().
 706    * @param optname   same as the "optname" parameter in getsockopt().
 707    * @param optval    the option value to set.
 708    * @return          same as the return value of setsockopt().
 709    */
 710   template <typename T>
 711   int setSockOpt(int  level,  int  optname,  const T *optval) {
 712     return setsockopt(fd_, level, optname, optval, sizeof(T));
 713   }
 714
 715   /**
 716    * Virtual method for reading a socket option returning integer
 717    * value, which is the most typical case. Convenient for overriding
 718    * and mocking.
 719    *
 720    * @param level     same as the "level" parameter in getsockopt().
 721    * @param optname   same as the "optname" parameter in getsockopt().
 722    * @param optval    same as "optval" parameter in getsockopt().
 723    * @param optlen    same as "optlen" parameter in getsockopt().
 724    * @return          same as the return value of getsockopt().
 725    */
 726   virtual int
 727   getSockOptVirtual(int level, int optname, void* optval, socklen_t* optlen) {
 728     return getsockopt(fd_, level, optname, optval, optlen);
 729   }
 730
 731   /**
 732    * Virtual method for setting a socket option accepting integer
 733    * value, which is the most typical case. Convenient for overriding
 734    * and mocking.
 735    *
 736    * @param level     same as the "level" parameter in setsockopt().
 737    * @param optname   same as the "optname" parameter in setsockopt().
 738    * @param optval    same as "optval" parameter in setsockopt().
 739    * @param optlen    same as "optlen" parameter in setsockopt().
 740    * @return          same as the return value of setsockopt().
 741    */
 742   virtual int setSockOptVirtual(
 743       int level,
 744       int optname,
 745       void const* optval,
 746       socklen_t optlen) {
 747     return setsockopt(fd_, level, optname, optval, optlen);
 748   }
 749
 750   /**
 751    * Set pre-received data, to be returned to read callback before any data
 752    * from the socket.
 753    */
 754   virtual void setPreReceivedData(std::unique_ptr<IOBuf> data) {
 755     if (preReceivedData_) {
 756       preReceivedData_->prependChain(std::move(data));
 757     } else {
 758       preReceivedData_ = std::move(data);
 759     }
 760   }
 761
 762   /**
 763    * Enables TFO behavior on the AsyncSocket if FOLLY_ALLOW_TFO
 764    * is set.
 765    */
 766   void enableTFO() {
 767     // No-op if folly does not allow tfo
 768 #if FOLLY_ALLOW_TFO
 769     tfoEnabled_ = true;
 770 #endif
 771   }
 772
 773   void disableTransparentTls() {
 774     noTransparentTls_ = true;
 775   }
 776
 777   void disableTSocks() {
 778     noTSocks_ = true;
 779   }
 780
 781   enum class StateEnum : uint8_t {
 782     UNINIT,
 783     CONNECTING,
 784     ESTABLISHED,
 785     CLOSED,
 786     ERROR,
 787     FAST_OPEN,
 788   };
 789
 790   void setBufferCallback(BufferCallback* cb);
 791
 792   // Callers should set this prior to connecting the socket for the safest
 793   // behavior.
 794   void setEvbChangedCallback(std::unique_ptr<EvbChangeCallback> cb) {
 795     evbChangeCb_ = std::move(cb);
 796   }
 797
 798   /**
 799    * Attempt to cache the current local and peer addresses (if not already
 800    * cached) so that they are available from getPeerAddress() and
 801    * getLocalAddress() even after the socket is closed.
 802    */
 803   void cacheAddresses();
 804
 805   /**
 806    * Returns true if there is any zero copy write in progress
 807    * Needs to be called from within the socket's EVB thread
 808    */
 809   bool isZeroCopyWriteInProgress() const noexcept;
 810
 811   /**
 812    * Tries to process the msg error queue
 813    * And returns true if there are no more zero copy writes in progress
 814    */
 815   bool processZeroCopyWriteInProgress() noexcept;
 816
 817   /**
 818    * writeReturn is the total number of bytes written, or WRITE_ERROR on error.
 819    * If no data has been written, 0 is returned.
 820    * exception is a more specific exception that cause a write error.
 821    * Not all writes have exceptions associated with them thus writeReturn
 822    * should be checked to determine whether the operation resulted in an error.
 823    */
 824   struct WriteResult {
 825     explicit WriteResult(ssize_t ret) : writeReturn(ret) {}
 826
 827     WriteResult(ssize_t ret, std::unique_ptr<const AsyncSocketException> e)
 828         : writeReturn(ret), exception(std::move(e)) {}
 829
 830     ssize_t writeReturn;
 831     std::unique_ptr<const AsyncSocketException> exception;
 832   };
 833
 834   /**
 835    * readReturn is the number of bytes read, or READ_EOF on EOF, or
 836    * READ_ERROR on error, or READ_BLOCKING if the operation will
 837    * block.
 838    * exception is a more specific exception that may have caused a read error.
 839    * Not all read errors have exceptions associated with them thus readReturn
 840    * should be checked to determine whether the operation resulted in an error.
 841    */
 842   struct ReadResult {
 843     explicit ReadResult(ssize_t ret) : readReturn(ret) {}
 844
 845     ReadResult(ssize_t ret, std::unique_ptr<const AsyncSocketException> e)
 846         : readReturn(ret), exception(std::move(e)) {}
 847
 848     ssize_t readReturn;
 849     std::unique_ptr<const AsyncSocketException> exception;
 850   };
 851
 852   /**
 853    * A WriteRequest object tracks information about a pending write operation.
 854    */
 855   class WriteRequest {
 856    public:
 857     WriteRequest(AsyncSocket* socket, WriteCallback* callback) :
 858       socket_(socket), callback_(callback) {}
 859
 860     virtual void start() {}
 861
 862     virtual void destroy() = 0;
 863
 864     virtual WriteResult performWrite() = 0;
 865
 866     virtual void consume() = 0;
 867
 868     virtual bool isComplete() = 0;
 869
 870     WriteRequest* getNext() const {
 871       return next_;
 872     }
 873
 874     WriteCallback* getCallback() const {
 875       return callback_;
 876     }
 877
 878     uint32_t getTotalBytesWritten() const {
 879       return totalBytesWritten_;
 880     }
 881
 882     void append(WriteRequest* next) {
 883       assert(next_ == nullptr);
 884       next_ = next;
 885     }
 886
 887     void fail(const char* fn, const AsyncSocketException& ex) {
 888       socket_->failWrite(fn, ex);
 889     }
 890
 891     void bytesWritten(size_t count) {
 892       totalBytesWritten_ += uint32_t(count);
 893       socket_->appBytesWritten_ += count;
 894     }
 895
 896    protected:
 897     // protected destructor, to ensure callers use destroy()
 898     virtual ~WriteRequest() {}
 899
 900     AsyncSocket* socket_;         ///< parent socket
 901     WriteRequest* next_{nullptr};          ///< pointer to next WriteRequest
 902     WriteCallback* callback_;     ///< completion callback
 903     uint32_t totalBytesWritten_{0};  ///< total bytes written
 904   };
 905
 906  protected:
 907   enum ReadResultEnum {
 908     READ_EOF = 0,
 909     READ_ERROR = -1,
 910     READ_BLOCKING = -2,
 911     READ_NO_ERROR = -3,
 912   };
 913
 914   enum WriteResultEnum {
 915     WRITE_ERROR = -1,
 916   };
 917
 918   /**
 919    * Protected destructor.
 920    *
 921    * Users of AsyncSocket must never delete it directly.  Instead, invoke
 922    * destroy() instead.  (See the documentation in DelayedDestruction.h for
 923    * more details.)
 924    */
 925   ~AsyncSocket() override;
 926
 927   friend std::ostream& operator << (std::ostream& os, const StateEnum& state);
 928
 929   enum ShutdownFlags {
 930     /// shutdownWrite() called, but we are still waiting on writes to drain
 931     SHUT_WRITE_PENDING = 0x01,
 932     /// writes have been completely shut down
 933     SHUT_WRITE = 0x02,
 934     /**
 935      * Reads have been shutdown.
 936      *
 937      * At the moment we don't distinguish between remote read shutdown
 938      * (received EOF from the remote end) and local read shutdown.  We can
 939      * only receive EOF when a read callback is set, and we immediately inform
 940      * it of the EOF.  Therefore there doesn't seem to be any reason to have a
 941      * separate state of "received EOF but the local side may still want to
 942      * read".
 943      *
 944      * We also don't currently provide any API for only shutting down the read
 945      * side of a socket.  (This is a no-op as far as TCP is concerned, anyway.)
 946      */
 947     SHUT_READ = 0x04,
 948   };
 949
 950   class BytesWriteRequest;
 951
 952   class WriteTimeout : public AsyncTimeout {
 953    public:
 954     WriteTimeout(AsyncSocket* socket, EventBase* eventBase)
 955       : AsyncTimeout(eventBase)
 956       , socket_(socket) {}
 957
 958     void timeoutExpired() noexcept override {
 959       socket_->timeoutExpired();
 960     }
 961
 962    private:
 963     AsyncSocket* socket_;
 964   };
 965
 966   class IoHandler : public EventHandler {
 967    public:
 968     IoHandler(AsyncSocket* socket, EventBase* eventBase)
 969       : EventHandler(eventBase, -1)
 970       , socket_(socket) {}
 971     IoHandler(AsyncSocket* socket, EventBase* eventBase, int fd)
 972       : EventHandler(eventBase, fd)
 973       , socket_(socket) {}
 974
 975     void handlerReady(uint16_t events) noexcept override {
 976       socket_->ioReady(events);
 977     }
 978
 979    private:
 980     AsyncSocket* socket_;
 981   };
 982
 983   void init();
 984
 985   class ImmediateReadCB : public folly::EventBase::LoopCallback {
 986    public:
 987     explicit ImmediateReadCB(AsyncSocket* socket) : socket_(socket) {}
 988     void runLoopCallback() noexcept override {
 989       DestructorGuard dg(socket_);
 990       socket_->checkForImmediateRead();
 991     }
 992    private:
 993     AsyncSocket* socket_;
 994   };
 995
 996   /**
 997    * Schedule checkForImmediateRead to be executed in the next loop
 998    * iteration.
 999    */
1000   void scheduleImmediateRead() noexcept {
1001     if (good()) {
1002       eventBase_->runInLoop(&immediateReadHandler_);
1003     }
1004   }
1005
1006   /**
1007    * Schedule handleInitalReadWrite to run in the next iteration.
1008    */
1009   void scheduleInitialReadWrite() noexcept {
1010     if (good()) {
1011       DestructorGuard dg(this);
1012       eventBase_->runInLoop([this, dg] {
1013         if (good()) {
1014           handleInitialReadWrite();
1015         }
1016       });
1017     }
1018   }
1019
1020   // event notification methods
1021   void ioReady(uint16_t events) noexcept;
1022   virtual void checkForImmediateRead() noexcept;
1023   virtual void handleInitialReadWrite() noexcept;
1024   virtual void prepareReadBuffer(void** buf, size_t* buflen);
1025   virtual size_t handleErrMessages() noexcept;
1026   virtual void handleRead() noexcept;
1027   virtual void handleWrite() noexcept;
1028   virtual void handleConnect() noexcept;
1029   void timeoutExpired() noexcept;
1030
1031   /**
1032    * Attempt to read from the socket.
1033    *
1034    * @param buf      The buffer to read data into.
1035    * @param buflen   The length of the buffer.
1036    *
1037    * @return Returns a read result. See read result for details.
1038    */
1039   virtual ReadResult performRead(void** buf, size_t* buflen, size_t* offset);
1040
1041   /**
1042    * Populate an iovec array from an IOBuf and attempt to write it.
1043    *
1044    * @param callback Write completion/error callback.
1045    * @param vec      Target iovec array; caller retains ownership.
1046    * @param count    Number of IOBufs to write, beginning at start of buf.
1047    * @param buf      Chain of iovecs.
1048    * @param flags    set of flags for the underlying write calls, like cork
1049    */
1050   void writeChainImpl(WriteCallback* callback, iovec* vec,
1051                       size_t count, std::unique_ptr<folly::IOBuf>&& buf,
1052                       WriteFlags flags);
1053
1054   /**
1055    * Write as much data as possible to the socket without blocking,
1056    * and queue up any leftover data to send when the socket can
1057    * handle writes again.
1058    *
1059    * @param callback The callback to invoke when the write is completed.
1060    * @param vec      Array of buffers to write; this method will make a
1061    *                 copy of the vector (but not the buffers themselves)
1062    *                 if the write has to be completed asynchronously.
1063    * @param count    Number of elements in vec.
1064    * @param buf      The IOBuf that manages the buffers referenced by
1065    *                 vec, or a pointer to nullptr if the buffers are not
1066    *                 associated with an IOBuf.  Note that ownership of
1067    *                 the IOBuf is transferred here; upon completion of
1068    *                 the write, the AsyncSocket deletes the IOBuf.
1069    * @param flags    Set of write flags.
1070    */
1071   void writeImpl(WriteCallback* callback, const iovec* vec, size_t count,
1072                  std::unique_ptr<folly::IOBuf>&& buf,
1073                  WriteFlags flags = WriteFlags::NONE);
1074
1075   /**
1076    * Attempt to write to the socket.
1077    *
1078    * @param vec             The iovec array pointing to the buffers to write.
1079    * @param count           The length of the iovec array.
1080    * @param flags           Set of write flags.
1081    * @param countWritten    On return, the value pointed to by this parameter
1082    *                          will contain the number of iovec entries that were
1083    *                          fully written.
1084    * @param partialWritten  On return, the value pointed to by this parameter
1085    *                          will contain the number of bytes written in the
1086    *                          partially written iovec entry.
1087    *
1088    * @return Returns a WriteResult. See WriteResult for more details.
1089    */
1090   virtual WriteResult performWrite(
1091       const iovec* vec,
1092       uint32_t count,
1093       WriteFlags flags,
1094       uint32_t* countWritten,
1095       uint32_t* partialWritten);
1096
1097   /**
1098    * Sends the message over the socket using sendmsg
1099    *
1100    * @param msg       Message to send
1101    * @param msg_flags Flags to pass to sendmsg
1102    */
1103   AsyncSocket::WriteResult
1104   sendSocketMessage(int fd, struct msghdr* msg, int msg_flags);
1105
1106   virtual ssize_t tfoSendMsg(int fd, struct msghdr* msg, int msg_flags);
1107
1108   int socketConnect(const struct sockaddr* addr, socklen_t len);
1109
1110   virtual void scheduleConnectTimeout();
1111   void registerForConnectEvents();
1112
1113   bool updateEventRegistration();
1114
1115   /**
1116    * Update event registration.
1117    *
1118    * @param enable Flags of events to enable. Set it to 0 if no events
1119    * need to be enabled in this call.
1120    * @param disable Flags of events
1121    * to disable. Set it to 0 if no events need to be disabled in this
1122    * call.
1123    *
1124    * @return true iff the update is successful.
1125    */
1126   bool updateEventRegistration(uint16_t enable, uint16_t disable);
1127
1128   // Actually close the file descriptor and set it to -1 so we don't
1129   // accidentally close it again.
1130   void doClose();
1131
1132   // error handling methods
1133   void startFail();
1134   void finishFail();
1135   void finishFail(const AsyncSocketException& ex);
1136   void invokeAllErrors(const AsyncSocketException& ex);
1137   void fail(const char* fn, const AsyncSocketException& ex);
1138   void failConnect(const char* fn, const AsyncSocketException& ex);
1139   void failRead(const char* fn, const AsyncSocketException& ex);
1140   void failErrMessageRead(const char* fn, const AsyncSocketException& ex);
1141   void failWrite(const char* fn, WriteCallback* callback, size_t bytesWritten,
1142                  const AsyncSocketException& ex);
1143   void failWrite(const char* fn, const AsyncSocketException& ex);
1144   void failAllWrites(const AsyncSocketException& ex);
1145   virtual void invokeConnectErr(const AsyncSocketException& ex);
1146   virtual void invokeConnectSuccess();
1147   void invalidState(ConnectCallback* callback);
1148   void invalidState(ErrMessageCallback* callback);
1149   void invalidState(ReadCallback* callback);
1150   void invalidState(WriteCallback* callback);
1151
1152   std::string withAddr(const std::string& s);
1153
1154   void cacheLocalAddress() const;
1155   void cachePeerAddress() const;
1156
1157   bool isZeroCopyRequest(WriteFlags flags);
1158
1159   bool isZeroCopyMsg(const cmsghdr& cmsg) const;
1160   void processZeroCopyMsg(const cmsghdr& cmsg);
1161
1162   uint32_t getNextZeroCopyBufId() {
1163     return zeroCopyBufId_++;
1164   }
1165   void adjustZeroCopyFlags(folly::WriteFlags& flags);
1166   void addZeroCopyBuf(std::unique_ptr<folly::IOBuf>&& buf);
1167   void addZeroCopyBuf(folly::IOBuf* ptr);
1168   void setZeroCopyBuf(std::unique_ptr<folly::IOBuf>&& buf);
1169   bool containsZeroCopyBuf(folly::IOBuf* ptr);
1170   void releaseZeroCopyBuf(uint32_t id);
1171
1172   // a folly::IOBuf can be used in multiple partial requests
1173   // there is a that maps a buffer id to a raw folly::IOBuf ptr
1174   // and another one that adds a ref count for a folly::IOBuf that is either
1175   // the original ptr or nullptr
1176   uint32_t zeroCopyBufId_{0};
1177
1178   struct IOBufInfo {
1179     uint32_t count_{0};
1180     std::unique_ptr<folly::IOBuf> buf_;
1181   };
1182
1183   std::unordered_map<uint32_t, folly::IOBuf*> idZeroCopyBufPtrMap_;
1184   std::unordered_map<folly::IOBuf*, IOBufInfo> idZeroCopyBufInfoMap_;
1185
1186   StateEnum state_;                      ///< StateEnum describing current state
1187   uint8_t shutdownFlags_;                ///< Shutdown state (ShutdownFlags)
1188   uint16_t eventFlags_;                  ///< EventBase::HandlerFlags settings
1189   int fd_;                               ///< The socket file descriptor
1190   mutable folly::SocketAddress addr_;    ///< The address we tried to connect to
1191   mutable folly::SocketAddress localAddr_;
1192                                          ///< The address we are connecting from
1193   uint32_t sendTimeout_;                 ///< The send timeout, in milliseconds
1194   uint16_t maxReadsPerEvent_;            ///< Max reads per event loop iteration
1195
1196   bool isBufferMovable_{false};
1197
1198   int8_t readErr_{READ_NO_ERROR}; ///< The read error encountered, if any
1199
1200   EventBase* eventBase_;                 ///< The EventBase
1201   WriteTimeout writeTimeout_;            ///< A timeout for connect and write
1202   IoHandler ioHandler_;                  ///< A EventHandler to monitor the fd
1203   ImmediateReadCB immediateReadHandler_; ///< LoopCallback for checking read
1204
1205   ConnectCallback* connectCallback_;     ///< ConnectCallback
1206   ErrMessageCallback* errMessageCallback_; ///< TimestampCallback
1207   SendMsgParamsCallback* ///< Callback for retrieving
1208       sendMsgParamCallback_; ///< ::sendmsg() parameters
1209   ReadCallback* readCallback_;           ///< ReadCallback
1210   WriteRequest* writeReqHead_;           ///< Chain of WriteRequests
1211   WriteRequest* writeReqTail_;           ///< End of WriteRequest chain
1212   std::weak_ptr<ShutdownSocketSet> wShutdownSocketSet_;
1213   size_t appBytesReceived_;              ///< Num of bytes received from socket
1214   size_t appBytesWritten_;               ///< Num of bytes written to socket
1215
1216   // Pre-received data, to be returned to read callback before any data from the
1217   // socket.
1218   std::unique_ptr<IOBuf> preReceivedData_;
1219
1220   std::chrono::steady_clock::time_point connectStartTime_;
1221   std::chrono::steady_clock::time_point connectEndTime_;
1222
1223   std::chrono::milliseconds connectTimeout_{0};
1224
1225   std::unique_ptr<EvbChangeCallback> evbChangeCb_{nullptr};
1226
1227   BufferCallback* bufferCallback_{nullptr};
1228   bool tfoEnabled_{false};
1229   bool tfoAttempted_{false};
1230   bool tfoFinished_{false};
1231   bool noTransparentTls_{false};
1232   bool noTSocks_{false};
1233   // Whether to track EOR or not.
1234   bool trackEor_{false};
1235   bool zeroCopyEnabled_{false};
1236   bool zeroCopyVal_{false};
1237 };
1238 #ifdef _MSC_VER
1239 #pragma vtordisp(pop)
1240 #endif
1241
1242 } // namespace folly