基于深度强化学习的云工作流调度.zip

import gym, random, copy, xlwt from gym import spaces from gym.utils import seeding import numpy as np import networkx as nx from . import utils class clusterEnv(gym.Env): metadata = {"render.modes": ["human", "rgb_array"], "video.frames_per_second": 50} def __init__(self): self.M = 30 # 动作列表长度，不包括-1 self.t_unit = 10 # 时间单位，用于生成DAG节点的任务需求时间，80%的概率平均分布于t-3t 20的概率平均分布在10t-15t self.cpu_res_unit = 100 # CPU资源单位，用于生成DAG节点的CPU占用需求，50%的概率的任务为CPU资源需求密集型，随机占用0.25r-0.5r 50%的概率随机占用0.05r-0.01r self.memory_res_unit = 100 # Memory资源单位，用于生成DAG节点的memoory占用需求，50%的概率的任务为memory资源需求密集型，随机占用0.25r-0.5r 50%的概率随机占用0.05r-0.01r self.time, self.cpu_res, self.memory_res = 0, 100, 100 ##状态信息 self.time = 0 # 整体环境时钟，DAG执行已花费时间 self.cpu_res = 100 # 当前计算资源的CPU容量 self.memory_res = 100 # 当前计算资源的memory容量 self.b_level = None # b-level值 self.children_num = None # 子节点数 self.edges = [] # 随机生成DAG的信息 self.available_list = [-1] * self.M # 调度输入个数 尽量大于readylist 取self.M，不足补-1 self.duration = [] # 随机生成DAG的时间占用信息 self.demand = [] # backlot的总资源占用信息 self.wait_duration = [-1] * self.M # 随机生成的DAG时间占用信息 self.cpu_demand = [-1] * self.M # 随机生成的DAG cpu 资源占用信息 self.memory_demand = [-1] * self.M # 随机生成的DAG memory 资源占用信息 self.ready_list = [] # 满足依赖关系的DAG,可能被执行，但不满足 self.done_job = [] # 已完成的任务ID self.position = [] # 当前DAG画图坐标 self.backlot_time = 0 # backlot的总时间占用信息 self.backlot_cpu_res = 0 # backlot的总CPU占用信息 self.backlot_memory_res = 0 # backlot的总memory占用信息 ##状态转移信息和中间变量 self.tasks = [] # 计算资源上挂起的任务 self.tasks_remaing_time = {} # 计算资源上挂起的任务剩余执行时间 self.seed1 = 0 self.viewer = None self.state = None self.edges_lib = [] self.duration_lib = [] self.demand_lib = [] self.DAGsize = 30 self.load_train_dataset(self.DAGsize) # self.load_test_dataset(self.DAGsize) def set_state(self, state): self.state = copy.deepcopy(state) def set_ready_list(self, ready_list): self.ready_list = copy.deepcopy(ready_list) def set_done_job(self, done_job): self.done_job = copy.deepcopy(done_job) def set_tasks(self, tasks): self.tasks = copy.deepcopy(tasks) def set_wait_duration(self, wait_duration): self.wait_duration = copy.deepcopy(wait_duration) def set_cpu_demand(self, cpu_demand): self.cpu_demand = copy.deepcopy(cpu_demand) def set_memory_demand(self, memory_demand): self.memory_demand = copy.deepcopy(memory_demand) def set_tasks_remaing_time(self, tasks_remaing_time): self.tasks_remaing_time = copy.deepcopy(tasks_remaing_time) def set_cpu_res(self, cpu_res): self.cpu_res = cpu_res def set_memory_res(self, memory_res): self.memory_res = memory_res def set_time(self, time): self.time = time def load_train_dataset(self, DAGsize): ##########################################training################################ print('train datasheet lib.') edges_lib_path = '/Users/livion/Documents/GitHub/Cloud-Workflow-Scheduling-base-on-Deep-Reinforcement-Learning/npy/train_datasheet/' + str( DAGsize) + '/edges' + str(DAGsize) + '_lib.npy' duration_lib_path = '/Users/livion/Documents/GitHub/Cloud-Workflow-Scheduling-base-on-Deep-Reinforcement-Learning/npy/train_datasheet/' + str( DAGsize) + '/duration' + str(DAGsize) + '_lib.npy' demand_lib_path = '/Users/livion/Documents/GitHub/Cloud-Workflow-Scheduling-base-on-Deep-Reinforcement-Learning/npy/train_datasheet/' + str( DAGsize) + '/demand' + str(DAGsize) + '_lib.npy' # edges_lib_path = '/Users/livion/Documents/GitHub/Cloud-Workflow-Scheduling-base-on-Deep-Reinforcement-Learning/npy/train_datasheet/'+'test'+'/edges' +'test' +'_lib.npy' # duration_lib_path = '/Users/livion/Documents/GitHub/Cloud-Workflow-Scheduling-base-on-Deep-Reinforcement-Learning/npy/train_datasheet/'+'test'+'/duration' + 'test' +'_lib.npy' # demand_lib_path = '/Users/livion/Documents/GitHub/Cloud-Workflow-Scheduling-base-on-Deep-Reinforcement-Learning/npy/train_datasheet/'+'test'+'/demand'+'test'+'_lib.npy' self.edges_lib = np.load(edges_lib_path, allow_pickle=True).tolist() self.duration_lib = np.load(duration_lib_path, allow_pickle=True).tolist() self.demand_lib = np.load(demand_lib_path, allow_pickle=True).tolist() print('load completed.') return def load_test_dataset(self, DAGsize): #########################################testing################################ print('test datasheet loaded.') edges_lib_path = '/Users/livion/Documents/GitHub/Cloud-Workflow-Scheduling-base-on-Deep-Reinforcement-Learning/npy/test_datasheet/' + str( DAGsize) + '/edges' + str(DAGsize) + '_lib.npy' duration_lib_path = '/Users/livion/Documents/GitHub/Cloud-Workflow-Scheduling-base-on-Deep-Reinforcement-Learning/npy/test_datasheet/' + str( DAGsize) + '/duration' + str(DAGsize) + '_lib.npy' demand_lib_path = '/Users/livion/Documents/GitHub/Cloud-Workflow-Scheduling-base-on-Deep-Reinforcement-Learning/npy/test_datasheet/' + str( DAGsize) + '/demand' + str(DAGsize) + '_lib.npy' self.edges_lib = np.load(edges_lib_path, allow_pickle=True).tolist() self.duration_lib = np.load(duration_lib_path, allow_pickle=True).tolist() self.demand_lib = np.load(demand_lib_path, allow_pickle=True).tolist() print('load completed.') return def return_dim_info(self): return (3 + self.M * 3 + 5), self.M + 1 def _search_for_predecessor(self, node, edges): ''' 寻找前继节点 :param node: 需要查找的节点id :param edges: DAG边信息 :return: node的前继节点id列表 ''' map = {} if node == 'Start': return print("error, 'Start' node do not have predecessor!") for i in range(len(edges)): if edges[i][1] in map.keys(): map[edges[i][1]].append(edges[i][0]) else: map[edges[i][1]] = [edges[i][0]] succ = map[node] return succ def _search_for_successors(self, node, edges): ''' 寻找后续节点 :param node: 需要查找的节点id :param edges: DAG边信息(注意最好传列表的值（edges[:]）进去而不是传列表的地址（edges）!!!) :return: node的后续节点id列表 ''' map = {} if node == 'Exit': return print("error, 'Exit' node do not have successors!") for i in range(len(edges)): if edges[i][0] in map.keys(): map[edges[i][0]].append(edges[i][1]) else: map[edges[i][0]] = [edges[i][1]] pred = map[node] return pred def _update_ready_list(self, ready_list, done_job, edges): ''' 根据已完成的任务更新当前可以执行的task列表，满足DAG的依赖关系。并不表明它可以被执行，因为还要受资源使用情况限制 :param ready_l