stack_analysis_parallel.py

import pandas as pd
import numpy as np
# from sklearn import preprocessing
import matplotlib.pyplot as plt 
import json
import os
import glob
import multiprocessing as mp
from concurrent.futures import ProcessPoolExecutor, as_completed
from functools import partial
from config import *
from utils import *

interrupt_lock_primitives = ["_raw_spin_lock", "_raw_spin_lock_bh", "_raw_spin_lock_irq", "_raw_spin_lock_irqsave", "_raw_spin_trylock", "_raw_spin_trylock_bh",
                        "_raw_spin_lock_nested", "_raw_spin_lock_irqsave_nested", 
                        "_raw_read_lock", "_raw_write_lock", "_raw_read_lock_bh", "_raw_write_lock_bh",
                        "_raw_read_lock_irq", "_raw_write_lock_irq", "_raw_read_lock_irqsave", "_raw_write_lock_irqsave"]

# non_spin_lock_primitives = ["mutex", "rt_mutex", "read_rw", "write_rw", "read_rw_bh", "write_rw_bh", "read_rw_irq", "write_rw_irq", "read_rw_irqsave", "write_rw_irqsave", 
#                             "read_sem", "write_sem"]

non_spin_lock_primitives = ["mutex", "read_rw", "write_rw", "read_rw_bh", "write_rw_bh", "read_rw_irq", "write_rw_irq", "read_rw_irqsave", "write_rw_irqsave", 
                            "read_sem", "write_sem"]


lock_primitives  = [ {"name": "spin", "primitive": "_raw_spin_lock", "type": "raw_spinlock_t"},
    {"name": "irq_spin", "primitive": "_raw_spin_lock_irq", "type": "raw_spinlock_t"},
    {"name": "bh_spin", "primitive": "_raw_spin_lock_bh", "type": "raw_spinlock_t"},
    {"name": "irqsave_spin", "primitive": "_raw_spin_lock_irqsave", "type": "raw_spinlock_t"},
    {"name": "try_spin", "primitive": "_raw_spin_trylock", "type": "raw_spinlock_t"},
    {"name": "try_bh_spin", "primitive": "_raw_spin_trylock_bh", "type": "raw_spinlock_t"},
    {"name": "nested_spin", "primitive": "_raw_spin_lock_nested", "type": "raw_spinlock_t"},
    {"name": "nested_irqsave_spin", "primitive": "_raw_spin_lock_irqsave_nested", "type": "raw_spinlock_t"},
    {"name": "mutex", "primitive": "mutex", "type": "mutex"},
    # {"name": "rt_mutex", "primitive": "rt_mutex", "type": "rt_mutex"},
    {"name": "read_rw", "primitive": "_raw_read_lock", "type": "rwlock_t"},
    {"name": "write_rw", "primitive": "_raw_write_lock", "type": "rwlock_t"},
    {"name": "read_rw_bh", "primitive": "_raw_read_lock_bh", "type": "rwlock_t"},
    {"name": "write_rw_bh", "primitive": "_raw_write_lock_bh", "type": "rwlock_t"},
    {"name": "read_rw_irq", "primitive": "_raw_read_lock_irq", "type": "rwlock_t"},
    {"name": "write_rw_irq", "primitive": "_raw_write_lock_irq", "type": "rwlock_t"},
    {"name": "read_rw_irqsave", "primitive": "_raw_read_lock_irqsave", "type": "rwlock_t"},
    {"name": "write_rw_irqsave", "primitive": "_raw_write_lock_irqsave", "type": "rwlock_t"},
    {"name": "read_sem", "primitive": "read_sem", "type": "rw_semaphore"},
    {"name": "write_sem", "primitive": "write_sem", "type": "rw_semaphore"}
]

# test_spin_lock_primitives  = [{"name": "irq_spin", "primitive": "_raw_spin_lock_irq"}]

# selected_microbench_options = "sysbench", "mem_micro", "file_metadata"

total_runs = 3
start_range = 1

#runtimes = ["host", "runc", "runsc-kvm", "fc"]
runtimes = ["fc"]


global_stack_file = "cloud_workloads_stack_details.csv" # micro_stack_details.csv, funcBench_stack_details, cloud_workloads_stack_details, fork, funcbench_stressng_stack_details
global_no_match_file = "cloud_workloads_no_match.csv" # micro_no_match.csv, funcBench_no_match.csv, cloud_workloads_no_match, fork_no_match, funcbench_stressng_no_match
stress_ng_stressors_enabled = True  # True if stress-ng stressors are available, False if not

trace_base_dir = trace_dir + "ubuntu22/"

global_processed_stacks_path = trace_dir + "ubuntu22/global_processed_traces"

# Process ID for unique temporary files
current_process_id = None

#functionBench = ["image_processing", "video_processing", "model_training", "face_detection", "cnn", "rnn"]
functionBench = ["face_detection", "cnn", "rnn", "image_processing", "video_processing", "model_training"]

# cloud_workloads = ["graph_analytics", "data_caching", "in_memory_analytics"] #, "media_streaming", "data_analytics", "data_serving", "web_search", "web_serving"]
#cloud_workloads = ["graph_analytics", "data_caching", "media_streaming", "data_serving", "data_analytics"] #, "in_memory_analytics"
cloud_workloads = ["in_memory_analytics"]
stress_ng_stressors = ["fork", "fstat", "getdent", "hdd", "mem"]
#stress_ng_stressors = ["hdd"]

# cloud_workloads = ["data_caching_no_warmup"] 

mem_micro_benchmarks = ["1MB_total_16GB", "8KB_total_16GB"]
file_metadata_micro_benchmarks = ["listdirs", "createfiles", "deletefiles"]

interrupt_handlers = ["__softirqentry_text_start", "hrtimer_interrupt", "scheduler_tick", "htab_map_update_elem", "rcu_core", "__common_interrupt", "common_interrupt", "asm_common_interrupt",
                       "ghes_notify_nmi", "nmi_handle", "bpf_prog_6deef7357e7b4530", "asm_sysvec_apic_timer_interrupt", "sysvec_apic_timer_interrupt", "__do_softirq", "run_timer_softirq", "__run_timers"]
locks_to_remove = ["ghes_notify_lock_irq", "&htab->lockdep_key", ""]


# This version subracts from the address based on what is added to the function address
# For "_raw_spin_lock+0x5 [kernel]",
# the address is 0xffffffff81cd0c35+0x5, so we need to subtract 0x5 from the address if the match is not found
# This is because the address is the start of the function + the offset

def process_single_trace(args):
    """Wrapper function for parallel processing of a single trace directory"""
    trace_dir, runtime, selected_microbench = args
    print(f"Processing trace: {trace_dir} (Runtime: {runtime}, Benchmark: {selected_microbench})")
    
    if not os.path.exists(trace_dir):
        print(f"Path does not exist: {trace_dir}")
        return
    
    try:
        # Set unique process ID for temporary files
        global current_process_id
        current_process_id = mp.current_process().pid
        
        locks_stack_processing(trace_dir, runtime, selected_microbench)
        interrupt_locks(trace_dir, runtime, selected_microbench)
        print(f"Completed processing: {trace_dir}")
    except Exception as e:
        print(f"Error processing {trace_dir}: {str(e)}")

def locks_stack_processing(trace_dir, runtime, selected_microbench):

    # stack_details_file = global_processed_stacks_path + "/stack_details.csv"
    # global_stack_no_match_file = global_processed_stacks_path + "/no_match.csv"

    stack_details_file = global_processed_stacks_path + "/" + global_stack_file
    global_stack_no_match_file = global_processed_stacks_path + "/" + global_no_match_file


    # Ensure global processed stacks directory exists
    if not os.path.exists(global_processed_stacks_path):
        os.makedirs(global_processed_stacks_path)
    
    # Create process-specific temporary files for global data
    global current_process_id
    if current_process_id is None:
        current_process_id = mp.current_process().pid
    
    temp_stack_details_file = stack_details_file.replace('.csv', f'_temp_{current_process_id}.csv')
    temp_global_no_match_file = global_stack_no_match_file.replace('.csv', f'_temp_{current_process_id}.csv')
    
    # Initialize temporary files if they don't exist
    if not os.path.exists(temp_stack_details_file):
        stack_details_df = pd.DataFrame(columns=["lock_primitive", "lock_name", "start_func", "start_func_addr", "start_file", "start_line", "end_func", "end_file", "end_line", "direct_start_func_match", "struct_name", "struct_path"])
        stack_details_df.to_csv(temp_stack_details_file, index=False)
    
    if not os.path.exists(temp_global_no_match_file):
        global_no_match_df = pd.DataFrame(columns=["stack_id", "lock_name", "lock_primitive", "runtime", "benchmark", "trace_dir"])
        global_no_match_df.to_csv(temp_global_no_match_file, index=False)
  
   
    

    # iterate over all lock primitives
    for primitive in lock_primitives:

        stack_details_df = pd.read_csv(temp_stack_details_file)

        primitive_name = primitive["name"]
        
        # read the stack trace file
        # print("Processing primitive: ", primitive["name"])
        stack_trace_file = trace_dir + "/" + primitive["name"] + "_filtered_stacks.json"
        print("Stack trace file: ", stack_trace_file)

        lock_type = primitive["type"]

        # load filtered locks file
        filtered_locks_file = trace_dir + "/" + lock_type +  "_filtered.csv"
        filtered_locks_df = pd.read_csv(filtered_locks_file)

        # if filtered locks file is empty, skip the primitive
        if filtered_locks_df.empty:
            continue

        # check if columns start_func, start_func_addr, start_file, start_line, end_func,end_file,end_line, struct_name,struct_path  exist in filtered_locks_df, if not then add the columns with empty values
        if "start_func" not in filtered_locks_df.columns:
            filtered_locks_df["start_func"] = ""
    
        if "start_func_addr" not in filtered_locks_df.columns:
            filtered_locks_df["start_func_addr"] = ""
        
        if "start_file" not in filtered_locks_df.columns:
            filtered_locks_df["start_file"] = ""
        
        if "start_line" not in filtered_locks_df.columns:
            filtered_locks_df["start_line"] = ""
        
        if "end_func" not in filtered_locks_df.columns:
            filtered_locks_df["end_func"] = ""
        
        if "end_file" not in filtered_locks_df.columns:
            filtered_locks_df["end_file"] = ""
        
        if "end_line" not in filtered_locks_df.columns:
            filtered_locks_df["end_line"] = ""
        
        if "struct_name" not in filtered_locks_df.columns:
            filtered_locks_df["struct_name"] = ""
        
        if "struct_path" not in filtered_locks_df.columns:
            filtered_locks_df["struct_path"] = ""


        # create another file to store ignored stacks for different primitives
        no_match_lock_primitive_file = trace_dir + "/" + primitive["name"] + "_stack_no_match.csv"

        # delete the file if it exists
        if os.path.exists(no_match_lock_primitive_file):
            os.remove(no_match_lock_primitive_file)
        
        # add column headers if the file does not exist
        if not os.path.exists(no_match_lock_primitive_file):
            no_match_lock_primitive_df = pd.DataFrame(columns=["stack_id", "lock_name", "lock_primitive", "runtime", "benchmark"])
            no_match_lock_primitive_df.to_csv(no_match_lock_primitive_file, index=False)

        
        # stack_trace_file = "/mydata/locks-access-traces/runc/mem_micro/cat_enabled/8KB_total_16GB/traces_run_1/spin_filtered_stacks.json"
         
        # print("Stack trace file: ", stack_trace_file)
        lock_primitive = primitive["primitive"]
        # load json file
        total = 0
        not_found = 0
        func_matches = 0
        with open(stack_trace_file) as f:
            stacks = json.load(f)

            # iterate through the stacks
            for key, value in stacks.items():

                # print key type
                # for debugging purposes
                # if total > 10:
                #     break
                # stack_details_df = pd.read_csv(stack_details_file)
                total += 1
                print("Processing stack: ", key)

                lock_name = value["locks"]
                

        
                # print(locks[0])
                addresses = value["addresses"]
                kernel_stack = value["kernel_stack"]

                # remove the last element of the kernel stack
                kernel_stack = kernel_stack[:-1]

                # print("Kernel stack: ", kernel_stack)

                # skip stacks with negative stacks
                if int(key) < 0 or len(lock_name) > 1:
                    # skipped += 1
                    print("Skipping stack: ", key)
                    
                    # add to no match file
                    no_match_lock_primitive_df = pd.read_csv(no_match_lock_primitive_file)
                    new_row = pd.DataFrame({
                    "stack_id": [int(key)],
                    "lock_name": [lock_name],
                    "lock_primitive": [primitive_name],
                    "runtime": [runtime],
                    "benchmark": [selected_microbench]
                    })
                    no_match_lock_primitive_df = pd.concat([no_match_lock_primitive_df, new_row], ignore_index=True)
                    no_match_lock_primitive_df.to_csv(no_match_lock_primitive_file, index=False)
                    continue
                
                lock_name = lock_name[0]
                # check the id for the lock primitive in the stack trace
                idx = 0
                start_func = kernel_stack[idx].split("+")[idx]
                # print("function: ", func)
                # print(lock_primitive)

                
                # Find the location of the primitive for spin locks 
                if lock_primitive in interrupt_lock_primitives:
                    while start_func != lock_primitive and idx < len(kernel_stack) - 1:
                        # print("function: ", func)
                        idx += 1
                        start_func = kernel_stack[idx].split("+")[0]
                    
                    # print("index of lock primitive: ", idx)
                    # print("length of kernel stack: ", len(kernel_stack))
                    # print("index: ", idx)
                    
                    # Whether match is found or not, don't count 
                    if idx == len(kernel_stack) - 1:
                        not_found += 1
                        # print("Lock primitive not found in stack")
                        # write no match to the lock primitive file
                        # print("No match for lock primitive: ", lock_primitive)
                        no_match_lock_primitive_df = pd.read_csv(no_match_lock_primitive_file)
                        new_row = pd.DataFrame({
                        "stack_id": [int(key)],
                        "lock_name": [lock_name],
                        "lock_primitive": [primitive_name],
                        "runtime": [runtime],
                        "benchmark": [selected_microbench]
                        })
                        no_match_lock_primitive_df = pd.concat([no_match_lock_primitive_df, new_row], ignore_index=True)
                        no_match_lock_primitive_df.to_csv(no_match_lock_primitive_file, index=False)
                        continue
                
                
                # print(type(lock_name))

                # print("found lock primitive", start_func)
                
                # extract the address at index + 1 of the lock primitive to get the start function for the search
                idx += 1
                # print("Index: ", idx)
                start_addr = addresses[idx]
                start_func =  kernel_stack[idx].split("+")[0]
                # subtract_hex = kernel_stack[idx].split("+")[-1]

                if "." in start_func:
                    start_func = start_func.split(".")[0]

                
                # print("here")

                # print("Start function: ", start_func)
                # print("Start address: ", start_addr)
                # print("Lock name: ", lock_name)

                # print the types 
                # print(type(start_addr))
                # print(type(start_func))
                # print(type(lock_name))
                

                # print(stack_details_df.dtypes)

                # check the stack_details df for lock name, start_func and start_addr and if it exists skip
                
                # check if data frame is empty, this df is used to incrementally add entries to the stack details file
                if not stack_details_df.empty:
                    # print("inside stack df check")
                    
                    filtered_rows = stack_details_df[
                        (stack_details_df["lock_name"].astype(str) == str(lock_name)) & 
                        (stack_details_df["start_func"].astype(str) == str(start_func)) 
                    ]
                    
                    if filtered_rows.shape[0] > 0:
                        # print("Entry already exists in stack details")
                        
                        # Extract the first matching row
                        row = filtered_rows.iloc[0]
                        # print("Row", row)
                        
                        # Update the filtered_locks_df with the values from the row
                        filtered_locks_df.loc[filtered_locks_df["stack_id"] == int(key), "start_func"] = row["start_func"]
                        filtered_locks_df.loc[filtered_locks_df["stack_id"] == int(key), "start_func_addr"] = row["start_func_addr"]
                        filtered_locks_df.loc[filtered_locks_df["stack_id"] == int(key), "start_file"] = row["start_file"]
                        filtered_locks_df.loc[filtered_locks_df["stack_id"] == int(key), "start_line"] = row["start_line"]
                        
                        continue
                        # print(type(stack_details_df["lock_name"]))
        

                    else:
                        print("Entry does not exist in stack details")

                func_name, file, line = run_addr2line(start_addr)
                # func_name, file, line = run_eu_addr2line(addresses[idx])
                
                # direct match
                if start_func == func_name:
                    # print("Found the start function: ", start_func)
                    # print(start_addr)

                    func_matches += 1
                    # add the entry to the stack details df

                    # check for discriminator in the line 
                    if "discriminator" in line:
                        line = line.split("(discriminator")[0]
                        # remove white spaces
                        line = line.strip()

                    new_row = pd.DataFrame({
                    "lock_primitive": [primitive_name],
                    "lock_name": [lock_name],
                    "start_func": [start_func],
                    "start_func_addr": [str(start_addr)],
                    "start_file": [file],
                    "start_line": [line],
                    "direct_start_func_match": [1]
                    })
                    # stack_details_df = stack_details_df.append({"lock_name": lock_name, "start_func": start_func, "start_func_addr": start_addr, "start_file": file, "start_line": line}, ignore_index=True)
                    stack_details_df = pd.concat([stack_details_df, new_row], ignore_index=True)

                    # add the entry to filtered locks df at stackid == key 
                    filtered_locks_df.loc[filtered_locks_df["stack_id"] == int(key), "start_func"] = start_func
                    filtered_locks_df.loc[filtered_locks_df["stack_id"] == int(key), "start_func_addr"] = start_addr
                    filtered_locks_df.loc[filtered_locks_df["stack_id"] == int(key), "start_file"] = file
                    filtered_locks_df.loc[filtered_locks_df["stack_id"] == int(key), "start_line"] = line
                    # stack_details_df.to_csv(stack_details_file, index=False)

                else:

                    # check if func_name is None as some addresses might resolve to ?? ??:0
                    if func_name is None:
                        # print("Function name is None")
                        # print("address length: ", len(addresses))   
                        while func_name is None and idx < (len(addresses)-1):
                            # print("index: ", idx)
                            
                            idx += 1
                            func_name, file, line = run_addr2line(addresses[idx]) 
                            #func_name, file, line = run_eu_addr2line(addresses[idx])
                            

                        
                    # all addresses resolved to ??:0
                    if idx == len(addresses):
                        # print("No function found")

                        # write no match to the lock primitive file 
                        no_match_lock_primitive_df = pd.read_csv(no_match_lock_primitive_file)
                        new_row = pd.DataFrame({
                        "stack_id": [int(key)],
                        "lock_name": [lock_name],
                        "lock_primitive": [primitive_name],
                        "runtime": [runtime],
                        "benchmark": [selected_microbench]
                        })
                        no_match_lock_primitive_df = pd.concat([no_match_lock_primitive_df, new_row], ignore_index=True)
                        no_match_lock_primitive_df.to_csv(no_match_lock_primitive_file, index=False)

                        # Write to process-specific temporary global no match file
                        global_no_match_df = pd.read_csv(temp_global_no_match_file)
                        new_row = pd.DataFrame({
                        "stack_id": [int(key)],
                        "lock_name": [lock_name],
                        "lock_primitive": [primitive_name],
                        "runtime": [runtime],
                        "benchmark": [selected_microbench],
                        "trace_dir": [trace_dir]
                        })
                        global_no_match_df = pd.concat([global_no_match_df, new_row], ignore_index=True)
                        global_no_match_df.to_csv(temp_global_no_match_file, index=False)

                        continue
                    
                    # incase the idx changed, re-read the start_func and start_addr
                    start_func = kernel_stack[idx].split("+")[0]
                    if "." in start_func:
                        start_func = start_func.split(".")[0]
                    
                    subtract_hex = kernel_stack[idx].split("+")[-1]
                    # remove everthing after [ in the subtract_hex
                    subtract_hex = subtract_hex.split("[")[0].strip()
                    start_addr = addresses[idx]

                    # print("Function name mismatch: ", start_func, func_name)
                    # print("Subtract hex: ", subtract_hex)

                    # subtract the hex value from the address
                    start_addr = hex(int(start_addr, 16) - int(subtract_hex, 16))

                    func_name, file, line = run_addr2line(start_addr)
                    # func_name, file, line = run_eu_addr2line(addresses[idx])

                    if start_func == func_name: 
                    
                        start_func = func_name
                        # print("stop: ", stop)
                        # print("Function match in else, name: ", func_name)
                        # print(start_addr)
                        func_matches += 1

                        # check for discriminator in the line 
                        if "discriminator" in line:
                            line = line.split("(discriminator")[0]
                            # remove white spaces
                            line = line.strip()

                        # add the entry to the stack details df
                        new_row = pd.DataFrame({
                        "lock_primitive": [primitive_name],
                        "lock_name": [lock_name],
                        "start_func": [start_func],
                        "start_func_addr": [str(start_addr)],
                        "start_file": [file],
                        "start_line": [line],
                        "direct_start_func_match": [0]
                        })

                        stack_details_df = pd.concat([stack_details_df, new_row], ignore_index=True)
                        # stack_details_df.to_csv(stack_details_file, index=False)

                        # add the entry to filtered locks df at stackid == key
                        filtered_locks_df.loc[filtered_locks_df["stack_id"] == int(key), "start_func"] = start_func
                        filtered_locks_df.loc[filtered_locks_df["stack_id"] == int(key), "start_func_addr"] = start_addr
                        filtered_locks_df.loc[filtered_locks_df["stack_id"] == int(key), "start_file"] = file
                        filtered_locks_df.loc[filtered_locks_df["stack_id"] == int(key), "start_line"] = line
                    
                    else:
                        # idx += 1
                        # print("Function match not found in 20 iterations")
                        # print("match not found")
                        # open match_not_found file and add the entry

                        # write no match to the lock primitive file 
                        no_match_lock_primitive_df = pd.read_csv(no_match_lock_primitive_file)
                        new_row = pd.DataFrame({
                        "stack_id": [int(key)],
                        "lock_name": [lock_name],
                        "lock_primitive": [primitive_name],
                        "runtime": [runtime],
                        "benchmark": [selected_microbench]
                        })
                        no_match_lock_primitive_df = pd.concat([no_match_lock_primitive_df, new_row], ignore_index=True)
                        no_match_lock_primitive_df.to_csv(no_match_lock_primitive_file, index=False)

                        # Write to process-specific temporary global no match file
                        global_no_match_df = pd.read_csv(temp_global_no_match_file)
                        new_row = pd.DataFrame({
                        "stack_id": [int(key)],
                        "lock_name": [lock_name],
                        "lock_primitive": [primitive_name],
                        "runtime": [runtime],
                        "benchmark": [selected_microbench],
                        "trace_dir": [trace_dir]
                        })
                        global_no_match_df = pd.concat([global_no_match_df, new_row], ignore_index=True)
                        global_no_match_df.to_csv(temp_global_no_match_file, index=False)
            


                    # stack_details_df = stack_details_df.append({"lock_name": lock_name, "lock_addr": key, "start_func": start_func, "start_func_addr": start_addr, "start_file": file, "start_line": line}, ignore_index=True)
                 

                # idx = 1
                # start_func = kernel_stack[idx].split("+")[0]
        
        # print("Total: ", total)
        # print("Not found: ", not_found)
        # print("Function matches: ", func_matches)
        
        # Write to process-specific temporary file
        stack_details_df.to_csv(temp_stack_details_file, index=False)
        
        filtered_locks_df.to_csv(filtered_locks_file, index=False)

def interrupt_locks(trace_dir, runtime, selected_microbench):
    # iterate over all lock primitives

    for primitive in lock_primitives:

        primitive_name = primitive["name"]
        
        # read the stack trace file
        # print("Processing primitive: ", primitive["name"])
        stack_trace_file = trace_dir + "/" + primitive["name"] + "_filtered_stacks.json"

        # create another file to store ignored stacks for different primitives
        interrupt_lock_primitive_file = trace_dir + "/" + primitive["name"] + "_stack_interrupt.csv"

        #delete the file if it exists
        if os.path.exists(interrupt_lock_primitive_file):
            os.remove(interrupt_lock_primitive_file)
        
        # add column headers if the file does not exist
        if not os.path.exists(interrupt_lock_primitive_file):
            no_match_lock_primitive_df = pd.DataFrame(columns=["stack_id", "lock_name", "lock_primitive", "runtime", "benchmark"])
            no_match_lock_primitive_df.to_csv(interrupt_lock_primitive_file, index=False)

        
        # stack_trace_file = "/mydata/locks-access-traces/runc/mem_micro/cat_enabled/8KB_total_16GB/traces_run_1/spin_filtered_stacks.json"
         
        # print("Stack trace file: ", stack_trace_file)
        lock_primitive = primitive["primitive"]
        # load json file
        total = 0
        not_found = 0
        func_matches = 0
        with open(stack_trace_file) as f:
            stacks = json.load(f)

            # iterate through the stacks
            for key, value in stacks.items():
                # for debugging purposes
                # if total > 10:
                #     break
                # stack_details_df = pd.read_csv(stack_details_file)
                total += 1
                # print("Processing stack: ", key)

                print("Processing stack: ", key)
                print("Primitive: ", primitive_name)

                lock_name = value["locks"]
                

        
                # print(locks[0])
                addresses = value["addresses"]
                kernel_stack = value["kernel_stack"]

                # remove the last element of the kernel stack
                kernel_stack = kernel_stack[:-1]

                # print("Kernel stack: ", kernel_stack)

                # skip stacks that are not from syscall entry
                if int(key) < 0 or len(lock_name) > 1:
                    # skipped += 1
                    print("Skipping stack: ", key)
                    continue
                
                lock_name = lock_name[0]
                # check the id for the lock primitive in the stack trace
                idx = 0
                start_func = kernel_stack[idx].split("+")[idx]
                # print("function: ", func)
                # print(lock_primitive)

                
                # check for spin locks 
                if lock_primitive in interrupt_lock_primitives:
                    while start_func != lock_primitive and idx < len(kernel_stack) - 1:
                        # print("function: ", func)
                        idx += 1
                        start_func = kernel_stack[idx].split("+")[0]
                    
                    # print("index of lock primitive: ", idx)
                    # print("length of kernel stack: ", len(kernel_stack))
                    # print("index: ", idx)
                    # if idx == len(kernel_stack) - 1:
                    if start_func != lock_primitive:
                        not_found += 1
                        # print("Lock primitive not found in stack")
                        continue

                    # iterate through the kernel stack to find the interrupt handler
                    while start_func not in interrupt_handlers and idx < len(kernel_stack) - 1:
                        idx += 1
                        start_func = kernel_stack[idx].split("+")[0]      
                        print("start func: ", start_func)
                    # print("index of lock primitive: ", idx)                  
                    

                    # if idx == len(kernel_stack) - 1:
                    if start_func not in interrupt_handlers:
                        not_found += 1
                        # print("start func:", start_func)
                        # print("Interrupt handler not found in stack")
                        continue
                    else:
                        # print("Interrupt handler found: ", start_func)
                        # write a new row to the interrupt lock file
                        interrupt_lock_primitive_df = pd.read_csv(interrupt_lock_primitive_file)
                        new_row = pd.DataFrame({
                        "stack_id": [key],
                        "lock_name": [lock_name],
                        "lock_primitive": [primitive_name],
                        "runtime": [runtime],
                        "benchmark": [selected_microbench]
                        })
                        # print(interrupt_lock_primitive_file)
                        interrupt_lock_primitive_df = pd.concat([interrupt_lock_primitive_df, new_row], ignore_index=True)
                        interrupt_lock_primitive_df.to_csv(interrupt_lock_primitive_file, index=False)


def combine_temporary_files():
    """Combine all temporary files created by parallel processes"""
    
    # Ensure global processed stacks directory exists
    if not os.path.exists(global_processed_stacks_path):
        os.makedirs(global_processed_stacks_path)
    
    stack_details_file = global_processed_stacks_path + "/" + global_stack_file
    global_stack_no_match_file = global_processed_stacks_path + "/" + global_no_match_file
    
    # Find all temporary stack details files
    temp_stack_files = []
    temp_no_match_files = []
    
    temp_stack_pattern = stack_details_file.replace('.csv', '_temp_*.csv')
    temp_no_match_pattern = global_stack_no_match_file.replace('.csv', '_temp_*.csv')
    
    temp_stack_files = glob.glob(temp_stack_pattern)
    temp_no_match_files = glob.glob(temp_no_match_pattern)
    
    # Combine stack details files
    if temp_stack_files:
        combined_stack_df = pd.DataFrame()
        for temp_file in temp_stack_files:
            if os.path.exists(temp_file):
                temp_df = pd.read_csv(temp_file)
                combined_stack_df = pd.concat([combined_stack_df, temp_df], ignore_index=True)
                os.remove(temp_file)  # Clean up temporary file
        
        # Remove duplicates and save
        if not combined_stack_df.empty:
            combined_stack_df = combined_stack_df.drop_duplicates()
            
            # If main file exists, combine with it
            if os.path.exists(stack_details_file):
                existing_df = pd.read_csv(stack_details_file)
                combined_stack_df = pd.concat([existing_df, combined_stack_df], ignore_index=True)
                combined_stack_df = combined_stack_df.drop_duplicates()
            
            combined_stack_df.to_csv(stack_details_file, index=False)
            print(f"Combined {len(temp_stack_files)} stack details files into {stack_details_file}")
    
    # Combine no match files
    if temp_no_match_files:
        combined_no_match_df = pd.DataFrame()
        for temp_file in temp_no_match_files:
            if os.path.exists(temp_file):
                temp_df = pd.read_csv(temp_file)
                combined_no_match_df = pd.concat([combined_no_match_df, temp_df], ignore_index=True)
                os.remove(temp_file)  # Clean up temporary file
        
        # Remove duplicates and save
        if not combined_no_match_df.empty:
            combined_no_match_df = combined_no_match_df.drop_duplicates()
            
            # If main file exists, combine with it
            if os.path.exists(global_stack_no_match_file):
                existing_df = pd.read_csv(global_stack_no_match_file)
                combined_no_match_df = pd.concat([existing_df, combined_no_match_df], ignore_index=True)
                combined_no_match_df = combined_no_match_df.drop_duplicates()
            
            combined_no_match_df.to_csv(global_stack_no_match_file, index=False)
            print(f"Combined {len(temp_no_match_files)} no match files into {global_stack_no_match_file}")

def main():
    warmup_phases = [True , False]
    # extract key,`` value pairs from the dictionary
    selected_benchmark = "cloud_workloads" # options: "mem_micro", "file_metadata", "sysbench", "stress-ng-fork", "stress-ng-futex", "funcBench", "cloud_workloads"

    benchmark_traces_path = {}

    if selected_benchmark == "mem_micro" or selected_benchmark == "mem_micro_sysdig" or selected_benchmark == "mem_micro_new":
        for runtime in runtimes:
            for mem_bench in mem_micro_benchmarks:
                trace_path = trace_base_dir + runtime + "/" + selected_benchmark + "/cat_hyperthreading_enabled/" + mem_bench + "/"

                if runtime in benchmark_traces_path:
                    benchmark_traces_path[runtime].append(trace_path)
                else:
                    benchmark_traces_path[runtime] = [trace_path]

    elif selected_benchmark == "file_metadata" or selected_benchmark == "file_metadata_new":

        for runtime in runtimes:
            for file_bench in file_metadata_micro_benchmarks:
                trace_path = trace_base_dir + runtime + "/" + selected_benchmark + "/cat_hyperthreading_enabled/" + file_bench + "/"
                # add to the dictionary of benchmark_traces_path
                # check if the key exists
                if runtime in benchmark_traces_path:
                    benchmark_traces_path[runtime].append(trace_path)
                else:
                    benchmark_traces_path[runtime] = [trace_path]

    elif selected_benchmark == "funcBench":
        print("Processing funcBench traces")
        if not stress_ng_stressors_enabled:
            for runtime in runtimes:
                for func_bench in functionBench:
                    trace_path = trace_base_dir + runtime + "/" + selected_benchmark + "/cat_hyperthreading_enabled/" + func_bench + "/"
                    # add to the dictionary of benchmark_traces_path
                    # check if the key exists
                    if runtime in benchmark_traces_path:
                        benchmark_traces_path[runtime].append(trace_path)
                    else:
                        benchmark_traces_path[runtime] = [trace_path]
        else:
            print("Using stress-ng stressors for funcBench")
            for runtime in runtimes:
                for func_bench in functionBench:
                    for stressor in stress_ng_stressors:
                        trace_path = trace_base_dir + runtime + "/" + selected_benchmark + "/cat_hyperthreading_enabled/" + func_bench + "/stress_ng_traces/" + stressor + "/"
                        # add to the dictionary of benchmark_traces_path
                        # check if the key exists
                        if runtime in benchmark_traces_path:
                            benchmark_traces_path[runtime].append(trace_path)
                        else:
                            benchmark_traces_path[runtime] = [trace_path]

    elif selected_benchmark == "cloud_workloads":
        print("here")
        for runtime in runtimes:
            for cloud_workload in cloud_workloads:
                # stress_ng_stressor is enabled
                if stress_ng_stressors_enabled:
                    for stress_ng_stressor in stress_ng_stressors:
                        trace_path = trace_base_dir + runtime + "/cloud-workloads-re-run/cat_hyperthreading_enabled/" + cloud_workload + "/" + stress_ng_stressor + "/"
                        # add to the dictionary of microbench_traces_path
                        # check if the key exists
                        if runtime in benchmark_traces_path:
                            benchmark_traces_path[runtime].append(trace_path)
                        else:
                            benchmark_traces_path[runtime] = [trace_path]
                else:
                    print("Not used, modify if two instances of cloud workloads are traced")
    else:
        for runtime in runtimes:
            trace_path = trace_base_dir + runtime + "/" + selected_benchmark  + "/cat_hyperthreading_enabled/"
            # add to the dictionary of benchmark_traces_path
            benchmark_traces_path[runtime] = [trace_path]

    print(benchmark_traces_path)

    # Collect all processing tasks
    processing_tasks = []
    
    for warmup_phase in warmup_phases:
        print("Warmup phase: ", warmup_phase)
        for runtime, paths in benchmark_traces_path.items():
            for run in range(start_range, total_runs+1):
                for path in paths:
                    run_path = path + ("traces_without_warmup_run_" if warmup_phase else "traces_with_warmup_run_") + str(run)
                    processing_tasks.append((run_path, runtime, selected_benchmark))

    print(f"Total processing tasks: {len(processing_tasks)}")
    
    # Determine number of parallel processes (use 70% of available CPU cores)
    num_processes = max(1, int(mp.cpu_count() * 0.70))
    print(f"Using {num_processes} parallel processes")

    # Process tasks in parallel
    if len(processing_tasks) > 0:
        with ProcessPoolExecutor(max_workers=num_processes) as executor:
            # Submit all tasks
            future_to_task = {executor.submit(process_single_trace, task): task for task in processing_tasks}
            
            # Process completed tasks
            completed = 0
            for future in as_completed(future_to_task):
                task = future_to_task[future]
                try:
                    future.result()  # This will raise an exception if the task failed
                    completed += 1
                    print(f"Progress: {completed}/{len(processing_tasks)} tasks completed")
                except Exception as exc:
                    print(f"Task {task[0]} generated an exception: {exc}")
    
    print("All processing tasks completed!")
    
    # Combine all temporary files
    print("Combining temporary files...")
    combine_temporary_files()
    print("Temporary files combined successfully!")

# call main 
if __name__ == "__main__":
    main()