kvthread.hh

/* Masstree
 * Eddie Kohler, Yandong Mao, Robert Morris
 * Copyright (c) 2012-2014 President and Fellows of Harvard College
 * Copyright (c) 2012-2014 Massachusetts Institute of Technology
 *
 * Permission is hereby granted, free of charge, to any person obtaining a
 * copy of this software and associated documentation files (the "Software"),
 * to deal in the Software without restriction, subject to the conditions
 * listed in the Masstree LICENSE file. These conditions include: you must
 * preserve this copyright notice, and you cannot mention the copyright
 * holders in advertising related to the Software without their permission.
 * The Software is provided WITHOUT ANY WARRANTY, EXPRESS OR IMPLIED. This
 * notice is a summary of the Masstree LICENSE file; the license in that file
 * is legally binding.
 */
#ifndef KVTHREAD_HH
#define KVTHREAD_HH 1
#include "mtcounters.hh"
#include "compiler.hh"
#include "circular_int.hh"
#include "timestamp.hh"
#include <assert.h>
#include <pthread.h>
#include <sys/mman.h>
#include <stdlib.h>

class threadinfo;
class loginfo;

extern volatile uint64_t globalepoch;    // global epoch, updated regularly
extern volatile bool recovering;

struct memdebug {
#if HAVE_MEMDEBUG
    enum {
        magic_value = 389612313 /* = 0x17390319 */,
        magic_free_value = 2015593488 /* = 0x78238410 */
    };
    int magic;
    int freetype;
    size_t size;
    int after_rcu;
    int line;
    const char* file;

    static void* make(void* p, size_t size, int freetype) {
        if (p) {
            memdebug *m = reinterpret_cast<memdebug *>(p);
            m->magic = magic_value;
            m->freetype = freetype;
            m->size = size;
            m->after_rcu = 0;
            m->line = 0;
            m->file = 0;
            return m + 1;
        } else
            return p;
    }
    static void set_landmark(void* p, const char* file, int line) {
        if (p) {
            memdebug* m = reinterpret_cast<memdebug*>(p) - 1;
            m->file = file;
            m->line = line;
        }
    }
    static void *check_free(void *p, size_t size, int freetype) {
        memdebug *m = reinterpret_cast<memdebug *>(p) - 1;
        free_checks(m, size, freetype, false, "deallocate");
        m->magic = magic_free_value;
        return m;
    }
    static void check_rcu(void *p, size_t size, int freetype) {
        memdebug *m = reinterpret_cast<memdebug *>(p) - 1;
        free_checks(m, size, freetype, false, "deallocate_rcu");
        m->after_rcu = 1;
    }
    static void *check_free_after_rcu(void *p, int freetype) {
        memdebug *m = reinterpret_cast<memdebug *>(p) - 1;
        free_checks(m, 0, freetype, true, "free_after_rcu");
        m->magic = magic_free_value;
        return m;
    }
    static bool check_use(const void *p, int type) {
        const memdebug *m = reinterpret_cast<const memdebug *>(p) - 1;
        return m->magic == magic_value && (type == 0 || (m->freetype >> 8) == type);
    }
    static bool check_use(const void *p, int type1, int type2) {
        const memdebug *m = reinterpret_cast<const memdebug *>(p) - 1;
        return m->magic == magic_value
            && ((m->freetype >> 8) == type1 || (m->freetype >> 8) == type2);
    }
    static void assert_use(const void *p, memtag tag) {
        if (!check_use(p, tag))
            hard_assert_use(p, tag, (memtag) -1);
    }
    static void assert_use(const void *p, memtag tag1, memtag tag2) {
        if (!check_use(p, tag1, tag2))
            hard_assert_use(p, tag1, tag2);
    }
  private:
    static void free_checks(const memdebug *m, size_t size, int freetype,
                            int after_rcu, const char *op) {
        if (m->magic != magic_value
            || m->freetype != freetype
            || (!after_rcu && m->size != size)
            || m->after_rcu != after_rcu)
            hard_free_checks(m, freetype, size, after_rcu, op);
    }
    void landmark(char* buf, size_t size) const;
    static void hard_free_checks(const memdebug* m, size_t size, int freetype,
                                 int after_rcu, const char* op);
    static void hard_assert_use(const void* ptr, memtag tag1, memtag tag2);
#else
    static void *make(void *p, size_t, int) {
        return p;
    }
    static void set_landmark(void*, const char*, int) {
    }
    static void *check_free(void *p, size_t, int) {
        return p;
    }
    static void check_rcu(void *, size_t, int) {
    }
    static void *check_free_after_rcu(void *p, int) {
        return p;
    }
    static bool check_use(void *, memtag) {
        return true;
    }
    static bool check_use(void *, memtag, memtag) {
        return true;
    }
    static void assert_use(void *, memtag) {
    }
    static void assert_use(void *, memtag, memtag) {
    }
#endif
};

enum {
#if HAVE_MEMDEBUG
    memdebug_size = sizeof(memdebug)
#else
    memdebug_size = 0
#endif
};

struct limbo_element {
    void *ptr_;
    int freetype_;
    uint64_t epoch_;
};

struct limbo_group {
    enum { capacity = (4076 - sizeof(limbo_group *)) / sizeof(limbo_element) };
    int head_;
    int tail_;
    limbo_element e_[capacity];
    limbo_group *next_;
    limbo_group()
        : head_(0), tail_(0), next_() {
    }
    void push_back(void *ptr, int freetype, uint64_t epoch) {
        assert(tail_ < capacity);
        e_[tail_].ptr_ = ptr;
        e_[tail_].freetype_ = freetype;
        e_[tail_].epoch_ = epoch;
        ++tail_;
    }
};

template <int N> struct has_threadcounter {
    static bool test(threadcounter ci) {
        return unsigned(ci) < unsigned(N);
    }
};
template <> struct has_threadcounter<0> {
    static bool test(threadcounter) {
        return false;
    }
};

struct rcu_callback {
    virtual ~rcu_callback() {
    }
    virtual void operator()(threadinfo& ti) = 0;
};

class threadinfo {
  public:
    enum {
        TI_MAIN, TI_PROCESS, TI_LOG, TI_CHECKPOINT
    };

    static threadinfo *make(int purpose, int index);
    // XXX destructor
    static pthread_key_t key;

    // thread information
    int purpose() const {
        return purpose_;
    }
    int index() const {
        return index_;
    }
    loginfo* logger() const {
        return logger_;
    }
    void set_logger(loginfo* logger) {
        assert(!logger_ && logger);
        logger_ = logger;
    }
    static threadinfo *allthreads;
    threadinfo* next() const {
        return next_;
    }

    // timestamps
    kvtimestamp_t operation_timestamp() const {
        return timestamp();
    }
    kvtimestamp_t update_timestamp() const {
        return ts_;
    }
    kvtimestamp_t update_timestamp(kvtimestamp_t x) const {
        if (circular_int<kvtimestamp_t>::less_equal(ts_, x))
            // x might be a marker timestamp; ensure result is not
            ts_ = (x | 1) + 1;
        return ts_;
    }
    kvtimestamp_t update_timestamp(kvtimestamp_t x, kvtimestamp_t y) const {
        if (circular_int<kvtimestamp_t>::less(x, y))
            x = y;
        if (circular_int<kvtimestamp_t>::less_equal(ts_, x))
            // x might be a marker timestamp; ensure result is not
            ts_ = (x | 1) + 1;
        return ts_;
    }
    void increment_timestamp() {
        ts_ += 2;
    }
    void advance_timestamp(kvtimestamp_t x) {
        if (circular_int<kvtimestamp_t>::less(ts_, x))
            ts_ = x;
    }

    // event counters
    void mark(threadcounter ci) {
        if (has_threadcounter<int(ncounters)>::test(ci))
            ++counters_[ci];
    }
    void mark(threadcounter ci, int64_t delta) {
        if (has_threadcounter<int(ncounters)>::test(ci))
            counters_[ci] += delta;
    }
    bool has_counter(threadcounter ci) const {
        return has_threadcounter<int(ncounters)>::test(ci);
    }
    uint64_t counter(threadcounter ci) const {
        return has_threadcounter<int(ncounters)>::test(ci) ? counters_[ci] : 0;
    }

    struct accounting_relax_fence_function {
        threadinfo *ti_;
        threadcounter ci_;
        accounting_relax_fence_function(threadinfo *ti, threadcounter ci)
            : ti_(ti), ci_(ci) {
        }
        void operator()() {
            relax_fence();
            ti_->mark(ci_);
        }
    };
    /** @brief Return a function object that calls mark(ci); relax_fence().
     *
     * This function object can be used to count the number of relax_fence()s
     * executed. */
    accounting_relax_fence_function accounting_relax_fence(threadcounter ci) {
        return accounting_relax_fence_function(this, ci);
    }

    struct stable_accounting_relax_fence_function {
        threadinfo *ti_;
        stable_accounting_relax_fence_function(threadinfo *ti)
            : ti_(ti) {
        }
        template <typename V>
        void operator()(V v) {
            relax_fence();
            ti_->mark(threadcounter(tc_stable + (v.isleaf() << 1) + v.splitting()));
        }
    };
    /** @brief Return a function object that calls mark(ci); relax_fence().
     *
     * This function object can be used to count the number of relax_fence()s
     * executed. */
    stable_accounting_relax_fence_function stable_fence() {
        return stable_accounting_relax_fence_function(this);
    }

    accounting_relax_fence_function lock_fence(threadcounter ci) {
        return accounting_relax_fence_function(this, ci);
    }

    // memory allocation
    void* allocate(size_t sz, memtag tag) {
        void *p = malloc(sz + memdebug_size);
        p = memdebug::make(p, sz, tag << 8);
        if (p)
            mark(threadcounter(tc_alloc + (tag > memtag_value)), sz);
        return p;
    }
    void deallocate(void* p, size_t sz, memtag tag) {
        // in C++ allocators, 'p' must be nonnull
        assert(p);
        p = memdebug::check_free(p, sz, tag << 8);
        free(p);
        mark(threadcounter(tc_alloc + (tag > memtag_value)), -sz);
    }
    void deallocate_rcu(void *p, size_t sz, memtag tag) {
        assert(p);
        memdebug::check_rcu(p, sz, tag << 8);
        record_rcu(p, tag << 8);
        mark(threadcounter(tc_alloc + (tag > memtag_value)), -sz);
    }

    void* pool_allocate(size_t sz, memtag tag) {
        int nl = (sz + memdebug_size + CACHE_LINE_SIZE - 1) / CACHE_LINE_SIZE;
        assert(nl <= pool_max_nlines);
        if (unlikely(!pool_[nl - 1]))
            refill_pool(nl);
        void *p = pool_[nl - 1];
        if (p) {
            pool_[nl - 1] = *reinterpret_cast<void **>(p);
            p = memdebug::make(p, sz, (tag << 8) + nl);
            mark(threadcounter(tc_alloc + (tag > memtag_value)),
                 nl * CACHE_LINE_SIZE);
        }
        return p;
    }
    void pool_deallocate(void* p, size_t sz, memtag tag) {
        int nl = (sz + memdebug_size + CACHE_LINE_SIZE - 1) / CACHE_LINE_SIZE;
        assert(p && nl <= pool_max_nlines);
        p = memdebug::check_free(p, sz, (tag << 8) + nl);
        if (use_pool()) {
            *reinterpret_cast<void **>(p) = pool_[nl - 1];
            pool_[nl - 1] = p;
        } else
            free(p);
        mark(threadcounter(tc_alloc + (tag > memtag_value)),
             -nl * CACHE_LINE_SIZE);
    }
    void pool_deallocate_rcu(void* p, size_t sz, memtag tag) {
        int nl = (sz + memdebug_size + CACHE_LINE_SIZE - 1) / CACHE_LINE_SIZE;
        assert(p && nl <= pool_max_nlines);
        memdebug::check_rcu(p, sz, (tag << 8) + nl);
        record_rcu(p, (tag << 8) + nl);
        mark(threadcounter(tc_alloc + (tag > memtag_value)),
             -nl * CACHE_LINE_SIZE);
    }

    // RCU
    void rcu_start() {
        if (gc_epoch_ != globalepoch)
            gc_epoch_ = globalepoch;
    }
    void rcu_stop() {
        if (limbo_epoch_ && (gc_epoch_ - limbo_epoch_) > 1)
            hard_rcu_quiesce();
        gc_epoch_ = 0;
    }
    void rcu_quiesce() {
        rcu_start();
        if (limbo_epoch_ && (gc_epoch_ - limbo_epoch_) > 2)
            hard_rcu_quiesce();
    }
    typedef ::rcu_callback rcu_callback;
    void rcu_register(rcu_callback* cb) {
        record_rcu(cb, -1);
    }

    // thread management
    void run();
    int run(void* (*thread_func)(threadinfo*), void* thread_data = 0);
    pthread_t threadid() const {
        return threadid_;
    }
    void* thread_data() const {
        return thread_data_;
    }

    static threadinfo *current() {
        return (threadinfo *) pthread_getspecific(key);
    }

    void report_rcu(void *ptr) const;
    static void report_rcu_all(void *ptr);

  private:
    union {
        struct {
            uint64_t gc_epoch_;
            uint64_t limbo_epoch_;
            loginfo *logger_;

            threadinfo *next_;
            int purpose_;
            int index_;         // the index of a udp, logging, tcp,
                                // checkpoint or recover thread

            pthread_t threadid_;
        };
        char padding1[CACHE_LINE_SIZE];
    };

  private:
    enum { pool_max_nlines = 20 };
    void *pool_[pool_max_nlines];

    limbo_group *limbo_head_;
    limbo_group *limbo_tail_;
    mutable kvtimestamp_t ts_;

    //enum { ncounters = (int) tc_max };
    enum { ncounters = 0 };
    uint64_t counters_[ncounters];

    void* (*thread_func_)(threadinfo*);
    void* thread_data_;

    void refill_pool(int nl);
    void refill_rcu();

    void free_rcu(void *p, int freetype) {
        if ((freetype & 255) == 0) {
            p = memdebug::check_free_after_rcu(p, freetype);
            ::free(p);
        } else if (freetype == -1)
            (*static_cast<rcu_callback *>(p))(*this);
        else {
            p = memdebug::check_free_after_rcu(p, freetype);
            int nl = freetype & 255;
            *reinterpret_cast<void **>(p) = pool_[nl - 1];
            pool_[nl - 1] = p;
        }
    }

    void record_rcu(void* ptr, int freetype) {
        if (recovering && freetype == (memtag_value << 8)) {
            free_rcu(ptr, freetype);
            return;
        }
        if (limbo_tail_->tail_ == limbo_tail_->capacity)
            refill_rcu();
        uint64_t epoch = globalepoch;
        limbo_tail_->push_back(ptr, freetype, epoch);
        if (!limbo_epoch_)
            limbo_epoch_ = epoch;
    }

#if ENABLE_ASSERTIONS
    static int no_pool_value;
    static bool use_pool() {
        return !no_pool_value;
    }
#else
    static bool use_pool() {
        return true;
    }
#endif

    void hard_rcu_quiesce();
    static void* thread_trampoline(void*);
    friend class loginfo;
};

#endif