proxygen: proxygen::FlowControlFilter Class Reference

#include <FlowControlFilter.h>

Inheritance diagram for proxygen::FlowControlFilter:

Classes
class	Callback

Public Member Functions
	FlowControlFilter (Callback &callback, folly::IOBufQueue &writeBuf, HTTPCodec *codec, uint32_t recvCapacity=0)

void	setReceiveWindowSize (folly::IOBufQueue &writeBuf, uint32_t capacity)

bool	ingressBytesProcessed (folly::IOBufQueue &writeBuf, uint32_t delta)

uint32_t	getAvailableSend () const

bool	isReusable () const override

void	onBody (StreamID stream, std::unique_ptr< folly::IOBuf > chain, uint16_t padding) override

void	onWindowUpdate (StreamID stream, uint32_t amount) override

size_t	generateBody (folly::IOBufQueue &writeBuf, StreamID stream, std::unique_ptr< folly::IOBuf > chain, folly::Optional< uint8_t > padding, bool eom) override

size_t	generateWindowUpdate (folly::IOBufQueue &writeBuf, StreamID stream, uint32_t delta) override

Public Member Functions inherited from proxygen::PassThroughHTTPCodecFilter
	PassThroughHTTPCodecFilter (bool calls=true, bool callbacks=true)

void	onMessageBegin (StreamID stream, HTTPMessage *msg) override

void	onPushMessageBegin (StreamID stream, StreamID assocStream, HTTPMessage *msg) override

void	onExMessageBegin (StreamID stream, StreamID controlStream, bool unidirectional, HTTPMessage *msg) override

void	onHeadersComplete (StreamID stream, std::unique_ptr< HTTPMessage > msg) override

void	onBody (StreamID stream, std::unique_ptr< folly::IOBuf > chain, uint16_t padding) override

void	onChunkHeader (StreamID stream, size_t length) override

void	onChunkComplete (StreamID stream) override

void	onTrailersComplete (StreamID stream, std::unique_ptr< HTTPHeaders > trailers) override

void	onMessageComplete (StreamID stream, bool upgrade) override

void	onFrameHeader (StreamID stream_id, uint8_t flags, uint64_t length, uint8_t type, uint16_t version=0) override

void	onError (StreamID stream, const HTTPException &error, bool newStream=false) override

void	onAbort (StreamID stream, 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 (StreamID stream, uint32_t amount) override

void	onSettings (const SettingsList &settings) override

void	onSettingsAck () override

void	onPriority (StreamID stream, const HTTPMessage::HTTPPriority &pri) override

bool	onNativeProtocolUpgrade (StreamID stream, CodecProtocol protocol, const std::string &protocolString, HTTPMessage &msg) override

void	onGenerateFrameHeader (StreamID streamID, uint8_t type, uint64_t length, uint16_t version) 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

uint32_t	numIncomingStreams () const override

HPACKTableInfo	getHPACKTableInfo () const override

CodecProtocol	getProtocol () const override

const std::string &	getUserAgent () const override

TransportDirection	getTransportDirection () const override

bool	supportsStreamFlowControl () const override

bool	supportsSessionFlowControl () const override

StreamID	createStream () override

void	setCallback (HTTPCodec::Callback *callback) override

bool	isBusy () const override

void	setParserPaused (bool paused) override

size_t	onIngress (const folly::IOBuf &buf) override

void	onIngressEOF () override

bool	onIngressUpgradeMessage (const HTTPMessage &msg) override

bool	isReusable () const override

bool	isWaitingToDrain () const override

bool	closeOnEgressComplete () const override

bool	supportsParallelRequests () const override

bool	supportsPushTransactions () const override

size_t	generateConnectionPreface (folly::IOBufQueue &writeBuf) override

void	generateHeader (folly::IOBufQueue &writeBuf, StreamID stream, const HTTPMessage &msg, bool eom, HTTPHeaderSize *size) override

void	generatePushPromise (folly::IOBufQueue &writeBuf, StreamID stream, const HTTPMessage &msg, StreamID assocStream, bool eom, HTTPHeaderSize *size) override

void	generateExHeader (folly::IOBufQueue &writeBuf, StreamID stream, const HTTPMessage &msg, const HTTPCodec::ExAttributes &exAttributes, bool eom, HTTPHeaderSize *size) override

size_t	generateBody (folly::IOBufQueue &writeBuf, StreamID stream, std::unique_ptr< folly::IOBuf > chain, folly::Optional< uint8_t > padding, bool eom) override

size_t	generateChunkHeader (folly::IOBufQueue &writeBuf, StreamID stream, size_t length) override

size_t	generateChunkTerminator (folly::IOBufQueue &writeBuf, StreamID stream) override

size_t	generateTrailers (folly::IOBufQueue &writeBuf, StreamID stream, const HTTPHeaders &trailers) override

size_t	generateEOM (folly::IOBufQueue &writeBuf, StreamID stream) override

size_t	generateRstStream (folly::IOBufQueue &writeBuf, StreamID stream, ErrorCode statusCode) override

size_t	generateGoaway (folly::IOBufQueue &writeBuf, StreamID lastStream, ErrorCode statusCode, std::unique_ptr< folly::IOBuf > debugData=nullptr) override

size_t	generatePingRequest (folly::IOBufQueue &writeBuf) override

size_t	generatePingReply (folly::IOBufQueue &writeBuf, uint64_t uniqueID) override

size_t	generateSettings (folly::IOBufQueue &writeBuf) override

size_t	generateSettingsAck (folly::IOBufQueue &writeBuf) override

size_t	generateWindowUpdate (folly::IOBufQueue &writeBuf, StreamID stream, uint32_t delta) override

size_t	generatePriority (folly::IOBufQueue &writeBuf, StreamID stream, const HTTPMessage::HTTPPriority &pri) override

size_t	generateCertificateRequest (folly::IOBufQueue &writeBuf, uint16_t requestId, std::unique_ptr< folly::IOBuf > chain) override

size_t	generateCertificate (folly::IOBufQueue &writeBuf, uint16_t certId, std::unique_ptr< folly::IOBuf > certData) override

HTTPSettings *	getEgressSettings () override

const HTTPSettings *	getIngressSettings () const override

void	setHeaderCodecStats (HeaderCodec::Stats *stats) override

void	enableDoubleGoawayDrain () override

HTTPCodec::StreamID	getLastIncomingStreamID () const override

uint32_t	getDefaultWindowSize () const override

size_t	addPriorityNodes (PriorityQueue &queue, folly::IOBufQueue &writeBuf, uint8_t maxLevel) override

StreamID	mapPriorityToDependency (uint8_t priority) const override

int8_t	mapDependencyToPriority (StreamID parent) const override

Public Member Functions inherited from proxygen::GenericFilter< T1, T2, set_callback, TakeOwnership, Dp >
	GenericFilter (bool calls, bool callbacks)

	~GenericFilter () override

void	append (Filter *nextFilter)

Private Attributes
Callback &	notify_

Window	recvWindow_

Window	sendWindow_

int32_t	toAck_ {0}

bool	error_:1

bool	sendsBlocked_:1

Additional Inherited Members
Public Types inherited from proxygen::GenericFilter< T1, T2, set_callback, TakeOwnership, Dp >
using	Filter = GenericFilter< T1, T2, set_callback, TakeOwnership, Dp >

Public Attributes inherited from proxygen::GenericFilter< T1, T2, set_callback, TakeOwnership, Dp >
const bool	kWantsCalls_

const bool	kWantsCallbacks_

Protected Member Functions inherited from proxygen::GenericFilter< T1, T2, set_callback, TakeOwnership, Dp >
void	setCallbackInternal (T2 *cb)

void	drop ()

Protected Attributes inherited from proxygen::GenericFilter< T1, T2, set_callback, TakeOwnership, Dp >
T1 *	call_ {nullptr}

T2 *	callback_ {nullptr}

Detailed Description

This class implements the logic for managing per-session flow control. Not every codec is interested in per-session flow control, so this filter can only be added in that case or else it is an error.

Definition at line 26 of file FlowControlFilter.h.

Constructor & Destructor Documentation

proxygen::FlowControlFilter::FlowControlFilter	(	Callback &	callback,
		folly::IOBufQueue &	writeBuf,
		HTTPCodec *	codec,
		uint32_t	recvCapacity = `0`
	)

explicit

Construct a flow control filter.

Parameters

callback	A channel to be notified when the window is not full anymore.
writeBuf	The buffer to write egress on. This constructor may generate a window update frame on this buffer.
codec	The codec implementation.
recvCapacity	The initial size of the conn-level recv window. It must be >= codec->getDefaultWindowSize(), or it will generate an immediate window update into writeBuf. 0 means use the codec default.

Definition at line 23 of file FlowControlFilter.cpp.

References proxygen::HTTPCodec::generateWindowUpdate(), proxygen::PassThroughHTTPCodecFilter::getDefaultWindowSize(), proxygen::HTTPCodec::getDefaultWindowSize(), recvWindow_, and proxygen::Window::setCapacity().

                                                            :
     notify_(callback),
     recvWindow_(codec->getDefaultWindowSize()),
     sendWindow_(codec->getDefaultWindowSize()),
     error_(false),
     sendsBlocked_(false) {
   if (recvCapacity > 0) {
     if (recvCapacity < codec->getDefaultWindowSize()) {
       VLOG(4) << "Ignoring low conn-level recv window size of " << recvCapacity;
     } else if (recvCapacity > codec->getDefaultWindowSize()) {
       auto delta = recvCapacity - codec->getDefaultWindowSize();
       VLOG(4) << "Incrementing default conn-level recv window by " << delta;
       CHECK(recvWindow_.setCapacity(recvCapacity));
       codec->generateWindowUpdate(writeBuf, 0, delta);
     }
   }
 }

Member Function Documentation

size_t proxygen::FlowControlFilter::generateBody	(	folly::IOBufQueue &	writeBuf,
		StreamID	stream,
		std::unique_ptr< folly::IOBuf >	chain,
		folly::Optional< uint8_t >	padding,
		bool	eom
	)

override

Definition at line 145 of file FlowControlFilter.cpp.

References proxygen::GenericFilter< T1, T2, set_callback, TakeOwnership, Dp >::call_, folly::IOBuf::computeChainDataLength(), proxygen::Window::getNonNegativeSize(), proxygen::Window::getSize(), folly::gen::move, notify_, proxygen::FlowControlFilter::Callback::onConnectionSendWindowClosed(), proxygen::Window::reserve(), sendsBlocked_, sendWindow_, and uint8_t.

                                                  {
   uint8_t padLen = padding ? *padding : 0;
   bool success = sendWindow_.reserve(
     chain->computeChainDataLength() + padLen);
   VLOG(5) << "Sending " << chain->computeChainDataLength()
           << " bytes, sendWindow=" << sendWindow_.getSize();
 
   // In the future, maybe make this DCHECK
   CHECK(success) << "Session-level send window underflowed! "
                  << "Too much data sent without WINDOW_UPDATES!";
 
   if (sendWindow_.getNonNegativeSize() == 0) {
     // Need to inform when the send window is no longer full
     VLOG(4) << "Send window closed";
     sendsBlocked_ = true;
     notify_.onConnectionSendWindowClosed();
   }
 
   return call_->generateBody(writeBuf, stream, std::move(chain), padding,
                              eom);
 }

size_t proxygen::FlowControlFilter::generateWindowUpdate	(	folly::IOBufQueue &	writeBuf,
		StreamID	stream,
		uint32_t	delta
	)

override

Definition at line 171 of file FlowControlFilter.cpp.

References proxygen::GenericFilter< T1, T2, set_callback, TakeOwnership, Dp >::call_.

                                                                {
   CHECK(stream) << " someone tried to manually manipulate a conn-level window";
   return call_->generateWindowUpdate(writeBuf, stream, delta);
 }

uint32_t proxygen::FlowControlFilter::getAvailableSend ( ) const

Returns: the number of bytes available in the connection-level send window

Definition at line 85 of file FlowControlFilter.cpp.

References proxygen::Window::getNonNegativeSize(), and sendWindow_.

Referenced by proxygen::HTTPSession::getNextToSend(), proxygen::HTTPSession::isConnWindowFull(), and proxygen::HTTPSession::runLoopCallback().

                                                    {
   return sendWindow_.getNonNegativeSize();
 }

bool proxygen::FlowControlFilter::ingressBytesProcessed	(	folly::IOBufQueue &	writeBuf,
		uint32_t	delta
	)

Notify the flow control filter that some ingress bytes were processed. If the number of bytes to acknowledge exceeds half the receive window's capacity, a WINDOW_UPDATE frame will be written.

Parameters

writeBuf	The buffer to write egress on. This function may generate a WINDOW_UPDATE on this buffer.
delta	The number of bytes that were processed.

Returns: true iff we wrote a WINDOW_UPDATE frame to the write buf.

Definition at line 69 of file FlowControlFilter.cpp.

References proxygen::GenericFilter< T1, T2, set_callback, TakeOwnership, Dp >::call_, proxygen::Window::free(), proxygen::Window::getCapacity(), recvWindow_, toAck_, and uint32_t.

Referenced by proxygen::HTTPSession::notifyIngressBodyProcessed(), and proxygen::HTTPSession::onBody().

                                                               {
   toAck_ += delta;
   bool willAck = (toAck_ > 0 &&
                   uint32_t(toAck_) > recvWindow_.getCapacity() / 2);
   VLOG(4) << "processed " << delta << " toAck_=" << toAck_
           << " bytes, will ack=" << willAck;
   if (willAck) {
     CHECK(recvWindow_.free(toAck_));
     call_->generateWindowUpdate(writeBuf, 0, toAck_);
     toAck_ = 0;
     return true;
   }
   return false;
 }

bool proxygen::FlowControlFilter::isReusable ( ) const

override

Definition at line 89 of file FlowControlFilter.cpp.

References proxygen::GenericFilter< T1, T2, set_callback, TakeOwnership, Dp >::call_, and error_.

                                          {
   if (error_) {
     return false;
   }
   return call_->isReusable();
 }

void proxygen::FlowControlFilter::onBody	(	StreamID	stream,
		std::unique_ptr< folly::IOBuf >	chain,
		uint16_t	padding
	)

override

Definition at line 96 of file FlowControlFilter.cpp.

References proxygen::GenericFilter< T1, T2, set_callback, TakeOwnership, Dp >::callback_, folly::IOBuf::computeChainDataLength(), error_, proxygen::Window::free(), proxygen::Window::getSize(), folly::gen::move, recvWindow_, proxygen::Window::reserve(), toAck_, and uint64_t.

                                                  {
   uint64_t amount = chain->computeChainDataLength();
   if (!recvWindow_.reserve(amount + padding)) {
     error_ = true;
     HTTPException ex = getException(
       folly::to<std::string>(
         "Failed to reserve receive window, window size=",
         recvWindow_.getSize(), ", amount=", amount));
     callback_->onError(0, ex, false);
   } else {
     if (VLOG_IS_ON(4) && recvWindow_.getSize() == 0) {
       VLOG(4) << "recvWindow full";
     }
     toAck_ += padding;
     CHECK(recvWindow_.free(padding));
     callback_->onBody(stream, std::move(chain), padding);
   }
 }

void proxygen::FlowControlFilter::onWindowUpdate	(	StreamID	stream,
		uint32_t	amount
	)

override

Definition at line 117 of file FlowControlFilter.cpp.

References proxygen::GenericFilter< T1, T2, set_callback, TakeOwnership, Dp >::callback_, error_, proxygen::Window::free(), proxygen::Window::getNonNegativeSize(), proxygen::Window::getOutstanding(), proxygen::Window::getSize(), notify_, proxygen::FlowControlFilter::Callback::onConnectionSendWindowOpen(), sendsBlocked_, and sendWindow_.

                                                                        {
   if (!stream) {
     bool success = sendWindow_.free(amount);
     VLOG(4) << "Remote side ack'd " << amount << " bytes, sendWindow=" <<
       sendWindow_.getSize();
     if (!success) {
       LOG(WARNING) << "Remote side sent connection-level WINDOW_UPDATE "
                    << "that could not be applied. Aborting session.";
       // If something went wrong applying the flow control change, abort
       // the entire session.
       error_ = true;
       HTTPException ex = getException(
         folly::to<std::string>(
           "Failed to update send window, outstanding=",
           sendWindow_.getOutstanding(), ", amount=", amount));
       callback_->onError(stream, ex, false);
     }
     if (sendsBlocked_ && sendWindow_.getNonNegativeSize()) {
       VLOG(4) << "Send window opened";
       sendsBlocked_ = false;
       notify_.onConnectionSendWindowOpen();
     }
     // Don't forward.
   } else {
     callback_->onWindowUpdate(stream, amount);
   }
 }

void proxygen::FlowControlFilter::setReceiveWindowSize	(	folly::IOBufQueue &	writeBuf,
		uint32_t	capacity
	)

Modify the session receive window

Parameters

writeBuf	The buffer to write egress on. This constructor may generate a window update frame on this buffer.
capacity	The initial size of the conn-level recv window. It must be >= the codec default.

Definition at line 44 of file FlowControlFilter.cpp.

References proxygen::GenericFilter< T1, T2, set_callback, TakeOwnership, Dp >::call_, proxygen::Window::getCapacity(), int32_t, recvWindow_, proxygen::Window::setCapacity(), and toAck_.

Referenced by proxygen::HTTPSession::onNativeProtocolUpgradeImpl(), and proxygen::HTTPSession::startNow().

                                                                 {
   if (capacity < recvWindow_.getCapacity()) {
     VLOG(4) << "Ignoring low conn-level recv window size of " << capacity;
     return;
   }
   int32_t delta = capacity - recvWindow_.getCapacity();
   if (delta < 0) {
     // For now, we're disallowing shrinking the window, since it can lead
     // to FLOW_CONTROL_ERRORs if there is data in flight.
     VLOG(4) << "Refusing to shrink the recv window";
     return;
   }
   VLOG(4) << "Incrementing default conn-level recv window by " << delta;
   if (!recvWindow_.setCapacity(capacity)) {
     VLOG(2) << "Failed setting conn-level recv window capacity to " << capacity;
     return;
   }
   toAck_ += delta;
   if (toAck_ > 0) {
     call_->generateWindowUpdate(writeBuf, 0, delta);
     toAck_ = 0;
   }
 }