BwTransactionTable.hpp

//
// Created by zhou822 on 11/20/22.
//

#ifndef LIVEGRAPH_BWTRANSACTIONTABLE_HPP
#define LIVEGRAPH_BWTRANSACTIONTABLE_HPP
#include <atomic>
#include <cassert>
#include <unordered_map>
#include "tbb/concurrent_hash_map.h"
#include <Libraries//parallel_hashmap/phmap.h>
#include "bwgraph.hpp"
#include "utils.hpp"
namespace bwgraph{
#define ABORT 0x7FFFFFFFFFFFFFFF
#define IN_PROGRESS 0
#define WORKER_THREAD_NUM 64
#define USING_PONTER_ENTRY true
#define TXN_TABLE_TEST true
#define OLD_TBB_MAP false
    /* todo::
     * alternative design: let each thread has a fixed-sized table. It generates transaction entries treating the table as a fixed-sized circular array. Entries are just in the table.
     * Ideally lazy updates and transaction creations will balance out so we always have space for new transactions. However. if indeed the table is full, the owner thread will try to do lazy
     * updates to clear up space for one entry.
     */
#if OLD_TBB_MAP
    struct txn_entry{
        txn_entry() {}
        txn_entry(uint64_t ts):readTS(ts){
            valid.store(false);
            this->status.store(IN_PROGRESS);
            this->op_count.store(0);
        }
        txn_entry(const txn_entry& other){
            valid.store(false);
            status.store(other.status.load());
            readTS = other.readTS;
            op_count.store(other.op_count);
        }
        txn_entry operator = (txn_entry const &other){
            valid.store(other.valid.load());
            status.store(other.status.load());
            readTS = other.readTS;
            op_count.store(other.op_count.load());
        }
        void reduce_op_count(int32_t count){
            op_count.fetch_sub(count);
        }
        std::atomic_bool valid;
        std::atomic_uint64_t status;//predefine abort and in progress ts values. All other values in status refer to the commit ts.
        uint64_t readTS;
        std::atomic_int32_t op_count;
    };
    /*
     * per worker thread transaction table
     * will be preallocated and address arrays are stored locally per thread.
     */
#if USING_PONTER_ENTRY
    class TransactionTable{
    public:
        void putTxn(uint64_t txnID, uint64_t readTS){
            txn_entry* entry = new txn_entry(readTS);
            tbb::concurrent_hash_map<uint64_t, txn_entry*>::accessor ac;
            txnTable.insert(ac,txnID);
            ac->second = entry;
            //txnTable.emplace(txnID,entry);
        }
        /*
         * this function is used for lazy updates
         */
        uint64_t peek_txn_status(uint64_t txnID){
            tbb::concurrent_hash_map<uint64_t, txn_entry*>::const_accessor cac;
            if(!txnTable.find(cac,txnID)){
                assert(false);//we always insert into txn table at txn start, we cannot delete an entry until all its lazy updates are finished
            }else{
                return cac->second->status;
            }
        }
        void lazy_update(uint64_t txnID,int32_t successful_updates){
            tbb::concurrent_hash_map<uint64_t, txn_entry*>::const_accessor cac;
            txnTable.find(cac,txnID);
            int32_t remaining_op=cac->second->op_count.fetch_sub(successful_updates);
            assert(remaining_op>=0);
            if(!remaining_op){
                delete cac->second;
                cac.release();
               txnTable.erase(txnID);
            }
        }
        void commitTxn(uint64_t txnID, uint64_t writeTS){
            tbb::concurrent_hash_map<uint64_t, txn_entry*>::const_accessor cac;
#if TXN_TABLE_TEST
            assert(txnTable.find(cac,txnID));
#else
            txnTable.find(cac,txnID);
#endif
            uint64_t originalStatus = cac->second->status.load();
#if TXN_TABLE_TEST
            assert(originalStatus==IN_PROGRESS);
#endif
            cac->second->status.compare_exchange_strong(originalStatus,writeTS);
        }

        void abortTxn(uint64_t txnID){
            tbb::concurrent_hash_map<uint64_t, txn_entry*>::const_accessor cac;
#if TXN_TABLE_TEST
            assert(txnTable.find(cac,txnID));
#else
            txnTable.find(cac,txnID);
#endif
            uint64_t originalStatus = cac->second->status.load();
#if TXN_TABLE_TEST
            assert(originalStatus==IN_PROGRESS);
#endif
            cac->second->status.compare_exchange_strong(originalStatus,ABORT);
        }
    private:
       tbb::concurrent_hash_map<uint64_t, txn_entry*>txnTable;
    };
#else
    class TransactionTable{
    public:
        void putTxn(uint64_t txnID, uint64_t readTS){
            txn_entry entry(readTS);
            txnTable.emplace(txnID,entry);
        }
        /*
         * this function is used for lazy updates
         */
        uint64_t peek_txn_status(uint64_t txnID){
            if(txnTable.count(txnID)==0){
                assert(false);//we always insert into txn table at txn start, we cannot delete an entry until all its lazy updates are finished
            }else{
                txn_entry& entry_ref = txnTable.at(txnID);//todo: compare with directly making a copy
                if(entry_ref.valid){
                    return entry_ref.status;//return the abort, in progress or committed at a special ts accordingly.
                }
                return IN_PROGRESS;//the txn entry is still being setup.
            }
        }
        void lazy_update(uint64_t txnID,int32_t successful_updates){
            txn_entry& entry_ref = txnTable.at(txnID);
            int32_t remaining_op=entry_ref.op_count.fetch_sub(successful_updates);
            assert(remaining_op>=0);
            if(!remaining_op){
                entry_ref.valid.store(false);
                txnTable.erase(txnID);
            }
        }
    private:
        tbb::concurrent_hash_map<uint64_t, txn_entry> txnTable;
    };
#endif
#else
struct txn_table_entry{
    txn_table_entry(){
        status.store(IN_PROGRESS);
    };
    txn_table_entry(uint64_t new_txn_id):txn_id(new_txn_id){
        status.store(IN_PROGRESS);
    }
    txn_table_entry(const txn_table_entry& other){
        txn_id = other.txn_id;
        status.store( other.status);
        op_count.store( other.op_count);
    }
    txn_table_entry& operator = (const txn_table_entry& other){
        txn_id = other.txn_id;
        status.store( other.status);
        op_count.store( other.op_count);
        return *this;
    }
    bool reduce_op_count(int64_t num){
        if(op_count.fetch_sub(num)==num)
        {
            return true;
        }
#if TXN_TABLE_TEST
        assert(op_count.load()>=0);
        if(op_count.load()==0){
            printf("found 0\n");
        }
#endif
        return false;
    }
    void commit(uint64_t ts){
        status.store(ts);
    }
    void abort(){
        status.store(ABORT);
    }
    uint64_t txn_id;
    std::atomic_uint64_t status;
    std::atomic_int64_t op_count;
    char padding[40];
};
static_assert(sizeof(txn_table_entry)==64);
#if USING_PONTER_ENTRY
using entry_ptr = txn_table_entry*;
using Map = phmap::parallel_flat_hash_map<
    uint64_t,
    entry_ptr ,
    phmap::priv::hash_default_hash<uint64_t>,
    phmap::priv::hash_default_eq<uint64_t>,
    std::allocator<std::pair<const uint64_t, entry_ptr>>,
    12,
    std::mutex>;
class TransactionTable{
public:
    TransactionTable(int32_t thread_id){
        owner_id = thread_id;
    }
    /*
     * return true if the table does contain the txn id, and thus act based on the status
     * return false (very rarely) if the status fetch happens before the lazy update took effect but then the thread is scheduled out:
     * Thread 1 finds in progress delta
     * Thread 2 lazy updates
     * Thread 2 does other work
     * Thread 2 updates op_count and erases the entry
     * Thread 1 fetches status --> no entry exists
     */
    bool get_status(uint64_t txn_id,uint64_t& status_result){
        return local_table.if_contains(txn_id,[&status_result](typename Map::value_type& pair){status_result=pair.second->status.load();});
    }
    void reduce_op_count(uint64_t txn_id,int64_t op_count){
        entry_ptr ptr;
#if TXN_TABLE_TEST
        assert( local_table.if_contains(txn_id,[&ptr](typename Map::value_type& pair){ ptr = pair.second;}));
        assert(ptr->op_count>0);
#else
        local_table.if_contains(txn_id,[&ptr](typename Map::value_type& pair){ptr = pair.second;});
#endif
        if(ptr->reduce_op_count(op_count)){
            local_table.erase(txn_id);
            //printf("something\n");
            delete ptr;//always safe, no other thread should ever need to peek this status anymore
        }
    }
    entry_ptr put_entry(uint64_t txn_id){
        entry_ptr ptr = new txn_table_entry(txn_id);
#if TXN_TABLE_TEST
        assert(!local_table.contains(txn_id));
#endif
        local_table.emplace(txn_id,ptr);
        return ptr;
    }
    //commit txn status is done by the manager updating the txn entry status

    //should contain this txn id
    void test_contain(uint64_t txn_id){
        local_table.contains(txn_id);
        //assert(local_table.contains(txn_id));
    }
    void commit_txn(entry_ptr ptr, uint64_t op_count, uint64_t commit_ts){
#if TXN_TABLE_TEST
        test_contain(ptr->txn_id);
#endif
        ptr->op_count = op_count;
        ptr->status.store(commit_ts);
    }
    void abort_txn(entry_ptr ptr, uint64_t op_count){
#if TXN_TABLE_TEST
        test_contain(ptr->txn_id);
#endif
        if(!op_count){
            local_table.erase(ptr->txn_id);
        }
        ptr->op_count = op_count;
        ptr->status.store(ABORT);
    }

    int32_t owner_id;
    Map local_table;
};
//can be stored within the global graph
class TransactionTables{
public:
    std::vector<TransactionTable>tables;
    TransactionTables()
    {
        for(int i=0;i<THREAD_NUM;i++){
            tables.push_back(TransactionTable(i));
        }
    }
    bool contains(uint64_t txn_id){
        int32_t thread_id = get_threadID(txn_id);
        return tables[thread_id].local_table.contains(txn_id);
    }
    bool get_status(uint64_t txn_id,uint64_t& status_result){
        int32_t thread_id = get_threadID(txn_id);
        return tables[thread_id].get_status(txn_id,status_result);
    }
    void reduce_op_count(uint64_t txn_id,int64_t op_count){
        int32_t thread_id = get_threadID(txn_id);
        tables[thread_id].reduce_op_count(txn_id, op_count);
    }
    entry_ptr put_entry(uint64_t txn_id){
        int32_t thread_id = get_threadID(txn_id);
        return tables[thread_id].put_entry(txn_id);
    }
    void commit_txn(entry_ptr ptr, uint64_t op_count, uint64_t commit_ts){
#if TXN_TABLE_TEST
        assert(op_count>0);
        int32_t thread_id = get_threadID(ptr->txn_id);
        tables[thread_id].commit_txn(ptr,op_count,commit_ts);
#else
        ptr->op_count = op_count;
        ptr->status.store(commit_ts);
#endif
    }
    void abort_txn(entry_ptr ptr, uint64_t op_count){
#if TXN_TABLE_TEST
        int32_t thread_id = get_threadID(ptr->txn_id);
        tables[thread_id].abort_txn(ptr,op_count);
#else
        ptr->op_count = op_count;
        ptr->status.store(ABORT);
#endif
    }
    int32_t all_remaining_transactions(){
        int32_t count=0;
        for(int i=0; i<tables.size();i++){
            count+=tables.at(i).local_table.size();
        }
        return count;
    }
    entry_ptr get_entry(uint64_t txn_id){
        int32_t thread_id = get_threadID(txn_id);
        return tables[thread_id].local_table.find(txn_id)->second;
    }
};
#else

#endif
#endif

}
#endif // LIVEGRAPH_BWTRANSACTIONTABLE_HPP