云计算任务调理仿真02

前面已经分享过一个仿真项目，但是基于policy gradient方法实现的，考虑到很多人从零到一实现DQN方法有点难度，以是这次分享一个基于DQN实现的仿真项目，非常简单。

这里之以是简单主要得益于它是用pytorch实现的，而pytorch各个版本之间差异不是非常大，可以互用。
这里没有之前那么复杂的建模，首先是任务类

1. class Task(object):
2.     # 任务类
3.     def __init__(self, jobID, index, CPU, RAM, disk, runtime, status):
4.         import time
5.         self.parent = []
6.         self.child = []
7.         self.jobID = jobID
8.         self.index = index
9.         self.CPU = CPU
10.         self.RAM = RAM
11.         self.disk = disk
12.         self.status = status  # -1: rejected, 0: finished, 1: ready, 2: running
13.         self.runtime = runtime
14.         self.ddl = time.time() + self.runtime * 5
15.         self.endtime = 0

复制代码

然后构建DAG，因为云计算中的任务大多是具有关联性的，是有向无环图

1. class DAG(object):
2.     def __init__(self, fname, num_task):
4.         self.fname = fname
5.         # 任务数量
6.         self.num_task = num_task
7.         self.job = []
8.         self.task = []
10.     def readfile(self):
11.         # 读取任务数据
12.         num_task = 0
13.         with open(self.fname, 'r') as f:
14.             task = []
15.             for line in f:
16.                 if line[0] == 'J':
17.                     if len(task) != 0:
18.                         self.job.append(task)
19.                         task = []
20.                 else:
21.                     info = list(line.split(','))
22.                     # 任务的信息，jobid，index就是任务的标识，cpu,内存，硬盘，
23.                     # 外加一个状态jobID, index, CPU, RAM, disk, runtime, status)
24.                     task.append \
25.                         (Task(info[5], info[6], float(info[3]), float(info[4]), float(info[8]), float(info[2]), 1))
26.                     num_task += 1
27.                 if num_task == self.num_task:
28.                     break
29.             if len(task) != 0:
30.                 self.job.append(task)
32.     def checkRing(self, parent, child):
33.         # 检查无环
34.         if parent.index == child.index:
35.             return True
36.         if len(child.child) == 0:
37.             return False
38.         for c in child.child:
39.             if self.checkRing(parent, c):
40.                 return True
41.         return False
43.     def buildDAG(self):
44.         # 构建有向无环图
45.         import random
46.         for job in self.job:
47.             for task in job:
48.                 i = random.randint(-len(job), len(job) - 1)
49.                 if i < 0:
50.                     continue
51.                 parent = job[i]
52.                 if self.checkRing(parent, task) == False:
53.                     task.parent.append(parent)
54.                     parent.child.append(task)
55. ……
56. ……

复制代码

环境类，定义云计算资源，以及调理过程中状态的转移，练习过程等等

1. class environment(object):
3.     def __init__(self, scale, fname, num_task, num_server):
5.         self.scale = scale
6.         self.fname = fname
7.         self.task = []
8.         self.dag = DAG(self.fname, num_task)  # 根据task数量构建dag
9.         # 设置每个服务器上虚拟机的数量
10.         self.VMNum = 5
11.         self.rej = 0
12.         # 任务数量和服务器数量是通过参数传递的
13.         self.num_task = num_task
14.         self.severNum = num_server
15.         # 而集群数量是通过计算出来的
16.         if num_server <= 50:
17.             self.farmNum = 1
18.         else:
19.             if int(self.severNum / 50) * 50 < num_server:
20.                 self.farmNum = int(self.severNum / 50) + 1
21.             else:
22.                 self.farmNum = int(self.severNum / 50)
24.         self.remainFarm = []
25.         self.FarmResources = []
26.         self.severs = [[1, 1] for _ in range(self.severNum)]
27.         self.VMtask = []
28.         self.totalcost = 0
29.         #self.init_severs(num_server)
31.         self.losses_stage1 = []
32.         self.losses_stage2 = []
34.         print("Total Number of tasks: {0}".format(num_task))
36.     def init_severs(self, severNum):
37.         # 服务器，host，每个host上又可以虚拟出一定的虚拟机，然后虚拟机处理任务
38.         VM = [[[1.0 / self.VMNum, 1.0 / self.VMNum] for _ in range(self.VMNum)] for _ in range(severNum)]
39.         self.VMtask.append([[[] for _ in range(self.VMNum)] for _ in range(severNum)])
40.         return VM
41. ……
42. ……

复制代码

构建DQN的智能体，有Q值的计算和更新，才是基于值的强化学习方法

1. class Agent():
2.     def __init__(self, input_dims, n_actions, lr, gamma=0.99,
3.                  epsilon=1.0, eps_dec=1e-5, eps_min=0.01):
4.         self.lr = lr
5.         self.input_dims = input_dims
6.         self.n_actions = n_actions
7.         self.gamma = gamma
8.         self.epsilon = epsilon
9.         self.eps_dec = eps_dec
10.         self.eps_min = eps_min
11.         self.action_space = [i for i in range(self.n_actions)]
13.         self.Q = LinearDeepQNetwork(self.lr, self.n_actions, self.input_dims)
14.         self.losses = []
16.     def choose_action(self, state):
17.         if np.random.random() > self.epsilon:
18.             state1 = T.tensor(state, dtype=T.float).to(self.Q.device)
19.             actions = self.Q.forward(state1)
20.             #选最大的动作执行
21.             action = T.argmax(actions).item()
22.         else:
23.             action = np.random.choice(self.action_space)
25.         return action
27.     def decrement_epsilon(self):
28.         #贪心的变化
29.         self.epsilon = self.epsilon - self.eps_dec \
30.                         if self.epsilon > self.eps_min else self.eps_min
32.     def learn(self, state, action, reward, state_):
33.         self.Q.optimizer.zero_grad()
34.         states = T.tensor(state, dtype=T.float).to(self.Q.device)
35.         actions = T.tensor(action).to(self.Q.device)
36.         rewards = T.tensor(reward).to(self.Q.device)
37.         states_ = T.tensor(state_, dtype=T.float).to(self.Q.device)
39.         q_pred = self.Q.forward(states)[actions]
41.         q_next = self.Q.forward(states_).max()
43.         q_target = reward + self.gamma*q_next
45.         loss = self.Q.loss(q_target, q_pred).to(self.Q.device)
46.         loss.backward()
47.         self.Q.optimizer.step()
48.         self.decrement_epsilon()
49.         self.losses.append(loss.item())

复制代码

在此底子上，可以继续实现fixed-q-target和experience replay以及double QDN等优化
我添加了打印损失函数值的代码

以是为了方便程序的运行和跨时间段使用，修改等，发起用pytorch举行实现

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。