1
0
mirror of https://github.com/facebook/proxygen.git synced 2026-10-24 01:33:54 +03:00

remove HttpTxn::isPushed

Summary:
isPushed is always false now that we've removed push support in proxygen/lib
this simplifies all the gated conditions on isPushed

Reviewed By: jbeshay, Carl-Elkhoury

Differential Revision: D124175915

fbshipit-source-id: f8bb91d143da3051066e1c1bf585be0f8695471f
This commit is contained in:
Hani Damlaj
2026-10-14 21:31:15 -07:00
committed by meta-codesync[bot]
parent 6461cf9a84
commit e79598c3b3
6 changed files with 14 additions and 64 deletions

View File

@@ -730,10 +730,6 @@ size_t HQSession::HQStreamTransportBase::changePriority(
if (isUpstream(session_.direction_) && txn->isIngressEOMSeen()) {
return 0;
}
if (txn->isPushed()) {
auto pushId = txn->getID();
return session_.sendPushPriority(pushId, httpPriority);
}
return session_.sendPriority(txn->getID(), httpPriority);
}
@@ -1778,9 +1774,8 @@ void HQSession::onPriority(quic::StreamId streamId, const HTTPPriority& pri) {
return;
}
CHECK(sock_);
// This also covers push streams:
auto stream = findStream(streamId);
if (!stream || (!stream->txn_.isPushed() && !stream->hasHeaders_)) {
if (!stream || !stream->hasHeaders_) {
// We are supposed to drop the connection with HTTP_ID_ERROR if the streamId
// points to a control stream. But why should I spend CPU looking it up?
priorityUpdatesBuffer_.insert(streamId, pri);
@@ -2413,7 +2408,7 @@ void HQSession::HQStreamTransportBase::initCodec(
c->setActivationHook([this] { return setActiveCodec("self"); });
}
auto g = folly::makeGuard(setActiveCodec(__func__));
if (isUpstream(session_.direction_) || txn_.isPushed()) {
if (isUpstream(session_.direction_)) {
codecStreamId_ = codecFilterChain.createStream();
}
hasCodec_ = true;
@@ -2809,20 +2804,9 @@ void HQSession::HQStreamTransportBase::transactionTimeout(
VLOG(4) << __func__ << " txn=" << txn_;
DCHECK(txn == &txn_);
if (txn->isPushed()) {
if (!hasIngressStreamId()) {
// This transaction has not been assigned a stream id yet.
// Do not attempt to close the stream but do invoke
// the timeout on the txn
VLOG(3) << "Transaction timeout on pushedTxn pushId=" << txn->getID();
txn_.onIngressTimeout();
return;
}
}
// Verify that the transaction has egress or ingress stream
DCHECK(hasIngressStreamId() || hasEgressStreamId())
<< "Timeout on transaction without stream id txnID=" << txn->getID()
<< " isPushed=" << txn->isPushed();
<< "Timeout on transaction without stream id txnID=" << txn->getID();
// A transaction has timed out. If the transaction does not have
// a Handler yet, because we haven't yet received the full request
// headers, we give it a DirectResponseHandler that generates an
@@ -3091,9 +3075,7 @@ size_t HQSession::HQStreamTransportBase::sendAbort(
folly::to<std::string>("Application aborts, errorCode=",
getErrorCodeString(errorCode),
" txnID=",
txn->getID(),
" isPushed=",
txn->isPushed()));
txn->getID()));
}
size_t HQSession::HQStreamTransportBase::sendAbortImpl(HTTP3::ErrorCode code,

View File

@@ -67,11 +67,6 @@ void HTTPDownstreamSession::onHeadersSent(const HTTPMessage& headers,
}
bool HTTPDownstreamSession::allTransactionsStarted() const {
for (const auto& txn : transactions_) {
if (txn.second.isPushed() && !txn.second.isEgressStarted()) {
return false;
}
}
return true;
}

View File

@@ -570,14 +570,6 @@ void HTTPSession::onMessageBegin(HTTPCodec::StreamID streamID,
HTTPTransaction* txn = findTransaction(streamID);
if (txn) {
if (isDownstream() && txn->isPushed()) {
// Push streams are unidirectional (half-closed). If the downstream
// attempts to send ingress, abort with STREAM_CLOSED error.
HTTPException ex(HTTPException::Direction::INGRESS_AND_EGRESS,
"Downstream attempts to send ingress, abort.");
ex.setCodecStatusCode(ErrorCode::STREAM_CLOSED);
txn->onError(ex);
}
return; // If this transaction is already registered, no need to add it now
}
@@ -1162,11 +1154,6 @@ void HTTPSession::sendHeaders(HTTPTransaction* txn,
}
const uint64_t newOffset = sessionByteOffset();
// for push response count towards the MAX_CONCURRENT_STREAMS limit
if (isDownstream() && headers.isResponse() && txn->isPushed()) {
incrementOutgoingStreams(txn);
}
// For all upstream headers, addFirstHeaderByteEvent should be added
// For all downstream, only response headers need addFirstHeaderByteEvent
bool shouldAddFirstHeaderByteEvent =
@@ -1421,8 +1408,7 @@ size_t HTTPSession::sendCertificateRequest(
void HTTPSession::decrementTransactionCount(HTTPTransaction* txn,
bool ingressEOM,
bool egressEOM) {
if ((isUpstream() && !txn->isPushed()) ||
(isDownstream() && txn->isPushed())) {
if (isUpstream()) {
if (ingressEOM && txn->testAndClearIsCountedTowardsStreamLimit()) {
DCHECK_NE(outgoingStreams_, 0);
outgoingStreams_--;
@@ -2298,13 +2284,11 @@ HTTPTransaction* HTTPSession::createTransaction(
incrementSeqNo();
txn->setReceiveWindow(receiveStreamWindowSize_);
if (isUpstream() && !txn->isPushed()) {
if (isUpstream()) {
incrementOutgoingStreams(txn);
// do not count towards MAX_CONCURRENT_STREAMS for PUSH_PROMISE
} else if (!(isDownstream() && txn->isPushed())) {
} else {
incrementIncomingStreams(txn);
}
// Downstream push is counted in HTTPSession::sendHeaders
if (infoCallback_) {
infoCallback_->onTransactionAttached(*this);

View File

@@ -147,10 +147,8 @@ HTTPTransaction::HTTPTransaction(
stats_->recordTransactionOpened();
}
if (isDownstream() || !isPushed()) {
queueHandle_ =
egressQueue_.addTransaction(id_, priority, this, false, &insertDepth_);
}
queueHandle_ =
egressQueue_.addTransaction(id_, priority, this, false, &insertDepth_);
if (priority.streamDependency != egressQueue_.getRootId() &&
insertDepth_ == 1) {
priorityFallback_ = true;
@@ -206,14 +204,12 @@ void HTTPTransaction::onIngressHeadersComplete(
std::unique_ptr<HTTPMessage> msg) {
DestructorGuard g(this);
msg->setSeqNo(seqNo_);
if (isUpstream() && !isPushed() && msg->isResponse()) {
if (isUpstream() && msg->isResponse()) {
lastResponseStatus_ = msg->getStatusCode();
}
bool nonFinalPushHeaders = isPushed() && msg->isRequest();
if (!validateIngressStateTransition(
msg->isFinal() && !nonFinalPushHeaders
? IngressSmEvent::onFinalHeaders
: IngressSmEvent::onNonFinalHeaders)) {
msg->isFinal() ? IngressSmEvent::onFinalHeaders
: IngressSmEvent::onNonFinalHeaders)) {
return;
}
checkForUpgrade(*msg);
@@ -1006,7 +1002,7 @@ void HTTPTransaction::sendHeadersWithOptionalEOM(const HTTPMessage& headers,
return;
}
DCHECK(!isEgressComplete());
if (!headers.isRequest() && !isPushed()) {
if (!headers.isRequest()) {
lastResponseStatus_ = headers.getStatusCode();
}
checkForUpgrade(headers);

View File

@@ -1377,13 +1377,6 @@ class HTTPTransaction
return nullptr;
}
/**
* True if this transaction is a server push transaction
*/
bool isPushed() const {
return assocStreamId_.has_value();
}
/**
* Overrides the default idle timeout value.
*/

View File

@@ -151,7 +151,7 @@ HTTPTransaction::Handler* HTTPUpstreamSession::getTransactionTimeoutHandler(
bool HTTPUpstreamSession::allTransactionsStarted() const {
for (const auto& txn : transactions_) {
if (!txn.second.isPushed() && !txn.second.isEgressStarted()) {
if (!txn.second.isEgressStarted()) {
return false;
}
}