base.py

from typing import Any, Iterator

from .models.mem_access import MemAccess
from .models.page_table import Page, PageTable
from .models.vm_stats import VMStats


class VMSimBase:
    """Base class for the virtual memory simulation

    This class won't be used directly - it should be inherited by the four algorithms.

    Every algorithm will need to implement the `_select_victim_page()` function. Other functions,
    especially `_handle_page_hit()`, `_handle_page_fault_no_evict()`, and `_handle_page_fault_evict()`
    may be overridden if needed.

    The `sim()` function is the main entry point for the simulation. It will iterate through the trace file,
    and for each memory access log/count the outcomes.

    Most of the work is done in the `_access_mem()` function (which delegates to the other functions that are
    overridden by the algorithms).

    Sources:
        https://www.geeksforgeeks.org/page-replacement-algorithms-in-operating-systems
    """

    ALGORITHM: str = ""

    def __init__(self, num_frames: int) -> None:
        """Initialize the data structures needed for all algorithms (page table, sim stats, RAM)

        Args:
            num_frames (int): Number of frames in RAM
        """
        self.num_frames: int = num_frames

        # I don't actually initialize anything here because that's done in `_sim_setup()`
        self.page_table: PageTable
        # Using a dictionary for RAM is not accurate to a real OS but makes sense for a simulation
        # If I get pointed deducted for not using an array, I'm gonna be pissed 😡
        self.ram: dict[int, Page]  # {frame_num: Page}
        self.stats: VMStats

    def iter_trace_file(self, trace_file_path: str) -> Iterator[tuple[str, int]]:
        """Helper function to iterate through and parse the trace file as an iterator

        Args:
            trace_file_path (Path): Path to the trace file

        Yields:
            Iterator[tuple[str, int]]: (mode, address)
        """
        f = open(trace_file_path, "r")
        for line in f:
            line = line.strip()
            if len(line) == 0 or line[0] not in {"I", "S", "L", "M"}:
                # Ignore non-memory access lines
                continue

            mode = line[0]
            try:
                # Converts 8 digit hex to int
                address = int(line.split()[1][:8], 16)
            except ValueError:
                continue

            if mode == "M":  # Split modify into two operations - load and store
                yield "I", address
                yield "S", address
            else:  # Otherwise, return just one operation
                yield mode, address
        f.close()

    def sim(self, trace_file: str, verbose: bool = False) -> dict[str, Any]:
        """Run the simulation

        Args:
            trace_file (str): Path to the trace file
            verbose (bool): Whether to print verbose output (the individual memory accesses)

        Returns:
            dict[str, Any]: Information and statistics about the simulation
        """
        self._sim_setup(trace_file)

        if verbose is True:  # Table headers for verbose output 💅✨
            print(
                "| Position | Address    | Frame | Mode | Memory Access Result   |\n"
                "| -------- | ---------- | ----- | ---- | ---------------------- |\n"
            )

        # Iterate through the trace file, simulate memory accesses, update stats, and log access info
        pos = 0
        for mode, address in self.iter_trace_file(trace_file):
            page = self.page_table.get_page(address)
            if page is None:  # Create new page if it doesn't exist
                page = self.page_table.add_page(address)

            # Call the algorithm-specific memory access function, which will return an enum of the type of memory access that occurred
            res: MemAccess = self._access_mem(page)

            if mode == "S":
                # I would've put it this in some other more appropriate function, but that would mean having to pass the `mode` bit
                # throughout the call stack and I cba to do that
                page.modified = True

            # Update stats
            self.stats.mem_accesses += 1
            if res == MemAccess.PAGE_HIT:
                self.stats.page_hits += 1
            elif res == MemAccess.PAGE_FAULT_NO_EVICT:
                self.stats.no_evict_page_faults += 1
            elif res == MemAccess.PAGE_FAULT_DIRTY_EVICT:
                self.stats.dirty_evict_page_faults += 1
            elif res == MemAccess.PAGE_FAULT_CLEAN_EVICT:
                self.stats.clean_evict_page_faults += 1

            if verbose is True:  # Print the table row summarizing the memory access
                # This print statement will decimate execution time.
                # It would've been faster if I tab-dilemited the output but I like my tables pretty :3 so deal with it
                # pass -nv (not verbose) to disable printing this
                print(
                    f"| {pos:<8d} "
                    f"| 0x{address:<08X} "
                    f"| {page.frame:<5d} "
                    f"| {mode:<4} "
                    f"| {res.name:<22} "  # Uses the `MemAccess` enum name
                    f"|",
                )

            pos += 1

        # Table headers (footers?) again because your terminal history was likely decimated
        if verbose is True:
            print(
                "| -------- | ---------- | ----- | ---- | ---------------------- |\n"
                "| Position | Address    | Frame | Mode | Memory Access Result   |\n"
            )

        return dict(self)

    def _access_mem(self, page: Page) -> MemAccess:
        """Process a memory access

        Overridden by subclasses when needed.

        Time complexity: O(1) for

        Args:
            page (Page): The page being accessed

        Returns:
            MemAccess: The type of memory access that occurred
        """
        if page.frame is not None:  # Page hit
            return self._handle_page_hit(page)
        elif len(self.page_table.free_frames) > 0:  # Page fault but we have space
            return self._handle_page_fault_no_evict(page)
        else:  # Page fault and we need to evict a page
            # The rent-pigs haven't been tipping their landlords !!! select a victim (algorithm-specific) and evict it
            victim = self._select_victim_page()
            return self._handle_page_fault_evict(page, victim)

    def _select_victim_page(self) -> Page:
        """Select a page to evict

        Every algorithm will implement this differently.

        Returns:
            Page: The selected victim page
        """
        raise NotImplementedError

    def _handle_page_hit(self, page: Page) -> MemAccess:
        """Handle a page hit

        Overridden by subclasses when needed.

        Time complexity: O(1)

        Args:
            page (Page): The page being accessed

        Returns:
            MemAccess: It'll always be `MemAccess.PAGE_HIT`
        """
        return MemAccess.PAGE_HIT  # Default behavior is to do nothing

    def _handle_page_fault_no_evict(self, page: Page) -> MemAccess:
        """Handle a page fault that doesn't require eviction (RAM isn't full)

        Overridden by subclasses when needed.

        Time complexity: O(1)

        Args:
            page (Page): The page causing the fault

        Returns:
            MemAccess: It'll always be `MemAccess.PAGE_FAULT_NO_EVICT`
        """
        frame = self.page_table.allocate_frame(page)
        self.ram[frame] = page  # Add the page to RAM

        return MemAccess.PAGE_FAULT_NO_EVICT

    def _handle_page_fault_evict(self, page: Page, victim: Page) -> MemAccess:
        """Handle a page fault that requires eviction (RAM is full)

        Overridden by subclasses when needed.

        Time complexity: O(1) average case, O(n) worst case

        Args:
            page (Page): The page causing the fault
            victim (Page): The page that will be evicted

        Returns:
            MemAccess: Either `MemAccess.PAGE_FAULT_DIRTY_EVICT` or `MemAccess.PAGE_FAULT_CLEAN_EVICT`
        """
        modified = victim.modified  # Save the dirty bit so we know what to return

        self.ram.pop(victim.frame)  # Remove victim from RAM (if it exists)
        self.page_table.del_page(victim)  # Free the victim frame and invalidate it

        # Now that we've evicted the page, the rest of the process is the same as a no-evict page fault
        self._handle_page_fault_no_evict(page)

        if modified is True:
            return MemAccess.PAGE_FAULT_DIRTY_EVICT
        else:
            return MemAccess.PAGE_FAULT_CLEAN_EVICT

    def _sim_setup(self, trace_file: str) -> None:
        """Helper function to do any extra processing before the simulation starts

        Overridden by subclasses when needed.

        Args:
            trace_file (str): Path to the trace file
        """
        self.page_table = PageTable(self.num_frames)
        self.ram = {}
        self.stats = VMStats()

    def __iter__(self) -> Iterator[tuple[str, Any]]:
        """Makes VMSimBase iterable so I can put it into a dict with `dict(vmsim)`

        Yields:
            Iterator[tuple[str, Any]]: Iterable of the simulation info and statistics
        """
        yield "algorithm", self.ALGORITHM
        yield "num_frames", self.num_frames
        yield "page_table", dict(self.page_table)
        yield "stats", dict(self.stats)

    def __str__(self) -> str:
        """Makes VMSimBase passable to `str()`. So I can do `print(vmsim)`

        Returns:
            str: Simulation info and statistics
        """
        return (
            f"Algorithm:\t\t\t{self.ALGORITHM}\n"
            f"Number of frames:\t\t{self.num_frames}\n"
            f"{str(self.stats)}\n"
            f"{str(self.page_table)}"
        )