Introduce DataConsumerHandleTee

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 "config.h"
+#include "modules/fetch/DataConsumerTee.h"
+
+#include "core/dom/ActiveDOMObject.h"
+#include "core/dom/ExecutionContext.h"
+#include "platform/Task.h"
+#include "platform/ThreadSafeFunctional.h"
+#include "platform/heap/Handle.h"
+#include "public/platform/Platform.h"
+#include "public/platform/WebThread.h"
+#include "public/platform/WebTraceLocation.h"
+#include "wtf/Deque.h"
+#include "wtf/Functional.h"
+#include "wtf/ThreadSafeRefCounted.h"
+#include "wtf/ThreadingPrimitives.h"
+#include "wtf/Vector.h"
+
+namespace blink {
+
+using Result = WebDataConsumerHandle::Result;
+using Flags = WebDataConsumerHandle::Flags;
+
+namespace {
+
+// This file contains the "tee" implementation. There are several classes and
+// their relationship is complicated, so let me describe here.
+//
+// Tee::create function creates two DestinationHandles (destinations) from one
+// WebDataConsumerHandle (source). In fact, it uses a reader of the source
+// handle.
+//
+// SourceContext reads data from the source reader and enques it to two
+// destination contexts. Destination readers read data from its assocaited

[hiroshige, 2015/06/19 06:40:35]

nit: associated

[yhirano, 2015/06/19 07:27:52]

Done.

+// contexts. Here is an object graph.
+//
+//   R: the root object
+//  SR: the source reader
+//  SC: the SourceContext
+// DCn: nth DestinationContext
+// DRn: nth DestinationReader
+// DHn: nth DestinationHandle
+// ---------
+// (normal)
+//       ---> DC1 <--- DR1 / DH1
+//       |
+//       |
+// SR <--SC <-> R
+//       |
+//       |
+//       ---> DC2 <--- DR2 / DH2
+//
+// ---------
+//
+// The root object (R) refers to the SourceContext, and is referred by many
+// objects including the SourceContext. As the root object is a
+// ThreadSafeRefCounted that reference cycle keeps the entire pipe alive.
+// The root object only has "stop" function that breaks the reference cycle.
+// It will be called when:
+//  - The source context finishes reading,
+//  - The source context gets errored while reading,
+//  - The execution context associated with the source context is stopped or
+//  - All destination handles and readers are gone.
+//
+// ---------
+// (stopped)
+//       ---> DC1 <--- DR1 / DH1
+//       |
+//       |
+// SR <--SC --> R
+//       |
+//       |
+//       ---> DC2 <--- DR2 / DH2
+//
+// -------
+// When |stop| is called, no one has a strong reference to the source context
+// and it will be collected.
+//
+
+class SourceContext;
+
+class TeeRootObject final : public ThreadSafeRefCounted<TeeRootObject> {
+public:
+    static PassRefPtr<TeeRootObject> create() { return adoptRef(new TeeRootObject()); }
+
+    void initialize(SourceContext* context)

[hiroshige, 2015/06/19 06:40:35]

optional: I prefer names like |sourceContext|, |destinationContext| etc. rather
than just |context|, because there are three kinds of contexts in this file.

[yhirano, 2015/06/19 07:27:52]

Done in this class. I think calling a DestinationContext as context in
DestinationX classes is sane.

+    {
+        m_context = context;
+    }
+
+    // This function can be called from any thread.
+    void stop()
+    {
+        m_context = nullptr;
+    }
+
+private:
+    TeeRootObject() = default;
+
+    CrossThreadPersistent<SourceContext> m_context;
+};
+
+class DestinationTracker final : public ThreadSafeRefCounted<DestinationTracker> {
+public:
+    enum Identifier {
+        First,
+        Second,
+    };
+    static PassRefPtr<DestinationTracker> create(PassRefPtr<TeeRootObject> root) { return adoptRef(new DestinationTracker(root)); }
+    void didDetach(Identifier identifier)

[hiroshige, 2015/06/19 06:40:35]

How about calling m_root->stop() in the dtor?
Then we can remove Identifier, m_mutex, m_isFirst|SecondActive, didDetach().

[yhirano, 2015/06/19 07:27:52]

You're right, thanks! Done.

+    {
+        ASSERT(m_root);
+        MutexLocker locker(m_mutex);
+        if (identifier == First) {
+            m_isFirstActive = false;
+        } else {
+            ASSERT(identifier == Second);
+            m_isSecondActive = false;
+        }
+        if (!m_isFirstActive && !m_isSecondActive) {
+            m_root->stop();
+            m_root = nullptr;
+        }
+    }
+
+private:
+    explicit DestinationTracker(PassRefPtr<TeeRootObject> root) : m_root(root) , m_isFirstActive(true) , m_isSecondActive(true) { }
+
+    RefPtr<TeeRootObject> m_root;
+    Mutex m_mutex;
+    bool m_isFirstActive;
+    bool m_isSecondActive;
+};
+
+class DestinationContext final : public ThreadSafeRefCounted<DestinationContext> {
+public:
+    using Identifier = DestinationTracker::Identifier;
+    static PassRefPtr<DestinationContext> create(PassRefPtr<DestinationTracker> tracker, Identifier identifier) { return adoptRef(new DestinationContext(tracker, identifier)); }
+
+    void enqueue(const char* buffer, size_t size)
+    {
+        bool needsNotification = false;
+        {
+            MutexLocker locker(m_mutex);
+            needsNotification = m_queue.isEmpty();
+            OwnPtr<Vector<char>> data = adoptPtr(new Vector<char>);
+            data->append(buffer, size);
+            m_queue.append(data.release());
+        }
+        if (needsNotification)
+            notify();
+    }
+
+    void setResult(Result r)
+    {
+        ASSERT(r != WebDataConsumerHandle::Ok);
+        ASSERT(r != WebDataConsumerHandle::ShouldWait);
+        {
+            MutexLocker locker(m_mutex);
+            if (m_result != WebDataConsumerHandle::ShouldWait) {
+                // The result was already set.
+                return;
+            }
+            m_result = r;
+            if (r != WebDataConsumerHandle::Done && !m_isTwoPhaseReadInProgress)
+                m_queue.clear();
+        }
+        notify();
+    }
+
+    void notify()
+    {
+        {
+            MutexLocker locker(m_mutex);
+            if (!m_client) {
+                // No client is registered.
+                return;
+            }
+            ASSERT(m_readerThread);
+            if (!m_readerThread->isCurrentThread()) {
+                m_readerThread->postTask(FROM_HERE, new Task(threadSafeBind(&DestinationContext::notify, this)));
+                return;
+            }
+        }
+        // The reading thread is the current thread.
+        if (m_client)
+            m_client->didGetReadable();
+    }
+
+    // The following functions don't use lock.
+    void attachReader(WebDataConsumerHandle::Client* client)
+    {
+        ASSERT(!m_readerThread);
+        ASSERT(!m_client);
+        m_readerThread = Platform::current()->currentThread();
+        m_client = client;
+    }
+    void detachReader()
+    {
+        ASSERT(m_readerThread && m_readerThread->isCurrentThread());
+        m_readerThread = nullptr;
+        m_client = nullptr;
+        if (!m_isHandleActive)
+            detach();

[hiroshige, 2015/06/19 06:40:35]

optional: I prefer to create a tracker object that is referenced by
DestinationHandler and DestinationReader and references DestinationContext, and
call DestinationContext::detach() on the tracker's dtor; liveness management
using |m_isHandleActive| (as well as |m_isFirstActive| and |m_isSecondActive|)
seems somehow prone to error.
However, this increases the numbers of classes and objects...

[yhirano, 2015/06/19 07:27:52]

Done.

+    }
+    void detachHandle()
+    {
+        m_isHandleActive = false;
+        if (!m_readerThread)
+            detach();
+    }
+    const OwnPtr<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();
+        ASSERT(m_offset + size <= top->size());
+        if (top->size() <= m_offset + size) {
+            m_offset = 0;
+            m_queue.removeFirst();
+        } else {
+            m_offset += size;
+        }
+    }
+    Result result() { return m_result; }
+    Mutex& mutex() { return m_mutex; }
+
+private:
+    DestinationContext(PassRefPtr<DestinationTracker> tracker, Identifier identifier)
+        : m_tracker(tracker)
+        , m_identifier(identifier)
+        , m_result(WebDataConsumerHandle::ShouldWait)
+        , m_readerThread(nullptr)
+        , m_client(nullptr)
+        , m_offset(0)
+        , m_isTwoPhaseReadInProgress(false)
+        , m_isHandleActive(true)
+    {
+    }
+
+    void detach()
+    {
+        ASSERT(!m_client);
+        ASSERT(!m_readerThread);
+        ASSERT(!m_isHandleActive);
+        m_tracker->didDetach(m_identifier);
+        m_queue.clear();
+    }
+
+    RefPtr<DestinationTracker> m_tracker;
+    Identifier m_identifier;
+    Result m_result;
+    Deque<OwnPtr<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;
+    WebDataConsumerHandle::Client* m_client;
+    size_t m_offset;
+    bool m_isTwoPhaseReadInProgress;
+    bool m_isHandleActive;
+    Mutex m_mutex;
+};
+
+class DestinationReader final : public WebDataConsumerHandle::Reader {
+public:
+    DestinationReader(PassRefPtr<DestinationContext> context, WebDataConsumerHandle::Client* client) : m_context(context)
+    {
+        MutexLocker(m_context->mutex());
+        m_context->attachReader(client);
+        if (client)
+            Platform::current()->currentThread()->postTask(FROM_HERE, new Task(bind(&DestinationContext::notify, m_context)));
+    }
+    ~DestinationReader() override
+    {
+        MutexLocker(m_context->mutex());
+        m_context->detachReader();
+    }
+
+    Result read(void* buffer, size_t size, Flags, size_t* readSize) override
+    {
+        MutexLocker locker(m_context->mutex());
+        *readSize = 0;
+        if (m_context->isEmpty())
+            return m_context->result();
+
+        const OwnPtr<Vector<char>>& chunk = m_context->top();
+        size_t sizeToCopy = std::min(size, chunk->size() - m_context->offset());
+        std::copy(chunk->data() + m_context->offset(), chunk->data() + m_context->offset() + sizeToCopy, static_cast<char*>(buffer));
+        m_context->consume(sizeToCopy);
+        *readSize = sizeToCopy;
+        return WebDataConsumerHandle::Ok;
+    }
+
+    Result beginRead(const void** buffer, Flags, size_t* available) override
+    {
+        MutexLocker locker(m_context->mutex());
+        *available = 0;
+        *buffer = nullptr;
+        if (m_context->isEmpty())
+            return m_context->result();
+
+        const OwnPtr<Vector<char>>& chunk = m_context->top();
+        *available = chunk->size() - m_context->offset();
+        *buffer = chunk->data() + m_context->offset();
+        return WebDataConsumerHandle::Ok;
+    }
+
+    Result endRead(size_t readSize) override
+    {
+        MutexLocker locker(m_context->mutex());
+        if (m_context->isEmpty())
+            return WebDataConsumerHandle::UnexpectedError;
+        m_context->consume(readSize);
+        return WebDataConsumerHandle::Ok;
+    }
+
+private:
+    RefPtr<DestinationContext> m_context;
+};
+
+class DestinationHandle final : public WebDataConsumerHandle {
+public:
+    explicit DestinationHandle(PassRefPtr<DestinationContext> context) : m_context(context) { }
+    ~DestinationHandle() override
+    {
+        m_context->detachHandle();
+    }
+
+private:
+    DestinationReader* obtainReaderInternal(Client* client) { return new DestinationReader(m_context, client); }
+
+    RefPtr<DestinationContext> m_context;
+};
+
+// Bound to the created thread.
+class SourceContext final : public GarbageCollectedFinalized<SourceContext>, public ActiveDOMObject, public WebDataConsumerHandle::Client {
+    WILL_BE_USING_GARBAGE_COLLECTED_MIXIN(SourceContext);
+public:
+    SourceContext(
+        PassRefPtr<TeeRootObject> root,
+        PassOwnPtr<WebDataConsumerHandle> src,
+        PassRefPtr<DestinationContext> dest1,
+        PassRefPtr<DestinationContext> dest2,
+        ExecutionContext* executionContext)
+        : ActiveDOMObject(executionContext)
+        , m_root(root)
+        , m_reader(src->obtainReader(this))
+        , m_dest1(dest1)
+        , m_dest2(dest2)
+    {
+        suspendIfNeeded();
+    }
+    ~SourceContext() override
+    {
+        stopInternal();
+    }
+
+    void didGetReadable() override
+    {
+        ASSERT(m_reader);
+        Result r = WebDataConsumerHandle::Ok;
+        while (true) {
+            const void* buffer = nullptr;
+            size_t available = 0;
+            r = m_reader->beginRead(&buffer, WebDataConsumerHandle::FlagNone, &available);
+            if (r == WebDataConsumerHandle::ShouldWait)
+                return;
+            if (r != WebDataConsumerHandle::Ok)
+                break;
+            m_dest1->enqueue(static_cast<const char*>(buffer), available);
+            m_dest2->enqueue(static_cast<const char*>(buffer), available);
+            m_reader->endRead(available);
+        }
+        m_dest1->setResult(r);
+        m_dest2->setResult(r);
+        stopInternal();
+    }
+
+    void stop() override
+    {
+        stopInternal();
+        ActiveDOMObject::stop();
+    }
+
+    DEFINE_INLINE_VIRTUAL_TRACE()
+    {
+        ActiveDOMObject::trace(visitor);
+    }
+
+private:
+    void stopInternal()
+    {
+        if (!m_root)
+            return;
+        // When we already set a result, this result setting will be ignored.
+        m_dest1->setResult(WebDataConsumerHandle::UnexpectedError);
+        m_dest2->setResult(WebDataConsumerHandle::UnexpectedError);
+        m_root->stop();
+        m_root = nullptr;
+        m_reader = nullptr;
+        m_dest1 = nullptr;
+        m_dest2 = nullptr;
+    }
+
+    RefPtr<TeeRootObject> m_root;
+    OwnPtr<WebDataConsumerHandle::Reader> m_reader;
+    RefPtr<DestinationContext> m_dest1;
+    RefPtr<DestinationContext> m_dest2;
+};
+
+} // namespace
+
+void DataConsumerTee::create(ExecutionContext* executionContext, PassOwnPtr<WebDataConsumerHandle> src, OwnPtr<WebDataConsumerHandle>* dest1, OwnPtr<WebDataConsumerHandle>* dest2)
+{
+    RefPtr<TeeRootObject> root = TeeRootObject::create();
+    RefPtr<DestinationTracker> tracker = DestinationTracker::create(root);
+    RefPtr<DestinationContext> context1 = DestinationContext::create(tracker, DestinationTracker::First);
+    RefPtr<DestinationContext> context2 = DestinationContext::create(tracker, DestinationTracker::Second);
+
+    root->initialize(new SourceContext(root, src, context1, context2, executionContext));
+
+    *dest1 = adoptPtr(new DestinationHandle(context1));
+    *dest2 = adoptPtr(new DestinationHandle(context2));
+}
+
+} // namespace blink