Bug 1663718 - Don't put too much data in buffer when the data can't be written to socket r=dragana

The main reason we used too much memory is that we ignore the `NS_BASE_STREAM_WOULD_BLOCK` returned from socket output stream.
When the output stream is blocked, all the data is stored in the output queue and make the memory usage high.

Differential Revision: https://phabricator.services.mozilla.com/D89563

netwerk/protocol/websocket/WebSocketChannel.cpp

@@ -3862,6 +3862,10 @@ void WebSocketChannel::DoEnqueueOutgoingMessage() {
         mHdrOut, mHdrOutSize, (uint8_t*)mCurrentOut->BeginReading(),
         mCurrentOut->Length());
 
+    if (rv == NS_BASE_STREAM_WOULD_BLOCK) {
+      return;
+    }
+
     LOG(("WebSocketChannel::DoEnqueueOutgoingMessage: rv %" PRIx32 "\n",
          static_cast<uint32_t>(rv)));
 
@@ -3924,6 +3928,12 @@ WebSocketChannel::OnDataReceived(uint8_t* aData, uint32_t aCount) {
   return ProcessInput(aData, aCount);
 }
 
+NS_IMETHODIMP
+WebSocketChannel::OnReadyToSendData() {
+  DoEnqueueOutgoingMessage();
+  return NS_OK;
+}
+
 }  // namespace net
 }  // namespace mozilla
 

netwerk/protocol/websocket/WebSocketConnectionChild.cpp

@@ -174,6 +174,12 @@ WebSocketConnectionChild::OnDataReceived(uint8_t* aData, uint32_t aCount) {
   return NS_OK;
 }
 
+NS_IMETHODIMP
+WebSocketConnectionChild::OnReadyToSendData() {
+  // TODO: implement flow control between parent and socket process.
+  return NS_OK;
+}
+
 void WebSocketConnectionChild::ActorDestroy(ActorDestroyReason aWhy) {
   LOG(("WebSocketConnectionChild::ActorDestroy %p\n", this));
   if (mConnection) {

netwerk/protocol/websocket/nsIWebSocketConnection.idl

@@ -71,4 +71,9 @@ interface nsIWebSocketConnectionListener : nsISupports
 
     void onDataReceived([array, size_is(dataLength)]in uint8_t data,
                         in unsigned long dataLength);
+
+    /**
+     * Called to inform the listener that the outgoing data is ready to write.
+     */
+    void onReadyToSendData();
 };

netwerk/protocol/websocket/nsWebSocketConnection.cpp

@@ -20,7 +20,8 @@ nsWebSocketConnection::nsWebSocketConnection(
       mSocketIn(aInputStream),
       mSocketOut(aOutputStream),
       mWriteOffset(0),
-      mStartReadingCalled(false) {
+      mStartReadingCalled(false),
+      mOutputStreamBlocked(false) {
   LOG(("nsWebSocketConnection ctor %p\n", this));
 }
 
@@ -105,6 +106,10 @@ nsWebSocketConnection::EnqueueOutputData(const uint8_t* aHdrBuf,
     mSocketOut->AsyncWait(this, 0, 0, mEventTarget);
   }
 
+  if (mOutputStreamBlocked) {
+    return NS_BASE_STREAM_WOULD_BLOCK;
+  }
+
   return NS_OK;
 }
 
@@ -215,6 +220,8 @@ nsWebSocketConnection::OnOutputStreamReady(nsIAsyncOutputStream* aStream) {
     return NS_OK;
   }
 
+  mOutputStreamBlocked = false;
+
   while (!mOutputQueue.empty()) {
     const OutputData& data = mOutputQueue.front();
 
@@ -230,6 +237,7 @@ nsWebSocketConnection::OnOutputStreamReady(nsIAsyncOutputStream* aStream) {
 
     if (rv == NS_BASE_STREAM_WOULD_BLOCK) {
       mSocketOut->AsyncWait(this, 0, 0, mEventTarget);
+      mOutputStreamBlocked = true;
       return NS_OK;
     }
 
@@ -248,5 +256,7 @@ nsWebSocketConnection::OnOutputStreamReady(nsIAsyncOutputStream* aStream) {
     }
   }
 
+  Unused << mListener->OnReadyToSendData();
+
   return NS_OK;
 }

netwerk/protocol/websocket/nsWebSocketConnection.h

@@ -60,6 +60,7 @@ class nsWebSocketConnection : public nsIWebSocketConnection,
   size_t mWriteOffset;
   std::list<OutputData> mOutputQueue;
   bool mStartReadingCalled;
+  bool mOutputStreamBlocked;
 };
 
 }  // namespace net