m-labs · May 17, 2015 · May 17, 2015
diff --git a/artiq/frontend/artiq_client.py b/artiq/frontend/artiq_client.py
@@ -33,27 +33,23 @@ def get_argparser():
 
     parser_add = subparsers.add_parser("submit", help="submit an experiment")
     parser_add.add_argument(
-        "-T", "--timed", default=None, type=str,
-        help="run the experiment in timed mode. "
-             "argument specifies the time of the first run, "
-             "use 'now' to run immediately")
+        "-t", "--timed", default=None, type=str,
+        help="set a due date for the experiment")
+    parser_add.add_argument("-p", "--pipeline", default="main", type=str,
+                            help="pipeline to run the experiment in "
+                                 "(default: %(default)s)")
     parser_add.add_argument("-e", "--experiment", default=None,
                             help="experiment to run")
-    parser_add.add_argument("--rtr-group", default=None, type=str,
-                            help="real-time result group "
-                                 "(defaults to filename)")
     parser_add.add_argument("file",
                             help="file containing the experiment to run")
     parser_add.add_argument("arguments", nargs="*",
                             help="run arguments")
 
-    parser_cancel = subparsers.add_parser("cancel",
-                                          help="cancel an experiment")
-    parser_cancel.add_argument("-T", "--timed", default=False,
-                               action="store_true",
-                               help="cancel a timed experiment")
-    parser_cancel.add_argument("rid", type=int,
-                               help="run identifier (RID/TRID)")
+    parser_delete = subparsers.add_parser("delete",
+                                          help="delete an experiment "
+                                               "from the schedule")
+    parser_delete.add_argument("rid", type=int,
+                               help="run identifier (RID)")
 
     parser_set_device = subparsers.add_parser(
         "set-device", help="add or modify a device")
@@ -79,7 +75,7 @@ def get_argparser():
         "show", help="show schedule, devices or parameters")
     parser_show.add_argument(
         "what",
-        help="select object to show: queue/timed/devices/parameters")
+        help="select object to show: schedule/devices/parameters")
 
     return parser
 
@@ -99,30 +95,21 @@ def _action_submit(remote, args):
         print("Failed to parse run arguments")
         sys.exit(1)
 
-    run_params = {
+    expid = {
         "file": args.file,
         "experiment": args.experiment,
         "arguments": arguments,
-        "rtr_group": args.rtr_group if args.rtr_group is not None \
-                        else args.file
     }
     if args.timed is None:
-        rid = remote.run_queued(run_params)
-        print("RID: {}".format(rid))
+        due_date = None
     else:
-        if args.timed == "now":
-            next_time = None
-        else:
-            next_time = time.mktime(parse_date(args.timed).timetuple())
-        trid = remote.run_timed(run_params, next_time)
-        print("TRID: {}".format(trid))
+        due_date = time.mktime(parse_date(args.timed).timetuple())
+    rid = remote.submit(args.pipeline, expid, due_date)
+    print("RID: {}".format(rid))
 
 
-def _action_cancel(remote, args):
-    if args.timed:
-        remote.cancel_timed(args.rid)
-    else:
-        remote.cancel_queued(args.rid)
+def _action_delete(remote, args):
+    remote.delete(args.rid)
 
 
 def _action_set_device(remote, args):
@@ -141,41 +128,30 @@ def _action_del_parameter(remote, args):
     remote.delete(args.name)
 
 
-def _show_queue(queue):
+def _show_schedule(schedule):
     clear_screen()
-    if queue:
-        table = PrettyTable(["RID", "File", "Experiment", "Arguments"])
-        for rid, run_params in queue:
-            row = [rid, run_params["file"]]
-            if run_params["experiment"] is None:
+    if schedule:
+        l = sorted(schedule.items(),
+                   key=lambda x: (x[1]["due_date"] or 0, x[0]))
+        table = PrettyTable(["RID", "Pipeline", "    Status    ", "Due date",
+                             "File", "Experiment", "Arguments"])
+        for rid, v in l:
+            row = [rid, v["pipeline"], v["status"]]
+            if v["due_date"] is None:
                 row.append("")
             else:
-                row.append(run_params["experiment"])
-            row.append(format_arguments(run_params["arguments"]))
-            table.add_row(row)
-        print(table)
-    else:
-        print("Queue is empty")
-
-
-def _show_timed(timed):
-    clear_screen()
-    if timed:
-        table = PrettyTable(["Next run", "TRID", "File", "Experiment",
-                             "Arguments"])
-        sp = sorted(timed.items(), key=lambda x: (x[1][0], x[0]))
-        for trid, (next_run, run_params) in sp:
-            row = [time.strftime("%m/%d %H:%M:%S", time.localtime(next_run)),
-                   trid, run_params["file"]]
-            if run_params["experiment"] is None:
+                row.append(time.strftime("%m/%d %H:%M:%S",
+                           time.localtime(v["due_date"])))
+            row.append(v["expid"]["file"])
+            if v["expid"]["experiment"] is None:
                 row.append("")
             else:
-                row.append(run_params["experiment"])
-            row.append(format_arguments(run_params["arguments"]))
+                row.append(v["expid"]["experiment"])
+            row.append(format_arguments(v["expid"]["arguments"]))
             table.add_row(row)
         print(table)
     else:
-        print("No timed schedule")
+        print("Schedule is empty")
 
 
 def _show_devices(devices):
@@ -211,16 +187,6 @@ def _run_subscriber(host, port, subscriber):
         loop.close()
 
 
-def _show_list(args, notifier_name, display_fun):
-    l = []
-    def init_l(x):
-        l[:] = x
-        return l
-    subscriber = Subscriber(notifier_name, init_l,
-                            lambda mod: display_fun(l))
-    _run_subscriber(args.server, args.port, subscriber)
-
-
 def _show_dict(args, notifier_name, display_fun):
     d = dict()
     def init_d(x):
@@ -236,10 +202,8 @@ def main():
     args = get_argparser().parse_args()
     action = args.action.replace("-", "_")
     if action == "show":
-        if args.what == "queue":
-            _show_list(args, "queue", _show_queue)
-        elif args.what == "timed":
-            _show_dict(args, "timed", _show_timed)
+        if args.what == "schedule":
+            _show_dict(args, "schedule", _show_schedule)
         elif args.what == "devices":
             _show_dict(args, "devices", _show_devices)
         elif args.what == "parameters":
@@ -251,7 +215,7 @@ def main():
         port = 3251 if args.port is None else args.port
         target_name = {
             "submit": "master_schedule",
-            "cancel": "master_schedule",
+            "delete": "master_schedule",
             "set_device": "master_ddb",
             "del_device": "master_ddb",
             "set_parameter": "master_pdb",

diff --git a/artiq/frontend/artiq_master.py b/artiq/frontend/artiq_master.py
@@ -48,20 +48,15 @@ def main():
         loop = asyncio.get_event_loop()
     atexit.register(lambda: loop.close())
 
-    def run_cb(rid, run_params):
-        rtr.current_group = run_params["rtr_group"]
-    scheduler = Scheduler(run_cb, get_last_rid() + 1)
-    scheduler.worker_handlers = {
+    worker_handlers = {
         "req_device": ddb.request,
         "req_parameter": pdb.request,
         "set_parameter": pdb.set,
         "init_rt_results": rtr.init,
         "update_rt_results": rtr.update,
-        "scheduler_run_queued": scheduler.run_queued,
-        "scheduler_cancel_queued": scheduler.cancel_queued,
-        "scheduler_run_timed": scheduler.run_timed,
-        "scheduler_cancel_timed": scheduler.cancel_timed,
     }
+    scheduler = Scheduler(get_last_rid() + 1, worker_handlers)
+    worker_handlers["scheduler_submit"] = scheduler.submit
     scheduler.start()
     atexit.register(lambda: loop.run_until_complete(scheduler.stop()))
 
@@ -76,8 +71,7 @@ def run_cb(rid, run_params):
     atexit.register(lambda: loop.run_until_complete(server_control.stop()))
 
     server_notify = Publisher({
-        "queue": scheduler.queue,
-        "timed": scheduler.timed,
+        "schedule": scheduler.notifier,
         "devices": ddb.data,
         "parameters": pdb.data,
         "parameters_simplehist": simplephist.history,

diff --git a/artiq/gui/explorer.py b/artiq/gui/explorer.py
@@ -144,10 +144,9 @@ def run(self, widget):
                 arguments = {}
             else:
                 arguments = self.controls.get_arguments()
-            run_params = {
+            expid = {
                 "file": data["file"],
                 "experiment": data["experiment"],
-                "arguments": arguments,
-                "rtr_group": data["file"]
+                "arguments": arguments
             }
-            asyncio.Task(self.schedule_ctl.run_queued(run_params))
+            asyncio.async(self.schedule_ctl.submit("main", expid, None))
diff --git a/artiq/gui/scheduler.py b/artiq/gui/scheduler.py
@@ -3,37 +3,29 @@
 
 from gi.repository import Gtk
 
-from artiq.gui.tools import Window, ListSyncer, DictSyncer
+from artiq.gui.tools import Window, DictSyncer
 from artiq.protocols.sync_struct import Subscriber
 from artiq.tools import format_arguments
 
 
-class _QueueStoreSyncer(ListSyncer):
-    def convert(self, x):
-        rid, run_params = x
-        row = [rid, run_params["file"]]
-        if run_params["experiment"] is None:
-            row.append("")
-        else:
-            row.append(run_params["experiment"])
-        row.append(format_arguments(run_params["arguments"]))
-        return row
-
-
-class _TimedStoreSyncer(DictSyncer):
+class _ScheduleStoreSyncer(DictSyncer):
     def order_key(self, kv_pair):
-        # order by next run time, and then by TRID
-        return (kv_pair[1][0], kv_pair[0])
+        # order by due date, and then by RID
+        return (kv_pair[1]["due_date"] or 0, kv_pair[0])
 
-    def convert(self, trid, x):
-        next_run, run_params = x
-        row = [time.strftime("%m/%d %H:%M:%S", time.localtime(next_run)),
-               trid, run_params["file"]]
-        if run_params["experiment"] is None:
+    def convert(self, rid, v):
+        row = [rid, v["pipeline"], v["status"]]
+        if v["due_date"] is None:
             row.append("")
         else:
-            row.append(run_params["experiment"])
-        row.append(format_arguments(run_params["arguments"]))
+            row.append(time.strftime("%m/%d %H:%M:%S",
+                       time.localtime(v["due_date"])))
+        row.append(v["expid"]["file"])
+        if v["expid"]["experiment"] is None:
+            row.append("")
+        else:
+            row.append(v["expid"]["experiment"])
+        row.append(format_arguments(v["expid"]["arguments"]))
         return row
 
 
@@ -43,93 +35,41 @@ def __init__(self, schedule_ctl, **kwargs):
 
         Window.__init__(self,
                         title="Scheduler",
-                        default_size=(720, 570),
+                        default_size=(950, 570),
                         **kwargs)
 
         topvbox = Gtk.VBox(spacing=6)
         self.add(topvbox)
 
-        hbox = Gtk.HBox(spacing=6)
-        enable = Gtk.Switch(active=True)
-        label = Gtk.Label("Run experiments")
-        hbox.pack_start(label, False, False, 0)
-        hbox.pack_start(enable, False, False, 0)
-        topvbox.pack_start(hbox, False, False, 0)
-
-        notebook = Gtk.Notebook()
-        topvbox.pack_start(notebook, True, True, 0)
-
-        self.queue_store = Gtk.ListStore(int, str, str, str)
-        self.queue_tree = Gtk.TreeView(self.queue_store)
-        for i, title in enumerate(["RID", "File", "Experiment", "Arguments"]):
+        self.schedule_store = Gtk.ListStore(int, str, str, str, str, str, str)
+        self.schedule_tree = Gtk.TreeView(self.schedule_store)
+        for i, title in enumerate(["RID", "Pipeline", "Status", "Due date",
+                                   "File", "Experiment", "Arguments"]):
             renderer = Gtk.CellRendererText()
             column = Gtk.TreeViewColumn(title, renderer, text=i)
-            self.queue_tree.append_column(column)
+            self.schedule_tree.append_column(column)
         scroll = Gtk.ScrolledWindow()
-        scroll.add(self.queue_tree)
-        vbox = Gtk.VBox(spacing=6)
-        vbox.pack_start(scroll, True, True, 0)
-        hbox = Gtk.HBox(spacing=6)
-        button = Gtk.Button("Find")
-        hbox.pack_start(button, True, True, 0)
-        button = Gtk.Button("Move up")
-        hbox.pack_start(button, True, True, 0)
-        button = Gtk.Button("Move down")
-        hbox.pack_start(button, True, True, 0)
-        button = Gtk.Button("Remove")
-        button.connect("clicked", self.remove_queued)
-        hbox.pack_start(button, True, True, 0)
-        vbox.pack_start(hbox, False, False, 0)
-        vbox.set_border_width(6)
-        notebook.insert_page(vbox, Gtk.Label("Queue"), -1)
-
-        self.timed_store = Gtk.ListStore(str, int, str, str, str)
-        self.timed_tree = Gtk.TreeView(self.timed_store)
-        for i, title in enumerate(["Next run", "TRID", "File", "Experiment",
-                                   "Arguments"]):
-            renderer = Gtk.CellRendererText()
-            column = Gtk.TreeViewColumn(title, renderer, text=i)
-            self.timed_tree.append_column(column)
-        scroll = Gtk.ScrolledWindow()
-        scroll.add(self.timed_tree)
-        vbox = Gtk.VBox(spacing=6)
-        vbox.pack_start(scroll, True, True, 0)
-        button = Gtk.Button("Remove")
-        button.connect("clicked", self.remove_timed)
-        vbox.pack_start(button, False, False, 0)
-        vbox.set_border_width(6)
-        notebook.insert_page(vbox, Gtk.Label("Timed schedule"), -1)
-
-    def remove_queued(self, widget):
-        store, selected = self.queue_tree.get_selection().get_selected()
+        scroll.add(self.schedule_tree)
+        topvbox.pack_start(scroll, True, True, 0)
+        button = Gtk.Button("Delete")
+        button.connect("clicked", self.delete)
+        topvbox.pack_start(button, False, False, 0)
+        topvbox.set_border_width(6)
+
+    def delete(self, widget):
+        store, selected = self.schedule_tree.get_selection().get_selected()
         if selected is not None:
             rid = store[selected][0]
-            asyncio.Task(self.schedule_ctl.cancel_queued(rid))
-
-    def remove_timed(self, widget):
-        store, selected = self.timed_tree.get_selection().get_selected()
-        if selected is not None:
-            trid = store[selected][1]
-            asyncio.Task(self.schedule_ctl.cancel_timed(trid))
+            asyncio.async(self.schedule_ctl.delete(rid))
 
     @asyncio.coroutine
     def sub_connect(self, host, port):
-        self.queue_subscriber = Subscriber("queue", self.init_queue_store)
-        yield from self.queue_subscriber.connect(host, port)
-        try:
-            self.timed_subscriber = Subscriber("timed", self.init_timed_store)
-            yield from self.timed_subscriber.connect(host, port)
-        except:
-            yield from self.queue_subscriber.close()
-            raise
+        self.schedule_subscriber = Subscriber("schedule", self.init_schedule_store)
+        yield from self.schedule_subscriber.connect(host, port)
 
     @asyncio.coroutine
     def sub_close(self):
-        yield from self.timed_subscriber.close()
-        yield from self.queue_subscriber.close()
-
-    def init_queue_store(self, init):
-        return _QueueStoreSyncer(self.queue_store, init)
+        yield from self.schedule_subscriber.close()
 
-    def init_timed_store(self, init):
-        return _TimedStoreSyncer(self.timed_store, init)
+    def init_schedule_store(self, init):
+        return _ScheduleStoreSyncer(self.schedule_store, init)