Skip to content
Permalink

Comparing changes

Choose two branches to see what’s changed or to start a new pull request. If you need to, you can also or learn more about diff comparisons.

Open a pull request

Create a new pull request by comparing changes across two branches. If you need to, you can also . Learn more about diff comparisons here.
base repository: m-labs/artiq
Failed to load repositories. Confirm that selected base ref is valid, then try again.
Loading
base: 4e35a247d1fa
Choose a base ref
...
head repository: m-labs/artiq
Failed to load repositories. Confirm that selected head ref is valid, then try again.
Loading
compare: ec328cf5e18a
Choose a head ref
  • 3 commits
  • 4 files changed
  • 1 contributor

Commits on Oct 28, 2015

  1. master: simplify worker/parent RPC

    @sbourdeauducq
    sbourdeauducq committed Oct 28, 2015
    Copy the full SHA
    1ada15a View commit details

  2. master/scheduler: reduce logging severity of worker exception backtra… 

    …ces to debug
    @sbourdeauducq
    sbourdeauducq committed Oct 28, 2015
    Copy the full SHA
    828b48a View commit details

  3. master/worker_db: reduce exception backtrace clutter of get_dataset. C… 

    …loses #168
    @sbourdeauducq
    sbourdeauducq committed Oct 28, 2015
    Copy the full SHA
    ec328cf View commit details
Showing with 29 additions and 34 deletions.
  1. +6 −6 artiq/master/scheduler.py
  2. +2 −3 artiq/master/worker.py
  3. +2 −1 artiq/master/worker_db.py
  4. +19 −24 artiq/master/worker_impl.py
12 changes: 6 additions & 6 deletions artiq/master/scheduler.py
View file
Original file line number Diff line number Diff line change
@@ -229,8 +229,8 @@ async def _do(self):
await run.prepare()
except:
logger.error("got worker exception in prepare stage, "
"deleting RID %d",
run.rid, exc_info=True)
"deleting RID %d", run.rid)
logger.debug("worker exception details", exc_info=True)
self.delete_cb(run.rid)
else:
run.status = RunStatus.prepare_done
@@ -279,8 +279,8 @@ async def _do(self):
completed = await run.run()
except:
logger.error("got worker exception in run stage, "
"deleting RID %d",
run.rid, exc_info=True)
"deleting RID %d", run.rid)
logger.debug("worker exception details", exc_info=True)
self.delete_cb(run.rid)
else:
if completed:
@@ -317,8 +317,8 @@ async def _do(self):
await run.write_results()
except:
logger.error("got worker exception in analyze stage, "
"deleting RID %d",
run.rid, exc_info=True)
"deleting RID %d", run.rid)
logger.debug("worker exception details", exc_info=True)
self.delete_cb(run.rid)
else:
self.delete_cb(run.rid)
5 changes: 2 additions & 3 deletions artiq/master/worker.py
View file
Original file line number Diff line number Diff line change
@@ -160,8 +160,7 @@ async def _handle_worker_requests(self):
return True
elif action == "pause":
return False
del obj["action"]
if action == "create_watchdog":
elif action == "create_watchdog":
func = self.create_watchdog
elif action == "delete_watchdog":
func = self.delete_watchdog
@@ -172,7 +171,7 @@ async def _handle_worker_requests(self):
if getattr(func, "worker_pass_rid", False):
func = partial(func, self.rid)
try:
data = func(**obj)
data = func(*obj["args"], **obj["kwargs"])
reply = {"status": "ok", "data": data}
except:
reply = {"status": "failed",
3 changes: 2 additions & 1 deletion artiq/master/worker_db.py
View file
Original file line number Diff line number Diff line change
@@ -191,7 +191,8 @@ def get(self, key):
try:
return self.local[key]
except KeyError:
return self.ddb.get(key)
pass
return self.ddb.get(key)

def write_hdf5(self, f):
result_dict_to_hdf5(f, self.local)
43 changes: 19 additions & 24 deletions artiq/master/worker_impl.py
View file
Original file line number Diff line number Diff line change
@@ -26,12 +26,9 @@ class ParentActionError(Exception):
pass


def make_parent_action(action, argnames, exception=ParentActionError):
argnames = argnames.split()
def parent_action(*args):
request = {"action": action}
for argname, arg in zip(argnames, args):
request[argname] = arg
def make_parent_action(action, exception=ParentActionError):
def parent_action(*args, **kwargs):
request = {"action": action, "args": args, "kwargs": kwargs}
put_object(request)
reply = get_object()
if "action" in reply:
@@ -50,7 +47,7 @@ class LogForwarder:
def __init__(self):
self.buffer = ""

to_parent = staticmethod(make_parent_action("log", "message"))
to_parent = staticmethod(make_parent_action("log"))

def write(self, data):
self.buffer += data
@@ -64,18 +61,18 @@ def flush(self):


class ParentDeviceDB:
get_device_db = make_parent_action("get_device_db", "")
get = make_parent_action("get_device", "key", KeyError)
get_device_db = make_parent_action("get_device_db")
get = make_parent_action("get_device", KeyError)


class ParentDatasetDB:
get = make_parent_action("get_dataset", "key", KeyError)
update = make_parent_action("update_dataset", "mod")
get = make_parent_action("get_dataset", KeyError)
update = make_parent_action("update_dataset")


class Watchdog:
_create = make_parent_action("create_watchdog", "t")
_delete = make_parent_action("delete_watchdog", "wid")
_create = make_parent_action("create_watchdog")
_delete = make_parent_action("delete_watchdog")

def __init__(self, t):
self.t = t
@@ -91,15 +88,14 @@ def __exit__(self, type, value, traceback):


class Scheduler:
pause_noexc = staticmethod(make_parent_action("pause", ""))
pause_noexc = staticmethod(make_parent_action("pause"))

def pause(self):
if self.pause_noexc():
raise TerminationRequested

submit = staticmethod(make_parent_action("scheduler_submit",
"pipeline_name expid priority due_date flush"))
cancel = staticmethod(make_parent_action("scheduler_cancel", "rid"))
submit = staticmethod(make_parent_action("scheduler_submit"))
cancel = staticmethod(make_parent_action("scheduler_cancel"))

def set_run_info(self, pipeline_name, expid, priority):
self.pipeline_name = pipeline_name
@@ -120,22 +116,21 @@ def get_exp(file, class_name):
return getattr(module, class_name)


register_experiment = make_parent_action("register_experiment",
"class_name name arguments")
register_experiment = make_parent_action("register_experiment")


class ExamineDeviceMgr:
get_device_db = make_parent_action("get_device_db", "")
get_device_db = make_parent_action("get_device_db")

def get(self, name):
def get(name):
return None


class DummyDatasetMgr:
def set(self, key, value, broadcast=False, persist=False, save=True):
def set(key, value, broadcast=False, persist=False, save=True):
return None

def get(self, key):
def get(key):
pass


@@ -213,7 +208,7 @@ def main():
f.close()
put_object({"action": "completed"})
elif action == "examine":
examine(ExamineDeviceMgr(), DummyDatasetMgr(), obj["file"])
examine(ExamineDeviceMgr, DummyDatasetMgr, obj["file"])
put_object({"action": "completed"})
elif action == "terminate":
break