Introduce CompositeDataConsumerHandle.

Source/modules/fetch/CompositeDataConsumerHandle.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 "config.h"
+#include "modules/fetch/CompositeDataConsumerHandle.h"
+
+#include "platform/Task.h"
+#include "platform/ThreadSafeFunctional.h"
+#include "public/platform/Platform.h"
+#include "public/platform/WebThread.h"
+#include "public/platform/WebTraceLocation.h"
+#include "wtf/Locker.h"
+#include "wtf/ThreadSafeRefCounted.h"
+#include "wtf/ThreadingPrimitives.h"
+
+namespace blink {
+
+using Result = WebDataConsumerHandle::Result;
+
+namespace {
+
+class RepeatingReader final : public WebDataConsumerHandle::Reader {
+public:
+    explicit RepeatingReader(Result r) : m_result(r) { }
+    Result read(void*, size_t, WebDataConsumerHandle::Flags, size_t *readSize) override
+    {
+        *readSize = 0;
+        return m_result;
+    }
+    Result beginRead(const void**, WebDataConsumerHandle::Flags, size_t *available) override
+    {
+        *available = 0;
+        return m_result;
+    }
+    Result endRead(size_t) override
+    {
+        return WebDataConsumerHandle::UnexpectedError;
+    }
+
+private:
+    const Result m_result;
+};
+
+class WaitingHandle final : public WebDataConsumerHandle {
+private:
+    Reader* obtainReaderInternal(Client*) override { return new RepeatingReader(ShouldWait); }
+};
+
+class DoneHandle final : public WebDataConsumerHandle {
+private:
+    Reader* obtainReaderInternal(Client*) override { return new RepeatingReader(Done); }
+};
+
+} // namespace
+
+class CompositeDataConsumerHandle::ReaderImpl final : public WebDataConsumerHandle::Reader {
+public:
+    explicit ReaderImpl(PassRefPtr<Context>);
+    ~ReaderImpl() override;
+    Result read(void* data, size_t /* size */, Flags, size_t* readSize) override;
+    Result beginRead(const void** buffer, Flags, size_t* available) override;
+    Result endRead(size_t readSize) override;
+
+private:
+    RefPtr<Context> m_context;
+};
+
+class CompositeDataConsumerHandle::Context : public ThreadSafeRefCounted<Context> {
+public:
+    using Token = unsigned;
+    static PassRefPtr<Context> create(PassOwnPtr<WebDataConsumerHandle> handle) { return adoptRef(new Context(handle)); }
+    Reader* reader() { return m_reader.get(); }
+    PassOwnPtr<ReaderImpl> obtainReader(Client* client)
+    {
+        MutexLocker locker(m_mutex);
+        ++m_token;

[hiroshige, 2015/06/09 05:37:16]

Assert that there is no reader (e.g. |!m_reader|).

[yhirano, 2015/06/09 06:30:52]

Done.

+        m_client = client;
+        m_readerThread = Platform::current()->currentThread();
+        m_reader = m_handle->obtainReader(m_client);
+        return adoptPtr(new ReaderImpl(this));
+    }
+    void detachReader()
+    {
+        MutexLocker locker(m_mutex);

[hiroshige, 2015/06/09 05:37:15]

Assert that m_readerThread is the current thread.

[yhirano, 2015/06/09 06:30:52]

Done.

+        ++m_token;
+        m_reader = nullptr;
+        m_readerThread = nullptr;
+        m_client = nullptr;
+    }
+    void update(PassOwnPtr<WebDataConsumerHandle> handle)
+    {
+        MutexLocker locker(m_mutex);
+        m_handle = handle;
+        if (!m_readerThread) {
+            // There is no reader.
+            return;
+        }
+        ++m_token;
+        updateReaderNoLock(m_token);
+    }
+
+private:
+    explicit Context(PassOwnPtr<WebDataConsumerHandle> handle)
+        : m_handle(handle)
+        , m_readerThread(nullptr)
+        , m_client(nullptr)
+        , m_token(0)
+    {
+    }
+    void updateReader(Token token)
+    {
+        MutexLocker locker(m_mutex);
+        updateReaderNoLock(token);
+    }
+    void updateReaderNoLock(Token token)
+    {
+        if (token != m_token) {
+            // This request is not fresh. Ignore it.
+            return;
+        }
+        ASSERT(m_readerThread);
+        ASSERT(m_reader);
+        if (m_readerThread->isCurrentThread()) {
+            // Unregister the old one, then register the new one.
+            m_reader = nullptr;
+            m_reader = m_handle->obtainReader(m_client);
+            return;
+        }
+        ++m_token;
+        m_readerThread->postTask(FROM_HERE, new Task(threadSafeBind(&Context::updateReader, this, m_token)));

[hiroshige, 2015/06/09 05:37:15]

|this| might destructed before this task is executed:
1. handle->update() is called, and postTask() here is called.
2. handle is destructed.
3. reader is destructed, Context is destructed.
4. updateReader() is called on the invalidated Context.

How about passing RefPtr to |this|? (I think RefPtr/PassRefPtr to this can be
passed to threadSafeBind() because Context is ThreadSafeRefCounted, and we can
call static void updateReader(PassRefPtr<Context> that, Token token) via
postTask(), but not so sure. ping me offline if necessary)

[hiroshige, 2015/06/09 05:47:10]

Correction: The current code is fine.
CrossThreadCopier wraps pointers to ThreadSafeRefCounted by PassRefPtr, which is
stored as RefPtr in WTF::Function, thus the reference is retained until the task
execution.

+    }
+
+    OwnPtr<Reader> m_reader;
+    OwnPtr<WebDataConsumerHandle> m_handle;
+    // 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;
+    Client* m_client;
+    Token m_token;
+    Mutex m_mutex;
+};
+
+CompositeDataConsumerHandle::ReaderImpl::ReaderImpl(PassRefPtr<Context> context) : m_context(context) { }
+
+CompositeDataConsumerHandle::ReaderImpl::~ReaderImpl()
+{
+    m_context->detachReader();
+}
+
+Result CompositeDataConsumerHandle::ReaderImpl::read(void* data, size_t size, Flags flags, size_t* readSize)
+{
+    return m_context->reader()->read(data, size, flags, readSize);
+}
+
+Result CompositeDataConsumerHandle::ReaderImpl::beginRead(const void** buffer, Flags flags, size_t* available)
+{
+    return m_context->reader()->beginRead(buffer, flags, available);
+}
+
+Result CompositeDataConsumerHandle::ReaderImpl::endRead(size_t readSize)
+{
+    return m_context->reader()->endRead(readSize);
+}
+
+CompositeDataConsumerHandle::CompositeDataConsumerHandle(PassOwnPtr<WebDataConsumerHandle> handle)
+    : m_context(Context::create(handle)) { }
+
+CompositeDataConsumerHandle::~CompositeDataConsumerHandle() { }
+
+WebDataConsumerHandle::Reader* CompositeDataConsumerHandle::obtainReaderInternal(Client* client)
+{
+    return m_context->obtainReader(client).leakPtr();
+}
+
+void CompositeDataConsumerHandle::update(PassOwnPtr<WebDataConsumerHandle> handle)
+{
+    m_context->update(handle);
+}
+
+PassOwnPtr<WebDataConsumerHandle> CompositeDataConsumerHandle::createWaitingHandle()
+{
+    return adoptPtr(new WaitingHandle);
+}
+
+PassOwnPtr<WebDataConsumerHandle> CompositeDataConsumerHandle::createDoneHandle()
+{
+    return adoptPtr(new DoneHandle);
+}
+
+} // namespace blink