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

third_party/WebKit/Source/modules/fetch/ReadableStreamDataConsumerHandle.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/ReadableStreamDataConsumerHandle.h"
 
 #include "bindings/core/v8/ExceptionState.h"
 #include "bindings/core/v8/ScopedPersistent.h"
 #include "bindings/core/v8/ScriptFunction.h"
 #include "bindings/core/v8/ScriptState.h"
@@ skipping 97 matching lines @@
 
     private:
         OnRejected(ScriptState* scriptState, PassRefPtr<ReadingContext> context)
             : ScriptFunction(scriptState), m_readingContext(context) {}
 
         RefPtr<ReadingContext> m_readingContext;
     };
 
     class ReaderImpl final : public FetchDataConsumerHandle::Reader {
     public:
-        ReaderImpl(PassRefPtr<ReadingContext> context, Client* client)
+        ReaderImpl(PassRefPtr<ReadingContext> context, Client* client, std::unique_ptr<WebTaskRunner> readerTaskRunner)
             : m_readingContext(context)
         {
-            m_readingContext->attachReader(client);
+            m_readingContext->attachReader(client, std::move(readerTaskRunner));
         }
         ~ReaderImpl() override
         {
             m_readingContext->detachReader();
         }
 
         Result beginRead(const void** buffer, Flags, size_t* available) override
         {
             return m_readingContext->beginRead(buffer, available);
         }
 
         Result endRead(size_t readSize) override
         {
             return m_readingContext->endRead(readSize);
         }
 
     private:
         RefPtr<ReadingContext> m_readingContext;
     };
 
     static PassRefPtr<ReadingContext> create(ScriptState* scriptState, ScriptValue streamReader)
     {
         return adoptRef(new ReadingContext(scriptState, streamReader));
     }
 
-    void attachReader(WebDataConsumerHandle::Client* client)
+    void attachReader(WebDataConsumerHandle::Client* client, std::unique_ptr<WebTaskRunner> readerTaskRunner)
     {
         m_client = client;
+        m_readerTaskRunner = std::move(readerTaskRunner);
         notifyLater();
     }
 
     void detachReader()
     {
         m_client = nullptr;
     }
 
     Result beginRead(const void** buffer, size_t* available)
     {
@@ skipping 75 matching lines @@
     void notify()
     {
         if (!m_client)
             return;
         m_client->didGetReadable();
     }
 
     void notifyLater()
     {
         ASSERT(m_client);
-        Platform::current()->currentThread()->getWebTaskRunner()->postTask(BLINK_FROM_HERE, WTF::bind(&ReadingContext::notify, PassRefPtr<ReadingContext>(this)));
+        m_readerTaskRunner->postTask(BLINK_FROM_HERE, WTF::bind(&ReadingContext::notify, PassRefPtr<ReadingContext>(this)));

[haraken, 2016/07/27 11:05:25]

For example, you can avoid adding m_readerTaskRunner by using
TaskRunnerHelper::getUnthrottledTaskRunner(m_scriptState.get()).

[tzik, 2016/07/27 14:59:41]

Done.
Frame/Document/ScriptState seem not available in other use cases.

     }
 
 private:
     ReadingContext(ScriptState* scriptState, ScriptValue streamReader)
         : m_reader(scriptState->isolate(), streamReader.v8Value())
         , m_scriptState(scriptState)
         , m_client(nullptr)
         , m_pendingOffset(0)
         , m_isReading(false)
         , m_isDone(false)
@@ skipping 17 matching lines @@
         data.GetParameter()->onCollected();
     }
 
     // |m_reader| is a weak persistent. It should be kept alive by someone
     // outside of ReadableStreamDataConsumerHandle.
     // Holding a ScopedPersistent here is safe in terms of cross-world wrapper
     // leakage because we read only Uint8Array chunks from the reader.
     ScopedPersistent<v8::Value> m_reader;
     RefPtr<ScriptState> m_scriptState;
     WebDataConsumerHandle::Client* m_client;
+    std::unique_ptr<WebTaskRunner> m_readerTaskRunner;
     Persistent<DOMUint8Array> m_pendingBuffer;
     size_t m_pendingOffset;
     bool m_isReading;
     bool m_isDone;
     bool m_hasError;
 };
 
 ReadableStreamDataConsumerHandle::ReadableStreamDataConsumerHandle(ScriptState* scriptState, ScriptValue streamReader)
     : m_readingContext(ReadingContext::create(scriptState, streamReader))
 {
 }
 ReadableStreamDataConsumerHandle::~ReadableStreamDataConsumerHandle() = default;
 
-std::unique_ptr<FetchDataConsumerHandle::Reader> ReadableStreamDataConsumerHandle::obtainFetchDataReader(Client* client)
+std::unique_ptr<FetchDataConsumerHandle::Reader> ReadableStreamDataConsumerHandle::obtainFetchDataReader(Client* client, std::unique_ptr<WebTaskRunner> readerTaskRunner)
 {
-    return WTF::wrapUnique(new ReadingContext::ReaderImpl(m_readingContext, client));
+    return WTF::wrapUnique(new ReadingContext::ReaderImpl(m_readingContext, client, std::move(readerTaskRunner)));
 }
 
 } // namespace blink