Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Test Scheduler by the state progress of each RID #1920

Open
wants to merge 35 commits into
base: master
Choose a base branch
from
Open

Test Scheduler by the state progress of each RID #1920

Changes from 18 commits
Commits
Show all changes
35 commits
Select commit Hold shift + click to select a range
9191d74
Add alternative to test case
Jun 21, 2022
6f89438
Revert "Add alternative to test case"
Jun 22, 2022
0c1da82
Modify expect and _get_basic_steps and create last_state_RID
Jun 22, 2022
ea53aba
Modify notify to suit the test case and change order of a function
Jun 22, 2022
52c6bb3
Re-order the parameters in _get_basic_steps
Jun 22, 2022
a9f1217
Remove space in args in _get_basic_steps
Jun 22, 2022
a09236d
Rename last_state to expect_next_state and Add comments
Jun 23, 2022
04a3c73
Flake8 correction
Jun 23, 2022
6f00de0
Keep track of each RID and its change from pending to preparing
Jun 23, 2022
0488c94
test_scheduler: Skip next check after reached done state
Jul 12, 2022
e9b1383
test_scheduler: Add SchedulerMonitor
Jul 13, 2022
1b06ab3
test_scheduler: Rename SchedulerMonitor.schedulers to experiments
Jul 13, 2022
a148b3e
SchedulerMonitor: Add records with time element
Jul 14, 2022
0b91f53
Add missing colon
Jul 14, 2022
99ffd59
SchedulerMonitor: Add status_record and related assert
Jul 14, 2022
c5f97a0
Rename scheduler_mon and records
Jul 14, 2022
b52d9fa
SchedulerMonitor: Add get_status_order
Jul 14, 2022
6c42c46
SchedulerMonitor: Add end_condition args
Jul 14, 2022
6c06512
Remove () after SchedulerMonitor
Jul 15, 2022
f8f0d34
SchedulerMonitor: Structure exp_flow
Jul 15, 2022
8c2501f
SchedulerMonitor: Add wait_until() and flags
Jul 18, 2022
bdad183
test_pending_priority: Replace done with monitor.wait_until
Jul 18, 2022
d7630a9
test_scheduler: Use monitor.wait_until and assert with basic_flow
Jul 18, 2022
f9bce0c
test_scheduler: remove _get_basic_steps
Jul 20, 2022
566909d
test_flush: test with different priority case
Jul 20, 2022
4357546
test_pause: Add comments and assert before request termination
Jul 21, 2022
c06a090
test_pending: Use asyncio.wait and update monitor
Jul 21, 2022
cd0465b
Scheduler: Remove status_records, get_in_time and get_out_time
Jul 29, 2022
ada6d27
scheduler: Remove end_condition and bool finished
Jul 29, 2022
0efa3db
scheduler: Replace notify by monitor.record
Jul 29, 2022
ee7d6b0
SchedulerMonitor: Apply flow check in record
Jul 29, 2022
e5bcbdf
Scheduler: Rename var for better understanding
Jul 29, 2022
36a36c3
Scheduler_monitor: Put flow_map into class attribute
Aug 1, 2022
a6df1ed
test_scheduler: Put duplicated code to setUp() and tearDown()
Aug 5, 2022
81a6559
test_scheduler: Create and apply AssertScheduler
Aug 5, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
198 changes: 76 additions & 122 deletions artiq/test/test_scheduler.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,7 +6,10 @@

from artiq.experiment import *
from artiq.master.scheduler import Scheduler
from sipyco.sync_struct import process_mod

basic_flow = ["pending", "preparing", "prepare_done", "running",
"run_done", "analyzing", "deleting"]

class EmptyExperiment(EnvExperiment):
def build(self):
Expand Down Expand Up @@ -50,11 +53,11 @@ def _get_expid(name):
}


def _get_basic_steps(rid, expid, priority=0, flush=False):
def _get_basic_steps(rid, expid, priority=0, flush=False, due_date=None):
return [
{"action": "setitem", "key": rid, "value":
{"pipeline": "main", "status": "pending", "priority": priority,
"expid": expid, "due_date": None, "flush": flush,
"expid": expid, "due_date": due_date, "flush": flush,
"repo_msg": None},
"path": []},
{"action": "setitem", "key": "status", "value": "preparing",
Expand Down Expand Up @@ -82,6 +85,59 @@ def get(self):
self._next_rid += 1
return rid

class SchedulerMonitor():
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Remove ()

def __init__(self, end_condition = "deleting"):
self.experiments = {}
self.last_status = {}
self.exp_flow = {}
self.status_records = {
"pending": [],
"preparing": [],
"prepare_done": [],
"running": [],
"run_done": [],
"analyzing": [],
"deleting": [],
"paused": []
}
self.finished = False
self.end_condition = end_condition

def record(self):
for key, value in self.experiments.items():
if key not in self.last_status.keys():
self.last_status[key] = ""
self.exp_flow[key] = []
current_status = self.experiments[key]["status"]
if current_status != self.last_status[key]:
self.last_status[key] = current_status
self.exp_flow[key].append(time())
self.exp_flow[key].append(current_status)
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

No. Structure it.

self.status_records[current_status].append(key)

if current_status == self.end_condition:
self.finished = True
return

def get_in_time(self, rid, status):
return self.exp_flow[rid][self.exp_flow[rid].index(status)-1]

def get_out_time(self, rid, status):
if self.exp_flow[rid][-1] == status:
return "never"
else:
return self.exp_flow[rid][self.exp_flow[rid].index(status) + 1]

def get_exp_order(self, status):
return self.status_records[status]

def get_status_order(self, rid):
result_list = self.exp_flow[rid].copy()
for item in result_list:
if isinstance(item, float):
result_list.remove(item)
return result_list


class SchedulerCase(unittest.TestCase):
def setUp(self):
Expand All @@ -96,6 +152,7 @@ def test_steps(self):
expect = _get_basic_steps(1, expid)
done = asyncio.Event()
expect_idx = 0

def notify(mod):
nonlocal expect_idx
self.assertEqual(mod, expect[expect_idx])
Expand Down Expand Up @@ -144,129 +201,17 @@ def test_pending_priority(self):
late = time() + 100000
early = time() + 1

expect = [
{
"path": [],
"action": "setitem",
"value": {
"repo_msg": None,
"priority": low_priority,
"pipeline": "main",
"due_date": None,
"status": "pending",
"expid": expid_bg,
"flush": False
},
"key": 0
},
{
"path": [],
"action": "setitem",
"value": {
"repo_msg": None,
"priority": high_priority,
"pipeline": "main",
"due_date": late,
"status": "pending",
"expid": expid_empty,
"flush": False
},
"key": 1
},
{
"path": [],
"action": "setitem",
"value": {
"repo_msg": None,
"priority": middle_priority,
"pipeline": "main",
"due_date": early,
"status": "pending",
"expid": expid_empty,
"flush": False
},
"key": 2
},
{
"path": [0],
"action": "setitem",
"value": "preparing",
"key": "status"
},
{
"path": [0],
"action": "setitem",
"value": "prepare_done",
"key": "status"
},
{
"path": [0],
"action": "setitem",
"value": "running",
"key": "status"
},
{
"path": [2],
"action": "setitem",
"value": "preparing",
"key": "status"
},
{
"path": [2],
"action": "setitem",
"value": "prepare_done",
"key": "status"
},
{
"path": [0],
"action": "setitem",
"value": "paused",
"key": "status"
},
{
"path": [2],
"action": "setitem",
"value": "running",
"key": "status"
},
{
"path": [2],
"action": "setitem",
"value": "run_done",
"key": "status"
},
{
"path": [0],
"action": "setitem",
"value": "running",
"key": "status"
},
{
"path": [2],
"action": "setitem",
"value": "analyzing",
"key": "status"
},
{
"path": [2],
"action": "setitem",
"value": "deleting",
"key": "status"
},
{
"path": [],
"action": "delitem",
"key": 2
},
]
monitor = SchedulerMonitor()

done = asyncio.Event()
expect_idx = 0

def notify(mod):
nonlocal expect_idx
self.assertEqual(mod, expect[expect_idx])
expect_idx += 1
if expect_idx >= len(expect):
process_mod(monitor.experiments, mod)
monitor.record()

if monitor.finished:
done.set()

scheduler.notifier.publish = notify

scheduler.start()
Expand All @@ -279,6 +224,15 @@ def notify(mod):
scheduler.notifier.publish = None
loop.run_until_complete(scheduler.stop())

# Assert
self.assertEqual(monitor.get_exp_order("preparing"), [0, 2])
self.assertEqual(monitor.get_out_time(1, "pending"), "never",
"RID 1 has left pending")
basic_flow = ["pending", "preparing", "prepare_done", "running",
"run_done", "analyzing", "deleting"]
self.assertEqual(monitor.get_status_order(2), basic_flow,
"RID 2 did not go through all stage")

def test_pause(self):
loop = self.loop

Expand Down