Arachne.h

/* Copyright (c) 2015-2016 Stanford University
 *
 * Permission to use, copy, modify, and distribute this software for any
 * purpose with or without fee is hereby granted, provided that the above
 * copyright notice and this permission notice appear in all copies.
 *
 * THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR(S) DISCLAIM ALL WARRANTIES
 * WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
 * MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL AUTHORS BE LIABLE FOR
 * ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
 * WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN
 * ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
 * OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
 */

#ifndef ARACHNE_H_
#define ARACHNE_H_

#include <assert.h>
#include <stdio.h>
#include <functional>
#include <vector>
#include <mutex>
#include <deque>
#include <atomic>
#include <queue>
#include <string>


#include "../PerfUtils/Cycles.h"
#include "Common.h"

namespace Arachne {

using PerfUtils::Cycles;

// Forward declare to break circular dependency between ThreadContext and
// ConditionVariable
struct ThreadContext;

// This is used in createThread.
extern volatile uint32_t numCores;
extern volatile uint32_t maxNumCores;

// Used in inline functions.
extern FILE* errorStream;
void dispatch();

struct ThreadId {
    ThreadContext* context;
    uint32_t generation;

    ThreadId(ThreadContext* context, uint32_t generation)
        : context(context)
        , generation(generation) { }

    ThreadId()
        : context(NULL)
        , generation(0) { }

    bool
    operator==(const ThreadId& other) const {
        return context == other.context && generation == other.generation;
    }

    bool
    operator!=(const ThreadId& other) const {
        return !(*this == other);
    }
};

void init(int* argcp = NULL, const char** argv = NULL);
void shutDown();
void waitForTermination();
void yield();
void sleep(uint64_t ns);

inline void block() {
    dispatch();
}
void signal(ThreadId id);
void join(ThreadId id);
ThreadId getThreadId();

void setErrorStream(FILE* ptr);
void testInit();
void testDestroy();

class SpinLock {
 public:
    explicit SpinLock(std::string name) : state(false), name(name) {}
    SpinLock() : state(false), name("unnamed") {}
    ~SpinLock(){}

    inline void
    lock() {
        uint64_t startOfContention = 0;
        while (state.exchange(true, std::memory_order_acquire) != false) {
            if (startOfContention == 0) {
                startOfContention = Cycles::rdtsc();
            } else {
                uint64_t now = Cycles::rdtsc();
                if (Cycles::toSeconds(now - startOfContention) > 1.0) {
                    fprintf(errorStream,
                            "%s SpinLock locked for one second; deadlock?\n",
                            name.c_str());
                    startOfContention = now;
                }
            }
            yield();
        }
    }

    inline bool
    try_lock() {
        // If the original value was false, then we successfully acquired the
        // lock. Otherwise we failed.
        return !state.exchange(true, std::memory_order_acquire);
    }

    inline void
    unlock() {
        state.store(false, std::memory_order_release);
    }

    inline void
    setName(std::string name) {
        this->name = name;
    }

 private:
    // Implements the lock: false means free, true means locked
    std::atomic<bool> state;

    // Descriptive name for this SpinLock. Used to identify the purpose of
    // the lock, what it protects, where it exists in the codebase, etc.
    //
    // Used to identify the lock when reporting a potential deadlock.
    std::string name;
};

class ConditionVariable {
 public:
    ConditionVariable();
    ~ConditionVariable();
    void notifyOne();
    void notifyAll();
    template <typename LockType> void wait(LockType& lock);
    template <typename LockType> void waitFor(LockType& lock, uint64_t ns);
 private:
    // Ordered collection of threads that are waiting on this condition
    // variable. Threads are processed from this list in FIFO order when a
    // notifyOne() is called.
    std::deque<ThreadId> blockedThreads;
    DISALLOW_COPY_AND_ASSIGN(ConditionVariable);
};

const Arachne::ThreadId NullThread;
// The declarations in following section are private to the thread library.

struct ThreadInvocationEnabler {
    virtual void runThread() = 0;
    virtual ~ThreadInvocationEnabler() { }
};

template<typename F>
struct ThreadInvocation : public ThreadInvocationEnabler {
    F mainFunction;

    explicit ThreadInvocation(F mainFunction)
        : mainFunction(mainFunction) {
        static_assert(sizeof(ThreadInvocation<F>) <= CACHE_LINE_SIZE,
                "Arachne requires the function and arguments for a thread to "
                "fit within one cache line.");
    }

    void
    runThread() {
        mainFunction();
    }
};

struct ThreadContext {
    void* stack;

    void* sp;

    volatile uint64_t wakeupTimeInCycles;

    uint32_t generation;

    SpinLock joinLock;

    ConditionVariable joinCV;

    uint8_t idInCore;


    struct alignas(CACHE_LINE_SIZE) {
        char data[CACHE_LINE_SIZE];
    }
    threadInvocation;

    ThreadContext() = delete;
    ThreadContext(ThreadContext&) = delete;

    explicit ThreadContext(uint8_t idInCore);
};

// Largest number of Arachne threads that can be simultaneously created on each
// core.
const int maxThreadsPerCore = 56;

const size_t SpaceForSavedRegisters = 48;

const uint64_t BLOCKED = ~0L;

const uint64_t UNOCCUPIED = ~0L - 1;

void schedulerMainLoop();
void swapcontext(void **saved, void **target);
void threadMain(int id);

extern thread_local int kernelThreadId;
extern thread_local ThreadContext *loadedContext;
extern thread_local ThreadContext** localThreadContexts;
extern std::vector<ThreadContext**> allThreadContexts;

struct MaskAndCount{
    uint64_t occupied : 56;
    uint8_t numOccupied : 8;
};

extern std::vector< std::atomic<MaskAndCount> *> occupiedAndCount;
extern thread_local std::atomic<MaskAndCount> *localOccupiedAndCount;

#ifdef TEST
static std::deque<uint64_t> mockRandomValues;
#endif

inline uint64_t
random(void) {
#ifdef TEST
    if (!mockRandomValues.empty()) {
        uint64_t returnValue = mockRandomValues.front();
        mockRandomValues.pop_front();
        return returnValue;
    }
#endif

    // This function came from the following site.
    // http://stackoverflow.com/a/1640399/391161
    //
    // It was chosen because it was advertised to be fast, but this fact has
    // not yet been verified or disproved through experiments.
    static uint64_t x = 123456789, y = 362436069, z = 521288629;
    uint64_t t;
    x ^= x << 16;
    x ^= x >> 5;
    x ^= x << 1;

    t = x;
    x = y;
    y = z;
    z = t ^ x ^ y;

    return z;
}

template<typename _Callable, typename... _Args>
ThreadId
createThread(int kId, _Callable&& __f, _Args&&... __args) {
    if (kId == -1)
        kId = kernelThreadId;

    auto task = std::bind(
            std::forward<_Callable>(__f), std::forward<_Args>(__args)...);

    bool success;
    uint32_t index;
    do {
        // Each iteration through this loop makes one attempt to enqueue the
        // task to the specified core. Multiple iterations are required only if
        // there is contention for the core's state variables.
        MaskAndCount slotMap = *occupiedAndCount[kId];
        MaskAndCount oldSlotMap = slotMap;

        if (slotMap.numOccupied >= maxThreadsPerCore)
            return NullThread;

        // Search for a non-occupied slot and attempt to reserve the slot
        index = 0;
        while ((slotMap.occupied & (1L << index)) && index < maxThreadsPerCore)
            index++;

        slotMap.occupied =
            (slotMap.occupied | (1L << index)) & 0x00FFFFFFFFFFFFFF;
        slotMap.numOccupied++;

        success = occupiedAndCount[kId]->compare_exchange_strong(oldSlotMap,
                slotMap);
    } while (!success);

    // Copy the thread invocation into the byte array.
    new (&allThreadContexts[kId][index]->threadInvocation)
        Arachne::ThreadInvocation<decltype(task)>(task);
    allThreadContexts[kId][index]->wakeupTimeInCycles = 0;
    return ThreadId(allThreadContexts[kId][index],
            allThreadContexts[kId][index]->generation);
}

// The ends the private section of the thread library.

template<typename _Callable, typename... _Args>
ThreadId
createThread(_Callable&& __f, _Args&&... __args) {
    // Find a kernel thread to enqueue to by picking two at random and choosing
    // the one with the fewest Arachne threads.
    int kId;
    int choice1 = static_cast<int>(random()) % numCores;
    int choice2 = static_cast<int>(random()) % numCores;
    while (choice2 == choice1 && numCores > 1)
        choice2 = static_cast<int>(random()) % numCores;

    if (occupiedAndCount[choice1]->load().numOccupied <
            occupiedAndCount[choice2]->load().numOccupied)
        kId = choice1;
    else
        kId = choice2;

    return createThread(kId, __f, __args...);
}

template <typename LockType> void
ConditionVariable::wait(LockType& lock) {
#if TIME_TRACE
    TimeTrace::record("Wait on Core %d", kernelThreadId);
#endif
    blockedThreads.push_back(
            ThreadId(loadedContext, loadedContext->generation));
    lock.unlock();
    dispatch();
#if TIME_TRACE
    TimeTrace::record("About to acquire lock after waking up");
#endif
    lock.lock();
}

template <typename LockType> void
ConditionVariable::waitFor(LockType& lock, uint64_t ns) {
    blockedThreads.push_back(
            ThreadId(loadedContext, loadedContext->generation));
    loadedContext->wakeupTimeInCycles =
        Cycles::rdtsc() + Cycles::fromNanoseconds(ns);
    lock.unlock();
    dispatch();
    lock.lock();
}

} // namespace Arachne

#endif // ARACHNE_H_