folly/io/async/AsyncTransport.h

   1 /*
   2  * Copyright 2015 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 #include <sys/uio.h>
  21
  22 #include <folly/io/async/DelayedDestruction.h>
  23 #include <folly/io/async/EventBase.h>
  24 #include <folly/io/async/AsyncSocketBase.h>
  25
  26 namespace folly {
  27
  28 class AsyncSocketException;
  29 class EventBase;
  30 class IOBuf;
  31 class SocketAddress;
  32
  33 /*
  34  * flags given by the application for write* calls
  35  */
  36 enum class WriteFlags : uint32_t {
  37   NONE = 0x00,
  38   /*
  39    * Whether to delay the output until a subsequent non-corked write.
  40    * (Note: may not be supported in all subclasses or on all platforms.)
  41    */
  42   CORK = 0x01,
  43   /*
  44    * for a socket that has ACK latency enabled, it will cause the kernel
  45    * to fire a TCP ESTATS event when the last byte of the given write call
  46    * will be acknowledged.
  47    */
  48   EOR = 0x02,
  49 };
  50
  51 /*
  52  * union operator
  53  */
  54 inline WriteFlags operator|(WriteFlags a, WriteFlags b) {
  55   return static_cast<WriteFlags>(
  56     static_cast<uint32_t>(a) | static_cast<uint32_t>(b));
  57 }
  58
  59 /*
  60  * intersection operator
  61  */
  62 inline WriteFlags operator&(WriteFlags a, WriteFlags b) {
  63   return static_cast<WriteFlags>(
  64     static_cast<uint32_t>(a) & static_cast<uint32_t>(b));
  65 }
  66
  67 /*
  68  * exclusion parameter
  69  */
  70 inline WriteFlags operator~(WriteFlags a) {
  71   return static_cast<WriteFlags>(~static_cast<uint32_t>(a));
  72 }
  73
  74 /*
  75  * unset operator
  76  */
  77 inline WriteFlags unSet(WriteFlags a, WriteFlags b) {
  78   return a & ~b;
  79 }
  80
  81 /*
  82  * inclusion operator
  83  */
  84 inline bool isSet(WriteFlags a, WriteFlags b) {
  85   return (a & b) == b;
  86 }
  87
  88
  89 /**
  90  * AsyncTransport defines an asynchronous API for streaming I/O.
  91  *
  92  * This class provides an API to for asynchronously waiting for data
  93  * on a streaming transport, and for asynchronously sending data.
  94  *
  95  * The APIs for reading and writing are intentionally asymmetric.  Waiting for
  96  * data to read is a persistent API: a callback is installed, and is notified
  97  * whenever new data is available.  It continues to be notified of new events
  98  * until it is uninstalled.
  99  *
 100  * AsyncTransport does not provide read timeout functionality, because it
 101  * typically cannot determine when the timeout should be active.  Generally, a
 102  * timeout should only be enabled when processing is blocked waiting on data
 103  * from the remote endpoint.  For server-side applications, the timeout should
 104  * not be active if the server is currently processing one or more outstanding
 105  * requests on this transport.  For client-side applications, the timeout
 106  * should not be active if there are no requests pending on the transport.
 107  * Additionally, if a client has multiple pending requests, it will ususally
 108  * want a separate timeout for each request, rather than a single read timeout.
 109  *
 110  * The write API is fairly intuitive: a user can request to send a block of
 111  * data, and a callback will be informed once the entire block has been
 112  * transferred to the kernel, or on error.  AsyncTransport does provide a send
 113  * timeout, since most callers want to give up if the remote end stops
 114  * responding and no further progress can be made sending the data.
 115  */
 116 class AsyncTransport : public DelayedDestruction, public AsyncSocketBase {
 117  public:
 118   typedef std::unique_ptr<AsyncTransport, Destructor> UniquePtr;
 119
 120   /**
 121    * Close the transport.
 122    *
 123    * This gracefully closes the transport, waiting for all pending write
 124    * requests to complete before actually closing the underlying transport.
 125    *
 126    * If a read callback is set, readEOF() will be called immediately.  If there
 127    * are outstanding write requests, the close will be delayed until all
 128    * remaining writes have completed.  No new writes may be started after
 129    * close() has been called.
 130    */
 131   virtual void close() = 0;
 132
 133   /**
 134    * Close the transport immediately.
 135    *
 136    * This closes the transport immediately, dropping any outstanding data
 137    * waiting to be written.
 138    *
 139    * If a read callback is set, readEOF() will be called immediately.
 140    * If there are outstanding write requests, these requests will be aborted
 141    * and writeError() will be invoked immediately on all outstanding write
 142    * callbacks.
 143    */
 144   virtual void closeNow() = 0;
 145
 146   /**
 147    * Reset the transport immediately.
 148    *
 149    * This closes the transport immediately, sending a reset to the remote peer
 150    * if possible to indicate abnormal shutdown.
 151    *
 152    * Note that not all subclasses implement this reset functionality: some
 153    * subclasses may treat reset() the same as closeNow().  Subclasses that use
 154    * TCP transports should terminate the connection with a TCP reset.
 155    */
 156   virtual void closeWithReset() {
 157     closeNow();
 158   }
 159
 160   /**
 161    * Perform a half-shutdown of the write side of the transport.
 162    *
 163    * The caller should not make any more calls to write() or writev() after
 164    * shutdownWrite() is called.  Any future write attempts will fail
 165    * immediately.
 166    *
 167    * Not all transport types support half-shutdown.  If the underlying
 168    * transport does not support half-shutdown, it will fully shutdown both the
 169    * read and write sides of the transport.  (Fully shutting down the socket is
 170    * better than doing nothing at all, since the caller may rely on the
 171    * shutdownWrite() call to notify the other end of the connection that no
 172    * more data can be read.)
 173    *
 174    * If there is pending data still waiting to be written on the transport,
 175    * the actual shutdown will be delayed until the pending data has been
 176    * written.
 177    *
 178    * Note: There is no corresponding shutdownRead() equivalent.  Simply
 179    * uninstall the read callback if you wish to stop reading.  (On TCP sockets
 180    * at least, shutting down the read side of the socket is a no-op anyway.)
 181    */
 182   virtual void shutdownWrite() = 0;
 183
 184   /**
 185    * Perform a half-shutdown of the write side of the transport.
 186    *
 187    * shutdownWriteNow() is identical to shutdownWrite(), except that it
 188    * immediately performs the shutdown, rather than waiting for pending writes
 189    * to complete.  Any pending write requests will be immediately failed when
 190    * shutdownWriteNow() is called.
 191    */
 192   virtual void shutdownWriteNow() = 0;
 193
 194   /**
 195    * Determine if transport is open and ready to read or write.
 196    *
 197    * Note that this function returns false on EOF; you must also call error()
 198    * to distinguish between an EOF and an error.
 199    *
 200    * @return  true iff the transport is open and ready, false otherwise.
 201    */
 202   virtual bool good() const = 0;
 203
 204   /**
 205    * Determine if the transport is readable or not.
 206    *
 207    * @return  true iff the transport is readable, false otherwise.
 208    */
 209   virtual bool readable() const = 0;
 210
 211   /**
 212    * Determine if the there is pending data on the transport.
 213    *
 214    * @return  true iff the if the there is pending data, false otherwise.
 215    */
 216   virtual bool isPending() const {
 217     return readable();
 218   }
 219   /**
 220    * Determine if transport is connected to the endpoint
 221    *
 222    * @return  false iff the transport is connected, otherwise true
 223    */
 224   virtual bool connecting() const = 0;
 225
 226   /**
 227    * Determine if an error has occurred with this transport.
 228    *
 229    * @return  true iff an error has occurred (not EOF).
 230    */
 231   virtual bool error() const = 0;
 232
 233   /**
 234    * Attach the transport to a EventBase.
 235    *
 236    * This may only be called if the transport is not currently attached to a
 237    * EventBase (by an earlier call to detachEventBase()).
 238    *
 239    * This method must be invoked in the EventBase's thread.
 240    */
 241   virtual void attachEventBase(EventBase* eventBase) = 0;
 242
 243   /**
 244    * Detach the transport from its EventBase.
 245    *
 246    * This may only be called when the transport is idle and has no reads or
 247    * writes pending.  Once detached, the transport may not be used again until
 248    * it is re-attached to a EventBase by calling attachEventBase().
 249    *
 250    * This method must be called from the current EventBase's thread.
 251    */
 252   virtual void detachEventBase() = 0;
 253
 254   /**
 255    * Determine if the transport can be detached.
 256    *
 257    * This method must be called from the current EventBase's thread.
 258    */
 259   virtual bool isDetachable() const = 0;
 260
 261   /**
 262    * Set the send timeout.
 263    *
 264    * If write requests do not make any progress for more than the specified
 265    * number of milliseconds, fail all pending writes and close the transport.
 266    *
 267    * If write requests are currently pending when setSendTimeout() is called,
 268    * the timeout interval is immediately restarted using the new value.
 269    *
 270    * @param milliseconds  The timeout duration, in milliseconds.  If 0, no
 271    *                      timeout will be used.
 272    */
 273   virtual void setSendTimeout(uint32_t milliseconds) = 0;
 274
 275   /**
 276    * Get the send timeout.
 277    *
 278    * @return Returns the current send timeout, in milliseconds.  A return value
 279    *         of 0 indicates that no timeout is set.
 280    */
 281   virtual uint32_t getSendTimeout() const = 0;
 282
 283   /**
 284    * Get the address of the local endpoint of this transport.
 285    *
 286    * This function may throw AsyncSocketException on error.
 287    *
 288    * @param address  The local address will be stored in the specified
 289    *                 SocketAddress.
 290    */
 291   virtual void getLocalAddress(SocketAddress* address) const = 0;
 292
 293   virtual void getAddress(SocketAddress* address) const {
 294     getLocalAddress(address);
 295   }
 296
 297   /**
 298    * Get the address of the remote endpoint to which this transport is
 299    * connected.
 300    *
 301    * This function may throw AsyncSocketException on error.
 302    *
 303    * @param address  The remote endpoint's address will be stored in the
 304    *                 specified SocketAddress.
 305    */
 306   virtual void getPeerAddress(SocketAddress* address) const = 0;
 307
 308   /**
 309    * @return True iff end of record tracking is enabled
 310    */
 311   virtual bool isEorTrackingEnabled() const = 0;
 312
 313   virtual void setEorTracking(bool track) = 0;
 314
 315   virtual size_t getAppBytesWritten() const = 0;
 316   virtual size_t getRawBytesWritten() const = 0;
 317   virtual size_t getAppBytesReceived() const = 0;
 318   virtual size_t getRawBytesReceived() const = 0;
 319
 320  protected:
 321   virtual ~AsyncTransport() = default;
 322 };
 323
 324 // Transitional intermediate interface. This is deprecated.
 325 // Wrapper around folly::AsyncTransport, that includes read/write callbacks
 326 class AsyncTransportWrapper : virtual public AsyncTransport {
 327  public:
 328   typedef std::unique_ptr<AsyncTransportWrapper, Destructor> UniquePtr;
 329
 330   class ReadCallback {
 331    public:
 332     virtual ~ReadCallback() = default;
 333
 334     /**
 335      * When data becomes available, getReadBuffer() will be invoked to get the
 336      * buffer into which data should be read.
 337      *
 338      * This method allows the ReadCallback to delay buffer allocation until
 339      * data becomes available.  This allows applications to manage large
 340      * numbers of idle connections, without having to maintain a separate read
 341      * buffer for each idle connection.
 342      *
 343      * It is possible that in some cases, getReadBuffer() may be called
 344      * multiple times before readDataAvailable() is invoked.  In this case, the
 345      * data will be written to the buffer returned from the most recent call to
 346      * readDataAvailable().  If the previous calls to readDataAvailable()
 347      * returned different buffers, the ReadCallback is responsible for ensuring
 348      * that they are not leaked.
 349      *
 350      * If getReadBuffer() throws an exception, returns a nullptr buffer, or
 351      * returns a 0 length, the ReadCallback will be uninstalled and its
 352      * readError() method will be invoked.
 353      *
 354      * getReadBuffer() is not allowed to change the transport state before it
 355      * returns.  (For example, it should never uninstall the read callback, or
 356      * set a different read callback.)
 357      *
 358      * @param bufReturn getReadBuffer() should update *bufReturn to contain the
 359      *                  address of the read buffer.  This parameter will never
 360      *                  be nullptr.
 361      * @param lenReturn getReadBuffer() should update *lenReturn to contain the
 362      *                  maximum number of bytes that may be written to the read
 363      *                  buffer.  This parameter will never be nullptr.
 364      */
 365     virtual void getReadBuffer(void** bufReturn, size_t* lenReturn) = 0;
 366
 367     /**
 368      * readDataAvailable() will be invoked when data has been successfully read
 369      * into the buffer returned by the last call to getReadBuffer().
 370      *
 371      * The read callback remains installed after readDataAvailable() returns.
 372      * It must be explicitly uninstalled to stop receiving read events.
 373      * getReadBuffer() will be called at least once before each call to
 374      * readDataAvailable().  getReadBuffer() will also be called before any
 375      * call to readEOF().
 376      *
 377      * @param len       The number of bytes placed in the buffer.
 378      */
 379     virtual void readDataAvailable(size_t len) noexcept = 0;
 380
 381     /**
 382      * readEOF() will be invoked when the transport is closed.
 383      *
 384      * The read callback will be automatically uninstalled immediately before
 385      * readEOF() is invoked.
 386      */
 387     virtual void readEOF() noexcept = 0;
 388
 389     /**
 390      * readError() will be invoked if an error occurs reading from the
 391      * transport.
 392      *
 393      * The read callback will be automatically uninstalled immediately before
 394      * readError() is invoked.
 395      *
 396      * @param ex        An exception describing the error that occurred.
 397      */
 398     virtual void readErr(const AsyncSocketException& ex) noexcept = 0;
 399   };
 400
 401   class WriteCallback {
 402    public:
 403     virtual ~WriteCallback() = default;
 404
 405     /**
 406      * writeSuccess() will be invoked when all of the data has been
 407      * successfully written.
 408      *
 409      * Note that this mainly signals that the buffer containing the data to
 410      * write is no longer needed and may be freed or re-used.  It does not
 411      * guarantee that the data has been fully transmitted to the remote
 412      * endpoint.  For example, on socket-based transports, writeSuccess() only
 413      * indicates that the data has been given to the kernel for eventual
 414      * transmission.
 415      */
 416     virtual void writeSuccess() noexcept = 0;
 417
 418     /**
 419      * writeError() will be invoked if an error occurs writing the data.
 420      *
 421      * @param bytesWritten      The number of bytes that were successfull
 422      * @param ex                An exception describing the error that occurred.
 423      */
 424     virtual void writeErr(size_t bytesWritten,
 425                           const AsyncSocketException& ex) noexcept = 0;
 426   };
 427
 428   // Read/write methods that aren't part of AsyncTransport
 429   virtual void setReadCB(ReadCallback* callback) = 0;
 430   virtual ReadCallback* getReadCallback() const = 0;
 431
 432   virtual void write(WriteCallback* callback, const void* buf, size_t bytes,
 433                      WriteFlags flags = WriteFlags::NONE) = 0;
 434   virtual void writev(WriteCallback* callback, const iovec* vec, size_t count,
 435                       WriteFlags flags = WriteFlags::NONE) = 0;
 436   virtual void writeChain(WriteCallback* callback,
 437                           std::unique_ptr<IOBuf>&& buf,
 438                           WriteFlags flags = WriteFlags::NONE) = 0;
 439 };
 440
 441 } // folly