Page Replacement Algorithms - Jonh Alexis Buot (LaplaceXD)

# ===========================================================
# Authors:  Jonh Alexis Buot
# Github:   https://github.com/LaplaceXD
# Date:     November 27, 2023
# Course:   CS 3104 - Operating Systems
# Activity: Page Replacement Algorithms
# ===========================================================

def fifo(frame_len, pages):
    memory = [None] * frame_len
    arrival_queue = []

    state = arrival_queue
    for time, arrived_page in enumerate(pages, start=1):
        replaced_page = None
        is_fault = False

        if arrived_page not in memory:
            is_fault = True
            mem_len = len(memory) - memory.count(None)

            frame = mem_len
            if mem_len == frame_len:
                replaced_page = arrival_queue.pop(0)
                frame = memory.index(replaced_page)

            memory[frame] = arrived_page
            arrival_queue.append(arrived_page)

        yield time, memory.copy(), state.copy(), arrived_page, replaced_page, is_fault

def lru(frame_len, pages):
    memory = [None] * frame_len
    usage_queue = []

    state = usage_queue
    for time, arrived_page in enumerate(pages, start=1):
        replaced_page = None
        is_fault = False

        if arrived_page not in memory:
            is_fault = True
            mem_len = len(memory) - memory.count(None)

            frame = mem_len
            if mem_len == frame_len:
                replaced_page = usage_queue.pop(0)
                frame = memory.index(replaced_page)

            memory[frame] = arrived_page
        else:
            usage_queue.remove(arrived_page)

        usage_queue.append(arrived_page)
        yield time, memory.copy(), state.copy(), arrived_page, replaced_page, is_fault

def lfu(frame_len, pages):
    memory = [None] * frame_len
    frequency_counts = [0] * frame_len
    arrival_queue = []

    state = []
    for time, arrived_page in enumerate(pages, start=1):
        replaced_page = None
        is_fault = False

        if arrived_page not in memory:
            is_fault = True
            mem_len = len(memory) - memory.count(None)

            frame = mem_len
            if mem_len == frame_len:
                replaced_page, _ = state.pop(0)
                frame = memory.index(replaced_page)
                arrival_queue.remove(replaced_page)

            memory[frame] = arrived_page
            frequency_counts[frame] = 1
            arrival_queue.append(arrived_page)
        else:
            frequency_counts[memory.index(arrived_page)] += 1

        page_freqs = zip((m for m in memory if m is not None), frequency_counts)
        state = sorted(page_freqs, key=lambda pf : (pf[1], arrival_queue.index(pf[0])))
        yield time, memory.copy(), state.copy(), arrived_page, replaced_page, is_fault

def optimal(frame_len, pages):
    memory = [None] * frame_len
    page_intervals = pages.copy()
    arrival_queue = []

    state = []
    for time, arrived_page in enumerate(pages, start=1):
        replaced_page = None
        is_fault = False

        if arrived_page in page_intervals:
            page_intervals.remove(arrived_page)

        if arrived_page not in memory:
            is_fault = True
            mem_len = len(memory) - memory.count(None)

            frame = mem_len
            if mem_len == frame_len:
                replaced_page = state.pop(0)
                frame = memory.index(replaced_page)
                arrival_queue.remove(replaced_page)

            memory[frame] = arrived_page
            arrival_queue.append(arrived_page)

        state = [page for page in memory if page is not None]
        # Sort by shortest next interval, if they no longer have a next interval,
        # then sort by their arrival time
        state.sort(key=lambda page : (page_intervals.index(page) if page in page_intervals else len(page_intervals), -1 * arrival_queue.index(page)), reverse=True)
        yield time, memory.copy(), state.copy(), arrived_page, replaced_page, is_fault

def create_mem_log(time, memory, state, inserted_page, replaced_page, is_fault, mem_log_pad, state_log_pad, time_log_pad):
    status = "❌ FAULT " if is_fault else "🎯 HIT   "
    action = "loaded" if is_fault else "found"
    stringed_memory = str(memory).replace("None", "-")

    page_inserted = "Page {}".format(inserted_page)
    involved_frame = "Frame {}".format(str(memory.index(inserted_page) + 1).zfill(3))
    replaced_clause = " replacing page {}".format(replaced_page) if replaced_page is not None else ""

    replacement_log = "{}: {} {} in {}{}".format(status, page_inserted, action, involved_frame, replaced_clause)
    state_log = "{:>{}}".format(str(state), state_log_pad) if state_log_pad is not None else str(state)
    time_log = "{:>{}}".format("[{}]".format(time), time_log_pad + 2) if time_log_pad is not None else "[{}]".format(time)
    mem_visual_log = "{:>{}}".format(stringed_memory, mem_log_pad) if mem_log_pad is not None else str(stringed_memory)

    return "{} {} {} {} {}".format(time_log, "({})".format(inserted_page), mem_visual_log, state_log, replacement_log)

pages_ref = input("Page references (separate by spaces): ")
num_of_frames = int(input("Number of Frames: "))

pages = pages_ref.split(" ")
pages = [int(page) if page.isdigit() else page for page in pages]

algs = {
    fifo: "First-In, First-Out (FIFO)",
    lru: "Least Recently Used (LRU)",
    lfu: "Least Frequently Used (LFU)",
    optimal: "Optimal"
}

for alg_fn, alg_name in algs.items():
    print("=====", alg_name, "Simulation =====")
    print("Legend: [<Time>] <Inserted Page> <Memory Visual> <State (After Replacement)> <Replacement Log>")
    snapshots = [snapshot for snapshot in alg_fn(num_of_frames, pages)]
    time_pad = max(len(str(s[0])) for s in snapshots)
    mem_visual_pad = max(len(str(s[1]).replace("None", "-")) for s in snapshots)
    mem_state_pad = max(len(str(s[2])) for s in snapshots)

    print(*(create_mem_log(*s, mem_visual_pad, mem_state_pad, time_pad) for s in snapshots), sep="\n")
    faults = [s[5] for s in snapshots].count(True)
    hits = len(snapshots) - faults
    hit_percent = "{:.2f}%".format(hits * 100 / len(snapshots))
    fault_percent = "{:.2f}%".format(faults * 100 / len(snapshots))
    print("{:15}: {:<30}".format("No. of Hits", hits), end="")
    print("{:15}: {:<30}".format("Hit Ratio", hit_percent), end="\n")
    print("{:15}: {:<30}".format("No. of Faults", faults), end="")
    print("{:15}: {:<30}".format("Fault Ratio", fault_percent), end="\n")
    print("\n" + "-" * 25 + "\n")