Merge pull request #47743 from fastly/jkarneges/writebyteschanged

base writeBytesAvailable on credits and emit signal when changed
fastly · Sep 19, 2023 · 3832ec9 · 3832ec9
2 parents 65b1234 + 335135b
commit 3832ec9
Show file tree

Hide file tree

Showing 3 changed files with 46 additions and 20 deletions.
diff --git a/src/corelib/httprequest.h b/src/corelib/httprequest.h
@@ -1,5 +1,6 @@
 /*
  * Copyright (C) 2012-2016 Fanout, Inc.
+ * Copyright (C) 2023 Fastly, Inc.
  *
  * This file is part of Pushpin.
  *
@@ -88,6 +89,7 @@ class HttpRequest : public QObject
 	void readyRead();
 	// indicates output data written and/or output finished
 	void bytesWritten(int count);
+	void writeBytesChanged();
 	void paused();
 	void error();
 };

diff --git a/src/corelib/zhttprequest.cpp b/src/corelib/zhttprequest.cpp
@@ -96,6 +96,7 @@ class ZhttpRequest::Private : public QObject
 	bool pendingUpdate;
 	bool needPause;
 	bool readableChanged;
+	bool writableChanged;
 	bool errored;
 	ErrorCondition errorCondition;
 	RTimer *expireTimer;
@@ -127,6 +128,7 @@ class ZhttpRequest::Private : public QObject
 		pendingUpdate(false),
 		needPause(false),
 		readableChanged(false),
+		writableChanged(false),
 		errored(false),
 		expireTimer(0),
 		keepAliveTimer(0),
@@ -153,6 +155,7 @@ class ZhttpRequest::Private : public QObject
 	{
 		needPause = false;
 		readableChanged = false;
+		writableChanged = false;
 
 		if(expireTimer)
 		{
@@ -425,7 +428,9 @@ class ZhttpRequest::Private : public QObject
 				// also send credits if we need to.
 
 				QByteArray buf = requestBodyBuf.take(outCredits);
+
 				outCredits -= buf.size();
+				writableChanged = true;
 
 				ZhttpRequestPacket p;
 				p.type = ZhttpRequestPacket::Data;
@@ -466,7 +471,11 @@ class ZhttpRequest::Private : public QObject
 				ZhttpResponsePacket packet;
 				packet.type = ZhttpResponsePacket::Data;
 				packet.body = responseBodyBuf.take(outCredits);
+
 				outCredits -= packet.body.size();
+				if(!packet.body.isEmpty())
+					writableChanged = true;
+
 				packet.more = (!responseBodyBuf.isEmpty() || !bodyFinished);
 
 				writePacket(packet);
@@ -580,20 +589,24 @@ class ZhttpRequest::Private : public QObject
 			}
 
 			if(packet.credits > 0)
+			{
 				outCredits += packet.credits;
+				writableChanged = true;
+			}
 
 			if(!packet.body.isEmpty() || (!done && haveRequestBody))
-			{
 				readableChanged = true;
+
+			if(readableChanged || writableChanged)
 				update();
-			}
 		}
 		else if(packet.type == ZhttpRequestPacket::Credit)
 		{
 			if(packet.credits > 0)
 			{
 				outCredits += packet.credits;
-				tryWrite();
+				writableChanged = true;
+				update();
 			}
 		}
 		else if(packet.type == ZhttpRequestPacket::KeepAlive)
@@ -742,26 +755,19 @@ class ZhttpRequest::Private : public QObject
 
 			responseBodyBuf += packet.body;
 
-			if(!doReq && packet.credits > 0)
+			if(packet.more)
 			{
-				outCredits += packet.credits;
-				if(outCredits > 0)
+				if(!doReq && packet.credits > 0)
 				{
-					// try to write anything that was waiting on credits
-					QPointer<QObject> self = this;
-					tryWrite();
-					if(!self)
-						return;
+					outCredits += packet.credits;
+					writableChanged = true;
 				}
-			}
 
-			if(packet.more)
-			{
 				if(needToSendHeaders || !packet.body.isEmpty())
-				{
 					readableChanged = true;
+
+				if(readableChanged || writableChanged)
 					update();
-				}
 			}
 			else
 			{
@@ -776,8 +782,8 @@ class ZhttpRequest::Private : public QObject
 			if(packet.credits > 0)
 			{
 				outCredits += packet.credits;
-				if(outCredits > 0)
-					tryWrite();
+				writableChanged = true;
+				update();
 			}
 		}
 		else if(packet.type == ZhttpResponsePacket::KeepAlive)
@@ -1036,6 +1042,12 @@ public slots:
 		else if(state == ClientRequesting)
 		{
 			tryWrite();
+
+			if(writableChanged)
+			{
+				writableChanged = false;
+				emit q->writeBytesChanged();
+			}
 		}
 		else if(state == ClientReceiving)
 		{
@@ -1127,6 +1139,12 @@ public slots:
 		else if(state == ServerResponding)
 		{
 			tryWrite();
+
+			if(writableChanged)
+			{
+				writableChanged = false;
+				emit q->writeBytesChanged();
+			}
 		}
 	}
 
@@ -1296,8 +1314,10 @@ int ZhttpRequest::bytesAvailable() const
 
 int ZhttpRequest::writeBytesAvailable() const
 {
-	if(d->responseBodyBuf.size() <= IDEAL_CREDITS)
-		return (IDEAL_CREDITS - d->responseBodyBuf.size());
+	if(d->server && d->responseBodyBuf.size() < d->outCredits)
+		return d->outCredits - d->responseBodyBuf.size();
+	else if(!d->server && d->requestBodyBuf.size() < d->outCredits)
+		return d->outCredits - d->requestBodyBuf.size();
 	else
 		return 0;
 }

diff --git a/src/handler/httpsession.cpp b/src/handler/httpsession.cpp
@@ -200,6 +200,7 @@ class HttpSession::Private : public QObject
 
 		req->setParent(this);
 		connect(req, &ZhttpRequest::bytesWritten, this, &Private::req_bytesWritten);
+		connect(req, &ZhttpRequest::writeBytesChanged, this, &Private::req_writeBytesChanged);
 		connect(req, &ZhttpRequest::error, this, &Private::req_error);
 
 		timer = new RTimer(this);
@@ -1420,7 +1421,10 @@ private slots:
 			doFinish();
 			return;
 		}
+	}
 
+	void req_writeBytesChanged()
+	{
 		if(state == SendingFirstInstructResponse)
 		{
 			tryWriteFirstInstructResponse();