Implement a threadsafe & revised version of http2PushedStream.

netwerk/protocol/http/Http2Push.cpp

@@ -30,8 +30,8 @@ class CallChannelOnPush final : public Runnable {
                     Http2PushedStream *pushStream)
     : mAssociatedChannel(associatedChannel)
     , mPushedURI(pushedURI)
-    , mPushedStream(pushStream)
   {
+    mPushedStreamWrapper = new Http2PushedStreamWrapper(pushStream);
   }
 
   NS_IMETHOD Run() override
@@ -40,21 +40,94 @@ class CallChannelOnPush final : public Runnable {
     RefPtr<nsHttpChannel> channel;
     CallQueryInterface(mAssociatedChannel, channel.StartAssignment());
     MOZ_ASSERT(channel);
-    if (channel && NS_SUCCEEDED(channel->OnPush(mPushedURI, mPushedStream))) {
+    if (channel && NS_SUCCEEDED(channel->OnPush(mPushedURI, mPushedStreamWrapper))) {
       return NS_OK;
     }
 
     LOG3(("Http2PushedStream Orphan %p failed OnPush\n", this));
-    mPushedStream->OnPushFailed();
+    mPushedStreamWrapper->OnPushFailed();
     return NS_OK;
   }
 
 private:
   nsCOMPtr<nsIHttpChannelInternal> mAssociatedChannel;
   const nsCString mPushedURI;
-  Http2PushedStream *mPushedStream;
+  RefPtr<Http2PushedStreamWrapper> mPushedStreamWrapper;
 };
 
+// Because WeakPtr isn't thread-safe we must ensure that the object is destroyed
+// on the socket thread, so any Release() called on a different thread is
+// dispatched to the socket thread.
+bool Http2PushedStreamWrapper::DispatchRelease() {
+  if (PR_GetCurrentThread() == gSocketThread) {
+    return false;
+  }
+
+  gSocketTransportService->Dispatch(
+      NewNonOwningRunnableMethod(this, &Http2PushedStreamWrapper::Release),
+      NS_DISPATCH_NORMAL);
+
+  return true;
+}
+
+NS_IMPL_ADDREF(Http2PushedStreamWrapper)
+NS_IMETHODIMP_(MozExternalRefCountType)
+Http2PushedStreamWrapper::Release() {
+  nsrefcnt count = mRefCnt - 1;
+  if (DispatchRelease()) {
+    // Redispatched to the socket thread.
+    return count;
+  }
+
+  MOZ_ASSERT(0 != mRefCnt, "dup release");
+  count = --mRefCnt;
+  NS_LOG_RELEASE(this, count, "Http2PushedStreamWrapper");
+
+  if (0 == count) {
+    mRefCnt = 1;
+    delete (this);
+    return 0;
+  }
+
+  return count;
+}
+
+NS_INTERFACE_MAP_BEGIN(Http2PushedStreamWrapper)
+NS_INTERFACE_MAP_END
+
+Http2PushedStreamWrapper::Http2PushedStreamWrapper(
+    Http2PushedStream* aPushStream) {
+  MOZ_ASSERT(PR_GetCurrentThread() == gSocketThread, "not on socket thread");
+  mStream = aPushStream;
+  mRequestString = aPushStream->GetRequestString();
+}
+
+Http2PushedStreamWrapper::~Http2PushedStreamWrapper() {
+  MOZ_ASSERT(PR_GetCurrentThread() == gSocketThread, "not on socket thread");
+}
+
+Http2PushedStream* Http2PushedStreamWrapper::GetStream() {
+  MOZ_ASSERT(PR_GetCurrentThread() == gSocketThread, "not on socket thread");
+  if (mStream) {
+    Http2Stream* stream = mStream;
+    return static_cast<Http2PushedStream*>(stream);
+  }
+  return nullptr;
+}
+
+void Http2PushedStreamWrapper::OnPushFailed() {
+  if (PR_GetCurrentThread() == gSocketThread) {
+    if (mStream) {
+      Http2Stream* stream = mStream;
+      static_cast<Http2PushedStream*>(stream)->OnPushFailed();
+    }
+  } else {
+    gSocketTransportService->Dispatch(
+        NewRunnableMethod(this, &Http2PushedStreamWrapper::OnPushFailed),
+        NS_DISPATCH_NORMAL);
+  }
+}
+
 //////////////////////////////////////////
 // Http2PushedStream
 //////////////////////////////////////////

netwerk/protocol/http/Http2Push.h

@@ -123,6 +123,24 @@ private:
   uint32_t mBufferedHTTP1Consumed;
 };
 
+class Http2PushedStreamWrapper : public nsISupports {
+public:
+  NS_DECL_THREADSAFE_ISUPPORTS
+  bool DispatchRelease();
+
+  explicit Http2PushedStreamWrapper(Http2PushedStream* aPushStream);
+
+  nsCString& GetRequestString() { return mRequestString; }
+  Http2PushedStream* GetStream();
+  void OnPushFailed();
+
+private:
+  virtual ~Http2PushedStreamWrapper();
+
+  nsCString mRequestString;
+  WeakPtr<Http2Stream> mStream;
+};
+
 } // namespace net
 } // namespace mozilla
 

netwerk/protocol/http/Http2Session.cpp

@@ -380,12 +380,24 @@ Http2Session::AddStream(nsAHttpTransaction *aHttpTransaction,
 
   if (mClosed || mShouldGoAway) {
     nsHttpTransaction *trans = aHttpTransaction->QueryHttpTransaction();
-    if (trans && !trans->GetPushedStream()) {
-      LOG3(("Http2Session::AddStream %p atrans=%p trans=%p session unusable - resched.\n",
-            this, aHttpTransaction, trans));
-      aHttpTransaction->SetConnection(nullptr);
-      gHttpHandler->InitiateTransaction(trans, trans->Priority());
-      return true;
+    if (trans) {
+      RefPtr<Http2PushedStreamWrapper> pushedStreamWrapper;
+      pushedStreamWrapper = trans->GetPushedStream();
+      if (!pushedStreamWrapper || !pushedStreamWrapper->GetStream()) {
+        LOG3(
+            ("Http2Session::AddStream %p atrans=%p trans=%p session unusable - "
+             "resched.\n", this, aHttpTransaction, trans));
+        aHttpTransaction->SetConnection(nullptr);
+        nsresult rv =
+            gHttpHandler->InitiateTransaction(trans, trans->Priority());
+        if (NS_FAILED(rv)) {
+          LOG3(
+              ("Http2Session::AddStream %p atrans=%p trans=%p failed to "
+               "initiate transaction (%08x).\n",
+               this, aHttpTransaction, trans, static_cast<uint32_t>(rv)));
+        }
+        return true;
+      }
     }
   }
 

netwerk/protocol/http/Http2Stream.cpp

@@ -442,12 +442,14 @@ Http2Stream::ParseHttpRequestHeaders(const char *buf,
       requestContext->GetSpdyPushCache(&cache);
     }
 
+    RefPtr<Http2PushedStreamWrapper> pushedStreamWrapper;
     Http2PushedStream *pushedStream = nullptr;
 
     // If a push stream is attached to the transaction via onPush, match only with that
     // one. This occurs when a push was made with in conjunction with a nsIHttpPushListener
     nsHttpTransaction *trans = mTransaction->QueryHttpTransaction();
-    if (trans && (pushedStream = trans->TakePushedStream())) {
+    if (trans && (pushedStreamWrapper = trans->TakePushedStream()) &&
+        (pushedStream = pushedStreamWrapper->GetStream())) {
       if (pushedStream->mSession == mSession) {
         LOG3(("Pushed Stream match based on OnPush correlation %p", pushedStream));
       } else {

netwerk/protocol/http/Http2Stream.h

@@ -28,8 +28,10 @@ class Http2Decompressor;
 class Http2Stream
   : public nsAHttpSegmentReader
   , public nsAHttpSegmentWriter
+  , public SupportsWeakPtr<Http2Stream>
 {
 public:
+  MOZ_DECLARE_WEAKREFERENCE_TYPENAME(Http2Stream)
   NS_DECL_NSAHTTPSEGMENTREADER
   NS_DECL_NSAHTTPSEGMENTWRITER
 

netwerk/protocol/http/nsHttpChannel.cpp

@@ -7820,7 +7820,7 @@ nsHttpChannel::AwaitingCacheCallbacks()
 }
 
 void
-nsHttpChannel::SetPushedStream(Http2PushedStream *stream)
+nsHttpChannel::SetPushedStream(Http2PushedStreamWrapper *stream)
 {
     MOZ_ASSERT(stream);
     MOZ_ASSERT(!mPushedStream);
@@ -7828,7 +7828,7 @@ nsHttpChannel::SetPushedStream(Http2PushedStream *stream)
 }
 
 nsresult
-nsHttpChannel::OnPush(const nsACString &url, Http2PushedStream *pushedStream)
+nsHttpChannel::OnPush(const nsACString &url, Http2PushedStreamWrapper *pushedStream)
 {
     MOZ_ASSERT(NS_IsMainThread());
     LOG(("nsHttpChannel::OnPush [this=%p]\n", this));

netwerk/protocol/http/nsHttpChannel.h

@@ -126,7 +126,7 @@ public:
                           const nsID& aChannelId,
                           nsContentPolicyType aContentPolicyType) override;
 
-    nsresult OnPush(const nsACString &uri, Http2PushedStream *pushedStream);
+    nsresult OnPush(const nsACString &uri, Http2PushedStreamWrapper *pushedStream);
 
     static bool IsRedirectStatus(uint32_t status);
 
@@ -448,7 +448,7 @@ private:
     nsresult OpenCacheInputStream(nsICacheEntry* cacheEntry, bool startBuffering,
                                   bool checkingAppCacheEntry);
 
-    void SetPushedStream(Http2PushedStream *stream);
+    void SetPushedStream(Http2PushedStreamWrapper *stream);
 
     void SetDoNotTrack();
 
@@ -578,9 +578,10 @@ private:
     nsTArray<nsContinueRedirectionFunc> mRedirectFuncStack;
 
     // Needed for accurate DNS timing
-    RefPtr<nsDNSPrefetch>           mDNSPrefetch;
+    RefPtr<nsDNSPrefetch>             mDNSPrefetch;
 
-    Http2PushedStream                 *mPushedStream;
+    RefPtr<Http2PushedStreamWrapper>  mPushedStream;
+    
     // True if the channel's principal was found on a phishing, malware, or
     // tracking (if tracking protection is enabled) blocklist
     bool                              mLocalBlocklist;

netwerk/protocol/http/nsHttpConnectionMgr.cpp

@@ -1827,13 +1827,18 @@ nsHttpConnectionMgr::ProcessNewTransaction(nsHttpTransaction *trans)
 
     trans->SetPendingTime();
 
-    Http2PushedStream *pushedStream = trans->GetPushedStream();
-    if (pushedStream) {
-        LOG(("  ProcessNewTransaction %p tied to h2 session push %p\n",
-             trans, pushedStream->Session()));
-        return pushedStream->Session()->
-            AddStream(trans, trans->Priority(), false, nullptr) ?
-            NS_OK : NS_ERROR_UNEXPECTED;
+    RefPtr<Http2PushedStreamWrapper> pushedStreamWrapper =
+        trans->GetPushedStream();
+    if (pushedStreamWrapper) {
+      Http2PushedStream* pushedStream = pushedStreamWrapper->GetStream();
+      if (pushedStream) {
+        LOG(("  ProcessNewTransaction %p tied to h2 session push %p\n", trans,
+             pushedStream->Session()));
+        return pushedStream->Session()->AddStream(trans, trans->Priority(), false,
+                                                  nullptr)
+                   ? NS_OK
+                   : NS_ERROR_UNEXPECTED;
+      }
     }
 
     nsresult rv = NS_OK;

netwerk/protocol/http/nsHttpTransaction.h

@@ -131,14 +131,14 @@ public:
 
     nsHttpTransaction *QueryHttpTransaction() override { return this; }
 
-    Http2PushedStream *GetPushedStream() { return mPushedStream; }
-    Http2PushedStream *TakePushedStream()
-    {
-        Http2PushedStream *r = mPushedStream;
-        mPushedStream = nullptr;
-        return r;
+    already_AddRefed<Http2PushedStreamWrapper> GetPushedStream() {
+      return do_AddRef(mPushedStream);
     }
-    void SetPushedStream(Http2PushedStream *push) { mPushedStream = push; }
+    already_AddRefed<Http2PushedStreamWrapper> TakePushedStream() {
+      return mPushedStream.forget();
+    }
+
+    void SetPushedStream(Http2PushedStreamWrapper* push) { mPushedStream = push; }
     uint32_t InitialRwin() const { return mInitialRwin; };
     bool ChannelPipeFull() { return mWaitingOnPipeOut; }
 
@@ -264,7 +264,7 @@ private:
     // so far been skipped.
     uint32_t                        mInvalidResponseBytesRead;
 
-    Http2PushedStream               *mPushedStream;
+    RefPtr<Http2PushedStreamWrapper> mPushedStream;
     uint32_t                        mInitialRwin;
 
     nsHttpChunkedDecoder            *mChunkedDecoder;