folly/io/async/AsyncTransport.h

   1 /*
   2  * Copyright 2017 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 <memory>
  20
  21 #include <folly/io/IOBuf.h>
  22 #include <folly/io/async/AsyncSocketBase.h>
  23 #include <folly/io/async/DelayedDestruction.h>
  24 #include <folly/io/async/EventBase.h>
  25 #include <folly/ssl/OpenSSLPtrTypes.h>
  26 #include <folly/portability/OpenSSL.h>
  27 #include <folly/portability/SysUio.h>
  28
  29 constexpr bool kOpenSslModeMoveBufferOwnership =
  30 #ifdef SSL_MODE_MOVE_BUFFER_OWNERSHIP
  31   true
  32 #else
  33   false
  34 #endif
  35 ;
  36
  37 namespace folly {
  38
  39 class AsyncSocketException;
  40 class EventBase;
  41 class SocketAddress;
  42
  43 /*
  44  * flags given by the application for write* calls
  45  */
  46 enum class WriteFlags : uint32_t {
  47   NONE = 0x00,
  48   /*
  49    * Whether to delay the output until a subsequent non-corked write.
  50    * (Note: may not be supported in all subclasses or on all platforms.)
  51    */
  52   CORK = 0x01,
  53   /*
  54    * for a socket that has ACK latency enabled, it will cause the kernel
  55    * to fire a TCP ESTATS event when the last byte of the given write call
  56    * will be acknowledged.
  57    */
  58   EOR = 0x02,
  59   /*
  60    * this indicates that only the write side of socket should be shutdown
  61    */
  62   WRITE_SHUTDOWN = 0x04,
  63 };
  64
  65 /*
  66  * union operator
  67  */
  68 inline WriteFlags operator|(WriteFlags a, WriteFlags b) {
  69   return static_cast<WriteFlags>(
  70     static_cast<uint32_t>(a) | static_cast<uint32_t>(b));
  71 }
  72
  73 /*
  74  * compound assignment union operator
  75  */
  76 inline WriteFlags& operator|=(WriteFlags& a, WriteFlags b) {
  77   a = a | b;
  78   return a;
  79 }
  80
  81 /*
  82  * intersection operator
  83  */
  84 inline WriteFlags operator&(WriteFlags a, WriteFlags b) {
  85   return static_cast<WriteFlags>(
  86     static_cast<uint32_t>(a) & static_cast<uint32_t>(b));
  87 }
  88
  89 /*
  90  * compound assignment intersection operator
  91  */
  92 inline WriteFlags& operator&=(WriteFlags& a, WriteFlags b) {
  93   a = a & b;
  94   return a;
  95 }
  96
  97 /*
  98  * exclusion parameter
  99  */
 100 inline WriteFlags operator~(WriteFlags a) {
 101   return static_cast<WriteFlags>(~static_cast<uint32_t>(a));
 102 }
 103
 104 /*
 105  * unset operator
 106  */
 107 inline WriteFlags unSet(WriteFlags a, WriteFlags b) {
 108   return a & ~b;
 109 }
 110
 111 /*
 112  * inclusion operator
 113  */
 114 inline bool isSet(WriteFlags a, WriteFlags b) {
 115   return (a & b) == b;
 116 }
 117
 118
 119 /**
 120  * AsyncTransport defines an asynchronous API for streaming I/O.
 121  *
 122  * This class provides an API to for asynchronously waiting for data
 123  * on a streaming transport, and for asynchronously sending data.
 124  *
 125  * The APIs for reading and writing are intentionally asymmetric.  Waiting for
 126  * data to read is a persistent API: a callback is installed, and is notified
 127  * whenever new data is available.  It continues to be notified of new events
 128  * until it is uninstalled.
 129  *
 130  * AsyncTransport does not provide read timeout functionality, because it
 131  * typically cannot determine when the timeout should be active.  Generally, a
 132  * timeout should only be enabled when processing is blocked waiting on data
 133  * from the remote endpoint.  For server-side applications, the timeout should
 134  * not be active if the server is currently processing one or more outstanding
 135  * requests on this transport.  For client-side applications, the timeout
 136  * should not be active if there are no requests pending on the transport.
 137  * Additionally, if a client has multiple pending requests, it will ususally
 138  * want a separate timeout for each request, rather than a single read timeout.
 139  *
 140  * The write API is fairly intuitive: a user can request to send a block of
 141  * data, and a callback will be informed once the entire block has been
 142  * transferred to the kernel, or on error.  AsyncTransport does provide a send
 143  * timeout, since most callers want to give up if the remote end stops
 144  * responding and no further progress can be made sending the data.
 145  */
 146 class AsyncTransport : public DelayedDestruction, public AsyncSocketBase {
 147  public:
 148   typedef std::unique_ptr<AsyncTransport, Destructor> UniquePtr;
 149
 150   /**
 151    * Close the transport.
 152    *
 153    * This gracefully closes the transport, waiting for all pending write
 154    * requests to complete before actually closing the underlying transport.
 155    *
 156    * If a read callback is set, readEOF() will be called immediately.  If there
 157    * are outstanding write requests, the close will be delayed until all
 158    * remaining writes have completed.  No new writes may be started after
 159    * close() has been called.
 160    */
 161   virtual void close() = 0;
 162
 163   /**
 164    * Close the transport immediately.
 165    *
 166    * This closes the transport immediately, dropping any outstanding data
 167    * waiting to be written.
 168    *
 169    * If a read callback is set, readEOF() will be called immediately.
 170    * If there are outstanding write requests, these requests will be aborted
 171    * and writeError() will be invoked immediately on all outstanding write
 172    * callbacks.
 173    */
 174   virtual void closeNow() = 0;
 175
 176   /**
 177    * Reset the transport immediately.
 178    *
 179    * This closes the transport immediately, sending a reset to the remote peer
 180    * if possible to indicate abnormal shutdown.
 181    *
 182    * Note that not all subclasses implement this reset functionality: some
 183    * subclasses may treat reset() the same as closeNow().  Subclasses that use
 184    * TCP transports should terminate the connection with a TCP reset.
 185    */
 186   virtual void closeWithReset() {
 187     closeNow();
 188   }
 189
 190   /**
 191    * Perform a half-shutdown of the write side of the transport.
 192    *
 193    * The caller should not make any more calls to write() or writev() after
 194    * shutdownWrite() is called.  Any future write attempts will fail
 195    * immediately.
 196    *
 197    * Not all transport types support half-shutdown.  If the underlying
 198    * transport does not support half-shutdown, it will fully shutdown both the
 199    * read and write sides of the transport.  (Fully shutting down the socket is
 200    * better than doing nothing at all, since the caller may rely on the
 201    * shutdownWrite() call to notify the other end of the connection that no
 202    * more data can be read.)
 203    *
 204    * If there is pending data still waiting to be written on the transport,
 205    * the actual shutdown will be delayed until the pending data has been
 206    * written.
 207    *
 208    * Note: There is no corresponding shutdownRead() equivalent.  Simply
 209    * uninstall the read callback if you wish to stop reading.  (On TCP sockets
 210    * at least, shutting down the read side of the socket is a no-op anyway.)
 211    */
 212   virtual void shutdownWrite() = 0;
 213
 214   /**
 215    * Perform a half-shutdown of the write side of the transport.
 216    *
 217    * shutdownWriteNow() is identical to shutdownWrite(), except that it
 218    * immediately performs the shutdown, rather than waiting for pending writes
 219    * to complete.  Any pending write requests will be immediately failed when
 220    * shutdownWriteNow() is called.
 221    */
 222   virtual void shutdownWriteNow() = 0;
 223
 224   /**
 225    * Determine if transport is open and ready to read or write.
 226    *
 227    * Note that this function returns false on EOF; you must also call error()
 228    * to distinguish between an EOF and an error.
 229    *
 230    * @return  true iff the transport is open and ready, false otherwise.
 231    */
 232   virtual bool good() const = 0;
 233
 234   /**
 235    * Determine if the transport is readable or not.
 236    *
 237    * @return  true iff the transport is readable, false otherwise.
 238    */
 239   virtual bool readable() const = 0;
 240
 241   /**
 242    * Determine if the transport is writable or not.
 243    *
 244    * @return  true iff the transport is writable, false otherwise.
 245    */
 246   virtual bool writable() const {
 247     // By default return good() - leave it to implementers to override.
 248     return good();
 249   }
 250
 251   /**
 252    * Determine if the there is pending data on the transport.
 253    *
 254    * @return  true iff the if the there is pending data, false otherwise.
 255    */
 256   virtual bool isPending() const {
 257     return readable();
 258   }
 259
 260   /**
 261    * Determine if transport is connected to the endpoint
 262    *
 263    * @return  false iff the transport is connected, otherwise true
 264    */
 265   virtual bool connecting() const = 0;
 266
 267   /**
 268    * Determine if an error has occurred with this transport.
 269    *
 270    * @return  true iff an error has occurred (not EOF).
 271    */
 272   virtual bool error() const = 0;
 273
 274   /**
 275    * Attach the transport to a EventBase.
 276    *
 277    * This may only be called if the transport is not currently attached to a
 278    * EventBase (by an earlier call to detachEventBase()).
 279    *
 280    * This method must be invoked in the EventBase's thread.
 281    */
 282   virtual void attachEventBase(EventBase* eventBase) = 0;
 283
 284   /**
 285    * Detach the transport from its EventBase.
 286    *
 287    * This may only be called when the transport is idle and has no reads or
 288    * writes pending.  Once detached, the transport may not be used again until
 289    * it is re-attached to a EventBase by calling attachEventBase().
 290    *
 291    * This method must be called from the current EventBase's thread.
 292    */
 293   virtual void detachEventBase() = 0;
 294
 295   /**
 296    * Determine if the transport can be detached.
 297    *
 298    * This method must be called from the current EventBase's thread.
 299    */
 300   virtual bool isDetachable() const = 0;
 301
 302   /**
 303    * Set the send timeout.
 304    *
 305    * If write requests do not make any progress for more than the specified
 306    * number of milliseconds, fail all pending writes and close the transport.
 307    *
 308    * If write requests are currently pending when setSendTimeout() is called,
 309    * the timeout interval is immediately restarted using the new value.
 310    *
 311    * @param milliseconds  The timeout duration, in milliseconds.  If 0, no
 312    *                      timeout will be used.
 313    */
 314   virtual void setSendTimeout(uint32_t milliseconds) = 0;
 315
 316   /**
 317    * Get the send timeout.
 318    *
 319    * @return Returns the current send timeout, in milliseconds.  A return value
 320    *         of 0 indicates that no timeout is set.
 321    */
 322   virtual uint32_t getSendTimeout() const = 0;
 323
 324   /**
 325    * Get the address of the local endpoint of this transport.
 326    *
 327    * This function may throw AsyncSocketException on error.
 328    *
 329    * @param address  The local address will be stored in the specified
 330    *                 SocketAddress.
 331    */
 332   virtual void getLocalAddress(SocketAddress* address) const = 0;
 333
 334   /**
 335    * Get the address of the remote endpoint to which this transport is
 336    * connected.
 337    *
 338    * This function may throw AsyncSocketException on error.
 339    *
 340    * @return         Return the local address
 341    */
 342   SocketAddress getLocalAddress() const {
 343     SocketAddress addr;
 344     getLocalAddress(&addr);
 345     return addr;
 346   }
 347
 348   void getAddress(SocketAddress* address) const override {
 349     getLocalAddress(address);
 350   }
 351
 352   /**
 353    * Get the address of the remote endpoint to which this transport is
 354    * connected.
 355    *
 356    * This function may throw AsyncSocketException on error.
 357    *
 358    * @param address  The remote endpoint's address will be stored in the
 359    *                 specified SocketAddress.
 360    */
 361   virtual void getPeerAddress(SocketAddress* address) const = 0;
 362
 363   /**
 364    * Get the address of the remote endpoint to which this transport is
 365    * connected.
 366    *
 367    * This function may throw AsyncSocketException on error.
 368    *
 369    * @return         Return the remote endpoint's address
 370    */
 371   SocketAddress getPeerAddress() const {
 372     SocketAddress addr;
 373     getPeerAddress(&addr);
 374     return addr;
 375   }
 376
 377   /**
 378    * Get the certificate used to authenticate the peer.
 379    */
 380   virtual ssl::X509UniquePtr getPeerCert() const { return nullptr; }
 381
 382   /**
 383    * The local certificate used for this connection. May be null
 384    */
 385   virtual const X509* getSelfCert() const {
 386     return nullptr;
 387   }
 388
 389   /**
 390    * @return True iff end of record tracking is enabled
 391    */
 392   virtual bool isEorTrackingEnabled() const = 0;
 393
 394   virtual void setEorTracking(bool track) = 0;
 395
 396   virtual size_t getAppBytesWritten() const = 0;
 397   virtual size_t getRawBytesWritten() const = 0;
 398   virtual size_t getAppBytesReceived() const = 0;
 399   virtual size_t getRawBytesReceived() const = 0;
 400
 401   class BufferCallback {
 402    public:
 403     virtual ~BufferCallback() {}
 404     virtual void onEgressBuffered() = 0;
 405     virtual void onEgressBufferCleared() = 0;
 406   };
 407
 408   /**
 409    * Callback class to signal when a transport that did not have replay
 410    * protection gains replay protection. This is needed for 0-RTT security
 411    * protocols.
 412    */
 413   class ReplaySafetyCallback {
 414    public:
 415     virtual ~ReplaySafetyCallback() = default;
 416
 417     /**
 418      * Called when the transport becomes replay safe.
 419      */
 420     virtual void onReplaySafe() = 0;
 421   };
 422
 423   /**
 424    * False if the transport does not have replay protection, but will in the
 425    * future.
 426    */
 427   virtual bool isReplaySafe() const { return true; }
 428
 429   /**
 430    * Set the ReplaySafeCallback on this transport.
 431    *
 432    * This should only be called if isReplaySafe() returns false.
 433    */
 434   virtual void setReplaySafetyCallback(ReplaySafetyCallback* callback) {
 435     if (callback) {
 436       CHECK(false) << "setReplaySafetyCallback() not supported";
 437     }
 438   }
 439
 440  protected:
 441   ~AsyncTransport() override = default;
 442 };
 443
 444 class AsyncReader {
 445  public:
 446   class ReadCallback {
 447    public:
 448     virtual ~ReadCallback() = default;
 449
 450     /**
 451      * When data becomes available, getReadBuffer() will be invoked to get the
 452      * buffer into which data should be read.
 453      *
 454      * This method allows the ReadCallback to delay buffer allocation until
 455      * data becomes available.  This allows applications to manage large
 456      * numbers of idle connections, without having to maintain a separate read
 457      * buffer for each idle connection.
 458      *
 459      * It is possible that in some cases, getReadBuffer() may be called
 460      * multiple times before readDataAvailable() is invoked.  In this case, the
 461      * data will be written to the buffer returned from the most recent call to
 462      * readDataAvailable().  If the previous calls to readDataAvailable()
 463      * returned different buffers, the ReadCallback is responsible for ensuring
 464      * that they are not leaked.
 465      *
 466      * If getReadBuffer() throws an exception, returns a nullptr buffer, or
 467      * returns a 0 length, the ReadCallback will be uninstalled and its
 468      * readError() method will be invoked.
 469      *
 470      * getReadBuffer() is not allowed to change the transport state before it
 471      * returns.  (For example, it should never uninstall the read callback, or
 472      * set a different read callback.)
 473      *
 474      * @param bufReturn getReadBuffer() should update *bufReturn to contain the
 475      *                  address of the read buffer.  This parameter will never
 476      *                  be nullptr.
 477      * @param lenReturn getReadBuffer() should update *lenReturn to contain the
 478      *                  maximum number of bytes that may be written to the read
 479      *                  buffer.  This parameter will never be nullptr.
 480      */
 481     virtual void getReadBuffer(void** bufReturn, size_t* lenReturn) = 0;
 482
 483     /**
 484      * readDataAvailable() will be invoked when data has been successfully read
 485      * into the buffer returned by the last call to getReadBuffer().
 486      *
 487      * The read callback remains installed after readDataAvailable() returns.
 488      * It must be explicitly uninstalled to stop receiving read events.
 489      * getReadBuffer() will be called at least once before each call to
 490      * readDataAvailable().  getReadBuffer() will also be called before any
 491      * call to readEOF().
 492      *
 493      * @param len       The number of bytes placed in the buffer.
 494      */
 495
 496     virtual void readDataAvailable(size_t len) noexcept = 0;
 497
 498     /**
 499      * When data becomes available, isBufferMovable() will be invoked to figure
 500      * out which API will be used, readBufferAvailable() or
 501      * readDataAvailable(). If isBufferMovable() returns true, that means
 502      * ReadCallback supports the IOBuf ownership transfer and
 503      * readBufferAvailable() will be used.  Otherwise, not.
 504
 505      * By default, isBufferMovable() always return false. If
 506      * readBufferAvailable() is implemented and to be invoked, You should
 507      * overwrite isBufferMovable() and return true in the inherited class.
 508      *
 509      * This method allows the AsyncSocket/AsyncSSLSocket do buffer allocation by
 510      * itself until data becomes available.  Compared with the pre/post buffer
 511      * allocation in getReadBuffer()/readDataAvailabe(), readBufferAvailable()
 512      * has two advantages.  First, this can avoid memcpy. E.g., in
 513      * AsyncSSLSocket, the decrypted data was copied from the openssl internal
 514      * buffer to the readbuf buffer.  With the buffer ownership transfer, the
 515      * internal buffer can be directly "moved" to ReadCallback. Second, the
 516      * memory allocation can be more precise.  The reason is
 517      * AsyncSocket/AsyncSSLSocket can allocate the memory of precise size
 518      * because they have more context about the available data than
 519      * ReadCallback.  Think about the getReadBuffer() pre-allocate 4072 bytes
 520      * buffer, but the available data is always 16KB (max OpenSSL record size).
 521      */
 522
 523     virtual bool isBufferMovable() noexcept {
 524       return false;
 525     }
 526
 527     /**
 528      * Suggested buffer size, allocated for read operations,
 529      * if callback is movable and supports folly::IOBuf
 530      */
 531
 532     virtual size_t maxBufferSize() const {
 533       return 64 * 1024; // 64K
 534     }
 535
 536     /**
 537      * readBufferAvailable() will be invoked when data has been successfully
 538      * read.
 539      *
 540      * Note that only either readBufferAvailable() or readDataAvailable() will
 541      * be invoked according to the return value of isBufferMovable(). The timing
 542      * and aftereffect of readBufferAvailable() are the same as
 543      * readDataAvailable()
 544      *
 545      * @param readBuf The unique pointer of read buffer.
 546      */
 547
 548     virtual void readBufferAvailable(std::unique_ptr<IOBuf> /*readBuf*/)
 549       noexcept {}
 550
 551     /**
 552      * readEOF() will be invoked when the transport is closed.
 553      *
 554      * The read callback will be automatically uninstalled immediately before
 555      * readEOF() is invoked.
 556      */
 557     virtual void readEOF() noexcept = 0;
 558
 559     /**
 560      * readError() will be invoked if an error occurs reading from the
 561      * transport.
 562      *
 563      * The read callback will be automatically uninstalled immediately before
 564      * readError() is invoked.
 565      *
 566      * @param ex        An exception describing the error that occurred.
 567      */
 568     virtual void readErr(const AsyncSocketException& ex) noexcept = 0;
 569   };
 570
 571   // Read methods that aren't part of AsyncTransport.
 572   virtual void setReadCB(ReadCallback* callback) = 0;
 573   virtual ReadCallback* getReadCallback() const = 0;
 574
 575  protected:
 576   virtual ~AsyncReader() = default;
 577 };
 578
 579 class AsyncWriter {
 580  public:
 581   class WriteCallback {
 582    public:
 583     virtual ~WriteCallback() = default;
 584
 585     /**
 586      * writeSuccess() will be invoked when all of the data has been
 587      * successfully written.
 588      *
 589      * Note that this mainly signals that the buffer containing the data to
 590      * write is no longer needed and may be freed or re-used.  It does not
 591      * guarantee that the data has been fully transmitted to the remote
 592      * endpoint.  For example, on socket-based transports, writeSuccess() only
 593      * indicates that the data has been given to the kernel for eventual
 594      * transmission.
 595      */
 596     virtual void writeSuccess() noexcept = 0;
 597
 598     /**
 599      * writeError() will be invoked if an error occurs writing the data.
 600      *
 601      * @param bytesWritten      The number of bytes that were successfull
 602      * @param ex                An exception describing the error that occurred.
 603      */
 604     virtual void writeErr(size_t bytesWritten,
 605                           const AsyncSocketException& ex) noexcept = 0;
 606   };
 607
 608   // Write methods that aren't part of AsyncTransport
 609   virtual void write(WriteCallback* callback, const void* buf, size_t bytes,
 610                      WriteFlags flags = WriteFlags::NONE) = 0;
 611   virtual void writev(WriteCallback* callback, const iovec* vec, size_t count,
 612                       WriteFlags flags = WriteFlags::NONE) = 0;
 613   virtual void writeChain(WriteCallback* callback,
 614                           std::unique_ptr<IOBuf>&& buf,
 615                           WriteFlags flags = WriteFlags::NONE) = 0;
 616
 617  protected:
 618   virtual ~AsyncWriter() = default;
 619 };
 620
 621 // Transitional intermediate interface. This is deprecated.
 622 // Wrapper around folly::AsyncTransport, that includes read/write callbacks
 623 class AsyncTransportWrapper : virtual public AsyncTransport,
 624                               virtual public AsyncReader,
 625                               virtual public AsyncWriter {
 626  public:
 627   using UniquePtr = std::unique_ptr<AsyncTransportWrapper, Destructor>;
 628
 629   // Alias for inherited members from AsyncReader and AsyncWriter
 630   // to keep compatibility.
 631   using ReadCallback    = AsyncReader::ReadCallback;
 632   using WriteCallback   = AsyncWriter::WriteCallback;
 633   void setReadCB(ReadCallback* callback) override = 0;
 634   ReadCallback* getReadCallback() const override = 0;
 635   void write(
 636       WriteCallback* callback,
 637       const void* buf,
 638       size_t bytes,
 639       WriteFlags flags = WriteFlags::NONE) override = 0;
 640   void writev(
 641       WriteCallback* callback,
 642       const iovec* vec,
 643       size_t count,
 644       WriteFlags flags = WriteFlags::NONE) override = 0;
 645   void writeChain(
 646       WriteCallback* callback,
 647       std::unique_ptr<IOBuf>&& buf,
 648       WriteFlags flags = WriteFlags::NONE) override = 0;
 649   /**
 650    * The transport wrapper may wrap another transport. This returns the
 651    * transport that is wrapped. It returns nullptr if there is no wrapped
 652    * transport.
 653    */
 654   virtual const AsyncTransportWrapper* getWrappedTransport() const {
 655     return nullptr;
 656   }
 657
 658   /**
 659    * In many cases when we need to set socket properties or otherwise access the
 660    * underlying transport from a wrapped transport. This method allows access to
 661    * the derived classes of the underlying transport.
 662    */
 663   template <class T>
 664   const T* getUnderlyingTransport() const {
 665     const AsyncTransportWrapper* current = this;
 666     while (current) {
 667       auto sock = dynamic_cast<const T*>(current);
 668       if (sock) {
 669         return sock;
 670       }
 671       current = current->getWrappedTransport();
 672     }
 673     return nullptr;
 674   }
 675
 676   template <class T>
 677   T* getUnderlyingTransport() {
 678     return const_cast<T*>(static_cast<const AsyncTransportWrapper*>(this)
 679         ->getUnderlyingTransport<T>());
 680   }
 681
 682   /**
 683    * Return the application protocol being used by the underlying transport
 684    * protocol. This is useful for transports which are used to tunnel other
 685    * protocols.
 686    */
 687   virtual std::string getApplicationProtocol() noexcept {
 688     return "";
 689   }
 690
 691   /**
 692    * Returns the name of the security protocol being used.
 693    */
 694   virtual std::string getSecurityProtocol() const { return ""; }
 695 };
 696
 697 } // folly