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

third_party/WebKit/Source/modules/fetch/DataConsumerHandleTestUtil.h

 // 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.
 
 #ifndef DataConsumerHandleTestUtil_h
 #define DataConsumerHandleTestUtil_h
 
 #include "bindings/core/v8/ScriptState.h"
 #include "core/testing/NullExecutionContext.h"
 #include "gin/public/isolate_holder.h"
@@ skipping 20 matching lines @@
 
 namespace blink {
 
 class DataConsumerHandleTestUtil {
     STATIC_ONLY(DataConsumerHandleTestUtil);
 public:
     class NoopClient final : public WebDataConsumerHandle::Client {
         DISALLOW_NEW();
     public:
         void didGetReadable() override { }
+        WebTaskRunner* getTaskRunner() override { return nullptr; }
     };
 
     // Thread has a WebThreadSupportingGC. It initializes / shutdowns
     // additional objects based on the given policy. The constructor and the
     // destructor blocks during the setup and the teardown.
     class Thread final {
         USING_FAST_MALLOC(Thread);
     public:
         // Initialization policy of a thread.
         enum InitializationPolicy {
             // Only garbage collection is supported.
             GarbageCollection,
             // Creating an isolate in addition to GarbageCollection.
             ScriptExecution,
             // Creating an execution context in addition to ScriptExecution.
             WithExecutionContext,
         };
 
         Thread(const char* name, InitializationPolicy = GarbageCollection);
         ~Thread();
 
         WebThreadSupportingGC* thread() { return m_thread.get(); }
+        WebTaskRunner* taskRunner() { return m_thread->platformThread().getWebTaskRunner(); }
         ExecutionContext* getExecutionContext() { return m_executionContext.get(); }
         ScriptState* getScriptState() { return m_scriptState.get(); }
         v8::Isolate* isolate() { return m_isolateHolder->isolate(); }
 
     private:
         void initialize();
         void shutdown();
 
         std::unique_ptr<WebThreadSupportingGC> m_thread;
         const InitializationPolicy m_initializationPolicy;
@@ skipping 47 matching lines @@
                 ASSERT(m_holder);
                 m_holder->readingThread()->postTask(location, std::move(task));
             }
             void postTaskToUpdatingThread(const WebTraceLocation& location, std::unique_ptr<CrossThreadClosure> task)
             {
                 MutexLocker locker(m_holderMutex);
                 ASSERT(m_holder);
                 m_holder->updatingThread()->postTask(location, std::move(task));
             }
 
+            WebTaskRunner* getReadingThreadTaskRunner()
+            {
+                return m_holder->readingThread()->platformThread().getWebTaskRunner();
+            }
+
         private:
             Context()
                 : m_holder(nullptr) { }
             String currentThreadName()
             {
                 MutexLocker locker(m_holderMutex);
                 if (m_holder) {
                     if (m_holder->readingThread()->isCurrentThread())
                         return "the reading thread";
                     if (m_holder->updatingThread()->isCurrentThread())
@@ skipping 123 matching lines @@
         ThreadingHandleNotificationTest() = default;
         void obtainReader()
         {
             m_reader = m_handle->obtainReader(this);
         }
         void didGetReadable() override
         {
             postTaskToReadingThread(BLINK_FROM_HERE, crossThreadBind(&Self::resetReader, wrapPassRefPtr(this)));
             postTaskToReadingThread(BLINK_FROM_HERE, crossThreadBind(&Self::signalDone, wrapPassRefPtr(this)));
         }
+        WebTaskRunner* getTaskRunner() override
+        {
+            return m_context->getReadingThreadTaskRunner();
+        }
 
         std::unique_ptr<WebDataConsumerHandle> m_handle;
     };
 
     class ThreadingHandleNoNotificationTest : public ThreadingTestBase, public WebDataConsumerHandle::Client {
     public:
         using Self = ThreadingHandleNoNotificationTest;
         static PassRefPtr<Self> create() { return adoptRef(new Self); }
 
         void run(std::unique_ptr<WebDataConsumerHandle> handle)
@@ skipping 10 matching lines @@
         void obtainReader()
         {
             m_reader = m_handle->obtainReader(this);
             m_reader = nullptr;
             postTaskToReadingThread(BLINK_FROM_HERE, crossThreadBind(&Self::signalDone, wrapPassRefPtr(this)));
         }
         void didGetReadable() override
         {
             ASSERT_NOT_REACHED();
         }
+        WebTaskRunner* getTaskRunner() override
+        {
+            return m_context->getReadingThreadTaskRunner();
+        }
 
         std::unique_ptr<WebDataConsumerHandle> m_handle;
     };
 
     class MockFetchDataConsumerHandle : public FetchDataConsumerHandle {
     public:
         static std::unique_ptr<::testing::StrictMock<MockFetchDataConsumerHandle>> create() { return wrapUnique(new ::testing::StrictMock<MockFetchDataConsumerHandle>); }
         MOCK_METHOD1(obtainFetchDataReader, std::unique_ptr<Reader>(Client*));
 
     private:
@@ skipping 26 matching lines @@
         DEFINE_INLINE_VIRTUAL_TRACE()
         {
             FetchDataLoader::Client::trace(visitor);
         }
 
         MOCK_METHOD1(didFetchDataLoadedBlobHandleMock, void(RefPtr<BlobDataHandle>));
         MOCK_METHOD1(didFetchDataLoadedArrayBufferMock, void(DOMArrayBuffer*));
         MOCK_METHOD1(didFetchDataLoadedString, void(const String&));
         MOCK_METHOD0(didFetchDataLoadStream, void());
         MOCK_METHOD0(didFetchDataLoadFailed, void());
+        MOCK_METHOD0(getTaskRunner, WebTaskRunner*());
 
         void didFetchDataLoadedArrayBuffer(DOMArrayBuffer* arrayBuffer) override
         {
             didFetchDataLoadedArrayBufferMock(arrayBuffer);
         }
         // In mock methods we use RefPtr<> rather than PassRefPtr<>.
         void didFetchDataLoadedBlobHandle(PassRefPtr<BlobDataHandle> blobDataHandle) override
         {
             didFetchDataLoadedBlobHandleMock(blobDataHandle);
         }
@@ skipping 53 matching lines @@
             Context();
             bool isEmpty() const { return m_commands.isEmpty(); }
             const Command& top();
             void consume(size_t);
             size_t offset() const { return m_offset; }
             void notify();
             void notifyInternal();
 
             Deque<Command> m_commands;
             size_t m_offset;
-            WebThread* m_readerThread;
+            std::unique_ptr<WebTaskRunner> m_readerTaskRunner;

[haraken, 2016/07/28 09:19:30]

Ditto.

[tzik, 2016/07/28 14:45:34]

This task runner seems to be used to identify attached reader, where each reader
has its own thread.

             Client* m_client;
             Result m_result;
             bool m_isHandleAttached;
             Mutex m_mutex;
             std::unique_ptr<WaitableEvent> m_detached;
         };
 
         Context* getContext() { return m_context.get(); }
         std::unique_ptr<Reader> obtainReader(Client*) override;
 
@@ skipping 19 matching lines @@
     };
 
     // HandleReader reads all data from the given WebDataConsumerHandle using
     // Reader::read on the thread on which it is created. When reading is done
     // or failed, it calls the given callback with the result.
     class HandleReader final : public WebDataConsumerHandle::Client {
         USING_FAST_MALLOC(HandleReader);
     public:
         using OnFinishedReading = WTF::Function<void(std::unique_ptr<HandleReadResult>)>;
 
-        HandleReader(std::unique_ptr<WebDataConsumerHandle>, std::unique_ptr<OnFinishedReading>);
+        HandleReader(std::unique_ptr<WebDataConsumerHandle>, std::unique_ptr<OnFinishedReading>, std::unique_ptr<WebTaskRunner> readerTaskRunner);
         void didGetReadable() override;
+        WebTaskRunner* getTaskRunner() override;
 
     private:
         void runOnFinishedReading(std::unique_ptr<HandleReadResult>);
 
+        std::unique_ptr<WebTaskRunner> m_readerTaskRunner;
         std::unique_ptr<WebDataConsumerHandle::Reader> m_reader;
         std::unique_ptr<OnFinishedReading> m_onFinishedReading;
         Vector<char> m_data;
     };
 
     // HandleTwoPhaseReader does the same as HandleReader, but it uses
     // |beginRead| / |endRead| instead of |read|.
     class HandleTwoPhaseReader final : public WebDataConsumerHandle::Client {
         USING_FAST_MALLOC(HandleTwoPhaseReader);
     public:
         using OnFinishedReading = WTF::Function<void(std::unique_ptr<HandleReadResult>)>;
 
-        HandleTwoPhaseReader(std::unique_ptr<WebDataConsumerHandle>, std::unique_ptr<OnFinishedReading>);
+        HandleTwoPhaseReader(std::unique_ptr<WebDataConsumerHandle>, std::unique_ptr<OnFinishedReading>, std::unique_ptr<WebTaskRunner>);
         void didGetReadable() override;
+        WebTaskRunner* getTaskRunner() override;
 
     private:
         void runOnFinishedReading(std::unique_ptr<HandleReadResult>);
 
+        std::unique_ptr<WebTaskRunner> m_readerTaskRunner;
         std::unique_ptr<WebDataConsumerHandle::Reader> m_reader;
         std::unique_ptr<OnFinishedReading> m_onFinishedReading;
         Vector<char> m_data;
     };
 
     // HandleReaderRunner<T> creates a dedicated thread and run T on the thread
     // where T is one of HandleReader and HandleTwophaseReader.
     template <typename T>
     class HandleReaderRunner final {
         STACK_ALLOCATED();
     public:
         explicit HandleReaderRunner(std::unique_ptr<WebDataConsumerHandle> handle)
             : m_thread(wrapUnique(new Thread("reading thread")))
             , m_event(wrapUnique(new WaitableEvent()))
             , m_isDone(false)
         {
-            m_thread->thread()->postTask(BLINK_FROM_HERE, crossThreadBind(&HandleReaderRunner::start, crossThreadUnretained(this), passed(std::move(handle))));
+            m_thread->thread()->postTask(BLINK_FROM_HERE, crossThreadBind(&HandleReaderRunner::start, crossThreadUnretained(this), passed(std::move(handle)), passed(m_thread->taskRunner()->clone())));
         }
         ~HandleReaderRunner()
         {
             wait();
         }
 
         std::unique_ptr<HandleReadResult> wait()
         {
             if (m_isDone)
                 return nullptr;
             m_event->wait();
             m_isDone = true;
             return std::move(m_result);
         }
 
     private:
-        void start(std::unique_ptr<WebDataConsumerHandle> handle)
+        void start(std::unique_ptr<WebDataConsumerHandle> handle, std::unique_ptr<WebTaskRunner> readerTaskRunner)
         {
-            m_handleReader = wrapUnique(new T(std::move(handle), WTF::bind(&HandleReaderRunner::onFinished, WTF::unretained(this))));
+            m_handleReader = wrapUnique(new T(std::move(handle), WTF::bind(&HandleReaderRunner::onFinished, WTF::unretained(this)), std::move(readerTaskRunner)));
         }
 
         void onFinished(std::unique_ptr<HandleReadResult> result)
         {
             m_handleReader = nullptr;
             m_result = std::move(result);
             m_event->signal();
         }
 
         std::unique_ptr<Thread> m_thread;
         std::unique_ptr<WaitableEvent> m_event;
         std::unique_ptr<HandleReadResult> m_result;
         bool m_isDone;
 
         std::unique_ptr<T> m_handleReader;
     };
 };
 
+std::unique_ptr<WebTaskRunner> getCurrentTaskRunner();
+
 } // namespace blink
 
 #endif // DataConsumerHandleTestUtil_h