python实现的最低松弛度优先(LLF)算法

#!/user/bin/env python
# -*- coding:utf-8 -*-

import sys


class TaskControlBlock:
    CURRENT_TIME = 0

    def __init__(self, dictionary,
                 name_list,
                 period_time,
                 central_processing_unit_time,
                 remain_time,
                 current_period,
                 laxity_time):
        for key in dictionary.keys():
            name_list.append(key)
            period_time.append(dictionary.get(key)[1])
            central_processing_unit_time.append(dictionary.get(key)[0])
            remain_time.append(dictionary.get(key)[0])
            current_period.append(1)
            laxity_time.append(dictionary.get(key)[1] - dictionary.get(key)[0] - self.CURRENT_TIME)

    @staticmethod
    def get_index_of_min(least_laxity_task_list, minimum):
        return least_laxity_task_list.index(minimum)

    @staticmethod
    def get_another_index_of_min(least_laxity_task_list, minimum):
        least_laxity_task_list[least_laxity_task_list.index(minimum)] = sys.maxsize
        return least_laxity_task_list.index(min(least_laxity_task_list))

    @staticmethod
    def is_exit(temp_list):
        for element in temp_list:
            if element != sys.maxsize:
                return False
        return True

    def scheduling(self, name_list,
                   period_time,
                   central_processing_unit_time,
                   remain_time,
                   current_period,
                   laxity_time):
        least_laxity_task = laxity_time.index(min(laxity_time))
        temp_list = []
        for i in laxity_time:
            temp_list.append(i)
        if self.CURRENT_TIME < period_time[least_laxity_task] * (current_period[least_laxity_task] - 1):
            while self.CURRENT_TIME < period_time[least_laxity_task] * 
                    (current_period[least_laxity_task] - 1):
                least_laxity_task = self.get_another_index_of_min(temp_list, min(temp_list))
                if self.is_exit(temp_list=temp_list):
                    exit(0)
            another_task = self.get_another_index_of_min(temp_list, min(temp_list))
            if remain_time[least_laxity_task] 
                    <= laxity_time[another_task]:
                running_time = remain_time[least_laxity_task]
            else:
                running_time = laxity_time[another_task]
            remain_time[least_laxity_task] -= running_time
            print(name_list[least_laxity_task], self.CURRENT_TIME, running_time)
            self.CURRENT_TIME += running_time
            if remain_time[least_laxity_task] == 0:
                current_period[least_laxity_task] += 1
                remain_time[least_laxity_task] = central_processing_unit_time[least_laxity_task]
            i = 0
            while i < laxity_time.__len__():
                laxity_time[i] = current_period[i] * period_time[i] - 
                                 remain_time[i] - self.CURRENT_TIME
                i += 1
            print(laxity_time)
        else:
            another_task = self.get_another_index_of_min(temp_list, min(temp_list))
            if remain_time[least_laxity_task] 
                    <= temp_list[another_task]:
                running_time = remain_time[least_laxity_task]
            else:
                running_time = laxity_time[another_task]
            remain_time[least_laxity_task] -= running_time
            print(name_list[least_laxity_task], self.CURRENT_TIME, running_time)
            self.CURRENT_TIME += running_time
            if remain_time[least_laxity_task] == 0:
                current_period[least_laxity_task] += 1
                remain_time[least_laxity_task] = central_processing_unit_time[least_laxity_task]
            i = 0
            while i < laxity_time.__len__():
                laxity_time[i] = current_period[i] * period_time[i] - 
                                 remain_time[i] - self.CURRENT_TIME
                i += 1
            print(laxity_time)


if __name__ == "__main__":
    task_dictionary = {"A": [10, 30],
                       "B": [25, 75],
                       "C": [30, 90],}
    current_time = 0
    name_list = []
    period_time = []
    central_processing_unit_time = []
    remain_time = []
    current_period = []
    laxity_time = []
    tcb = TaskControlBlock(task_dictionary,
                           name_list,
                           period_time,
                           central_processing_unit_time,
                           remain_time,
                           current_period,
                           laxity_time)

#!/user/bin/env python
# -*- coding:utf-8 -*-

import sys
import EarliestDeadlineFirst
import LeastLaxityFirst


class GetCloseTime:
    """ design the close time by itself """
    def __init__(self, dictionary):
        self.dictionary = dictionary

    def greatest_common_divisor(self, _left, _right):
        return _left if _right == 0 else self.greatest_common_divisor(_right, _left % _right)

    def lowest_common_multiple(self):
        temp_result = 1
        for value in self.dictionary.values():
            temp_result = value[1] * temp_result / self.greatest_common_divisor(value[1], temp_result)
        return temp_result


class TimeError(Exception):
    """
    Self-defined Exception :
        Judging whether the processing time and cycle of real-time tasks satisfy the conditions
    """
    def __init__(self, message):
        self.message = message

def is_execute(dictionary):
    sum = 0
    for value in dictionary.values():
        sum += value[0] / value[1]
    return sum

if __name__ == "__main__":
    task_dictionary = {"A": [10, 30],
                       "B": [20, 60],
                       "C": [30, 90]}
    if is_execute(task_dictionary) > 1:
        raise TimeError("error, scheduling finish!")
    user_choose = input("Please enter your choice"
                        "(1 : EarlistDeadlineFirist) or (2 : LeastLaxityFirst) : ")
    close_time_object = GetCloseTime(task_dictionary)
    close_time = close_time_object.lowest_common_multiple()
    if int(user_choose) == 1:
        current_time = 0
        name_list = []
        period_time = []
        central_processing_unit_time = []
        remain_time = []
        current_period = []
        tcb = EarliestDeadlineFirst.TaskControlBlock(task_dictionary,
                               name_list,
                               period_time,
                               central_processing_unit_time,
                               remain_time,
                               current_period)


        while tcb.CURRENT_TIME < close_time:
            tcb.scheduling(name_list,
                           period_time,
                           central_processing_unit_time,
                           remain_time,
                           current_period)
    else:
        current_time = 0
        name_list = []
        period_time = []
        central_processing_unit_time = []
        remain_time = []
        current_period = []
        laxity_time = []
        tcb = LeastLaxityFirst.TaskControlBlock(task_dictionary,
                               name_list,
                               period_time,
                               central_processing_unit_time,
                               remain_time,
                               current_period,
                               laxity_time)
        while tcb.CURRENT_TIME < close_time:
            tcb.scheduling(name_list,
                           period_time,
                           central_processing_unit_time,
                           remain_time,
                           current_period,
                           laxity_time)