HTTPSession.h

Go to the documentation of this file.

/*
 *  Copyright (c) 2015-present, Facebook, Inc.
 *  All rights reserved.
 *
 *  This source code is licensed under the BSD-style license found in the
 *  LICENSE file in the root directory of this source tree. An additional grant
 *  of patent rights can be found in the PATENTS file in the same directory.
 *
 */
#pragma once

#include <fizz/protocol/Certificate.h>
#include <fizz/record/Types.h>
#include <folly/IntrusiveList.h>
#include <folly/io/IOBufQueue.h>
#include <folly/io/async/EventBase.h>
#include <folly/io/async/HHWheelTimer.h>
#include <folly/Optional.h>
#include <proxygen/lib/http/HTTPConstants.h>
#include <proxygen/lib/http/HTTPHeaderSize.h>
#include <proxygen/lib/http/codec/FlowControlFilter.h>
#include <proxygen/lib/http/codec/HTTPCodec.h>
#include <proxygen/lib/http/codec/HTTPCodecFilter.h>
#include <proxygen/lib/http/session/ByteEventTracker.h>
#include <proxygen/lib/http/session/HTTPEvent.h>
#include <proxygen/lib/http/session/HTTPSessionBase.h>
#include <proxygen/lib/http/session/HTTPTransaction.h>
#include <proxygen/lib/http/session/SecondaryAuthManager.h>
#include <queue>
#include <set>
#include <folly/io/async/AsyncSocket.h>
#include <folly/io/async/AsyncSSLSocket.h>
#include <vector>
#include <proxygen/lib/utils/WheelTimerInstance.h>

namespace proxygen {

class HTTPSessionController;
class HTTPSessionStats;

#define PROXYGEN_HTTP_SESSION_USES_BASE  1

class HTTPSession:
  public HTTPSessionBase,
  public HTTPTransaction::Transport,
  public ByteEventTracker::Callback,
  protected folly::AsyncTransport::BufferCallback,
  protected HTTPPriorityMapFactoryProvider,
  private FlowControlFilter::Callback,
  private HTTPCodec::Callback,
  private folly::EventBase::LoopCallback,
  private folly::AsyncTransportWrapper::ReadCallback,
  private folly::AsyncTransport::ReplaySafetyCallback {
 public:
  using UniquePtr = std::unique_ptr<HTTPSession, Destructor>;

  HTTPSessionBase::SessionType getType() const noexcept override {
    return HTTPSessionBase::SessionType::HTTP;
  }

  folly::AsyncTransportWrapper* getTransport() override {
    return sock_.get();
  }

  folly::EventBase* getEventBase() const override {
    if (sock_) {
      return sock_->getEventBase();
    }
    return nullptr;
  }

  const folly::AsyncTransportWrapper* getTransport() const override {
    return sock_.get();
  }

  bool hasActiveTransactions() const override {
    return !transactions_.empty();
  }

  uint32_t getNumOutgoingStreams() const override {
    return outgoingStreams_;
  }

  uint32_t getNumIncomingStreams() const override {
    return incomingStreams_;
  }

  ByteEventTracker* getByteEventTracker() { return byteEventTracker_.get(); }

  void setByteEventTracker(std::shared_ptr<ByteEventTracker> byteEventTracker);

  void setSessionStats(HTTPSessionStats* stats) override;
  void setFlowControl(
   size_t initialReceiveWindow,
   size_t receiveStreamWindowSize,
   size_t receiveSessionWindowSize) override;

  void setEgressSettings(const SettingsList& inSettings) override;

  bool getHTTP2PrioritiesEnabled() const override {
    return HTTPSessionBase::getHTTP2PrioritiesEnabled();
  }

  void setHTTP2PrioritiesEnabled(bool enabled) override {
    HTTPSessionBase::setHTTP2PrioritiesEnabled(enabled);
  }

  const folly::SocketAddress& getLocalAddress() const noexcept override {
    return HTTPSessionBase::getLocalAddress();
  }

  const folly::SocketAddress& getPeerAddress() const noexcept override {
    return HTTPSessionBase::getPeerAddress();
  }

  const wangle::TransportInfo& getSetupTransportInfo() const noexcept
    override {
    return HTTPSessionBase::getSetupTransportInfo();
  }

  bool getCurrentTransportInfo(wangle::TransportInfo* tinfo) override;

  void setMaxConcurrentIncomingStreams(uint32_t num) override;

  void setEgressBytesLimit(uint64_t bytesLimit);

  void startNow() override;

  size_t sendSettings() override;

  size_t sendPing() override;

  HTTPCodec::StreamID sendPriority(http2::PriorityUpdate pri) override;

  size_t sendPriority(HTTPCodec::StreamID id, http2::PriorityUpdate pri)
    override;

  size_t sendCertificateRequest(
      std::unique_ptr<folly::IOBuf> certificateRequestContext,
      std::vector<fizz::Extension> extensions) override;

  // public ManagedConnection methods
  void timeoutExpired() noexcept override {
      readTimeoutExpired();
  }

  void describe(std::ostream& os) const override;
  bool isBusy() const override;
  void notifyPendingShutdown() override;
  void closeWhenIdle() override;
  void dropConnection() override;
  void dumpConnectionState(uint8_t loglevel) override;

  bool getCurrentTransportInfoWithoutUpdate(
    wangle::TransportInfo* tinfo) const override;

  void setHeaderCodecStats(HeaderCodec::Stats* stats) override {
    codec_->setHeaderCodecStats(stats);
  }

  void enableDoubleGoawayDrain() override {
    codec_->enableDoubleGoawayDrain();
  }

  bool connCloseByRemote() override {
    auto sock = getTransport()->getUnderlyingTransport<folly::AsyncSocket>();
    if (sock) {
      return sock->isClosedByPeer();
    }
    return false;
  }

  void setSecondAuthManager(
      std::unique_ptr<SecondaryAuthManager> secondAuthManager);

  SecondaryAuthManager* getSecondAuthManager() const;

 protected:
  HTTPSession(
      const WheelTimerInstance& timeout,
      folly::AsyncTransportWrapper::UniquePtr sock,
      const folly::SocketAddress& localAddr,
      const folly::SocketAddress& peerAddr,
      HTTPSessionController* controller,
      std::unique_ptr<HTTPCodec> codec,
      const wangle::TransportInfo& tinfo,
      InfoCallback* infoCallback);

  // thrift uses WheelTimer
  HTTPSession(
      folly::HHWheelTimer* transactionTimeouts,
      folly::AsyncTransportWrapper::UniquePtr sock,
      const folly::SocketAddress& localAddr,
      const folly::SocketAddress& peerAddr,
      HTTPSessionController* controller,
      std::unique_ptr<HTTPCodec> codec,
      const wangle::TransportInfo& tinfo,
      InfoCallback* infoCallback);

  ~HTTPSession() override;

  virtual void setupOnHeadersComplete(HTTPTransaction* txn,
                                      HTTPMessage* msg) = 0;

  virtual HTTPTransaction::Handler* getTransactionTimeoutHandler(
    HTTPTransaction* txn) = 0;

  virtual void onHeadersSent(
      const HTTPMessage& /* headers */,
      bool /* codecWasReusable */) {}

  virtual bool allTransactionsStarted() const = 0;

  void setNewTransactionPauseState(HTTPCodec::StreamID streamID);

  void onEgressMessageFinished(HTTPTransaction* txn,
                               bool withRST = false);

  std::unique_ptr<folly::IOBuf> getNextToSend(bool* cork, bool* eom);

  void decrementTransactionCount(HTTPTransaction* txn,
                                 bool ingressEOM, bool egressEOM);

  size_t getCodecSendWindowSize() const;

  size_t sendPriorityImpl(HTTPCodec::StreamID streamID,
                          http2::PriorityUpdate pri);

  bool onNativeProtocolUpgradeImpl(HTTPCodec::StreamID txn,
                                   std::unique_ptr<HTTPCodec> codec,
                                   const std::string& protocolString);

  virtual folly::Optional<const HTTPMessage::HTTPPriority> getHTTPPriority(
    uint8_t) override {
    return folly::none;
  }

  void readTimeoutExpired() noexcept;
  void writeTimeoutExpired() noexcept;
  void flowControlTimeoutExpired() noexcept;

  // AsyncTransportWrapper::ReadCallback methods
  void getReadBuffer(void** buf, size_t* bufSize) override;
  void readDataAvailable(size_t readSize) noexcept override;
  bool isBufferMovable() noexcept override;
  void readBufferAvailable(std::unique_ptr<folly::IOBuf>) noexcept override;
  void processReadData();
  void readEOF() noexcept override;
  void readErr(
      const folly::AsyncSocketException&) noexcept override;

  std::string getSecurityProtocol() const override {
    return sock_->getSecurityProtocol();
  }

  // HTTPCodec::Callback methods
  void onMessageBegin(HTTPCodec::StreamID streamID, HTTPMessage* msg) override;
  void onPushMessageBegin(HTTPCodec::StreamID streamID,
                          HTTPCodec::StreamID assocStreamID,
                          HTTPMessage* msg) override;
  void onExMessageBegin(HTTPCodec::StreamID streamID,
                        HTTPCodec::StreamID controlStream,
                        bool unidirectional,
                        HTTPMessage* msg) override;
  void onHeadersComplete(HTTPCodec::StreamID streamID,
                         std::unique_ptr<HTTPMessage> msg) override;
  void onBody(HTTPCodec::StreamID streamID,
              std::unique_ptr<folly::IOBuf> chain, uint16_t padding) override;
  void onChunkHeader(HTTPCodec::StreamID stream, size_t length) override;
  void onChunkComplete(HTTPCodec::StreamID stream) override;
  void onTrailersComplete(HTTPCodec::StreamID streamID,
      std::unique_ptr<HTTPHeaders> trailers) override;
  void onMessageComplete(HTTPCodec::StreamID streamID, bool upgrade) override;
  void onError(HTTPCodec::StreamID streamID,
               const HTTPException& error, bool newTxn) override;
  void onAbort(HTTPCodec::StreamID streamID,
               ErrorCode code) override;
  void onGoaway(uint64_t lastGoodStreamID,
                ErrorCode code,
                std::unique_ptr<folly::IOBuf> debugData = nullptr) override;
  void onPingRequest(uint64_t uniqueID) override;
  void onPingReply(uint64_t uniqueID) override;
  void onWindowUpdate(HTTPCodec::StreamID stream, uint32_t amount) override;
  void onSettings(const SettingsList& settings) override;
  void onSettingsAck()  override;
  void onPriority(HTTPCodec::StreamID stream,
                  const HTTPMessage::HTTPPriority&) override;
  void onCertificateRequest(uint16_t requestId,
                            std::unique_ptr<folly::IOBuf> authRequest) override;
  void onCertificate(uint16_t certId,
                     std::unique_ptr<folly::IOBuf> authenticator) override;
  uint32_t numOutgoingStreams() const override {
    return outgoingStreams_;
  }
  uint32_t numIncomingStreams() const override {
    return incomingStreams_;
  }

  // HTTPTransaction::Transport methods
  void pauseIngress(HTTPTransaction* txn) noexcept override;
  void resumeIngress(HTTPTransaction* txn) noexcept override;
  void transactionTimeout(HTTPTransaction* txn) noexcept override;
  void sendHeaders(HTTPTransaction* txn,
                   const HTTPMessage& headers,
                   HTTPHeaderSize* size,
                   bool includeEOM) noexcept override;
  size_t sendBody(HTTPTransaction* txn, std::unique_ptr<folly::IOBuf>,
                  bool includeEOM, bool trackLastByteFlushed) noexcept override;
  size_t sendChunkHeader(HTTPTransaction* txn,
                         size_t length) noexcept override;
  size_t sendChunkTerminator(HTTPTransaction* txn) noexcept override;
  size_t sendEOM(HTTPTransaction* txn,
                 const HTTPHeaders* trailers) noexcept override;
  size_t sendAbort(HTTPTransaction* txn,
                   ErrorCode statusCode) noexcept override;
  size_t sendPriority(HTTPTransaction* txn,
                      const http2::PriorityUpdate& pri) noexcept override;
  void detach(HTTPTransaction* txn) noexcept override;
  size_t sendWindowUpdate(HTTPTransaction* txn,
                          uint32_t bytes) noexcept override;
  void notifyPendingEgress() noexcept override;
  void notifyIngressBodyProcessed(uint32_t bytes) noexcept override;
  void notifyEgressBodyBuffered(int64_t bytes) noexcept override;
  HTTPTransaction* newPushedTransaction(
    HTTPCodec::StreamID assocStreamId,
    HTTPTransaction::PushHandler* handler) noexcept override;
  HTTPTransaction* newExTransaction(
    HTTPTransaction::Handler* handler,
    HTTPCodec::StreamID controlStream,
    bool unidirectional = false) noexcept override;

  const HTTPCodec& getCodec() const noexcept override {
    return codec_.getChainEnd();
  }

  const folly::AsyncTransportWrapper* getUnderlyingTransport()
      const noexcept override {
    return sock_.get();
  }

  bool isDraining() const override { return draining_; }

  void drain() override;

  void shutdownTransport(bool shutdownReads = true,
                         bool shutdownWrites = true,
                         const std::string& errorMsg = "");

  void shutdownTransportWithReset(ProxygenError errorCode,
                                  const std::string& errorMsg = "");

  // EventBase::LoopCallback methods
  void runLoopCallback() noexcept override;

  void scheduleWrite();

  void updateWriteCount();
  void updateWriteBufSize(int64_t delta);

  inline uint64_t sessionByteOffset() {
    return bytesScheduled_ + writeBuf_.chainLength();
  }

  void immediateShutdown();

  void checkForShutdown();

  HTTPTransaction* findTransaction(HTTPCodec::StreamID streamID);

  HTTPTransaction* createTransaction(
    HTTPCodec::StreamID streamID,
    const folly::Optional<HTTPCodec::StreamID>& assocStreamID,
    const folly::Optional<HTTPCodec::ExAttributes>& exAttributes,
    const http2::PriorityUpdate& priority = http2::DefaultPriority);

  void onWriteSuccess(uint64_t bytesWritten);

  void onWriteError(size_t bytesWritten,
      const folly::AsyncSocketException& ex);

  void onWriteCompleted();

  void pauseReads();

  void onSessionParseError(const HTTPException& error);

  void onNewTransactionParseError(HTTPCodec::StreamID streamID,
                                  const HTTPException& error);

  void resumeReads();

  bool hasMoreWrites() const;

  template<typename... Args1, typename... Args2>
  void invokeOnAllTransactions(void (HTTPTransaction::*fn)(Args1...),
                               Args2&&... args) {
    DestructorGuard g(this);
    std::vector<HTTPCodec::StreamID> ids;
    for (const auto& txn: transactions_) {
      ids.push_back(txn.first);
    }
    for (auto idit = ids.begin(); idit != ids.end() && !transactions_.empty();
         ++idit) {
      auto txn = findTransaction(*idit);
      if (txn != nullptr) {
        (txn->*fn)(std::forward<Args2>(args)...);
      }
    }
  }

  void resumeTransactions();

  void errorOnAllTransactions(ProxygenError err, const std::string& errorMsg);

  void errorOnTransactionIds(const std::vector<HTTPCodec::StreamID>& ids,
                             ProxygenError err,
                             const std::string& extraErrorMsg = "");

  void errorOnTransactionId(HTTPCodec::StreamID id, HTTPException ex);

  bool shouldShutdown() const;

  void drainImpl();

  void pauseReadsImpl();
  void resumeReadsImpl();

  bool readsUnpaused() const {
    return reads_ == SocketState::UNPAUSED;
  }

  bool readsPaused() const {
    return reads_ == SocketState::PAUSED;
  }

  bool readsShutdown() const {
    return reads_ == SocketState::SHUTDOWN;
  }

  bool writesUnpaused() const {
    return writes_ == SocketState::UNPAUSED;
  }

  bool writesPaused() const {
    return writes_ == SocketState::PAUSED;
  }

  bool writesShutdown() const {
    return writes_ == SocketState::SHUTDOWN;
  }

  void rescheduleLoopCallbacks() {
    if (!isLoopCallbackScheduled()) {
      sock_->getEventBase()->runInLoop(this);
    }

    if (shutdownTransportCb_ &&
        !shutdownTransportCb_->isLoopCallbackScheduled()) {
      sock_->getEventBase()->runInLoop(shutdownTransportCb_.get(), true);
    }
  }

  void cancelLoopCallbacks() {
    if (isLoopCallbackScheduled()) {
      cancelLoopCallback();
    }
    if (shutdownTransportCb_) {
      shutdownTransportCb_->cancelLoopCallback();
    }
  }

  // protected members
  class WriteTimeout :
      public folly::HHWheelTimer::Callback {
   public:
    explicit WriteTimeout(HTTPSession* session) : session_(session) {}
    ~WriteTimeout() override {}

    void timeoutExpired() noexcept override { session_->writeTimeoutExpired(); }
   private:
    HTTPSession* session_;
  };
  WriteTimeout writeTimeout_;

  folly::IOBufQueue writeBuf_{folly::IOBufQueue::cacheChainLength()};

  folly::IOBufQueue readBuf_{folly::IOBufQueue::cacheChainLength()};

  HTTP2PriorityQueue txnEgressQueue_;

  std::map<HTTPCodec::StreamID, HTTPTransaction> transactions_;

  uint32_t liveTransactions_{0};

  folly::AsyncTransportWrapper::UniquePtr sock_;

  WheelTimerInstance timeout_;

  unsigned numActiveWrites_{0};

  bool draining_:1;

  bool started_:1;

  bool writesDraining_:1;
  bool resetAfterDrainingWrites_:1;
  bool ingressError_:1;

 private:
  uint32_t getMaxConcurrentOutgoingStreamsRemote() const override {
    return maxConcurrentOutgoingStreamsRemote_;
  }

  bool isUpstream() const;
  bool isDownstream() const;

  // from folly::AsyncTransport::BufferCallback
  void onEgressBuffered() override;
  void onEgressBufferCleared() override;

  void setupCodec();
  void onSetSendWindow(uint32_t windowSize);
  void onSetMaxInitiatedStreams(uint32_t maxTxns);

  uint32_t getCertAuthSettingVal();

  bool verifyCertAuthSetting(uint32_t value);

  void addLastByteEvent(HTTPTransaction* txn, uint64_t byteNo) noexcept;

  void addAckToLastByteEvent(HTTPTransaction* txn,
                             const ByteEvent& lastByteEvent);

  void onConnectionSendWindowOpen() override;
  void onConnectionSendWindowClosed() override;

  HTTPCodec::StreamID getGracefulGoawayAck() const;

  void invalidStream(HTTPCodec::StreamID stream,
                     ErrorCode code = ErrorCode::_SPDY_INVALID_STREAM);

  http2::PriorityUpdate getMessagePriority(const HTTPMessage* msg);

  bool isConnWindowFull() const {
    return connFlowControl_ && connFlowControl_->getAvailableSend() == 0;
  }

  //ByteEventTracker::Callback functions
  void onPingReplyLatency(int64_t latency) noexcept override;
  void onLastByteEvent(HTTPTransaction* txn,
                       uint64_t offset, bool eomTracked) noexcept override;
  void onDeleteAckEvent() noexcept override;

  void commonEom(
      HTTPTransaction* txn,
      size_t encodedSize,
      bool piggybacked) noexcept;

  void addWaitingForReplaySafety(
      ReplaySafetyCallback* callback) noexcept override {
    if (sock_->isReplaySafe()) {
      callback->onReplaySafe();
    } else {
      waitingForReplaySafety_.push_back(callback);
    }
  }

  void removeWaitingForReplaySafety(
      ReplaySafetyCallback* callback) noexcept override {
    waitingForReplaySafety_.remove(callback);
  }

  bool needToBlockForReplaySafety() const override {
    return !waitingForReplaySafety_.empty();
  }

  void onReplaySafe() noexcept override;

  // Returns the number of streams that count against a pipelining limit.
  // Upstreams can't really pipleine (send responses before requests), so
  // count ANY stream against the limit.
  size_t getPipelineStreamCount() const {
    return isDownstream() ? incomingStreams_ : transactions_.size();
  }

  bool maybeResumePausedPipelinedTransaction(size_t oldStreamCount,
                                             uint32_t txnSeqn);

  void incrementOutgoingStreams();

  // private members

  std::list<ReplaySafetyCallback*> waitingForReplaySafety_;

  class WriteSegment :
    public folly::AsyncTransportWrapper::WriteCallback {
   public:
    WriteSegment(HTTPSession* session, uint64_t length);

    void setCork(bool cork) {
      if (cork) {
        flags_ = flags_ | folly::WriteFlags::CORK;
      } else {
        unSet(flags_, folly::WriteFlags::CORK);
      }
    }

    void setEOR(bool eor) {
      if (eor) {
        flags_ = flags_ | folly::WriteFlags::EOR;
      } else {
        unSet(flags_, folly::WriteFlags::EOR);
      }
    }

    void detach();

    folly::WriteFlags getFlags() {
      return flags_;
    }

    uint64_t getLength() const {
      return length_;
    }

    // AsyncTransport::WriteCallback methods
    void writeSuccess() noexcept override;
    void writeErr(
        size_t bytesWritten,
        const folly::AsyncSocketException&) noexcept override;

    folly::IntrusiveListHook listHook;
   private:

    void remove();

    HTTPSession* session_;
    uint64_t length_;
    folly::WriteFlags flags_{
      folly::WriteFlags::NONE};
  };
  using WriteSegmentList =
    folly::IntrusiveList<WriteSegment, &WriteSegment::listHook>;
  WriteSegmentList pendingWrites_;

  FlowControlFilter* connFlowControl_{nullptr};

  uint32_t maxConcurrentOutgoingStreamsRemote_{100000};

  uint32_t maxConcurrentIncomingStreams_{100};

  uint32_t outgoingStreams_{0};

  uint32_t incomingStreams_{0};

  uint64_t bytesWritten_{0};

  uint64_t bytesScheduled_{0};

  int64_t pendingWriteSizeDelta_{0};

  uint64_t bodyBytesPerWriteBuf_{0};

  HTTP2PriorityQueue::NextEgressResult nextEgressResults_;

  std::shared_ptr<ByteEventTracker> byteEventTracker_{nullptr};

  uint64_t egressBytesLimit_{0};

  // Flow control settings
  size_t initialReceiveWindow_{0};
  size_t receiveStreamWindowSize_{0};
  size_t receiveSessionWindowSize_{0};

  class ShutdownTransportCallback : public folly::EventBase::LoopCallback {
   public:
    explicit ShutdownTransportCallback(HTTPSession* session) :
        session_(session),
        dg_(std::make_unique<DestructorGuard>(session)) { }

    ~ShutdownTransportCallback() override { }

    void runLoopCallback() noexcept override {
        VLOG(4) << *session_ << " shutdown from onEgressMessageFinished";
        bool shutdownReads =
            session_->isDownstream() && !session_->ingressUpgraded_;
        session_->shutdownTransport(shutdownReads, true);
        dg_.reset();
    }

  private:
    HTTPSession* session_;
    std::unique_ptr<DestructorGuard> dg_;
  };
  std::unique_ptr<ShutdownTransportCallback> shutdownTransportCb_;

  class FlowControlTimeout : public folly::HHWheelTimer::Callback {
   public:
    explicit FlowControlTimeout(HTTPSession* session) : session_(session) {}
    ~FlowControlTimeout() override {}

    void timeoutExpired() noexcept override {
      session_->flowControlTimeoutExpired();
    }

    std::chrono::milliseconds getTimeoutDuration() const {
      return duration_;
    }

    void setTimeoutDuration(std::chrono::milliseconds duration) {
      duration_ = duration;
    }
   private:
    HTTPSession* session_;
    std::chrono::milliseconds duration_{std::chrono::milliseconds(0)};
  };
  FlowControlTimeout flowControlTimeout_;

  class DrainTimeout : public folly::HHWheelTimer::Callback {
   public:
    explicit DrainTimeout(HTTPSession* session) : session_(session) {}
    ~DrainTimeout() override {}

    void timeoutExpired() noexcept override {
      session_->closeWhenIdle();
    }
   private:
    HTTPSession* session_;
  };
  DrainTimeout drainTimeout_;

  enum SocketState {
    UNPAUSED = 0,
    PAUSED = 1,
    SHUTDOWN = 2,
  };

  SocketState reads_:2;
  SocketState writes_:2;

  bool ingressUpgraded_:1;
  bool resetSocketOnShutdown_:1;
  // indicates a fatal error that prevents further ingress data processing
  bool inLoopCallback_:1;
  bool inResume_:1;
  bool pendingPause_:1;

  // secondary authentication manager
  std::unique_ptr<SecondaryAuthManager> secondAuthManager_;
};

} // proxygen