Add Multi-Level-Feedback-Queue scheduling algorithm

scheduling/multi_level_feedback_queue.py

@@ -0,0 +1,319 @@
+from collections import deque
+
+
+class Process:
+    """
+    Process should have these fields below
+    """
+
+    def __init__(self, process_name: str, arrival_time: int, burst_time: int) -> None:
+        self.process_name = process_name  # process name
+        self.arrival_time = arrival_time  # arrival time of the process
+        # completion time of finished process or last interrupted time
+        self.stop_time = arrival_time
+        self.burst_time = burst_time  # remaining burst time
+        self.waiting_time = 0  # total time of the process wait in ready queue
+        self.turnaround_time = 0  # time from arrival time to completion time
+
+
+class MLFQ:
+    """
+    MLFQ(Multi Level Feedback Queue)
+    https://en.wikipedia.org/wiki/Multilevel_feedback_queue
+    This MLFQ has a lot of queue that has different priority
+    In this MLFQ,
+    The first Queue(0) to last second Queue(N-2) of MLFQ has Round Robin Algorithm
+    The last Queue(N-1) has First Come, First Served Algorithm
+    """
+
+    def __init__(
+        self,
+        number_of_queues: int,
+        time_slices: list[int],
+        queue: deque[Process],
+        current_time: int,
+    ) -> None:
+        # total number of mlfq's queues
+        self.number_of_queues = number_of_queues
+        # time slice of queues that round robin algorithm applied
+        self.time_slices = time_slices
+        # unfinished process is in this ready_queue
+        self.ready_queue = queue
+        # current time
+        self.current_time = current_time
+        # finished process is in this sequence queue
+        self.finish_queue: deque[Process] = deque()
+
+    def calculate_sequence_of_finish_queue(self) -> list[str]:
+        """
+        This method returns the sequence of finished processes
+        >>> P1 = Process("P1", 0, 53)
+        >>> P2 = Process("P2", 0, 17)
+        >>> P3 = Process("P3", 0, 68)
+        >>> P4 = Process("P4", 0, 24)
+        >>> mlfq = MLFQ(3, [17, 25], deque([P1, P2, P3, P4]), 0)
+        >>> _ = mlfq.multi_level_feedback_queue()
+        >>> mlfq.calculate_sequence_of_finish_queue()
+        ['P2', 'P4', 'P1', 'P3']
+        """
+        sequence = []
+        for i in range(len(self.finish_queue)):
+            sequence.append(self.finish_queue[i].process_name)
+        return sequence
+
+    def calculate_waiting_time(self, queue: list[Process]) -> list[int]:
+        """
+        This method calculates waiting time of processes
+        >>> P1 = Process("P1", 0, 53)
+        >>> P2 = Process("P2", 0, 17)
+        >>> P3 = Process("P3", 0, 68)
+        >>> P4 = Process("P4", 0, 24)
+        >>> mlfq = MLFQ(3, [17, 25], deque([P1, P2, P3, P4]), 0)
+        >>> _ = mlfq.multi_level_feedback_queue()
+        >>> mlfq.calculate_waiting_time([P1, P2, P3, P4])
+        [83, 17, 94, 101]
+        """
+        waiting_times = []
+        for i in range(len(queue)):
+            waiting_times.append(queue[i].waiting_time)
+        return waiting_times
+
+    def calculate_turnaround_time(self, queue: list[Process]) -> list[int]:
+        """
+        This method calculates turnaround time of processes
+        >>> P1 = Process("P1", 0, 53)
+        >>> P2 = Process("P2", 0, 17)
+        >>> P3 = Process("P3", 0, 68)
+        >>> P4 = Process("P4", 0, 24)
+        >>> mlfq = MLFQ(3, [17, 25], deque([P1, P2, P3, P4]), 0)
+        >>> _ = mlfq.multi_level_feedback_queue()
+        >>> mlfq.calculate_turnaround_time([P1, P2, P3, P4])
+        [136, 34, 162, 125]
+        """
+        turnaround_times = []
+        for i in range(len(queue)):
+            turnaround_times.append(queue[i].turnaround_time)
+        return turnaround_times
+
+    def calculate_completion_time(self, queue: list[Process]) -> list[int]:
+        """
+        This method calculates completion time of processes
+        >>> P1 = Process("P1", 0, 53)
+        >>> P2 = Process("P2", 0, 17)
+        >>> P3 = Process("P3", 0, 68)
+        >>> P4 = Process("P4", 0, 24)
+        >>> mlfq = MLFQ(3, [17, 25], deque([P1, P2, P3, P4]), 0)
+        >>> _ = mlfq.multi_level_feedback_queue()
+        >>> mlfq.calculate_turnaround_time([P1, P2, P3, P4])
+        [136, 34, 162, 125]
+        """
+        completion_times = []
+        for i in range(len(queue)):
+            completion_times.append(queue[i].stop_time)
+        return completion_times
+
+    def calculate_remaining_burst_time_of_processes(
+        self, queue: deque[Process]
+    ) -> list[int]:
+        """
+        This method calculate remaining burst time of processes
+        >>> P1 = Process("P1", 0, 53)
+        >>> P2 = Process("P2", 0, 17)
+        >>> P3 = Process("P3", 0, 68)
+        >>> P4 = Process("P4", 0, 24)
+        >>> mlfq = MLFQ(3, [17, 25], deque([P1, P2, P3, P4]), 0)
+        >>> finish_queue, ready_queue = mlfq.round_robin(deque([P1, P2, P3, P4]), 17)
+        >>> mlfq.calculate_remaining_burst_time_of_processes(mlfq.finish_queue)
+        [0]
+        >>> mlfq.calculate_remaining_burst_time_of_processes(ready_queue)
+        [36, 51, 7]
+        >>> finish_queue, ready_queue = mlfq.round_robin(ready_queue, 25)
+        >>> mlfq.calculate_remaining_burst_time_of_processes(mlfq.finish_queue)
+        [0, 0]
+        >>> mlfq.calculate_remaining_burst_time_of_processes(ready_queue)
+        [11, 26]
+        """
+        remaining_burst_times = []
+        for i in range(len(queue)):
+            remaining_burst_times.append(queue[i].burst_time)
+        return remaining_burst_times
+
+    def update_waiting_time(self, process: Process) -> int:
+        """
+        This method updates waiting times of unfinished processes
+        >>> P1 = Process("P1", 0, 53)
+        >>> P2 = Process("P2", 0, 17)
+        >>> P3 = Process("P3", 0, 68)
+        >>> P4 = Process("P4", 0, 24)
+        >>> mlfq = MLFQ(3, [17, 25], deque([P1, P2, P3, P4]), 0)
+        >>> mlfq.current_time = 10
+        >>> P1.stop_time = 5
+        >>> mlfq.update_waiting_time(P1)
+        5
+        """
+        process.waiting_time += self.current_time - process.stop_time
+        return process.waiting_time
+
+    def first_come_first_served(self, ready_queue: deque[Process]) -> deque[Process]:
+        """
+        FCFS(First Come, First Served)
+        FCFS will be applied to MLFQ's last queue
+        A first came process will be finished at first
+        >>> P1 = Process("P1", 0, 53)
+        >>> P2 = Process("P2", 0, 17)
+        >>> P3 = Process("P3", 0, 68)
+        >>> P4 = Process("P4", 0, 24)
+        >>> mlfq = MLFQ(3, [17, 25], deque([P1, P2, P3, P4]), 0)
+        >>> _ = mlfq.first_come_first_served(mlfq.ready_queue)
+        >>> mlfq.calculate_sequence_of_finish_queue()
+        ['P1', 'P2', 'P3', 'P4']
+        """
+        finished: deque[Process] = deque()  # sequence deque of finished process
+        while len(ready_queue) != 0:
+            cp = ready_queue.popleft()  # current process
+
+            # if process's arrival time is later than current time, update current time
+            if self.current_time < cp.arrival_time:
+                self.current_time += cp.arrival_time
+
+            # update waiting time of current process
+            self.update_waiting_time(cp)
+            # update current time
+            self.current_time += cp.burst_time
+            # finish the process and set the process's burst-time 0
+            cp.burst_time = 0
+            # set the process's turnaround time because it is finished
+            cp.turnaround_time = self.current_time - cp.arrival_time
+            # set the completion time
+            cp.stop_time = self.current_time
+            # add the process to queue that has finished queue
+            finished.append(cp)
+
+        self.finish_queue.extend(finished)  # add finished process to finish queue
+        # FCFS will finish all remaining processes
+        return finished
+
+    def round_robin(
+        self, ready_queue: deque[Process], time_slice: int
+    ) -> tuple[deque[Process], deque[Process]]:
+        """
+        RR(Round Robin)
+        RR will be applied to MLFQ's all queues except last queue
+        All processes can't use CPU for time more than time_slice
+        If the process consume CPU up to time_slice, it will go back to ready queue
+        >>> P1 = Process("P1", 0, 53)
+        >>> P2 = Process("P2", 0, 17)
+        >>> P3 = Process("P3", 0, 68)
+        >>> P4 = Process("P4", 0, 24)
+        >>> mlfq = MLFQ(3, [17, 25], deque([P1, P2, P3, P4]), 0)
+        >>> finish_queue, ready_queue = mlfq.round_robin(mlfq.ready_queue, 17)
+        >>> mlfq.calculate_sequence_of_finish_queue()
+        ['P2']
+        """
+        finished: deque[Process] = deque()  # sequence deque of terminated process
+        # just for 1 cycle and unfinished processes will go back to queue
+        for i in range(len(ready_queue)):
+            cp = ready_queue.popleft()  # current process
+
+            # if process's arrival time is later than current time, update current time
+            if self.current_time < cp.arrival_time:
+                self.current_time += cp.arrival_time
+
+            # update waiting time of unfinished processes
+            self.update_waiting_time(cp)
+            # if the burst time of process is bigger than time-slice
+            if cp.burst_time > time_slice:
+                # use CPU for only time-slice
+                self.current_time += time_slice
+                # update remaining burst time
+                cp.burst_time -= time_slice
+                # update end point time
+                cp.stop_time = self.current_time
+                # locate the process behind the queue because it is not finished
+                ready_queue.append(cp)
+            else:
+                # use CPU for remaining burst time
+                self.current_time += cp.burst_time
+                # set burst time 0 because the process is finished
+                cp.burst_time = 0
+                # set the finish time
+                cp.stop_time = self.current_time
+                # update the process' turnaround time because it is finished
+                cp.turnaround_time = self.current_time - cp.arrival_time
+                # add the process to queue that has finished queue
+                finished.append(cp)
+
+        self.finish_queue.extend(finished)  # add finished process to finish queue
+        # return finished processes queue and remaining processes queue
+        return finished, ready_queue
+
+    def multi_level_feedback_queue(self) -> deque[Process]:
+        """
+        MLFQ(Multi Level Feedback Queue)
+        >>> P1 = Process("P1", 0, 53)
+        >>> P2 = Process("P2", 0, 17)
+        >>> P3 = Process("P3", 0, 68)
+        >>> P4 = Process("P4", 0, 24)
+        >>> mlfq = MLFQ(3, [17, 25], deque([P1, P2, P3, P4]), 0)
+        >>> finish_queue = mlfq.multi_level_feedback_queue()
+        >>> mlfq.calculate_sequence_of_finish_queue()
+        ['P2', 'P4', 'P1', 'P3']
+        """
+
+        #  all queues except last one have round_robin algorithm
+        for i in range(self.number_of_queues - 1):
+            finished, self.ready_queue = self.round_robin(
+                self.ready_queue, self.time_slices[i]
+            )
+        #  the last queue has first_come_first_served algorithm
+        self.first_come_first_served(self.ready_queue)
+
+        return self.finish_queue
+
+
+if __name__ == "__main__":
+    import doctest
+
+    P1 = Process("P1", 0, 53)
+    P2 = Process("P2", 0, 17)
+    P3 = Process("P3", 0, 68)
+    P4 = Process("P4", 0, 24)
+    number_of_queues = 3
+    time_slices = [17, 25]
+    queue = deque([P1, P2, P3, P4])
+
+    if len(time_slices) != number_of_queues - 1:
+        exit()
+
+    doctest.testmod(extraglobs={"queue": deque([P1, P2, P3, P4])})
+
+    P1 = Process("P1", 0, 53)
+    P2 = Process("P2", 0, 17)
+    P3 = Process("P3", 0, 68)
+    P4 = Process("P4", 0, 24)
+    number_of_queues = 3
+    time_slices = [17, 25]
+    queue = deque([P1, P2, P3, P4])
+    mlfq = MLFQ(number_of_queues, time_slices, queue, 0)
+    finish_queue = mlfq.multi_level_feedback_queue()
+
+    # print total waiting times of processes(P1, P2, P3, P4)
+    print(
+        f"waiting time:\
+        \t\t\t{MLFQ.calculate_waiting_time(mlfq, [P1, P2, P3, P4])}"
+    )
+    # print completion times of processes(P1, P2, P3, P4)
+    print(
+        f"completion time:\
+        \t\t{MLFQ.calculate_completion_time(mlfq, [P1, P2, P3, P4])}"
+    )
+    # print total turnaround times of processes(P1, P2, P3, P4)
+    print(
+        f"turnaround time:\
+        \t\t{MLFQ.calculate_turnaround_time(mlfq, [P1, P2, P3, P4])}"
+    )
+    # print sequence of finished processes
+    print(
+        f"sequnece of finished processes:\
+        {mlfq.calculate_sequence_of_finish_queue()}"
+    )