lock_raw.h

Go to the documentation of this file.

/*
 * (c) Copyright 2014, Hewlett-Packard Development Company, LP
 */
#ifndef __LOCK_RAW_H
#define __LOCK_RAW_H

#include <cstdint>
#include <ostream>
#include <pthread.h>
#include "AtomicCounter.hpp"
#include "w_defines.h"
#include "w_okvl.h"
#include "w_rc.h"
#include "w_gc_pool_forest.h"
#include "w_markable_pointer.h"
#include "lsn.h"

inline void atomic_synchronize() {
    lintel::atomic_thread_fence(lintel::memory_order_seq_cst); // corresponds to full mfence
}

#ifdef PURE_SPIN_RAWLOCK

inline void atomic_synchronize_if_mutex() {}

#else // PURE_SPIN_RAWLOCK
inline void atomic_synchronize_if_mutex() { atomic_synchronize(); }
#endif // PURE_SPIN_RAWLOCK

struct RawLockBucket;
struct RawLockQueue;
struct RawLock;
struct RawXct;
class sm_options;

struct RawLock : public GcPoolEntry {
    enum LockState {
        UNUSED = 0,
        OBSOLETE,
        ACTIVE,
        WAITING,
    };

    uint32_t hash;

    LockState state;

    MarkablePointer<RawLock> next;

    RawXct* owner_xct;

    okvl_mode mode;

    RawLock* xct_previous;
    RawLock* xct_next;

    // another doubly linked list for RawXctLockHashMap
    RawLock* xct_hashmap_previous;
    RawLock* xct_hashmap_next;
};

std::ostream& operator<<(std::ostream& o, const RawLock& v);

const MarkablePointer<RawLock> NULL_RAW_LOCK;

struct RawLockQueue {
    struct Iterator {
        Iterator(const RawLockQueue* enclosure, RawLock* start_from);

        bool is_null() const {
            return current.is_null();
        }

        void next(bool& must_retry);

        const RawLockQueue* enclosure;

        RawLock* predecessor;

        MarkablePointer<RawLock> current;
    };

    w_error_codes acquire(RawXct* xct, uint32_t hash, const okvl_mode& mode,
                          int32_t timeout_in_ms, bool check, bool wait, bool acquire,
                          RawLock** out);

    w_error_codes retry_acquire(RawLock** lock, bool wait, bool acquire,
                                int32_t timeout_in_ms);

    w_error_codes complete_acquire(RawLock** lock, bool wait, bool acquire,
                                   int32_t timeout_in_ms);

    void release(RawLock* lock, const lsn_t& commit_lsn);

    void update_xlock_tag(const lsn_t& commit_lsn);

    void atomic_lock_insert(RawLock* new_lock);

    struct Compatibility {
        Compatibility(bool grant, bool deadlock, RawXct* block)
                : can_be_granted(grant),
                  deadlocked(deadlock),
                  blocker(block) {}

        bool can_be_granted;

        bool deadlocked;

        RawXct* blocker;
    };

    Compatibility check_compatiblity(RawLock* lock) const;

    bool peek_compatiblity(RawXct* xct, uint32_t hash, const okvl_mode& mode) const;

    w_error_codes wait_for(RawLock* new_lock, int32_t timeout_in_ms);

    RawLock* find_predecessor(RawLock* lock) const;

    RawLock* tail() const;

    bool delink(RawLock* predecessor, RawLock* target, RawLock* successor) const;

    // Helper function called by RawLockQueue::acquire
    // Based on the information in Compatibility, if the blocker txn is a loser txn and it is not
    // in the middle of rolling back, trigger the on_demand UNDO for the loser transaction
    // Return true if an on_demand UNDO operation was triggered and completed
    bool trigger_UNDO(Compatibility& compatibility);  // In: the current compatibility status of the requested lock

    mutable RawLock head;

    lsn_t x_lock_tag;

    // For on_demand and mixed UNDO counting purpose
    static int loser_count;
};

std::ostream& operator<<(std::ostream& o, const RawLockQueue& v);

const int RAW_XCT_LOCK_HASHMAP_SIZE = 1023;

class RawXctLockHashMap {
public:
    RawXctLockHashMap();

    ~RawXctLockHashMap();

    okvl_mode get_granted_mode(uint32_t lock_id) const;

    void reset();

    void push_front(RawLock* link);

    void remove(RawLock* link);

private:

    RawLock* _buckets[RAW_XCT_LOCK_HASHMAP_SIZE];

    static uint32_t _bucket_id(uint32_t lock_id) {
        return lock_id % RAW_XCT_LOCK_HASHMAP_SIZE;
    }
};

struct RawXct : GcPoolEntry {
    enum XctState {
        UNUSED = 0,
        ACTIVE,
        WAITING,
    };

    void init(gc_thread_id thread_id,
              GcPoolForest<RawLock>* lock_pool, gc_pointer_raw* lock_pool_next);

    void uninit();

    bool is_deadlocked(RawXct* first_blocker);

    void update_read_watermark(const lsn_t& tag) {
        if (read_watermark < tag) {
            read_watermark = tag;
        }
    }

    RawLock* allocate_lock(uint32_t hash,
                           const okvl_mode& mode, RawLock::LockState state);

    void deallocate_lock(RawLock* lock);

    void dump_lockinfo(std::ostream& out) const;

    bool has_locks() const {
        return private_first != nullptr;
    }

    gc_thread_id thread_id;

    GcPoolForest<RawLock>* lock_pool;

    gc_pointer_raw* lock_pool_next;

    XctState state;

    bool deadlock_detected_by_others;

    RawXct* blocker;

#ifndef PURE_SPIN_RAWLOCK

    pthread_cond_t              lock_wait_cond;
    pthread_mutex_t             lock_wait_mutex;
#endif // PURE_SPIN_RAWLOCK

    lsn_t read_watermark;

    RawXctLockHashMap private_hash_map;

    RawLock* private_first;

    RawLock* private_last;
};

std::ostream& operator<<(std::ostream& o, const RawXct& v);

class RawLockBackgroundThread {
public:
    RawLockBackgroundThread(const sm_options& options,
                            GcPoolForest<RawLock>* lock_pool, GcPoolForest<RawXct>* xct_pool);

    ~RawLockBackgroundThread();

    void start();

    void stop_synchronous();

    void wakeup();

    void run_main();

    static void* pthread_main(void* t);

protected:
    pthread_t _thread;

    pthread_attr_t _join_attr;

    pthread_mutex_t _interval_mutex;

    pthread_cond_t _interval_cond;

    bool _stop_requested;

    bool _running;

    int _dummy_lsn_lock;

    int _dummy_lsn_xct;

    uint32_t _generation_count;

    uint32_t _init_generation_count;

    uint32_t _free_segment_count;

    uint32_t _max_segment_count;

    uint32_t _internal_milliseconds;

    uint32_t _lockpool_initseg;

    uint32_t _xctpool_initseg;

    size_t _lockpool_segsize;

    size_t _xctpool_segsize;

    GcPoolForest<RawLock>* _lock_pool;

    GcPoolForest<RawXct>* _xct_pool;
};

#endif // __LOCK_RAW_H