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