1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
|
import random
import time
import torch
import torch.distributed.rpc as rpc
from torch.distributed.rpc import rpc_sync
from agent import AgentBase
class ObserverBase:
def __init__(self):
r"""
Inits observer class
"""
self.id = rpc.get_worker_info().id
def set_state(self, state_size, batch):
r"""
Further initializes observer to be aware of rpc environment
Args:
state_size (list): List of integers denoting dimensions of state
batch (bool): Whether agent will be using batch select action
"""
self.state_size = state_size
self.select_action = AgentBase.select_action_batch if batch else AgentBase.select_action_non_batch
def reset(self):
r"""
Resets state randomly
"""
state = torch.rand(self.state_size)
return state
def step(self, action):
r"""
Generates random state and reward
Args:
action (int): Int received from agent representing action to take on state
"""
state = torch.rand(self.state_size)
reward = random.randint(0, 1)
return state, reward
def run_ob_episode(self, agent_rref, n_steps):
r"""
Runs single observer episode where for n_steps, an action is selected
from the agent based on curent state and state is updated
Args:
agent_rref (RRef): Remote Reference to the agent
n_steps (int): Number of times to select an action to transform state per episode
"""
state, ep_reward = self.reset(), None
rewards = torch.zeros(n_steps)
observer_latencies = []
observer_throughput = []
for st in range(n_steps):
ob_latency_start = time.time()
action = rpc_sync(agent_rref.owner(), self.select_action, args=(
agent_rref, self.id, state))
ob_latency = time.time() - ob_latency_start
observer_latencies.append(ob_latency)
observer_throughput.append(1 / ob_latency)
state, reward = self.step(action)
rewards[st] = reward
return [rewards, ep_reward, observer_latencies, observer_throughput]
|
