smthread.h

Go to the documentation of this file.

/*
 * (c) Copyright 2011-2013, Hewlett-Packard Development Company, LP
 */

/* -*- mode:C++; c-basic-offset:4 -*-
     Shore-MT -- Multi-threaded port of the SHORE storage manager

                       Copyright (c) 2007-2009
      Data Intensive Applications and Systems Labaratory (DIAS)
               Ecole Polytechnique Federale de Lausanne

                         All Rights Reserved.

   Permission to use, copy, modify and distribute this software and
   its documentation is hereby granted, provided that both the
   copyright notice and this permission notice appear in all copies of
   the software, derivative works or modified versions, and any
   portions thereof, and that both notices appear in supporting
   documentation.

   This code is distributed in the hope that it will be useful, but
   WITHOUT ANY WARRANTY; without even the implied warranty of
   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. THE AUTHORS
   DISCLAIM ANY LIABILITY OF ANY KIND FOR ANY DAMAGES WHATSOEVER
   RESULTING FROM THE USE OF THIS SOFTWARE.
*/

/*<std-header orig-src='shore' incl-file-exclusion='SMTHREAD_H'>

 $Id: smthread.h,v 1.106 2010/12/08 17:37:43 nhall Exp $

SHORE -- Scalable Heterogeneous Object REpository

Copyright (c) 1994-99 Computer Sciences Department, University of
                      Wisconsin -- Madison
All Rights Reserved.

Permission to use, copy, modify and distribute this software and its
documentation is hereby granted, provided that both the copyright
notice and this permission notice appear in all copies of the
software, derivative works or modified versions, and any portions
thereof, and that both notices appear in supporting documentation.

THE AUTHORS AND THE COMPUTER SCIENCES DEPARTMENT OF THE UNIVERSITY
OF WISCONSIN - MADISON ALLOW FREE USE OF THIS SOFTWARE IN ITS
"AS IS" CONDITION, AND THEY DISCLAIM ANY LIABILITY OF ANY KIND
FOR ANY DAMAGES WHATSOEVER RESULTING FROM THE USE OF THIS SOFTWARE.

This software was developed with support by the Advanced Research
Project Agency, ARPA order number 018 (formerly 8230), monitored by
the U.S. Army Research Laboratory under contract DAAB07-91-C-Q518.
Further funding for this work was provided by DARPA through
Rome Research Laboratory Contract No. F30602-97-2-0247.

*/

#ifndef __SMTHREAD_H
#define __SMTHREAD_H

#include "w_defines.h"

/*  -- do not edit anything above this line --   </std-header>*/

#include <cassert>

#ifndef __W_H
#include "w.h"
#endif // __W_H
#ifndef __SM_BASE_H
#include "sm_base.h"
#endif // __SM_BASE_H

#include "w_bitvector.h"
#include <mutex>
#include <memory>
#include <list>

#include "timeout.h"
#include "latches.h"
#include "logrec.h"
#include "smstats.h"

class xct_t;
class lockid_t;
class smthread_t;

enum {
    FINGER_BITS = SM_DREADLOCK_FINGERS
};

typedef w_bitvector_t<SM_DREADLOCK_BITCOUNT> sm_thread_map_t;
typedef sm_thread_map_t atomic_thread_map_t;
/*
    We do NOT need synchronization for Dreadlock bitmap.
    We freshly re-compute bitmaps at each spin (no false-negatives in long run),
    and we tolerate occasional false positives.
    So, atomic_thread_map_t is no longer needed.

class  atomic_thread_map_t : public sm_thread_map_t {
private:
    // mutable srwlock_t   _map_lock;
public:

    bool has_reader() const {
        return _map_lock.has_reader();
    }
    bool has_writer() const {
        return _map_lock.has_writer();
    }
    void lock_for_read() const {
        _map_lock.acquire_read();
    }
    void lock_for_write() {
        _map_lock.acquire_write();
    }
    void unlock_reader() const{
        w_assert2(_map_lock.has_reader());
        _map_lock.release_read();
    }
    void unlock_writer() {
        w_assert2(_map_lock.has_writer());
        _map_lock.release_write();
    }
    atomic_thread_map_t () {
        w_assert1(_map_lock.has_reader() == false);
        w_assert1(_map_lock.has_writer() == false);
    }
    ~atomic_thread_map_t () {
        w_assert1(_map_lock.has_reader() == false);
        w_assert1(_map_lock.has_writer() == false);
    }

    atomic_thread_map_t &operator=(const atomic_thread_map_t &other) {
        // Copy only the bitmap portion; do not touch the
        // _map_lock
#if W_DEBUG_LEVEL > 0
        bool X=_map_lock.has_reader();
        bool Y=_map_lock.has_writer();
#endif
        copy(other);
#if W_DEBUG_LEVEL > 0
        w_assert1(_map_lock.has_reader() == X);
        w_assert1(_map_lock.has_writer() == Y);
#endif
        return *this;
    }
};
*/

class smthread_t {
    friend class smthread_init_t;

    struct tcb_t {
        // CS: moved out of xct_t
        logrec_t _logbuf;

        logrec_t _logbuf2; // for "piggy-backed" SSX

        xct_t* xct;

        int pin_count;      // number of rsrc_m pins
        int prev_pin_count; // previous # of rsrc_m pins
        int lock_timeout;    // timeout to use for lock acquisitions
        bool _in_sm;      // thread is in sm ss_m:: function
        bool _is_update_thread;// thread is in update function

        int16_t _depth; // how many "outer" this has
        tcb_t* _outer; // this forms a singly linked list

        sm_stats_t _TL_stats; // thread-local stats

        // for lock_head_t::my_lock::get_me
        queue_based_lock_t::ext_qnode _me1;

        queue_based_lock_t::ext_qnode _me2;

        queue_based_lock_t::ext_qnode _me3;

        queue_based_lock_t::ext_qnode _xlist_mutex_node;

        // force this to be 8-byte aligned:

        void create_TL_stats();

        void clear_TL_stats();

        void destroy_TL_stats();

        inline sm_stats_t& TL_stats() {
            return _TL_stats;
        }

        inline const sm_stats_t& TL_stats_const() const {
            return _TL_stats;
        }

        static long get_TL_stat(sm_stat_id stat) {
            return smthread_t::TL_stats()[enum_to_base(stat)];
        }

        tcb_t(tcb_t* outer) :
                xct(0),
                pin_count(0),
                prev_pin_count(0),
                lock_timeout(timeout_t::WAIT_FOREVER), // default for a thread
                _in_sm(false),
                _is_update_thread(false),
                _depth(outer == nullptr ? 1 : outer->_depth + 1),
                _outer(outer) {
            QUEUE_EXT_QNODE_INITIALIZE(_me1);
            QUEUE_EXT_QNODE_INITIALIZE(_me2);
            QUEUE_EXT_QNODE_INITIALIZE(_me3);
            QUEUE_EXT_QNODE_INITIALIZE(_xlist_mutex_node);

            create_TL_stats();
        }

        ~tcb_t() {
            destroy_TL_stats();
        }
    };

public:

    // bool               _try_initialize_fingerprint(); // true: failure false: ok
    // void               _initialize_fingerprint();
    // void               _uninitialize_fingerprint();
    // short              _fingerprint[FINGER_BITS]; // dreadlocks
    // atomic_thread_map_t  _fingerprint_map; // map containing only fingerprint

public:
    // const atomic_thread_map_t&  get_fingerprint_map() const
    //                         {   return _fingerprint_map; }
    static void timeout_to_timespec(int timeout_ms, struct timespec& when);

public:

    /* public for debugging */
    static void init_fingerprint_map();
    static void attach_xct(xct_t* x);

    static void detach_xct(xct_t* x);

    static inline
    int lock_timeout() {
        return tcb().lock_timeout;
    }

    static inline
    void set_lock_timeout(int i) {
        tcb().lock_timeout = i;
    }

    static logrec_t* get_logbuf() {
        return &tcb()._logbuf;
    }

    static logrec_t* get_logbuf2() {
        return &tcb()._logbuf2;
    }

    static inline xct_t* xct() {
        return tcb().xct;
    }

    static inline sm_stats_t& TL_stats() {
        return tcb().TL_stats();
    }

    static long get_TL_stat(sm_stat_id s) {
        return tcb().get_TL_stat(s);
    }

    static void add_from_TL_stats(sm_stats_t& w);

    // NOTE: These macros don't have to be atomic since these thread stats
    // are stored in the smthread and collected when the smthread's tcb is
    // destroyed.

#define GET_TSTAT(x) smthread_t::TL_stats()[enum_to_base(sm_stat_id::x)]

#define INC_TSTAT(x) smthread_t::TL_stats()[enum_to_base(sm_stat_id::x)]++

#define ADD_TSTAT(x, y) smthread_t::TL_stats()[enum_to_base(sm_stat_id::x)] += (y)

#define SET_TSTAT(x, y) smthread_t::TL_stats()[enum_to_base(sm_stat_id::x)] = (y)


    /*
     *  These functions are used to verify than nothing is
     *  left pinned accidentally.  Call mark_pin_count before an
     *  operation and check_pin_count after it with the expected
     *  number of pins that should not have been realeased.
     */
    static void mark_pin_count();

    static void check_pin_count(int change);

    static void check_actual_pin_count(int actual);

    static void incr_pin_count(int amount);

    static int pin_count();

    /*
     *  These functions are used to verify that a thread
     *  is only in one ss_m::, scan::, or pin:: function at a time.
     */
    static inline
    void in_sm(bool in) {
        tcb()._in_sm = in;
    }

    static inline
    bool is_in_sm() {
        return tcb()._in_sm;
    }

    static inline
    bool is_update_thread() {
        return tcb()._is_update_thread;
    }

    static inline
    void set_is_update_thread(bool in) {
        tcb()._is_update_thread = in;
    }

    static void no_xct(xct_t*);

    virtual void _dump(ostream&) const; // to be over-ridden
    /* thread-level block() and unblock aren't public or protected
       accessible.
       These methods are used by the lock manager.
       Otherwise, ordinarly pthreads sychronization variables
       are used.
    */
    // w_error_codes smthread_block(int WAIT_FOREVER,
    //                   const char * const caller = 0,
    //                   const void * id = 0);
    // w_rc_t            smthread_unblock(w_error_codes e);

    // int sampling;
// private:
    // w_error_codes _smthread_block( int WAIT_FOREVER,
    //                           const char * const why =0);
    // w_rc_t           _smthread_unblock(w_error_codes e);

public:
    static queue_based_lock_t::ext_qnode& get_me3() {
        return tcb()._me3;
    }

    static queue_based_lock_t::ext_qnode& get_me2() {
        return tcb()._me2;
    }

    static queue_based_lock_t::ext_qnode& get_me1() {
        return tcb()._me1;
    }

    static queue_based_lock_t::ext_qnode& get_xlist_mutex_node() {
        return tcb()._xlist_mutex_node;
    }

private:
    /* sm-specific block / unblock implementation */
    bool _waiting;

    static tcb_t*& tcb_ptr() {
        // static local var -- initialized only on first call
        static thread_local tcb_t* tcb = new tcb_t(nullptr);
        return tcb;
    }

    static tcb_t& tcb() {
        auto ret = tcb_ptr();
        w_assert3(ret);
        return *ret;
    }

public:
    static inline size_t get_tcb_depth() {
        return tcb()._depth;
    }

private:

    class GlobalThreadList {
    public:
        void add_me() {
            std::unique_lock<std::mutex> lck{_mutex};
            _tcb_list.push_back(tcb_ptr());
        }

        void remove_me() {
            std::unique_lock<std::mutex> lck{_mutex};
            auto iter = _tcb_list.begin();
            while (iter != _tcb_list.end()) {
                if (*iter == tcb_ptr()) {
                    delete *iter;
                    iter = _tcb_list.erase(iter);
                } else {
                    iter++;
                }
            }
        }

        template<typename F>
        void for_each_stats(F& f) {
            std::unique_lock<std::mutex> lck{_mutex};
            for (auto p : _tcb_list) {
                f(p->TL_stats());
            }
        }

    private:
        std::list<tcb_t*> _tcb_list;

        std::mutex _mutex;
    };

    static std::shared_ptr<GlobalThreadList> thread_list() {
        static auto ptr = std::make_shared<GlobalThreadList>();
        return ptr;
    }

public:

    static void add_me_to_thread_list() {
        thread_list()->add_me();
    }

    static void remove_me_from_thread_list() {
        thread_list()->remove_me();
    }

    template<typename F>
    static void for_each_thread_stats(F& f) {
        thread_list()->for_each_stats(f);
    }
};

inline xct_t* xct() {
    return smthread_t::xct();
}

class smthread_init_t {
public:
    smthread_init_t();

    ~smthread_init_t();

private:
    static int count;
};
inline void
smthread_t::mark_pin_count() {
    tcb().prev_pin_count = tcb().pin_count;
}

inline void
smthread_t::check_pin_count(int W_IFDEBUG4(change)) {
#if W_DEBUG_LEVEL > 3
    int diff = tcb().pin_count - tcb().prev_pin_count;
    if (change >= 0) {
        w_assert4(diff <= change);
    } else {
        w_assert4(diff >= change);
    }
#endif
}

inline void
smthread_t::check_actual_pin_count(int actual) {
    w_assert3(tcb().pin_count == actual);
}

inline void
smthread_t::incr_pin_count(int amount) {
    tcb().pin_count += amount;
}

inline int
smthread_t::pin_count() {
    return tcb().pin_count;
}

void
DumpBlockedThreads(ostream& o);

/*
 * redefine DBGTHRD to use our threads
 */
#ifdef DBGTHRD
#undef DBGTHRD
#endif
#define DBGTHRD(arg) DBG(<< " th." << std::this_thread::get_id() << " " arg)

/*<std-footer incl-file-exclusion='SMTHREAD_H'>  -- do not edit anything below this line -- */

#endif // __SMTHREAD_H /*</std-footer>*/