Introduce CompositeDataConsumerHandle.

Source/modules/fetch/CompositeDataConsumerHandle.cpp

Index: Source/modules/fetch/CompositeDataConsumerHandle.cpp
diff --git a/Source/modules/fetch/CompositeDataConsumerHandle.cpp b/Source/modules/fetch/CompositeDataConsumerHandle.cpp
new file mode 100644
index 0000000000000000000000000000000000000000..f34e002df689590acbd33ec4ccc2042a250b3d8b
--- /dev/null
+++ b/Source/modules/fetch/CompositeDataConsumerHandle.cpp
@@ -0,0 +1,171 @@
+// 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 WaitingHandle final : public WebDataConsumerHandle {
+public:
+    Result read(void*, size_t, Flags, size_t *readSize) override
+    {
+        *readSize = 0;
+        return ShouldWait;
+    }
+    Result beginRead(const void**, Flags, size_t *available) override
+    {
+        *available = 0;
+        return ShouldWait;
+    }
+    Result endRead(size_t) override
+    {
+        return UnexpectedError;
+    }
+    void registerClient(Client*) override { }
+    void unregisterClient() override { }
+};
+
+class DoneHandle final : public WebDataConsumerHandle {
+public:
+    Result read(void*, size_t, Flags, size_t *readSize) override
+    {
+        *readSize = 0;
+        return Done;
+    }
+    Result beginRead(const void**, Flags, size_t *available) override
+    {
+        *available = 0;
+        return Done;
+    }
+    Result endRead(size_t) override
+    {
+        return UnexpectedError;
+    }
+    void registerClient(Client*) override { }
+    void unregisterClient() override { }
+};
+
+} // namespace
+
+class CompositeDataConsumerHandle::MutexHolder : public ThreadSafeRefCounted<MutexHolder> {
+public:
+    static PassRefPtr<MutexHolder> create() { return adoptRef(new MutexHolder); }
+    Mutex& get() { return m_mutex; }
+    bool isActive() const { return m_isActive; }
+    void stop() { m_isActive = false; }
+
+private:
+    MutexHolder() : m_isActive(true) { }
+    Mutex m_mutex;
+    bool m_isActive;
+};
+
+CompositeDataConsumerHandle::CompositeDataConsumerHandle(PassOwnPtr<WebDataConsumerHandle> handle)
+    : m_handle(handle), m_mutex(MutexHolder::create()), m_client(nullptr), m_clientThread(nullptr) { }
+
+CompositeDataConsumerHandle::~CompositeDataConsumerHandle()
+{
+    // We need this lock in order to avoid race with |updateOnTheRightThread|.
+    MutexLocker locker(m_mutex->get());
+    m_mutex->stop();
+}
+
+Result CompositeDataConsumerHandle::read(void* data, size_t size, Flags flags, size_t* readSize)

[hiroshige, 2015/06/03 08:45:05]

How about adding ASSERT(!m_client || m_clientThread ==
Platform::current()->currentThread()) (or RELEASE_ASSERT) in read() and other
methods that have this thread restriction?

[yhirano, 2015/06/04 05:43:02]

Done.

+{
+    MutexLocker locker(m_mutex->get());
+    return m_handle->read(data, size, flags, readSize);
+}
+
+Result CompositeDataConsumerHandle::beginRead(const void** buffer, Flags flags, size_t* available)
+{
+    MutexLocker locker(m_mutex->get());
+    return m_handle->beginRead(buffer, flags, available);
+}
+
+Result CompositeDataConsumerHandle::endRead(size_t readSize)
+{
+    MutexLocker locker(m_mutex->get());
+    return m_handle->endRead(readSize);
+}
+
+void CompositeDataConsumerHandle::registerClient(Client* client)
+{
+    MutexLocker locker(m_mutex->get());
+    m_client = client;
+    m_clientThread = Platform::current()->currentThread();
+    ASSERT(m_clientThread);
+    m_handle->registerClient(client);
+}
+
+void CompositeDataConsumerHandle::unregisterClient()
+{
+    MutexLocker locker(m_mutex->get());
+    m_client = nullptr;
+    m_clientThread = nullptr;
+    m_handle->unregisterClient();
+}
+
+void CompositeDataConsumerHandle::update(PassOwnPtr<WebDataConsumerHandle> handle)

[hiroshige, 2015/06/03 10:35:56]

Multiple calls to update() might be executed out-of-order.
For example,

0. registerClient() on Thread A.
1. update(handle1) on Thread B. -> postTask to Thread A.
2. unregisterClient() on Thread A.
3. update(handle2) on Thread B, and m_handle = handle2 because |m_client| is
null here.
4. updateInternal(handle1) is called on Thread A (posted on Step 1), and
m_handle = handle1.

or

0. registerClient() on Thread A.
1. update(handle1) on Thread B. -> postTask to Thread A.
2. unregisterClient() on Thread A.
3. registerClient() on Thread B.
4. updateInternal(handle1) is called on Thread A (posted on Step 1) -> postTask
to Thread B again.
5. update(handle2) on Thread B, and m_handle = handle2.
6. updateInternal(handle1) on Thread B )posted on Step 4), and m_handle =
handle1.

The nature of thread restriction of WebDataConsumerHandle might make the
semantics complex, because "the thread that calls registerClient()" is
dynamically defined after handle's construction, can be changed, can be
non-existing e.g. before the first call to registerClient() / after the last
call to unregisterClient().

[yhirano, 2015/06/04 05:43:02]

You're right, thanks.
I fixed the problem by introducing a request token.

There is another issue that update will never take effect if client registration
/ unregistration is made multiple time on multiple threads (register on A ->
update on C -> unregister on A -> register on B -> unregister on B -> ...), but
I don't care about that pathological case.

+{
+    MutexLocker locker(m_mutex->get());
+    updateInternal(handle);
+}
+
+void CompositeDataConsumerHandle::updateInternal(PassOwnPtr<WebDataConsumerHandle> handle)

[hiroshige, 2015/06/03 08:45:05]

I think |handle| shouldn't have any client here, right?
If so, it would be good to mention this in the comment of update() and add
ASSERT()'s.

[yhirano, 2015/06/04 05:43:02]

Yes. Added a comment on the header. I cannot put an assertion because we don't
have such a predicate.

+{
+    if (!m_client) {
+        m_handle = handle;
+        return;
+    }
+    // We need to transfer the client from the old handle to the new handle
+    // on the thread on which the client is registered.
+    if (m_clientThread == Platform::current()->currentThread()) {
+        // ... and it is the curren thread.

[hiroshige, 2015/06/03 08:45:05]

nit: current.

[yhirano, 2015/06/04 05:43:02]

Done.

+        m_handle->unregisterClient();
+        m_handle = handle;
+        m_handle->registerClient(m_client);
+        return;
+    }
+    // Now we need to post a task to the right thread.
+    m_clientThread->postTask(FROM_HERE, new Task(threadSafeBind(
+        &updateOnTheRightThread, AllowCrossThreadAccess(this), handle, m_mutex)));
+}
+
+void CompositeDataConsumerHandle::updateOnTheRightThread(
+    CompositeDataConsumerHandle* that, PassOwnPtr<WebDataConsumerHandle> handle, PassRefPtr<MutexHolder> mutex)
+{
+    MutexLocker locker(mutex->get());
+    if (!mutex->isActive()) {
+        // The handle was already deleted. Do nothing.
+        return;
+    }
+    // |that| is protected by the lock (see the destructor).
+    that->updateInternal(handle);
+}
+
+PassOwnPtr<WebDataConsumerHandle> CompositeDataConsumerHandle::createWaitingHandle()
+{
+    return adoptPtr(new WaitingHandle);
+}
+
+PassOwnPtr<WebDataConsumerHandle> CompositeDataConsumerHandle::createDoneHandle()
+{
+    return adoptPtr(new DoneHandle);
+}
+
+} // namespace blink