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