Yarn: Add Ticket, Ticket::Queue

src/Yarn/Ticket.hpp

+// Copyright 2019 The SwiftShader Authors. All Rights Reserved.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//    http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#ifndef yarn_ticket_hpp
+#define yarn_ticket_hpp
+
+#include "Pool.hpp"
+#include "ConditionVariable.hpp"
+#include "Scheduler.hpp"
+
+namespace yarn {
+
+// Ticket is a synchronization primitive used to serially order execution.
+//
+// Tickets exist in 3 mutually exclusive states: Waiting, Called and Finished.
+//
+// Tickets are obtained from a Ticket::Queue, using the Ticket::Queue::take()
+// methods. The order in which tickets are taken from the queue dictates the
+// order in which they are called.
+//
+// The first ticket to be taken from a queue will be in the 'called' state,
+// others will be in the 'waiting' state until the previous ticket has finished.
+//
+// Ticket::wait() will block until the ticket is called.
+// Ticket::done() sets the ticket into the 'finished' state and calls the next
+// taken ticket from the queue.
+//
+// If a ticket is taken from a queue and does not have done() called before
+// its last reference is dropped, it will implicitly call done(), calling the
+// next ticket.
+//
+// Example:
+//
+//  void runTasksConcurrentThenSerially(int numConcurrentTasks)
+//  {
+//      yarn::Ticket::Queue queue;
+//      for (int i = 0; i < numConcurrentTasks; i++)
+//      {
+//          auto ticket = queue.take();
+//          yarn::schedule([=] {
+//              doConcurrentWork(); // <- function may be called concurrently
+//              ticket.wait(); // <- serialize tasks
+//              doSerialWork(); // <- function will not be called concurrently
+//              ticket.done(); // <- optional, as done() is called implicitly on dropping of last reference
+//          });
+//      }
+//  }
+class Ticket
+{
+    struct Shared;
+    struct Record;
+public:
+
+    // Queue hands out Tickets.
+    class Queue
+    {
+    public:
+        // take() returns a single ticket from the queue.
+        inline Ticket take();
+
+        // take() retrieves count tickets from the queue, calling f() with each
+        // retrieved ticket.
+        // F must be a function of the signature: void(Ticket&&)
+        template <typename F>
+        inline void take(size_t count, const F& f);
+
+    private:
+        std::shared_ptr<Shared> shared = std::make_shared<Shared>();
+        UnboundedPool<Record> pool;
+    };
+
+    inline Ticket() = default;
+    inline Ticket(const Ticket& other) = default;
+    inline Ticket(Ticket&& other) = default;
+    inline Ticket& operator = (const Ticket& other) = default;
+
+    // wait() blocks until the ticket is called.
+    inline void wait() const;
+
+    // done() marks the ticket as finished and calls the next ticket.
+    inline void done() const;
+
+    // onCall() registers the function f to be invoked when this ticket is
+    // called. If the ticket is already called prior to calling onCall(), then
+    // f() will be executed immediately.
+    // F must be a function of the signature: void F()
+    template<typename F>
+    inline void onCall(F&& f) const;
+
+private:
+    // Internal doubly-linked-list data structure. One per ticket instance.
+    struct Record
+    {
+        inline ~Record();
+
+        inline void done();
+        inline void callAndUnlock(std::unique_lock<std::mutex> &lock);
+
+        ConditionVariable isCalledCondVar;
+
+        std::shared_ptr<Shared> shared;
+        Record *next = nullptr; // guarded by shared->mutex
+        Record *prev = nullptr; // guarded by shared->mutex
+        inline void unlink(); // guarded by shared->mutex
+        Task onCall; // guarded by shared->mutex
+        bool isCalled = false; // guarded by shared->mutex
+        std::atomic<bool> isDone = { false };
+    };
+
+    // Data shared between all tickets and the queue.
+    struct Shared
+    {
+        std::mutex mutex;
+        Record tail;
+    };
+
+    inline Ticket(Loan<Record>&& record);
+
+    Loan<Record> record;
+};
+
+////////////////////////////////////////////////////////////////////////////////
+// Ticket
+////////////////////////////////////////////////////////////////////////////////
+
+Ticket::Ticket(Loan<Record>&& record) : record(std::move(record)) {}
+
+void Ticket::wait() const
+{
+    std::unique_lock<std::mutex> lock(record->shared->mutex);
+    record->isCalledCondVar.wait(lock, [this] { return record->isCalled; });
+}
+
+void Ticket::done() const
+{
+    record->done();
+}
+
+template<typename Function>
+void Ticket::onCall(Function&& f) const
+{
+    std::unique_lock<std::mutex> lock(record->shared->mutex);
+    if (record->isCalled)
+    {
+        yarn::schedule(std::move(f));
+        return;
+    }
+    if (record->onCall)
+    {
+        struct Joined
+        {
+            void operator() () const { a(); b(); }
+            Task a, b;
+        };
+        record->onCall = std::move(Joined{ std::move(record->onCall), std::move(f) });
+    }
+    else
+    {
+        record->onCall = std::move(f);
+    }
+}
+
+////////////////////////////////////////////////////////////////////////////////
+// Ticket::Queue
+////////////////////////////////////////////////////////////////////////////////
+
+Ticket Ticket::Queue::take()
+{
+    Ticket out;
+    take(1, [&](Ticket&& ticket) { out = std::move(ticket); });
+    return out;
+}
+
+template <typename F>
+void Ticket::Queue::take(size_t n, const F& f)
+{
+    Loan<Record> first, last;
+    pool.borrow(n, [&] (Loan<Record>&& record) {
+        Loan<Record> rec = std::move(record);
+        rec->shared = shared;
+        if (first.get() == nullptr)
+        {
+            first = rec;
+        }
+        if (last.get() != nullptr)
+        {
+            last->next = rec.get();
+            rec->prev = last.get();
+        }
+        last = rec;
+        f(std::move(Ticket(std::move(rec))));
+    });
+    last->next = &shared->tail;
+    std::unique_lock<std::mutex> lock(shared->mutex);
+    first->prev = shared->tail.prev;
+    shared->tail.prev = last.get();
+    if (first->prev == nullptr)
+    {
+        first->callAndUnlock(lock);
+    }
+    else
+    {
+        first->prev->next = first.get();
+    }
+}
+
+////////////////////////////////////////////////////////////////////////////////
+// Ticket::Record
+////////////////////////////////////////////////////////////////////////////////
+
+Ticket::Record::~Record()
+{
+    if (shared != nullptr)
+    {
+        done();
+    }
+}
+
+void Ticket::Record::done()
+{
+    if (isDone.exchange(true)) { return; }
+    std::unique_lock<std::mutex> lock(shared->mutex);
+    auto callNext = (prev == nullptr && next != nullptr) ? next : nullptr;
+    unlink();
+    if (callNext != nullptr) // lock needs to be held otherwise callNext might be destructed.
+    {
+        callNext->callAndUnlock(lock);
+    }
+}
+
+void Ticket::Record::callAndUnlock(std::unique_lock<std::mutex> &lock)
+{
+    if (isCalled) { return; }
+    isCalled = true;
+    Task task;
+    std::swap(task, onCall);
+    isCalledCondVar.notify_all();
+    lock.unlock();
+
+    if (task)
+    {
+        yarn::schedule(std::move(task));
+    }
+}
+
+void Ticket::Record::unlink()
+{
+    if (prev != nullptr) { prev->next = next; }
+    if (next != nullptr) { next->prev = prev; }
+    prev = nullptr;
+    next = nullptr;
+}
+
+} // namespace yarn
+
+#endif  // yarn_ticket_hpp

src/Yarn/Ticket_test.cpp

+// Copyright 2019 The SwiftShader Authors. All Rights Reserved.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//    http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#include "Yarn_test.hpp"
+
+#include "Yarn/Ticket.hpp"
+
+TEST_P(WithBoundScheduler, Ticket)
+{
+    yarn::Ticket::Queue queue;
+
+    constexpr int count = 1000;
+    std::atomic<int> next = { 0 };
+    int result[count] = {};
+
+    for (int i = 0; i < count; i++)
+    {
+        auto ticket = queue.take();
+        yarn::schedule([ticket, i, &result, &next] {
+            ticket.wait();
+            result[next++] = i;
+            ticket.done();
+        });
+    }
+
+    queue.take().wait();
+
+    for (int i = 0; i < count; i++)
+    {
+        ASSERT_EQ(result[i], i);
+    }
+}