Introduce CompositeDataConsumerHandle.

Source/modules/fetch/CompositeDataConsumerHandleTest.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 "platform/heap/Handle.h"
+#include "public/platform/Platform.h"
+#include "public/platform/WebThread.h"
+#include "public/platform/WebTraceLocation.h"
+#include "public/platform/WebWaitableEvent.h"
+#include "wtf/Locker.h"
+
+#include <gmock/gmock.h>
+#include <gtest/gtest.h>
+
+namespace blink {
+
+namespace {
+
+using Result = WebDataConsumerHandle::Result;
+using Flags = WebDataConsumerHandle::Flags;
+using ::testing::InSequence;
+using ::testing::Return;
+using ::testing::StrictMock;
+using Checkpoint = StrictMock<::testing::MockFunction<void(int)>>;
+
+const Result kShouldWait = WebDataConsumerHandle::ShouldWait;
+const Result kDone = WebDataConsumerHandle::Done;
+const Result kOk = WebDataConsumerHandle::Ok;
+const Result kUnexpectedError = WebDataConsumerHandle::UnexpectedError;
+const Flags kNone = WebDataConsumerHandle::FlagNone;
+
+class NoopClient final : public WebDataConsumerHandle::Client {
+public:
+    void didGetReadable() override { }
+};
+
+class MockReader : public WebDataConsumerHandle::Reader {
+public:
+    static PassOwnPtr<StrictMock<MockReader>> create() { return adoptPtr(new StrictMock<MockReader>); }
+
+    MOCK_METHOD4(read, Result(void*, size_t, Flags, size_t*));
+    MOCK_METHOD3(beginRead, Result(const void**, Flags, size_t*));
+    MOCK_METHOD1(endRead, Result(size_t));
+};
+
+class MockHandle : public WebDataConsumerHandle {
+public:
+    static PassOwnPtr<StrictMock<MockHandle>> create() { return adoptPtr(new StrictMock<MockHandle>); }
+
+    MOCK_METHOD1(obtainReaderInternal, Reader*(Client*));
+};
+
+class ThreadingTestBase : public ThreadSafeRefCounted<ThreadingTestBase> {
+public:
+    class Context : public ThreadSafeRefCounted<Context> {
+    public:
+        static PassRefPtr<Context> create() { return adoptRef(new Context); }
+        void recordAttach(const String& handle)
+        {
+            MutexLocker locker(m_loggingMutex);
+            m_result.append("A reader is attached to " + handle.isolatedCopy() + " on " + threadName() + ".\n");
+        }
+        void recordDetach(const String& handle)
+        {
+            MutexLocker locker(m_loggingMutex);
+            m_result.append("A reader is detached from " + handle.isolatedCopy() + " on " + threadName() + ".\n");
+        }
+
+        const String& result()
+        {
+            MutexLocker locker(m_loggingMutex);
+            return m_result;
+        }
+        WebThread* readingThread() { return m_readingThread.get(); }
+        WebThread* updatingThread() { return m_updatingThread.get(); }
+
+    private:
+        Context()
+        {
+            m_readingThread = adoptPtr(Platform::current()->createThread("reading thread"));

[hiroshige, 2015/06/15 07:29:37]

nit optional: can this be in the initializer list?

[yhirano, 2015/06/16 09:03:53]

Done.

+            m_updatingThread = adoptPtr(Platform::current()->createThread("updating thread"));
+        }
+        String threadName()

[hiroshige, 2015/06/15 07:29:37]

nit optional: currentThreadName() might be a better name.

[yhirano, 2015/06/16 09:03:53]

Done.

+        {
+            if (m_readingThread->isCurrentThread())
+                return "the reading thread";
+            if (m_updatingThread->isCurrentThread())
+                return "the updating thread";
+            return "an unknown thread";
+        }
+
+        OwnPtr<WebThread> m_readingThread;
+        OwnPtr<WebThread> m_updatingThread;
+        Mutex m_loggingMutex;
+        String m_result;
+    };
+
+    class ReaderImpl final : public WebDataConsumerHandle::Reader {
+    public:
+        ReaderImpl(const String& name, PassRefPtr<Context> context) : m_name(name.isolatedCopy()), m_context(context)
+        {
+            m_context->recordAttach(m_name);

[hiroshige, 2015/06/15 07:29:37]

Please take m_name.isolatedCopy() here (rather than in recordAttach()), so that
all the code for thread-safe handling of |m_name| is in ReaderImpl class.

[yhirano, 2015/06/16 09:03:53]

Done.

+        }
+        ~ReaderImpl() override { m_context->recordDetach(m_name); }

[hiroshige, 2015/06/15 07:29:37]

ditto.

[yhirano, 2015/06/16 09:03:53]

Done.

+        Result read(void*, size_t, Flags, size_t*) override { return kShouldWait; }
+        Result beginRead(const void**, Flags, size_t*) override { return kShouldWait; }
+        Result endRead(size_t) override { return kUnexpectedError; }
+
+    private:
+        const String m_name;
+        RefPtr<Context> m_context;
+    };
+    class DataConsumerHandle final : public WebDataConsumerHandle {
+    public:
+        DataConsumerHandle(const String& name, PassRefPtr<Context> context) : m_name(name.isolatedCopy()), m_context(context) { }
+
+    private:
+        Reader* obtainReaderInternal(Client*) { return new ReaderImpl(m_name, m_context); }
+        const String m_name;
+        RefPtr<Context> m_context;
+    };
+
+    void resetReader() { m_reader = nullptr; }
+    void signalDone() { m_waitableEvent->signal(); }
+    const String& result() { return m_context->result(); }
+    WebThread* readingThread() { return m_context->readingThread(); }
+    WebThread* updatingThread() { return m_context->updatingThread(); }
+
+protected:
+    RefPtr<Context> m_context;
+    OwnPtr<CompositeDataConsumerHandle> m_handle;
+    OwnPtr<WebDataConsumerHandle::Reader> m_reader;
+    OwnPtr<WebWaitableEvent> m_waitableEvent;
+    NoopClient m_client;
+};
+
+class ThreadingRegistrationTest : public ThreadingTestBase {
+public:
+    using Self = ThreadingRegistrationTest;
+    void run()
+    {
+        m_context = Context::create();
+        m_waitableEvent = adoptPtr(Platform::current()->createWaitableEvent());
+        m_handle = CompositeDataConsumerHandle::create(adoptPtr(new DataConsumerHandle("handle1", m_context)));
+
+        readingThread()->postTask(FROM_HERE, new Task(threadSafeBind(&Self::obtainReader, this)));
+
+        m_waitableEvent->wait();
+    }
+
+private:
+    void obtainReader()
+    {
+        m_reader = m_handle->obtainReader(&m_client);
+        updatingThread()->postTask(FROM_HERE, new Task(threadSafeBind(&Self::update, this)));
+    }
+    void update()
+    {
+        m_handle->update(adoptPtr(new DataConsumerHandle("handle2", m_context)));
+        readingThread()->postTask(FROM_HERE, new Task(threadSafeBind(&Self::resetReader, this)));
+        readingThread()->postTask(FROM_HERE, new Task(threadSafeBind(&Self::signalDone, this)));
+    }
+};
+
+class ThreadingRegistrationDeleteHandleTest : public ThreadingTestBase {
+public:
+    using Self = ThreadingRegistrationDeleteHandleTest;
+    void run()
+    {
+        m_context = Context::create();
+        m_waitableEvent = adoptPtr(Platform::current()->createWaitableEvent());
+        m_handle = CompositeDataConsumerHandle::create(adoptPtr(new DataConsumerHandle("handle1", m_context)));
+
+        readingThread()->postTask(FROM_HERE, new Task(threadSafeBind(&Self::obtainReader, this)));
+
+        m_waitableEvent->wait();
+    }
+
+private:
+    void obtainReader()
+    {
+        m_reader = m_handle->obtainReader(&m_client);
+        updatingThread()->postTask(FROM_HERE, new Task(threadSafeBind(&Self::update, this)));
+    }
+    void update()
+    {
+        m_handle->update(adoptPtr(new DataConsumerHandle("handle2", m_context)));
+        m_handle = nullptr;
+        readingThread()->postTask(FROM_HERE, new Task(threadSafeBind(&Self::resetReader, this)));
+        readingThread()->postTask(FROM_HERE, new Task(threadSafeBind(&Self::signalDone, this)));
+    }
+};
+
+class ThreadingRegistrationDeleteReaderTest : public ThreadingTestBase {

[hiroshige, 2015/06/15 07:29:37]

Could you add tests for the case where a reader is deleted and then a new one is
created (on readingThread(), and on another thread e.g. updatingThread()) while
updating?

[yhirano, 2015/06/16 09:03:53]

Added:  ThreadingUpdatingReaderWhileUpdatingTest

+public:
+    using Self = ThreadingRegistrationDeleteReaderTest;
+    void run()
+    {
+        m_context = Context::create();
+        m_waitableEvent = adoptPtr(Platform::current()->createWaitableEvent());
+        m_handle = CompositeDataConsumerHandle::create(adoptPtr(new DataConsumerHandle("handle1", m_context)));
+
+        readingThread()->postTask(FROM_HERE, new Task(threadSafeBind(&Self::obtainReader, this)));
+
+        m_waitableEvent->wait();
+    }
+
+private:
+    void obtainReader()
+    {
+        m_reader = m_handle->obtainReader(&m_client);
+        updatingThread()->postTask(FROM_HERE, new Task(threadSafeBind(&Self::update, this)));
+    }
+    void update()
+    {
+        readingThread()->postTask(FROM_HERE, new Task(threadSafeBind(&Self::resetReader, this)));
+        m_handle->update(adoptPtr(new DataConsumerHandle("handle2", m_context)));
+        readingThread()->postTask(FROM_HERE, new Task(threadSafeBind(&Self::resetReader, this)));
+        readingThread()->postTask(FROM_HERE, new Task(threadSafeBind(&Self::signalDone, this)));
+    }
+};
+
+class ThreadingRegistrationUpdateTwiceAtOneTimeTest : public ThreadingTestBase {
+public:
+    using Self = ThreadingRegistrationUpdateTwiceAtOneTimeTest;
+    void run()
+    {
+        m_context = Context::create();
+        m_waitableEvent = adoptPtr(Platform::current()->createWaitableEvent());
+        m_handle = CompositeDataConsumerHandle::create(adoptPtr(new DataConsumerHandle("handle1", m_context)));
+
+        readingThread()->postTask(FROM_HERE, new Task(threadSafeBind(&Self::obtainReader, this)));
+
+        m_waitableEvent->wait();
+    }
+
+private:
+    void obtainReader()
+    {
+        m_reader = m_handle->obtainReader(&m_client);
+        updatingThread()->postTask(FROM_HERE, new Task(threadSafeBind(&Self::update, this)));
+    }
+    void update()
+    {
+        m_handle->update(adoptPtr(new DataConsumerHandle("handle2", m_context)));
+        m_handle->update(adoptPtr(new DataConsumerHandle("handle3", m_context)));
+        readingThread()->postTask(FROM_HERE, new Task(threadSafeBind(&Self::resetReader, this)));
+        readingThread()->postTask(FROM_HERE, new Task(threadSafeBind(&Self::signalDone, this)));
+    }
+};
+
+class ThreadingDoneHandleNotificationTest : public ThreadingTestBase, public WebDataConsumerHandle::Client {

[hiroshige, 2015/06/15 07:29:37]

Could you add a test similar to ThreadingDoneHandleNotificationTest but:
void obtainReader()
{
    m_reader = m_handle->obtainReader(this);
    resetReader();
    postTask signalDone;
}
void didGetReadable() override
{
    ASSERT_NOT_REACHED();
}

i.e. reader is destructed before didGetReadable() notification and thus
didGetReadable() is not called (and the test doesn't crash by use-after-free of
the reader)

[yhirano, 2015/06/16 09:03:52]

Added: ThreadingDoneHandleNoNotificationTest

+public:
+    using Self = ThreadingDoneHandleNotificationTest;
+    void run()
+    {
+        m_context = Context::create();
+        m_waitableEvent = adoptPtr(Platform::current()->createWaitableEvent());
+        m_handle = CompositeDataConsumerHandle::create(CompositeDataConsumerHandle::createDoneHandle());
+
+        readingThread()->postTask(FROM_HERE, new Task(threadSafeBind(&Self::obtainReader, this)));
+
+        m_waitableEvent->wait();
+    }
+
+private:
+    void obtainReader()
+    {
+        m_reader = m_handle->obtainReader(this);
+    }
+    void didGetReadable() override
+    {
+        readingThread()->postTask(FROM_HERE, new Task(threadSafeBind(&Self::resetReader, this)));
+        readingThread()->postTask(FROM_HERE, new Task(threadSafeBind(&Self::signalDone, this)));

[hiroshige, 2015/06/15 07:29:37]

These tasks are same-thread tasks so please use bind instead of
threadSafeBind().

(in these cases, bind(&Self::..., this) is sufficient, rather than
bind(&Self::..., PassRefPtr<Self>(this)), because a reference to this is
retained by the caller of run().)

[yhirano, 2015/06/16 09:03:53]

Discussed offline. We use threadSafeBind for these use-cases, at least for now.

+    }
+};
+
+TEST(CompositeDataConsumerHandleTest, CreateWaitingHandle)
+{
+    char buffer[20];
+    const void* p = nullptr;
+    size_t size = 0;
+    OwnPtr<WebDataConsumerHandle> handle = CompositeDataConsumerHandle::createWaitingHandle();
+    OwnPtr<WebDataConsumerHandle::Reader> reader = handle->obtainReader(nullptr);
+
+    EXPECT_EQ(kShouldWait, reader->read(buffer, sizeof(buffer), kNone, &size));

[hiroshige, 2015/06/15 09:28:57]

Should we check |size| for read() and beginRead(), here and CreateDoneHandle?

[yhirano, 2015/06/16 09:03:52]

The behavior is not specified, so we don't have expectation.

+    EXPECT_EQ(kShouldWait, reader->beginRead(&p, kNone, &size));
+    EXPECT_EQ(kUnexpectedError, reader->endRead(99));
+}
+
+TEST(CompositeDataConsumerHandleTest, CreateDoneHandle)
+{
+    char buffer[20];
+    const void* p = nullptr;
+    size_t size = 0;
+    OwnPtr<WebDataConsumerHandle> handle = CompositeDataConsumerHandle::createDoneHandle();
+    OwnPtr<WebDataConsumerHandle::Reader> reader = handle->obtainReader(nullptr);
+
+    EXPECT_EQ(kDone, reader->read(buffer, sizeof(buffer), kNone, &size));
+    EXPECT_EQ(kDone, reader->beginRead(&p, kNone, &size));
+    EXPECT_EQ(kUnexpectedError, reader->endRead(99));
+    (void)kOk;

[hiroshige, 2015/06/15 07:29:37]

What is this line?

[yhirano, 2015/06/16 09:03:52]

Deleted.

+}
+
+TEST(CompositeDataConsumerHandleTest, Read)
+{
+    char buffer[20];
+    size_t size = 0;
+    NoopClient client;
+    Checkpoint checkpoint;
+
+    OwnPtr<MockHandle> handle1 = MockHandle::create();
+    OwnPtr<MockHandle> handle2 = MockHandle::create();
+    OwnPtr<MockReader> reader1 = MockReader::create();
+    OwnPtr<MockReader> reader2 = MockReader::create();
+
+    InSequence s;
+    EXPECT_CALL(checkpoint, Call(0));
+    EXPECT_CALL(*handle1, obtainReaderInternal(&client)).WillOnce(Return(reader1.get()));
+    EXPECT_CALL(checkpoint, Call(1));
+    EXPECT_CALL(*reader1, read(buffer, sizeof(buffer), kNone, &size)).WillOnce(Return(kOk));
+    EXPECT_CALL(checkpoint, Call(2));
+    EXPECT_CALL(*handle2, obtainReaderInternal(&client)).WillOnce(Return(reader2.get()));
+    EXPECT_CALL(checkpoint, Call(3));
+    EXPECT_CALL(*reader2, read(buffer, sizeof(buffer), kNone, &size)).WillOnce(Return(kOk));
+    EXPECT_CALL(checkpoint, Call(4));
+
+    // They are adopted by |obtainReader|.
+    (void)reader1.leakPtr();
+    (void)reader2.leakPtr();
+
+    OwnPtr<CompositeDataConsumerHandle> handle = CompositeDataConsumerHandle::create(handle1.release());
+    checkpoint.Call(0);
+    OwnPtr<CompositeDataConsumerHandle::Reader> reader = handle->obtainReader(&client);
+    checkpoint.Call(1);
+    EXPECT_EQ(kOk, reader->read(buffer, sizeof(buffer), kNone, &size));
+    checkpoint.Call(2);
+    handle->update(handle2.release());
+    checkpoint.Call(3);
+    EXPECT_EQ(kOk, reader->read(buffer, sizeof(buffer), kNone, &size));
+    checkpoint.Call(4);
+}
+
+TEST(CompositeDataConsumerHandleTest, TwoPhaseRead)
+{
+    const void* p = nullptr;
+    size_t size = 0;
+    Checkpoint checkpoint;
+
+    OwnPtr<MockHandle> handle1 = MockHandle::create();
+    OwnPtr<MockHandle> handle2 = MockHandle::create();
+    OwnPtr<MockReader> reader1 = MockReader::create();
+    OwnPtr<MockReader> reader2 = MockReader::create();
+
+    InSequence s;
+    EXPECT_CALL(checkpoint, Call(0));
+    EXPECT_CALL(*handle1, obtainReaderInternal(nullptr)).WillOnce(Return(reader1.get()));
+    EXPECT_CALL(checkpoint, Call(1));
+    EXPECT_CALL(*reader1, beginRead(&p, kNone, &size)).WillOnce(Return(kOk));
+    EXPECT_CALL(checkpoint, Call(2));
+    EXPECT_CALL(*reader1, endRead(0)).WillOnce(Return(kOk));
+    EXPECT_CALL(checkpoint, Call(3));
+    EXPECT_CALL(*handle2, obtainReaderInternal(nullptr)).WillOnce(Return(reader2.get()));
+    EXPECT_CALL(checkpoint, Call(4));
+    EXPECT_CALL(*reader2, beginRead(&p, kNone, &size)).WillOnce(Return(kOk));
+    EXPECT_CALL(checkpoint, Call(5));
+    EXPECT_CALL(*reader2, endRead(0)).WillOnce(Return(kOk));
+    EXPECT_CALL(checkpoint, Call(6));
+
+    // They are adopted by |obtainReader|.
+    (void)reader1.leakPtr();
+    (void)reader2.leakPtr();
+
+    OwnPtr<CompositeDataConsumerHandle> handle = CompositeDataConsumerHandle::create(handle1.release());
+    checkpoint.Call(0);
+    OwnPtr<CompositeDataConsumerHandle::Reader> reader = handle->obtainReader(nullptr);
+    checkpoint.Call(1);
+    EXPECT_EQ(kOk, reader->beginRead(&p, kNone, &size));
+    checkpoint.Call(2);
+    EXPECT_EQ(kOk, reader->endRead(0));
+    checkpoint.Call(3);
+    handle->update(handle2.release());
+    checkpoint.Call(4);
+    EXPECT_EQ(kOk, reader->beginRead(&p, kNone, &size));
+    checkpoint.Call(5);
+    EXPECT_EQ(kOk, reader->endRead(0));
+    checkpoint.Call(6);
+}
+
+TEST(CompositeDataConsumerHandleTest, HangingTwoPhaseRead)

[hiroshige, 2015/06/15 07:29:37]

Could you add a test similar to HangingTwoPhaseRead but the first beginRead()
returns non-OK (and thus corresponding endRead() is not called)?

[yhirano, 2015/06/16 09:03:53]

Done.

+{
+    const void* p = nullptr;
+    size_t size = 0;
+    Checkpoint checkpoint;
+
+    OwnPtr<MockHandle> handle1 = MockHandle::create();
+    OwnPtr<MockHandle> handle2 = MockHandle::create();
+    OwnPtr<MockReader> reader1 = MockReader::create();
+    OwnPtr<MockReader> reader2 = MockReader::create();
+
+    InSequence s;
+    EXPECT_CALL(checkpoint, Call(0));
+    EXPECT_CALL(*handle1, obtainReaderInternal(nullptr)).WillOnce(Return(reader1.get()));
+    EXPECT_CALL(checkpoint, Call(1));
+    EXPECT_CALL(*reader1, beginRead(&p, kNone, &size)).WillOnce(Return(kOk));
+    EXPECT_CALL(checkpoint, Call(2));
+    EXPECT_CALL(checkpoint, Call(3));
+    EXPECT_CALL(*reader1, endRead(0)).WillOnce(Return(kOk));
+    EXPECT_CALL(*handle2, obtainReaderInternal(nullptr)).WillOnce(Return(reader2.get()));
+    EXPECT_CALL(checkpoint, Call(4));
+    EXPECT_CALL(*reader2, beginRead(&p, kNone, &size)).WillOnce(Return(kOk));
+    EXPECT_CALL(checkpoint, Call(5));
+    EXPECT_CALL(*reader2, endRead(0)).WillOnce(Return(kOk));
+    EXPECT_CALL(checkpoint, Call(6));
+
+    // They are adopted by |obtainReader|.
+    (void)reader1.leakPtr();
+    (void)reader2.leakPtr();
+
+    OwnPtr<CompositeDataConsumerHandle> handle = CompositeDataConsumerHandle::create(handle1.release());
+    checkpoint.Call(0);
+    OwnPtr<CompositeDataConsumerHandle::Reader> reader = handle->obtainReader(nullptr);
+    checkpoint.Call(1);
+    EXPECT_EQ(kOk, reader->beginRead(&p, kNone, &size));
+    checkpoint.Call(2);
+    handle->update(handle2.release());
+    checkpoint.Call(3);
+    EXPECT_EQ(kOk, reader->endRead(0));
+    checkpoint.Call(4);
+    EXPECT_EQ(kOk, reader->beginRead(&p, kNone, &size));
+    checkpoint.Call(5);
+    EXPECT_EQ(kOk, reader->endRead(0));
+    checkpoint.Call(6);
+
+}
+
+TEST(CompositeDataConsumerHandleTest, RegisterClientOnDifferentThreads)
+{
+    ThreadingRegistrationTest test;
+    test.run();
+
+    EXPECT_EQ(
+        "A reader is attached to handle1 on the reading thread.\n"
+        "A reader is detached from handle1 on the reading thread.\n"
+        "A reader is attached to handle2 on the reading thread.\n"
+        "A reader is detached from handle2 on the reading thread.\n",
+        test.result());
+}
+
+TEST(CompositeDataConsumerHandleTest, DeleteHandleWhileUpdating)
+{
+    ThreadingRegistrationDeleteHandleTest test;
+    test.run();
+
+    EXPECT_EQ(
+        "A reader is attached to handle1 on the reading thread.\n"
+        "A reader is detached from handle1 on the reading thread.\n"
+        "A reader is attached to handle2 on the reading thread.\n"
+        "A reader is detached from handle2 on the reading thread.\n",
+        test.result());
+}
+
+TEST(CompositeDataConsumerHandleTest, DeleteReaderWhileUpdating)
+{
+    ThreadingRegistrationDeleteReaderTest test;
+    test.run();
+
+    EXPECT_EQ(
+        "A reader is attached to handle1 on the reading thread.\n"
+        "A reader is detached from handle1 on the reading thread.\n",
+        test.result());
+}
+
+TEST(CompositeDataConsumerHandleTest, UpdateTwiceAtOnce)
+{
+    ThreadingRegistrationUpdateTwiceAtOneTimeTest test;
+    test.run();
+
+    EXPECT_EQ(
+        "A reader is attached to handle1 on the reading thread.\n"
+        "A reader is detached from handle1 on the reading thread.\n"
+        "A reader is attached to handle3 on the reading thread.\n"
+        "A reader is detached from handle3 on the reading thread.\n",
+        test.result());
+}
+
+TEST(CompositeDataConsumerHandleTest, DoneHandleNotification)
+{
+    ThreadingDoneHandleNotificationTest test;
+    // Test this function returns.
+    test.run();
+}
+
+} // namespace
+
+} // namespace blink