python，python-scheduling.rar

from collections import deque class Process: 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 MLFQ has a lot of queues that have different priority In this MLFQ, The first Queue(0) to last second Queue(N-2) of MLFQ have 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] """ return [q.burst_time for q in queue] 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)