<filename>TCache/sys_op.py
""" TCache Workload System Operation """
from multiprocessing import Queue
import time
from utils.load_dataset import load_item_univ, load_txn_univ, load_txn_seq, load_ycsb_seq
from TCache.cache_alg import SingleAlg
from TCache.reorder_alg import ReorderAlg
import numpy as np
import json
import pdb
import os
import psycopg2
import happybase
from pymemcache.client import base
from pymemcache.client.base import PooledClient
import csv

class WorkloadManager(object):
    """Manage workload for TCache
    
    """
    def __init__(self, dataset_dir: str, sys_test=True) -> None:
        super().__init__()
        # load entire workload data from dataset_dir
        item_size_path = dataset_dir + '/item_size.pkl'
        cls_item_path = dataset_dir + '/cls_item.pkl'
        txn_item_path = dataset_dir + '/txn_item.pkl'
        id_seq_path = dataset_dir + '/id_seq.npy'
        flag_seq_path = dataset_dir + '/flag_seq.npy'
        self.item_size_dict, self.cls_item_dict = load_item_univ(item_size_path, cls_item_path)
        self.txn_item_dict = load_txn_univ(txn_item_path)
        self.txn_id_seq, self.write_flag_seq = load_txn_seq(id_seq_path, flag_seq_path)
        if sys_test:
            ycsb_seq_path = dataset_dir + '/transactions.dat'
            self.ycsb_id_2_key, self.ycsb_id_2_read, self.ycsb_id_2_write = load_ycsb_seq(ycsb_seq_path)
        self.get_workload_stats(print_stats=True)


    def get_workload_stats(self, print_stats=True) -> None:
        """ Get statistics for workload data. """
        query_num = len(self.item_size_dict)
        cls_num = len(self.cls_item_dict)
        seq_len = len(self.txn_id_seq)
        read_qry_cnt, wrt_qry_cnt = 0, 0
        wrt_txn_cnt = self.write_flag_seq.sum()
        read_txn_cnt = len(self.write_flag_seq) - wrt_txn_cnt
        # item_read_time_dict & item_write_time_dict just for workload statistics
        item_read_time_dict, item_write_time_dict = {i:[] for i in range(query_num)}, {i:[] for i in range(query_num)}
        for time_step in range(len(self.txn_id_seq)):
            txn_vec = self.txn_item_dict[self.txn_id_seq[time_step]]
            if self.write_flag_seq[time_step]:
                wrt_qry_cnt += np.sum(txn_vec)
                for item_id in np.where(txn_vec == 1)[0]:
                    item_write_time_dict[item_id].append(time_step)
            else:
                read_qry_cnt += np.sum(txn_vec)
                for item_id in np.where(txn_vec == 1)[0]:
                    item_read_time_dict[item_id].append(time_step)
        dist_read_qry_cnt, dist_wrt_qry_cnt = 0, 0
        for i in range(query_num):
            if len(item_read_time_dict[i]) > 0:
                dist_read_qry_cnt += 1
            if len(item_write_time_dict[i]) > 0:
                dist_wrt_qry_cnt += 1
        total_item_size = sum(self.item_size_dict.values())
        self.workload_stats = {'query_num': query_num, 'cls_num': cls_num, 
            'total_size': total_item_size, 'seq_len': seq_len,
            'read_txn_cnt': read_txn_cnt, 'write_txn_cnt': wrt_txn_cnt, 
            'read_qry_cnt': read_qry_cnt, 'write_qry_cnt': wrt_qry_cnt, 
            'unique_read_qry_cnt': dist_read_qry_cnt, 'unique_write_qry_cnt': dist_wrt_qry_cnt}

        workload_stats_columns = ['query_num', 'cls_num', 'total_size', 'seq_len',
               'read_txn_cnt', 'write_txn_cnt', 'read_qry_cnt', 'write_qry_cnt', 'unique_read_qry_cnt', 'unique_write_qry_cnt']
        dict_data = [
        {'query_num': query_num, 'cls_num': cls_num, 
            'total_size': total_item_size, 'seq_len': seq_len,
            'read_txn_cnt': read_txn_cnt, 'write_txn_cnt': wrt_txn_cnt, 
            'read_qry_cnt': read_qry_cnt, 'write_qry_cnt': wrt_qry_cnt, 
            'unique_read_qry_cnt': dist_read_qry_cnt, 'unique_write_qry_cnt': dist_wrt_qry_cnt}]
        csv_file = "data/res/workload_stats.csv"
        try:
            with open(csv_file, 'a', newline='') as p: # a means append
                writer = csv.DictWriter(p, fieldnames=workload_stats_columns)
                if os.stat(csv_file).st_size == 0: # if csv_file is empty, then add the header
                    writer.writeheader()
                for data in dict_data:
                    writer.writerow(data)
        except IOError:
            print("I/O error")   
            
        if print_stats:
            print(self.workload_stats)
            # json.dumps(self.workload_stats, indent=4)


class WorkloadTest(WorkloadManager):
    """ Create input for algorithm, operate Cache and DB accordingly. """
    def __init__(self, dataset_dir: str, sys_test=True) -> None:
        super().__init__(dataset_dir, sys_test)
        self.sys_test = sys_test
        self.orireor_txn_id_seq = np.zeros(self.workload_stats['seq_len'], dtype=int) # reorder seq based our strategy
        self.ran_txn_id_seq = np.zeros(self.workload_stats['seq_len'], dtype=int)
        self.readfir_txn_id_seq = np.zeros(self.workload_stats['seq_len'], dtype=int) # 
        self.wrtfir_txn_id_seq = np.zeros(self.workload_stats['seq_len'], dtype=int)

        # generate reordered txn_id_seq, write_flag_seq and txn_item_dict
        self.reor_txn_id_seq = np.zeros(self.workload_stats['seq_len'], dtype=int)
        self.reor_write_flag_seq = np.zeros(self.workload_stats['seq_len'], dtype=int)
        self.reor_txn_item_dict = {}

    def generate_reor_res_seq(self, batch_start, batch_size, orireor_txn_id_seq, write_flag_seq, txn_item_dict):
        """ generate reordered txn_id_seq, write_flag_seq and txn_item_dict based on orireor_txn_id_seq"""
        for ordered_txn_num in range(batch_size):
            self.reor_txn_id_seq[batch_start+ordered_txn_num] = batch_start+ordered_txn_num
            self.reor_write_flag_seq[batch_start+ordered_txn_num] = write_flag_seq[orireor_txn_id_seq[batch_start+ordered_txn_num]]
            self.reor_txn_item_dict[batch_start+ordered_txn_num] = txn_item_dict[orireor_txn_id_seq[batch_start+ordered_txn_num]]

    def batch_test(self, queue: Queue, csize: float, cache_scheme: str, alg_name: str, batch_size: int, reorder_flag:bool, liveness_bound:int, reorder_name:str, opt_len=500) -> dict:
        """ Test transaction sequence in batches. """
        assert self.workload_stats['seq_len'] % batch_size == 0   # transaction sequence consists of full batches by default
        cache_size =  int(csize * self.workload_stats['total_size'])
        assert 'Comb' not in alg_name   # TODO: support combiner style caching algorithm
        alg_obj = SingleAlg(alg_name)
        # using opt_len as findOB truncate length for optimization
        alg_obj.workload_init(cache_size, cache_scheme, item_num=self.workload_stats['query_num'], findOB_trunc=opt_len)
        batch_num = int(self.workload_stats['seq_len'] / batch_size)
        item_num = self.workload_stats['query_num']
        seq_start_time = time.time()
        off_liveness, ran_liveness, readfir_liveness, wrtfir_liveness = 0, 0, 0, 0
        off_max_liveness, ran_max_liveness, readfir_max_liveness, wrtfir_max_liveness = 0, 0, 0, 0
        for i in range(batch_num):
            batch_start, batch_end = i * batch_size, (i + 1) * batch_size
            # TODO: define a new reorder object, generate the reordered txn_id_seq[batch_start:batch_end] & write_flag_seq[batch_start:batch_end]
            if reorder_flag:
                print("reordering start")
                reorder_obj = ReorderAlg(cache_size)
                reorder_obj.init_read_write_time(item_num, txn_id_seq=self.txn_id_seq[batch_start:batch_end], 
                        write_flag_seq=self.write_flag_seq[batch_start:batch_end], 
                        txn_item_dict=self.txn_item_dict, batch_start=batch_start)
                if reorder_name in ['Greedy']:
                    # print("Greedy reorder algorithm")
                    reorder_obj.get_writetxn_bond_readtxn(batch_start=batch_start, batch_end=batch_end,
                                txn_id_seq=self.txn_id_seq[batch_start:batch_end], 
                                write_flag_seq=self.write_flag_seq[batch_start:batch_end], 
                                txn_item_dict=self.txn_item_dict, liveness_bound=liveness_bound) 
                    reorder_obj.pick_readtxn2wrttxn(batch_start, batch_size)
                    self.orireor_txn_id_seq = reorder_obj.reorder_read_main(item_num, batch_start, batch_size, self.txn_id_seq[batch_start:batch_end], 
                                                self.write_flag_seq[batch_start:batch_end], self.item_size_dict, self.txn_item_dict, self.orireor_txn_id_seq)
                    self.generate_reor_res_seq(batch_start, batch_size, self.orireor_txn_id_seq, self.write_flag_seq, self.txn_item_dict)
                    # compute the liveness based on the reordered seq
                    off_liveness_res = reorder_obj.get_livenessbound(item_num, batch_start=batch_start, 
                                txn_id_seq=self.txn_id_seq[batch_start:batch_end],
                                write_flag_seq=self.write_flag_seq[batch_start:batch_end],
                                txn_item_dict=self.txn_item_dict, reor_txn_id_seq=self.orireor_txn_id_seq[batch_start:batch_end],
                                liveness_bound=liveness_bound)
                    off_liveness += off_liveness_res[0]
                    off_max_liveness += off_liveness_res[1]
                if reorder_name in ['Random']:  
                    self.ran_txn_id_seq = reorder_obj.get_random_seq(batch_start, batch_end, self.ran_txn_id_seq)
                    ran_liveness_res = reorder_obj.get_livenessbound(item_num, batch_start=batch_start, 
                            txn_id_seq=self.txn_id_seq[batch_start:batch_end],
                            write_flag_seq=self.write_flag_seq[batch_start:batch_end],
                            txn_item_dict=self.txn_item_dict, reor_txn_id_seq=self.ran_txn_id_seq[batch_start:batch_end],
                            liveness_bound=liveness_bound)
                    ran_liveness += ran_liveness_res[0] 
                    ran_max_liveness += ran_liveness_res[1]      
                    self.generate_reor_res_seq(batch_start, batch_size, self.ran_txn_id_seq, self.write_flag_seq, self.txn_item_dict)

                if reorder_name in ['Readfir']:
                    self.readfir_txn_id_seq = reorder_obj.get_allreadfirst_seq(batch_start, batch_size, self.readfir_txn_id_seq)
                    readfir_liveness_res = reorder_obj.get_livenessbound(item_num, batch_start=batch_start, 
                            txn_id_seq=self.txn_id_seq[batch_start:batch_end],
                            write_flag_seq=self.write_flag_seq[batch_start:batch_end],
                            txn_item_dict=self.txn_item_dict, reor_txn_id_seq=self.readfir_txn_id_seq[batch_start:batch_end],
                            liveness_bound=liveness_bound)
                    readfir_liveness += readfir_liveness_res[0] 
                    readfir_max_liveness += readfir_liveness_res[1]
                    self.generate_reor_res_seq(batch_start, batch_size, self.readfir_txn_id_seq, self.write_flag_seq, self.txn_item_dict)

                if reorder_name in ['Wrtfir']:
                    self.wrtfir_txn_id_seq = reorder_obj.get_allwrtfirst_seq(batch_start, batch_size, self.wrtfir_txn_id_seq)
                    wrtfir_liveness_res = reorder_obj.get_livenessbound(item_num, batch_start=batch_start, 
                            txn_id_seq=self.txn_id_seq[batch_start:batch_end],
                            write_flag_seq=self.write_flag_seq[batch_start:batch_end],
                            txn_item_dict=self.txn_item_dict, reor_txn_id_seq=self.wrtfir_txn_id_seq[batch_start:batch_end],
                            liveness_bound=liveness_bound)
                    wrtfir_liveness += wrtfir_liveness_res[0]
                    wrtfir_max_liveness += wrtfir_liveness_res[1]
                    self.generate_reor_res_seq(batch_start, batch_size, self.wrtfir_txn_id_seq, self.write_flag_seq, self.txn_item_dict)
            else:
                self.reor_txn_id_seq = self.txn_id_seq
                self.reor_write_flag_seq = self.write_flag_seq
                self.reor_txn_item_dict = self.txn_item_dict
            
            # get item read and write time for current batch
            alg_obj.init_read_write_time(txn_id_seq=self.reor_txn_id_seq[batch_start:batch_end], 
                write_flag_seq=self.reor_write_flag_seq[batch_start:batch_end], 
                txn_item_dict=self.reor_txn_item_dict, batch_start=batch_start)
            for time_step in range(batch_start, batch_end):
                op_ret_dict = alg_obj.batch_step_process(time_step, batch_start, batch_end, self.reor_txn_id_seq[batch_start:batch_end], 
                    self.reor_write_flag_seq[batch_start:batch_end], self.item_size_dict, self.reor_txn_item_dict, self.cls_item_dict)
                if self.sys_test:
                    # TODO: carry out cache and DB operation here
                    tmp_op_dict = {}
                    if 'write_items' in op_ret_dict:
                        query_list = [self.ycsb_id_2_write[item_id] for item_id in np.where(op_ret_dict['write_items'] == 1)[0]]
                        tmp_op_dict['db_write'] = query_list
                        tmp_op_dict['db_write_key'] = [self.ycsb_id_2_key[item_id] for item_id in np.where(op_ret_dict['write_items'] == 1)[0]]
                        # self.db_write_txn(query_list)
                    if 'acc_update' in op_ret_dict:
                        item_to_load = [item_id for item_id in np.where(op_ret_dict['acc_update'] == 1)[0]]
                        query_list = [self.ycsb_id_2_read[item_id] for item_id in item_to_load]
                        tmp_op_dict['acc_update'] = query_list
                        tmp_op_dict['acc_update_key'] = [self.ycsb_id_2_key[item_id] for item_id in item_to_load]
                    if 'evict_from_cache' in op_ret_dict:
                        tmp_op_dict['cache_evict'] = [self.ycsb_id_2_key[item_id] for item_id in np.where(op_ret_dict['evict_from_cache'] == 1)[0]]
                    if 'read_on_miss' in op_ret_dict:
                        item_to_load = [item_id for item_id in np.where(op_ret_dict['read_on_miss'] == 1)[0]]
                        query_list = [self.ycsb_id_2_read[item_id] for item_id in item_to_load]
                        tmp_op_dict['read_on_miss'] = query_list
                        tmp_op_dict['read_on_miss_key'] = [self.ycsb_id_2_key[item_id] for item_id in item_to_load]
                        # self.load_to_cache(op_ret_dict['read_on_miss'])
                    if 'read_on_abort' in op_ret_dict:
                        item_to_load = [item_id for item_id in np.where(op_ret_dict['read_on_abort'] == 1)[0]]
                        query_list = [self.ycsb_id_2_read[item_id] for item_id in item_to_load]
                        tmp_op_dict['read_on_abort'] = query_list
                        tmp_op_dict['read_on_abort_key'] = [self.ycsb_id_2_key[item_id] for item_id in item_to_load]
                        # self.load_to_cache(op_ret_dict['read_on_abort'])
                    if 'read_from_cache' in op_ret_dict:
                        tmp_op_dict['cache_read'] = [self.ycsb_id_2_key[item_id] for item_id in np.where(op_ret_dict['read_from_cache'] == 1)[0]]
                        # txn_answ = self.cache_client.get_many([self.ycsb_id_2_key[item_id] for item_id in np.where(op_ret_dict['read_from_cache'] == 1)[0]])
                    queue.put(tmp_op_dict)
        queue.put(None)
        seq_end_time = time.time()
        alg_obj.print_stats()


        # print('off_liveness: {}, ran_liveness: {}, readfir_liveness: {}, wrtfir_liveness: {}'.format(
        #        off_liveness/batch_num, ran_liveness/batch_num, readfir_liveness/batch_num, wrtfir_liveness/batch_num))
        # only when compare the performance of the reorder strategies, this csv file is useful.
        res_columns = ['off_avg_liveness', 'ran_avg_liveness', 'readfir_avg_liveness', 'wrtfir_avg_liveness',
        'off_max_liveness', 'ran_max_liveness', 'readfir_max_liveness', 'wrtfir_max_liveness']
        dict_data = [
        {'off_avg_liveness': off_liveness/batch_num, 'ran_avg_liveness': ran_liveness/batch_num, 'readfir_avg_liveness': readfir_liveness/batch_num, 'wrtfir_avg_liveness': wrtfir_liveness/batch_num,
        'off_max_liveness': off_max_liveness/batch_num, 'ran_max_liveness': ran_max_liveness/batch_num, 'readfir_max_liveness': readfir_max_liveness/batch_num, 'wrtfir_max_liveness': wrtfir_max_liveness/batch_num}]
        csv_file = "data/res/liveness.csv"
        try:
            with open(csv_file, 'a', newline='') as p: # a means append
                writer = csv.DictWriter(p, fieldnames=res_columns)
                if os.stat(csv_file).st_size == 0: # if csv_file is empty, then add the header
                    writer.writeheader()
                for data in dict_data:
                    writer.writerow(data)
        except IOError:
            print("I/O error")  

        res_columns = ['alg_name', 'cache_scheme', 'cost', 'whole_cost',
               'ob_cnt', 'evict_cnt','cch_cnt']
        dict_data = [
        {'alg_name': alg_obj.alg_name, 'cache_scheme': alg_obj.cache_scheme, 'cost': alg_obj.cost, 'whole_cost': alg_obj.whole_cost,
               'ob_cnt': alg_obj.ob_cnt, 'evict_cnt': alg_obj.evict_cnt, 'cch_cnt': alg_obj.cch_cnt}]
        csv_file = "data/res/res.csv"
        try:
            with open(csv_file, 'a', newline='') as p: # a means append
                writer = csv.DictWriter(p, fieldnames=res_columns)
                if os.stat(csv_file).st_size == 0: # if csv_file is empty, then add the header
                    writer.writeheader()
                for data in dict_data:
                    writer.writerow(data)
        except IOError:
            print("I/O error")       

        print('ALG Total Time: {}'.format(seq_end_time - seq_start_time))
        # TODO: calculate throughput, save algorithm simulation result


class SystemOperator:
    """Operate DB and cache based on operations in queue."""
    def __init__(self, db_conn_param: dict, cache_conn_param: dict) -> None:
        self.db_sys = db_conn_param["db"]
        del db_conn_param['db']
        if self.db_sys == "postgresql":
            self.db_conn = psycopg2.connect(**db_conn_param)
            self.db_conn.set_session(autocommit=False)
            self.db_cursor = self.db_conn.cursor()
        else:
            assert self.db_sys == "hbase"
            # self.db_conn = happybase.Connection(host=db_conn_param["host"], port=db_conn_param["port"])
            self.db_conn = happybase.Connection(host=db_conn_param["host"], port=9090)
            self.table = self.db_conn.table('usertable')
        self.cache_sys = cache_conn_param["cache"]
        del cache_conn_param['cache']
        if self.cache_sys == "memcached":
            self.cache_client = base.Client((cache_conn_param['host'], int(cache_conn_param['port'])))
            # self.cache_client = PooledClient(('127.0.0.1', int(cache_conn_param['port'])), max_pool_size = 400)
            self.cache_client.flush_all()
    

    def operation_exe(self, op_queue: Queue):
        """Get operations from queue, execute them."""
        counter = 0
        db_read_time, db_write_time, cache_read_time, cache_write_time = 0, 0, 0, 0
        cache_counter = 0
        cache_list = []
        while True:
            op_dict = op_queue.get() # wait unless there is value in queue. Default: blocking
            if op_dict is None:
                print('counter: {}'.format(counter))
                print("cache_counter:", cache_counter)
                sys_end_time = time.time()
                if sys_start_time is not None:
                    print('System Execution Time: {}'.format(sys_end_time - sys_start_time))
                print('db_read_time: {}, db_write_time: {}'.format(db_read_time, db_write_time))
                print('cache_read_time: {}, cache_write_time: {}'.format(cache_read_time, cache_write_time))
                break
            else:
                if counter == 0:
                    sys_start_time = time.time()
                if 'db_write' in op_dict:
                    db_write_time += self.db_write_txn(op_dict['db_write'], op_dict['db_write_key'])
                if 'acc_update' in op_dict:
                    db_read_time += self.load_to_cache(op_dict['acc_update'], op_dict['acc_update_key'])
                if 'cache_evict' in op_dict:
                    tmp_start_time = time.time()
                    self.cache_client.delete_many(op_dict['cache_evict'])
                    cache_write_time += time.time() - tmp_start_time
                if 'read_on_miss' in op_dict:
                    db_read_time += self.load_to_cache(op_dict['read_on_miss'], op_dict['read_on_miss_key'])
                if 'read_on_abort' in op_dict:
                    db_read_time += self.load_to_cache(op_dict['read_on_abort'], op_dict['read_on_abort_key'])
                if 'cache_read' in op_dict:
                    cache_counter += 1
                    cache_list.append(op_dict['cache_read'][0])
                    # sumbit in a batch way
                    if (counter%100==0):
                        # print("*"*100)
                        tmp_start_time = time.time()
                        self.cache_client.get_many(cache_list) 
                        cache_read_time += time.time() - tmp_start_time
                    #tmp_start_time = time.time()
                    #self.cache_client.get_many(op_dict['cache_read']) # the type of op_dict['cache_read'] is list
                    #print('each get memcache time: {}, counter: {}'.format(time.time() - tmp_start_time, counter))
                    #if (time.time() - tmp_start_time>0.02):
                    #    cache_counter+=1
                    #cache_read_time += time.time() - tmp_start_time
                counter += 1
                
        

    def load_to_cache(self, query_list: list, key_str_list: list):
        """Execute read transaction at database and load results to cache."""
        tmp_load_start_time = time.time()
        if self.db_sys == "postgresql":
            try:
                txn_res_list = []
                for query_str in query_list:
                    self.db_cursor.execute(query_str)
                    txn_res_list.extend(self.db_cursor.fetchall())
                self.db_conn.commit()
                for query_result in txn_res_list:
                    self.cache_client.set(query_result[0], ''.join(query_result[1:]))
            except psycopg2.DatabaseError as e:
                print('PostgreSQL read transaction execution error')
                self.db_conn.rollback()
        else:   # HBase read transaction
            try:
                txn_res_list = []
                for (key_str, qual_bytes_list) in zip(key_str_list, query_list):
                    row = self.table.row(key_str.encode('utf-8'), columns=qual_bytes_list)
                    txn_res_list.append((key_str, b''.join(row.values()).decode('utf-8')))
                    self.cache_client.set_many(dict(txn_res_list))
            except ValueError:
                print('HBase value error.')
                pass
        tmp_load_end_time = time.time()
        return tmp_load_end_time - tmp_load_start_time
    

    def db_write_txn(self, query_list: list, key_str_list: list):
        """Execute write transaction at database."""
        tmp_write_start_time = time.time()
        if self.db_sys == "postgresql": # postgresql write transaction
            try:
                for query_str in query_list:
                    self.db_cursor.execute(query_str)
                self.db_conn.commit()
            except psycopg2.DatabaseError as e:
                print('PostgreSQL write transaction execution error')
                self.db_conn.rollback()
        else:   # HBase write transaction
            try:
                with self.table.batch(transaction=True) as bat:
                    for (key_str, qual_val_dict) in zip(key_str_list, query_list):
                        bat.put(key_str.encode('utf-8'), qual_val_dict)
            # TODO: check error handling here
            except ValueError:
                print('HBase value error.')
                pass
        tmp_write_end_time = time.time()
        return tmp_write_end_time - tmp_write_start_time


    def cleanup(self):
        """Close connection after testing."""
        self.db_conn.close()
        self.cache_client.flush_all()
        self.cache_client.close()

<filename>TCache/sys_op.py
""" TCache Workload System Operation """
from multiprocessing import Queue
import time
from utils.load_dataset import load_item_univ, load_txn_univ, load_txn_seq, load_ycsb_seq
from TCache.cache_alg import SingleAlg
from TCache.reorder_alg import ReorderAlg
import numpy as np
import json
import pdb
import os
import psycopg2
import happybase
from pymemcache.client import base
from pymemcache.client.base import PooledClient
import csv

class WorkloadManager(object):
    """Manage workload for TCache
    
    """
    def __init__(self, dataset_dir: str, sys_test=True) -> None:
        super().__init__()
        # load entire workload data from dataset_dir
        item_size_path = dataset_dir + '/item_size.pkl'
        cls_item_path = dataset_dir + '/cls_item.pkl'
        txn_item_path = dataset_dir + '/txn_item.pkl'
        id_seq_path = dataset_dir + '/id_seq.npy'
        flag_seq_path = dataset_dir + '/flag_seq.npy'
        self.item_size_dict, self.cls_item_dict = load_item_univ(item_size_path, cls_item_path)
        self.txn_item_dict = load_txn_univ(txn_item_path)
        self.txn_id_seq, self.write_flag_seq = load_txn_seq(id_seq_path, flag_seq_path)
        if sys_test:
            ycsb_seq_path = dataset_dir + '/transactions.dat'
            self.ycsb_id_2_key, self.ycsb_id_2_read, self.ycsb_id_2_write = load_ycsb_seq(ycsb_seq_path)
        self.get_workload_stats(print_stats=True)


    def get_workload_stats(self, print_stats=True) -> None:
        """ Get statistics for workload data. """
        query_num = len(self.item_size_dict)
        cls_num = len(self.cls_item_dict)
        seq_len = len(self.txn_id_seq)
        read_qry_cnt, wrt_qry_cnt = 0, 0
        wrt_txn_cnt = self.write_flag_seq.sum()
        read_txn_cnt = len(self.write_flag_seq) - wrt_txn_cnt
        # item_read_time_dict & item_write_time_dict just for workload statistics
        item_read_time_dict, item_write_time_dict = {i:[] for i in range(query_num)}, {i:[] for i in range(query_num)}
        for time_step in range(len(self.txn_id_seq)):
            txn_vec = self.txn_item_dict[self.txn_id_seq[time_step]]
            if self.write_flag_seq[time_step]:
                wrt_qry_cnt += np.sum(txn_vec)
                for item_id in np.where(txn_vec == 1)[0]:
                    item_write_time_dict[item_id].append(time_step)
            else:
                read_qry_cnt += np.sum(txn_vec)
                for item_id in np.where(txn_vec == 1)[0]:
                    item_read_time_dict[item_id].append(time_step)
        dist_read_qry_cnt, dist_wrt_qry_cnt = 0, 0
        for i in range(query_num):
            if len(item_read_time_dict[i]) > 0:
                dist_read_qry_cnt += 1
            if len(item_write_time_dict[i]) > 0:
                dist_wrt_qry_cnt += 1
        total_item_size = sum(self.item_size_dict.values())
        self.workload_stats = {'query_num': query_num, 'cls_num': cls_num, 
            'total_size': total_item_size, 'seq_len': seq_len,
            'read_txn_cnt': read_txn_cnt, 'write_txn_cnt': wrt_txn_cnt, 
            'read_qry_cnt': read_qry_cnt, 'write_qry_cnt': wrt_qry_cnt, 
            'unique_read_qry_cnt': dist_read_qry_cnt, 'unique_write_qry_cnt': dist_wrt_qry_cnt}

        workload_stats_columns = ['query_num', 'cls_num', 'total_size', 'seq_len',
               'read_txn_cnt', 'write_txn_cnt', 'read_qry_cnt', 'write_qry_cnt', 'unique_read_qry_cnt', 'unique_write_qry_cnt']
        dict_data = [
        {'query_num': query_num, 'cls_num': cls_num, 
            'total_size': total_item_size, 'seq_len': seq_len,
            'read_txn_cnt': read_txn_cnt, 'write_txn_cnt': wrt_txn_cnt, 
            'read_qry_cnt': read_qry_cnt, 'write_qry_cnt': wrt_qry_cnt, 
            'unique_read_qry_cnt': dist_read_qry_cnt, 'unique_write_qry_cnt': dist_wrt_qry_cnt}]
        csv_file = "data/res/workload_stats.csv"
        try:
            with open(csv_file, 'a', newline='') as p: # a means append
                writer = csv.DictWriter(p, fieldnames=workload_stats_columns)
                if os.stat(csv_file).st_size == 0: # if csv_file is empty, then add the header
                    writer.writeheader()
                for data in dict_data:
                    writer.writerow(data)
        except IOError:
            print("I/O error")   
            
        if print_stats:
            print(self.workload_stats)
            # json.dumps(self.workload_stats, indent=4)


class WorkloadTest(WorkloadManager):
    """ Create input for algorithm, operate Cache and DB accordingly. """
    def __init__(self, dataset_dir: str, sys_test=True) -> None:
        super().__init__(dataset_dir, sys_test)
        self.sys_test = sys_test
        self.orireor_txn_id_seq = np.zeros(self.workload_stats['seq_len'], dtype=int) # reorder seq based our strategy
        self.ran_txn_id_seq = np.zeros(self.workload_stats['seq_len'], dtype=int)
        self.readfir_txn_id_seq = np.zeros(self.workload_stats['seq_len'], dtype=int) # 
        self.wrtfir_txn_id_seq = np.zeros(self.workload_stats['seq_len'], dtype=int)

        # generate reordered txn_id_seq, write_flag_seq and txn_item_dict
        self.reor_txn_id_seq = np.zeros(self.workload_stats['seq_len'], dtype=int)
        self.reor_write_flag_seq = np.zeros(self.workload_stats['seq_len'], dtype=int)
        self.reor_txn_item_dict = {}

    def generate_reor_res_seq(self, batch_start, batch_size, orireor_txn_id_seq, write_flag_seq, txn_item_dict):
        """ generate reordered txn_id_seq, write_flag_seq and txn_item_dict based on orireor_txn_id_seq"""
        for ordered_txn_num in range(batch_size):
            self.reor_txn_id_seq[batch_start+ordered_txn_num] = batch_start+ordered_txn_num
            self.reor_write_flag_seq[batch_start+ordered_txn_num] = write_flag_seq[orireor_txn_id_seq[batch_start+ordered_txn_num]]
            self.reor_txn_item_dict[batch_start+ordered_txn_num] = txn_item_dict[orireor_txn_id_seq[batch_start+ordered_txn_num]]

    def batch_test(self, queue: Queue, csize: float, cache_scheme: str, alg_name: str, batch_size: int, reorder_flag:bool, liveness_bound:int, reorder_name:str, opt_len=500) -> dict:
        """ Test transaction sequence in batches. """
        assert self.workload_stats['seq_len'] % batch_size == 0   # transaction sequence consists of full batches by default
        cache_size =  int(csize * self.workload_stats['total_size'])
        assert 'Comb' not in alg_name   # TODO: support combiner style caching algorithm
        alg_obj = SingleAlg(alg_name)
        # using opt_len as findOB truncate length for optimization
        alg_obj.workload_init(cache_size, cache_scheme, item_num=self.workload_stats['query_num'], findOB_trunc=opt_len)
        batch_num = int(self.workload_stats['seq_len'] / batch_size)
        item_num = self.workload_stats['query_num']
        seq_start_time = time.time()
        off_liveness, ran_liveness, readfir_liveness, wrtfir_liveness = 0, 0, 0, 0
        off_max_liveness, ran_max_liveness, readfir_max_liveness, wrtfir_max_liveness = 0, 0, 0, 0
        for i in range(batch_num):
            batch_start, batch_end = i * batch_size, (i + 1) * batch_size
            # TODO: define a new reorder object, generate the reordered txn_id_seq[batch_start:batch_end] & write_flag_seq[batch_start:batch_end]
            if reorder_flag:
                print("reordering start")
                reorder_obj = ReorderAlg(cache_size)
                reorder_obj.init_read_write_time(item_num, txn_id_seq=self.txn_id_seq[batch_start:batch_end], 
                        write_flag_seq=self.write_flag_seq[batch_start:batch_end], 
                        txn_item_dict=self.txn_item_dict, batch_start=batch_start)
                if reorder_name in ['Greedy']:
                    # print("Greedy reorder algorithm")
                    reorder_obj.get_writetxn_bond_readtxn(batch_start=batch_start, batch_end=batch_end,
                                txn_id_seq=self.txn_id_seq[batch_start:batch_end], 
                                write_flag_seq=self.write_flag_seq[batch_start:batch_end], 
                                txn_item_dict=self.txn_item_dict, liveness_bound=liveness_bound) 
                    reorder_obj.pick_readtxn2wrttxn(batch_start, batch_size)
                    self.orireor_txn_id_seq = reorder_obj.reorder_read_main(item_num, batch_start, batch_size, self.txn_id_seq[batch_start:batch_end], 
                                                self.write_flag_seq[batch_start:batch_end], self.item_size_dict, self.txn_item_dict, self.orireor_txn_id_seq)
                    self.generate_reor_res_seq(batch_start, batch_size, self.orireor_txn_id_seq, self.write_flag_seq, self.txn_item_dict)
                    # compute the liveness based on the reordered seq
                    off_liveness_res = reorder_obj.get_livenessbound(item_num, batch_start=batch_start, 
                                txn_id_seq=self.txn_id_seq[batch_start:batch_end],
                                write_flag_seq=self.write_flag_seq[batch_start:batch_end],
                                txn_item_dict=self.txn_item_dict, reor_txn_id_seq=self.orireor_txn_id_seq[batch_start:batch_end],
                                liveness_bound=liveness_bound)
                    off_liveness += off_liveness_res[0]
                    off_max_liveness += off_liveness_res[1]
                if reorder_name in ['Random']:  
                    self.ran_txn_id_seq = reorder_obj.get_random_seq(batch_start, batch_end, self.ran_txn_id_seq)
                    ran_liveness_res = reorder_obj.get_livenessbound(item_num, batch_start=batch_start, 
                            txn_id_seq=self.txn_id_seq[batch_start:batch_end],
                            write_flag_seq=self.write_flag_seq[batch_start:batch_end],
                            txn_item_dict=self.txn_item_dict, reor_txn_id_seq=self.ran_txn_id_seq[batch_start:batch_end],
                            liveness_bound=liveness_bound)
                    ran_liveness += ran_liveness_res[0] 
                    ran_max_liveness += ran_liveness_res[1]      
                    self.generate_reor_res_seq(batch_start, batch_size, self.ran_txn_id_seq, self.write_flag_seq, self.txn_item_dict)

                if reorder_name in ['Readfir']:
                    self.readfir_txn_id_seq = reorder_obj.get_allreadfirst_seq(batch_start, batch_size, self.readfir_txn_id_seq)
                    readfir_liveness_res = reorder_obj.get_livenessbound(item_num, batch_start=batch_start, 
                            txn_id_seq=self.txn_id_seq[batch_start:batch_end],
                            write_flag_seq=self.write_flag_seq[batch_start:batch_end],
                            txn_item_dict=self.txn_item_dict, reor_txn_id_seq=self.readfir_txn_id_seq[batch_start:batch_end],
                            liveness_bound=liveness_bound)
                    readfir_liveness += readfir_liveness_res[0] 
                    readfir_max_liveness += readfir_liveness_res[1]
                    self.generate_reor_res_seq(batch_start, batch_size, self.readfir_txn_id_seq, self.write_flag_seq, self.txn_item_dict)

                if reorder_name in ['Wrtfir']:
                    self.wrtfir_txn_id_seq = reorder_obj.get_allwrtfirst_seq(batch_start, batch_size, self.wrtfir_txn_id_seq)
                    wrtfir_liveness_res = reorder_obj.get_livenessbound(item_num, batch_start=batch_start, 
                            txn_id_seq=self.txn_id_seq[batch_start:batch_end],
                            write_flag_seq=self.write_flag_seq[batch_start:batch_end],
                            txn_item_dict=self.txn_item_dict, reor_txn_id_seq=self.wrtfir_txn_id_seq[batch_start:batch_end],
                            liveness_bound=liveness_bound)
                    wrtfir_liveness += wrtfir_liveness_res[0]
                    wrtfir_max_liveness += wrtfir_liveness_res[1]
                    self.generate_reor_res_seq(batch_start, batch_size, self.wrtfir_txn_id_seq, self.write_flag_seq, self.txn_item_dict)
            else:
                self.reor_txn_id_seq = self.txn_id_seq
                self.reor_write_flag_seq = self.write_flag_seq
                self.reor_txn_item_dict = self.txn_item_dict
            
            # get item read and write time for current batch
            alg_obj.init_read_write_time(txn_id_seq=self.reor_txn_id_seq[batch_start:batch_end], 
                write_flag_seq=self.reor_write_flag_seq[batch_start:batch_end], 
                txn_item_dict=self.reor_txn_item_dict, batch_start=batch_start)
            for time_step in range(batch_start, batch_end):
                op_ret_dict = alg_obj.batch_step_process(time_step, batch_start, batch_end, self.reor_txn_id_seq[batch_start:batch_end], 
                    self.reor_write_flag_seq[batch_start:batch_end], self.item_size_dict, self.reor_txn_item_dict, self.cls_item_dict)
                if self.sys_test:
                    # TODO: carry out cache and DB operation here
                    tmp_op_dict = {}
                    if 'write_items' in op_ret_dict:
                        query_list = [self.ycsb_id_2_write[item_id] for item_id in np.where(op_ret_dict['write_items'] == 1)[0]]
                        tmp_op_dict['db_write'] = query_list
                        tmp_op_dict['db_write_key'] = [self.ycsb_id_2_key[item_id] for item_id in np.where(op_ret_dict['write_items'] == 1)[0]]
                        # self.db_write_txn(query_list)
                    if 'acc_update' in op_ret_dict:
                        item_to_load = [item_id for item_id in np.where(op_ret_dict['acc_update'] == 1)[0]]
                        query_list = [self.ycsb_id_2_read[item_id] for item_id in item_to_load]
                        tmp_op_dict['acc_update'] = query_list
                        tmp_op_dict['acc_update_key'] = [self.ycsb_id_2_key[item_id] for item_id in item_to_load]
                    if 'evict_from_cache' in op_ret_dict:
                        tmp_op_dict['cache_evict'] = [self.ycsb_id_2_key[item_id] for item_id in np.where(op_ret_dict['evict_from_cache'] == 1)[0]]
                    if 'read_on_miss' in op_ret_dict:
                        item_to_load = [item_id for item_id in np.where(op_ret_dict['read_on_miss'] == 1)[0]]
                        query_list = [self.ycsb_id_2_read[item_id] for item_id in item_to_load]
                        tmp_op_dict['read_on_miss'] = query_list
                        tmp_op_dict['read_on_miss_key'] = [self.ycsb_id_2_key[item_id] for item_id in item_to_load]
                        # self.load_to_cache(op_ret_dict['read_on_miss'])
                    if 'read_on_abort' in op_ret_dict:
                        item_to_load = [item_id for item_id in np.where(op_ret_dict['read_on_abort'] == 1)[0]]
                        query_list = [self.ycsb_id_2_read[item_id] for item_id in item_to_load]
                        tmp_op_dict['read_on_abort'] = query_list
                        tmp_op_dict['read_on_abort_key'] = [self.ycsb_id_2_key[item_id] for item_id in item_to_load]
                        # self.load_to_cache(op_ret_dict['read_on_abort'])
                    if 'read_from_cache' in op_ret_dict:
                        tmp_op_dict['cache_read'] = [self.ycsb_id_2_key[item_id] for item_id in np.where(op_ret_dict['read_from_cache'] == 1)[0]]
                        # txn_answ = self.cache_client.get_many([self.ycsb_id_2_key[item_id] for item_id in np.where(op_ret_dict['read_from_cache'] == 1)[0]])
                    queue.put(tmp_op_dict)
        queue.put(None)
        seq_end_time = time.time()
        alg_obj.print_stats()


        # print('off_liveness: {}, ran_liveness: {}, readfir_liveness: {}, wrtfir_liveness: {}'.format(
        #        off_liveness/batch_num, ran_liveness/batch_num, readfir_liveness/batch_num, wrtfir_liveness/batch_num))
        # only when compare the performance of the reorder strategies, this csv file is useful.
        res_columns = ['off_avg_liveness', 'ran_avg_liveness', 'readfir_avg_liveness', 'wrtfir_avg_liveness',
        'off_max_liveness', 'ran_max_liveness', 'readfir_max_liveness', 'wrtfir_max_liveness']
        dict_data = [
        {'off_avg_liveness': off_liveness/batch_num, 'ran_avg_liveness': ran_liveness/batch_num, 'readfir_avg_liveness': readfir_liveness/batch_num, 'wrtfir_avg_liveness': wrtfir_liveness/batch_num,
        'off_max_liveness': off_max_liveness/batch_num, 'ran_max_liveness': ran_max_liveness/batch_num, 'readfir_max_liveness': readfir_max_liveness/batch_num, 'wrtfir_max_liveness': wrtfir_max_liveness/batch_num}]
        csv_file = "data/res/liveness.csv"
        try:
            with open(csv_file, 'a', newline='') as p: # a means append
                writer = csv.DictWriter(p, fieldnames=res_columns)
                if os.stat(csv_file).st_size == 0: # if csv_file is empty, then add the header
                    writer.writeheader()
                for data in dict_data:
                    writer.writerow(data)
        except IOError:
            print("I/O error")  

        res_columns = ['alg_name', 'cache_scheme', 'cost', 'whole_cost',
               'ob_cnt', 'evict_cnt','cch_cnt']
        dict_data = [
        {'alg_name': alg_obj.alg_name, 'cache_scheme': alg_obj.cache_scheme, 'cost': alg_obj.cost, 'whole_cost': alg_obj.whole_cost,
               'ob_cnt': alg_obj.ob_cnt, 'evict_cnt': alg_obj.evict_cnt, 'cch_cnt': alg_obj.cch_cnt}]
        csv_file = "data/res/res.csv"
        try:
            with open(csv_file, 'a', newline='') as p: # a means append
                writer = csv.DictWriter(p, fieldnames=res_columns)
                if os.stat(csv_file).st_size == 0: # if csv_file is empty, then add the header
                    writer.writeheader()
                for data in dict_data:
                    writer.writerow(data)
        except IOError:
            print("I/O error")       

        print('ALG Total Time: {}'.format(seq_end_time - seq_start_time))
        # TODO: calculate throughput, save algorithm simulation result


class SystemOperator:
    """Operate DB and cache based on operations in queue."""
    def __init__(self, db_conn_param: dict, cache_conn_param: dict) -> None:
        self.db_sys = db_conn_param["db"]
        del db_conn_param['db']
        if self.db_sys == "postgresql":
            self.db_conn = psycopg2.connect(**db_conn_param)
            self.db_conn.set_session(autocommit=False)
            self.db_cursor = self.db_conn.cursor()
        else:
            assert self.db_sys == "hbase"
            # self.db_conn = happybase.Connection(host=db_conn_param["host"], port=db_conn_param["port"])
            self.db_conn = happybase.Connection(host=db_conn_param["host"], port=9090)
            self.table = self.db_conn.table('usertable')
        self.cache_sys = cache_conn_param["cache"]
        del cache_conn_param['cache']
        if self.cache_sys == "memcached":
            self.cache_client = base.Client((cache_conn_param['host'], int(cache_conn_param['port'])))
            # self.cache_client = PooledClient(('127.0.0.1', int(cache_conn_param['port'])), max_pool_size = 400)
            self.cache_client.flush_all()
    

    def operation_exe(self, op_queue: Queue):
        """Get operations from queue, execute them."""
        counter = 0
        db_read_time, db_write_time, cache_read_time, cache_write_time = 0, 0, 0, 0
        cache_counter = 0
        cache_list = []
        while True:
            op_dict = op_queue.get() # wait unless there is value in queue. Default: blocking
            if op_dict is None:
                print('counter: {}'.format(counter))
                print("cache_counter:", cache_counter)
                sys_end_time = time.time()
                if sys_start_time is not None:
                    print('System Execution Time: {}'.format(sys_end_time - sys_start_time))
                print('db_read_time: {}, db_write_time: {}'.format(db_read_time, db_write_time))
                print('cache_read_time: {}, cache_write_time: {}'.format(cache_read_time, cache_write_time))
                break
            else:
                if counter == 0:
                    sys_start_time = time.time()
                if 'db_write' in op_dict:
                    db_write_time += self.db_write_txn(op_dict['db_write'], op_dict['db_write_key'])
                if 'acc_update' in op_dict:
                    db_read_time += self.load_to_cache(op_dict['acc_update'], op_dict['acc_update_key'])
                if 'cache_evict' in op_dict:
                    tmp_start_time = time.time()
                    self.cache_client.delete_many(op_dict['cache_evict'])
                    cache_write_time += time.time() - tmp_start_time
                if 'read_on_miss' in op_dict:
                    db_read_time += self.load_to_cache(op_dict['read_on_miss'], op_dict['read_on_miss_key'])
                if 'read_on_abort' in op_dict:
                    db_read_time += self.load_to_cache(op_dict['read_on_abort'], op_dict['read_on_abort_key'])
                if 'cache_read' in op_dict:
                    cache_counter += 1
                    cache_list.append(op_dict['cache_read'][0])
                    # sumbit in a batch way
                    if (counter%100==0):
                        # print("*"*100)
                        tmp_start_time = time.time()
                        self.cache_client.get_many(cache_list) 
                        cache_read_time += time.time() - tmp_start_time
                    #tmp_start_time = time.time()
                    #self.cache_client.get_many(op_dict['cache_read']) # the type of op_dict['cache_read'] is list
                    #print('each get memcache time: {}, counter: {}'.format(time.time() - tmp_start_time, counter))
                    #if (time.time() - tmp_start_time>0.02):
                    #    cache_counter+=1
                    #cache_read_time += time.time() - tmp_start_time
                counter += 1
                
        

    def load_to_cache(self, query_list: list, key_str_list: list):
        """Execute read transaction at database and load results to cache."""
        tmp_load_start_time = time.time()
        if self.db_sys == "postgresql":
            try:
                txn_res_list = []
                for query_str in query_list:
                    self.db_cursor.execute(query_str)
                    txn_res_list.extend(self.db_cursor.fetchall())
                self.db_conn.commit()
                for query_result in txn_res_list:
                    self.cache_client.set(query_result[0], ''.join(query_result[1:]))
            except psycopg2.DatabaseError as e:
                print('PostgreSQL read transaction execution error')
                self.db_conn.rollback()
        else:   # HBase read transaction
            try:
                txn_res_list = []
                for (key_str, qual_bytes_list) in zip(key_str_list, query_list):
                    row = self.table.row(key_str.encode('utf-8'), columns=qual_bytes_list)
                    txn_res_list.append((key_str, b''.join(row.values()).decode('utf-8')))
                    self.cache_client.set_many(dict(txn_res_list))
            except ValueError:
                print('HBase value error.')
                pass
        tmp_load_end_time = time.time()
        return tmp_load_end_time - tmp_load_start_time
    

    def db_write_txn(self, query_list: list, key_str_list: list):
        """Execute write transaction at database."""
        tmp_write_start_time = time.time()
        if self.db_sys == "postgresql": # postgresql write transaction
            try:
                for query_str in query_list:
                    self.db_cursor.execute(query_str)
                self.db_conn.commit()
            except psycopg2.DatabaseError as e:
                print('PostgreSQL write transaction execution error')
                self.db_conn.rollback()
        else:   # HBase write transaction
            try:
                with self.table.batch(transaction=True) as bat:
                    for (key_str, qual_val_dict) in zip(key_str_list, query_list):
                        bat.put(key_str.encode('utf-8'), qual_val_dict)
            # TODO: check error handling here
            except ValueError:
                print('HBase value error.')
                pass
        tmp_write_end_time = time.time()
        return tmp_write_end_time - tmp_write_start_time


    def cleanup(self):
        """Close connection after testing."""
        self.db_conn.close()
        self.cache_client.flush_all()
        self.cache_client.close()

TCache Workload System Operation Manage workload for TCache # load entire workload data from dataset_dir Get statistics for workload data. # item_read_time_dict & item_write_time_dict just for workload statistics # a means append # if csv_file is empty, then add the header # json.dumps(self.workload_stats, indent=4) Create input for algorithm, operate Cache and DB accordingly. # reorder seq based our strategy # # generate reordered txn_id_seq, write_flag_seq and txn_item_dict generate reordered txn_id_seq, write_flag_seq and txn_item_dict based on orireor_txn_id_seq Test transaction sequence in batches. # transaction sequence consists of full batches by default # TODO: support combiner style caching algorithm # using opt_len as findOB truncate length for optimization # TODO: define a new reorder object, generate the reordered txn_id_seq[batch_start:batch_end] & write_flag_seq[batch_start:batch_end] # print("Greedy reorder algorithm") # compute the liveness based on the reordered seq # get item read and write time for current batch # TODO: carry out cache and DB operation here # self.db_write_txn(query_list) # self.load_to_cache(op_ret_dict['read_on_miss']) # self.load_to_cache(op_ret_dict['read_on_abort']) # txn_answ = self.cache_client.get_many([self.ycsb_id_2_key[item_id] for item_id in np.where(op_ret_dict['read_from_cache'] == 1)[0]]) # print('off_liveness: {}, ran_liveness: {}, readfir_liveness: {}, wrtfir_liveness: {}'.format( #        off_liveness/batch_num, ran_liveness/batch_num, readfir_liveness/batch_num, wrtfir_liveness/batch_num)) # only when compare the performance of the reorder strategies, this csv file is useful. # a means append # if csv_file is empty, then add the header # a means append # if csv_file is empty, then add the header # TODO: calculate throughput, save algorithm simulation result Operate DB and cache based on operations in queue. # self.db_conn = happybase.Connection(host=db_conn_param["host"], port=db_conn_param["port"]) # self.cache_client = PooledClient(('127.0.0.1', int(cache_conn_param['port'])), max_pool_size = 400) Get operations from queue, execute them. # wait unless there is value in queue. Default: blocking # sumbit in a batch way # print("*"*100) #tmp_start_time = time.time() #self.cache_client.get_many(op_dict['cache_read']) # the type of op_dict['cache_read'] is list #print('each get memcache time: {}, counter: {}'.format(time.time() - tmp_start_time, counter)) #if (time.time() - tmp_start_time>0.02): #    cache_counter+=1 #cache_read_time += time.time() - tmp_start_time Execute read transaction at database and load results to cache. # HBase read transaction Execute write transaction at database. # postgresql write transaction # HBase write transaction # TODO: check error handling here Close connection after testing.