Use per-frame TaskRunner instead of thread's default in DataConsumerHandle

third_party/WebKit/Source/modules/fetch/DataConsumerTee.cpp

 // Copyright 2015 The Chromium Authors. All rights reserved.
 // Use of this source code is governed by a BSD-style license that can be
 // found in the LICENSE file.
 
 #include "modules/fetch/DataConsumerTee.h"
 
 #include "core/dom/ActiveDOMObject.h"
 #include "core/dom/ExecutionContext.h"
+#include "core/dom/TaskRunnerHelper.h"
 #include "modules/fetch/DataConsumerHandleUtil.h"
 #include "modules/fetch/FetchBlobDataConsumerHandle.h"
 #include "platform/CrossThreadFunctional.h"
 #include "platform/heap/Handle.h"
 #include "public/platform/Platform.h"
 #include "public/platform/WebTaskRunner.h"
 #include "public/platform/WebThread.h"
 #include "public/platform/WebTraceLocation.h"
 #include "wtf/Deque.h"
 #include "wtf/Functional.h"
@@ skipping 157 matching lines @@
     }
 
     void notify()
     {
         {
             MutexLocker locker(m_mutex);
             if (!m_client) {
                 // No client is registered.
                 return;
             }
-            DCHECK(m_readerThread);
-            if (!m_readerThread->isCurrentThread()) {
-                m_readerThread->getWebTaskRunner()->postTask(BLINK_FROM_HERE, crossThreadBind(&DestinationContext::notify, wrapPassRefPtr(this)));
+            DCHECK(m_readerTaskRunner);
+            if (!m_readerTaskRunner->runsTasksOnCurrentThread()) {
+                m_readerTaskRunner->postTask(BLINK_FROM_HERE, crossThreadBind(&DestinationContext::notify, wrapPassRefPtr(this)));
                 return;
             }
         }
         // The reading thread is the current thread.
         if (m_client)
             m_client->didGetReadable();
     }
 
     Mutex& mutex() { return m_mutex; }
 
     // The following functions don't use lock. They should be protected by the
     // caller.
-    void attachReader(WebDataConsumerHandle::Client* client)
+    void attachReader(WebDataConsumerHandle::Client* client, std::unique_ptr<WebTaskRunner> readerTaskRunner)
     {
-        DCHECK(!m_readerThread);
+        DCHECK(!m_readerTaskRunner);
         DCHECK(!m_client);
-        m_readerThread = Platform::current()->currentThread();
+        m_readerTaskRunner = std::move(readerTaskRunner);
         m_client = client;
     }
     void detachReader()
     {
-        DCHECK(m_readerThread && m_readerThread->isCurrentThread());
-        m_readerThread = nullptr;
+        DCHECK(m_readerTaskRunner && m_readerTaskRunner->runsTasksOnCurrentThread());
+        m_readerTaskRunner = nullptr;
         m_client = nullptr;
     }
     const std::unique_ptr<Vector<char>>& top() const { return m_queue.first(); }
     bool isEmpty() const { return m_queue.isEmpty(); }
     size_t offset() const { return m_offset; }
     void consume(size_t size)
     {
         const auto& top = m_queue.first();
         DCHECK(m_offset <= m_offset + size);
         DCHECK(m_offset + size <= top->size());
         if (top->size() <= m_offset + size) {
             m_offset = 0;
             m_queue.removeFirst();
         } else {
             m_offset += size;
         }
     }
     Result getResult() { return m_result; }
 
 private:
     DestinationContext()
         : m_result(WebDataConsumerHandle::ShouldWait)
-        , m_readerThread(nullptr)
+        , m_readerTaskRunner(nullptr)
         , m_client(nullptr)
         , m_offset(0)
         , m_isTwoPhaseReadInProgress(false)
     {
     }
 
     void detach()
     {
         MutexLocker locker(m_mutex);
         DCHECK(!m_client);
-        DCHECK(!m_readerThread);
+        DCHECK(!m_readerTaskRunner);
         m_queue.clear();
     }
 
     Result m_result;
     Deque<std::unique_ptr<Vector<char>>> m_queue;
-    // Note: Holding a WebThread raw pointer is not generally safe, but we can
-    // do that in this case because:
-    //  1. Destructing a ReaderImpl when the bound thread ends is a user's
-    //     responsibility.
-    //  2. |m_readerThread| will never be used after the associated reader is
-    //     detached.
-    WebThread* m_readerThread;
+    std::unique_ptr<WebTaskRunner> m_readerTaskRunner;
     WebDataConsumerHandle::Client* m_client;
     size_t m_offset;
     bool m_isTwoPhaseReadInProgress;
     Mutex m_mutex;
 };
 
 class DestinationReader final : public WebDataConsumerHandle::Reader {
 public:
-    DestinationReader(PassRefPtr<DestinationContext::Proxy> contextProxy, WebDataConsumerHandle::Client* client)
+    DestinationReader(PassRefPtr<DestinationContext::Proxy> contextProxy, WebDataConsumerHandle::Client* client, std::unique_ptr<WebTaskRunner> readerTaskRunner)
         : m_contextProxy(contextProxy)
     {
         MutexLocker locker(context()->mutex());
-        context()->attachReader(client);
+        context()->attachReader(client, readerTaskRunner->clone());
         if (client) {
-            // We need to use crossThreadBind here to retain the context. Note
-            // |context()| return value is of type DestinationContext*, not
-            // PassRefPtr<DestinationContext>.
-            Platform::current()->currentThread()->getWebTaskRunner()->postTask(BLINK_FROM_HERE, crossThreadBind(&DestinationContext::notify, wrapPassRefPtr(context())));
+            readerTaskRunner->postTask(BLINK_FROM_HERE, bind(&DestinationContext::notify, wrapPassRefPtr(context())));
         }
     }
     ~DestinationReader() override
     {
         MutexLocker locker(context()->mutex());
         context()->detachReader();
     }
 
     Result beginRead(const void** buffer, Flags, size_t* available) override
     {
@@ skipping 24 matching lines @@
     RefPtr<DestinationContext::Proxy> m_contextProxy;
 };
 
 class DestinationHandle final : public WebDataConsumerHandle {
 public:
     static std::unique_ptr<WebDataConsumerHandle> create(PassRefPtr<DestinationContext::Proxy> contextProxy)
     {
         return wrapUnique(new DestinationHandle(contextProxy));
     }
 
-    std::unique_ptr<Reader> obtainReader(Client* client)
+    std::unique_ptr<Reader> obtainReader(Client* client, std::unique_ptr<WebTaskRunner> readerTaskRunner)
     {
-        return wrapUnique(new DestinationReader(m_contextProxy, client));
+        return wrapUnique(new DestinationReader(m_contextProxy, client, std::move(readerTaskRunner)));
     }
 
 private:
     DestinationHandle(PassRefPtr<DestinationContext::Proxy> contextProxy) : m_contextProxy(contextProxy) { }
     const char* debugName() const override { return "DestinationHandle"; }
 
     RefPtr<DestinationContext::Proxy> m_contextProxy;
 };
 
 // Bound to the created thread.
 class SourceContext final : public GarbageCollectedFinalized<SourceContext>, public ActiveDOMObject, public WebDataConsumerHandle::Client {
     USING_GARBAGE_COLLECTED_MIXIN(SourceContext);
 public:
     SourceContext(
         PassRefPtr<TeeRootObject> root,
         std::unique_ptr<WebDataConsumerHandle> src,
         PassRefPtr<DestinationContext> dest1,
         PassRefPtr<DestinationContext> dest2,
         ExecutionContext* executionContext)
         : ActiveDOMObject(executionContext)
         , m_root(root)
-        , m_reader(src->obtainReader(this))
+        , m_reader(src->obtainReader(this, TaskRunnerHelper::getUnthrottledTaskRunner(executionContext)->clone()))
         , m_dest1(dest1)
         , m_dest2(dest2)
     {
         suspendIfNeeded();
     }
     ~SourceContext() override
     {
         stopInternal();
     }
 
@@ skipping 60 matching lines @@
     RefPtr<DestinationContext> context2 = DestinationContext::create();
 
     root->initialize(new SourceContext(root, std::move(src), context1, context2, executionContext));
 
     *dest1 = DestinationHandle::create(DestinationContext::Proxy::create(context1, tracker));
     *dest2 = DestinationHandle::create(DestinationContext::Proxy::create(context2, tracker));
 }
 
 void DataConsumerTee::create(ExecutionContext* executionContext, std::unique_ptr<FetchDataConsumerHandle> src, std::unique_ptr<FetchDataConsumerHandle>* dest1, std::unique_ptr<FetchDataConsumerHandle>* dest2)
 {
-    RefPtr<BlobDataHandle> blobDataHandle = src->obtainFetchDataReader(nullptr)->drainAsBlobDataHandle(FetchDataConsumerHandle::Reader::AllowBlobWithInvalidSize);
+    std::unique_ptr<WebTaskRunner> readerTaskRunner = TaskRunnerHelper::getUnthrottledTaskRunner(executionContext)->clone();
+    RefPtr<BlobDataHandle> blobDataHandle = src->obtainFetchDataReader(nullptr, std::move(readerTaskRunner))->drainAsBlobDataHandle(FetchDataConsumerHandle::Reader::AllowBlobWithInvalidSize);
     if (blobDataHandle) {
         *dest1 = FetchBlobDataConsumerHandle::create(executionContext, blobDataHandle);
         *dest2 = FetchBlobDataConsumerHandle::create(executionContext, blobDataHandle);
         return;
     }
 
     std::unique_ptr<WebDataConsumerHandle> webDest1, webDest2;
     DataConsumerTee::create(executionContext, static_cast<std::unique_ptr<WebDataConsumerHandle>>(std::move(src)), &webDest1, &webDest2);
     *dest1 = createFetchDataConsumerHandleFromWebHandle(std::move(webDest1));
     *dest2 = createFetchDataConsumerHandleFromWebHandle(std::move(webDest2));
     return;
 }
 
 } // namespace blink