Bug 1668851 - Try to write data to socket directly without waiting r=necko-reviewers,dragana,remote-protocol-reviewers

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

devtools/client/netmonitor/test/browser.ini

@@ -256,7 +256,7 @@ skip-if = (os == 'win' && os_version == '6.1' && !debug) # Bug 1547150
 [browser_net_ws-clear.js]
 [browser_net_ws-connection-closed.js]
 [browser_net_ws-early-connection.js]
-skip-if = fission
+skip-if = fission || socketprocess_networking
 [browser_net_ws-filter-dropdown.js]
 [browser_net_ws-filter-regex.js]
 [browser_net_ws-limit-payload.js]
@@ -264,3 +264,4 @@ skip-if = fission
 [browser_net_ws-sse-persist-columns.js]
 [browser_net_ws-stomp-payload.js]
 [browser_net-ws-filter-freetext.js]
+skip-if = socketprocess_networking

netwerk/protocol/websocket/PWebSocketConnection.ipdl

@@ -19,6 +19,7 @@ parent:
   async OnTCPClosed();
   async OnDataReceived(uint8_t[] aData);
   async OnUpgradeFailed(nsresult aReason);
+  async OnDataSent();
 
 child:
   async EnqueueOutgoingData(uint8_t[] aData);

netwerk/protocol/websocket/WebSocketChannel.cpp

@@ -3852,7 +3852,7 @@ void WebSocketChannel::DoEnqueueOutgoingMessage() {
 
   if (!mCurrentOut) PrimeNewOutgoingMessage();
 
-  while (mCurrentOut && mConnection) {
+  if (mCurrentOut && mConnection) {
     LOG(
         ("WebSocketChannel::DoEnqueueOutgoingMessage: "
          "Try to send %u of hdr/copybreak and %u of data\n",
@@ -3862,10 +3862,6 @@ 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)));
 
@@ -3873,14 +3869,6 @@ void WebSocketChannel::DoEnqueueOutgoingMessage() {
       AbortSession(rv);
       return;
     }
-
-    if (!mStopped) {
-      mTargetThread->Dispatch(
-          new CallAcknowledge(this, mCurrentOut->OrigLength()),
-          NS_DISPATCH_NORMAL);
-    }
-    DeleteCurrentOutGoingMessage();
-    PrimeNewOutgoingMessage();
   }
 
   if (mReleaseOnTransmit) ReleaseSession();
@@ -3929,8 +3917,19 @@ WebSocketChannel::OnDataReceived(uint8_t* aData, uint32_t aCount) {
 }
 
 NS_IMETHODIMP
-WebSocketChannel::OnReadyToSendData() {
-  DoEnqueueOutgoingMessage();
+WebSocketChannel::OnDataSent() {
+  MOZ_ASSERT(OnSocketThread(), "not on socket thread");
+
+  if (mCurrentOut) {
+    if (!mStopped) {
+      mTargetThread->Dispatch(
+          new CallAcknowledge(this, mCurrentOut->OrigLength()),
+          NS_DISPATCH_NORMAL);
+    }
+    DeleteCurrentOutGoingMessage();
+    PrimeNewOutgoingMessage();
+  }
+
   return NS_OK;
 }
 

netwerk/protocol/websocket/WebSocketConnectionChild.cpp

@@ -175,8 +175,12 @@ WebSocketConnectionChild::OnDataReceived(uint8_t* aData, uint32_t aCount) {
 }
 
 NS_IMETHODIMP
-WebSocketConnectionChild::OnReadyToSendData() {
-  // TODO: implement flow control between parent and socket process.
+WebSocketConnectionChild::OnDataSent() {
+  LOG(("WebSocketConnectionChild::OnDataSent %p\n", this));
+
+  if (CanSend()) {
+    Unused << SendOnDataSent();
+  }
   return NS_OK;
 }
 

netwerk/protocol/websocket/WebSocketConnectionParent.cpp

@@ -130,6 +130,22 @@ mozilla::ipc::IPCResult WebSocketConnectionParent::RecvOnDataReceived(
   return IPC_OK();
 }
 
+mozilla::ipc::IPCResult WebSocketConnectionParent::RecvOnDataSent() {
+  LOG(("WebSocketConnectionParent::RecvOnDataSent %p\n", this));
+  MOZ_ASSERT(mEventTarget);
+
+  RefPtr<WebSocketConnectionParent> self = this;
+  auto task = [self{std::move(self)}]() {
+    if (self->mListener) {
+      Unused << self->mListener->OnDataSent();
+    }
+  };
+
+  DispatchHelper(mEventTarget, "WebSocketConnectionParent::RecvOnDataSent",
+                 std::move(task));
+  return IPC_OK();
+}
+
 void WebSocketConnectionParent::ActorDestroy(ActorDestroyReason aWhy) {
   LOG(("WebSocketConnectionParent::ActorDestroy %p aWhy=%d\n", this, aWhy));
   if (!mClosed) {

netwerk/protocol/websocket/WebSocketConnectionParent.h

@@ -45,6 +45,7 @@ class WebSocketConnectionParent final : public PWebSocketConnectionParent,
   mozilla::ipc::IPCResult RecvOnTCPClosed();
   mozilla::ipc::IPCResult RecvOnDataReceived(nsTArray<uint8_t>&& aData);
   mozilla::ipc::IPCResult RecvOnUpgradeFailed(const nsresult& aReason);
+  mozilla::ipc::IPCResult RecvOnDataSent();
 
   void ActorDestroy(ActorDestroyReason aWhy) override;
 

netwerk/protocol/websocket/nsIWebSocketConnection.idl

@@ -73,7 +73,8 @@ interface nsIWebSocketConnectionListener : nsISupports
                         in unsigned long dataLength);
 
     /**
-     * Called to inform the listener that the outgoing data is ready to write.
+     * Called to inform the listener that the outgoing data is written
+     * to socket.
      */
-    void onReadyToSendData();
+    void onDataSent();
 };

netwerk/protocol/websocket/nsWebSocketConnection.cpp

@@ -20,8 +20,7 @@ nsWebSocketConnection::nsWebSocketConnection(
       mSocketIn(aInputStream),
       mSocketOut(aOutputStream),
       mWriteOffset(0),
-      mStartReadingCalled(false),
-      mOutputStreamBlocked(false) {
+      mStartReadingCalled(false) {
   LOG(("nsWebSocketConnection ctor %p\n", this));
 }
 
@@ -80,12 +79,12 @@ nsWebSocketConnection::Close() {
 nsresult nsWebSocketConnection::EnqueueOutputData(nsTArray<uint8_t>&& aData) {
   MOZ_ASSERT(mEventTarget->IsOnCurrentThread());
 
-  mOutputQueue.emplace_back(std::move(aData));
-
-  if (mSocketOut) {
-    mSocketOut->AsyncWait(this, 0, 0, mEventTarget);
+  if (!mSocketOut) {
+    return NS_ERROR_NOT_AVAILABLE;
   }
 
+  mOutputQueue.emplace_back(std::move(aData));
+  OnOutputStreamReady(mSocketOut);
   return NS_OK;
 }
 
@@ -97,19 +96,16 @@ nsWebSocketConnection::EnqueueOutputData(const uint8_t* aHdrBuf,
   LOG(("nsWebSocketConnection::EnqueueOutputData %p\n", this));
   MOZ_ASSERT(mEventTarget->IsOnCurrentThread());
 
+  if (!mSocketOut) {
+    return NS_ERROR_NOT_AVAILABLE;
+  }
+
   nsTArray<uint8_t> data;
   data.AppendElements(aHdrBuf, aHdrBufLength);
   data.AppendElements(aPayloadBuf, aPayloadBufLength);
   mOutputQueue.emplace_back(std::move(data));
 
-  if (mSocketOut) {
-    mSocketOut->AsyncWait(this, 0, 0, mEventTarget);
-  }
-
-  if (mOutputStreamBlocked) {
-    return NS_BASE_STREAM_WOULD_BLOCK;
-  }
-
+  OnOutputStreamReady(mSocketOut);
   return NS_OK;
 }
 
@@ -220,8 +216,6 @@ nsWebSocketConnection::OnOutputStreamReady(nsIAsyncOutputStream* aStream) {
     return NS_OK;
   }
 
-  mOutputStreamBlocked = false;
-
   while (!mOutputQueue.empty()) {
     const OutputData& data = mOutputQueue.front();
 
@@ -237,7 +231,6 @@ nsWebSocketConnection::OnOutputStreamReady(nsIAsyncOutputStream* aStream) {
 
     if (rv == NS_BASE_STREAM_WOULD_BLOCK) {
       mSocketOut->AsyncWait(this, 0, 0, mEventTarget);
-      mOutputStreamBlocked = true;
       return NS_OK;
     }
 
@@ -253,10 +246,12 @@ nsWebSocketConnection::OnOutputStreamReady(nsIAsyncOutputStream* aStream) {
     if (toWrite == wrote) {
       mWriteOffset = 0;
       mOutputQueue.pop_front();
+      Unused << mListener->OnDataSent();
+    } else {
+      mSocketOut->AsyncWait(this, 0, 0, mEventTarget);
+      return NS_OK;
     }
   }
 
-  Unused << mListener->OnReadyToSendData();
-
   return NS_OK;
 }

netwerk/protocol/websocket/nsWebSocketConnection.h

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

remote/test/browser/browser.ini

@@ -9,6 +9,7 @@ support-files =
 
 [browser_agent.js]
 [browser_cdp.js]
+skip-if = socketprocess_networking
 [browser_main_target.js]
 skip-if = socketprocess_networking
 [browser_session.js]

remote/test/browser/input/browser.ini

@@ -13,4 +13,5 @@ support-files =
 [browser_dispatchKeyEvent.js]
 [browser_dispatchKeyEvent_events.js]
 [browser_dispatchKeyEvent_race.js]
+skip-if = socketprocess_networking
 [browser_dispatchMouseEvent.js]