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: b117b9320d0d
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: 090a7e587155
Choose a head ref
  • 2 commits
  • 19 files changed
  • 1 contributor

Commits on Oct 3, 2015

  1. use Python 3.5 coroutines

    @sbourdeauducq
    sbourdeauducq committed Oct 3, 2015
    Copy the full SHA
    f552d62 View commit details

  2. test/sync_struct: cleanup

    @sbourdeauducq
    sbourdeauducq committed Oct 3, 2015
    Copy the full SHA
    090a7e5 View commit details
85 changes: 38 additions & 47 deletions artiq/frontend/artiq_ctlmgr.py
View file
Original file line number Diff line number Diff line change
@@ -56,94 +56,88 @@ def __init__(self, name, ddb_entry):
self.process = None
self.launch_task = asyncio.Task(self.launcher())

@asyncio.coroutine
def end(self):
async def end(self):
self.launch_task.cancel()
yield from asyncio.wait_for(self.launch_task, None)
await asyncio.wait_for(self.launch_task, None)

@asyncio.coroutine
def _call_controller(self, method):
async def _call_controller(self, method):
remote = AsyncioClient()
yield from remote.connect_rpc(self.host, self.port, None)
await remote.connect_rpc(self.host, self.port, None)
try:
targets, _ = remote.get_rpc_id()
remote.select_rpc_target(targets[0])
r = yield from getattr(remote, method)()
r = await getattr(remote, method)()
finally:
remote.close_rpc()
return r

@asyncio.coroutine
def _ping(self):
async def _ping(self):
try:
ok = yield from asyncio.wait_for(self._call_controller("ping"),
self.ping_timeout)
ok = await asyncio.wait_for(self._call_controller("ping"),
self.ping_timeout)
if ok:
self.retry_timer_cur = self.retry_timer
return ok
except:
return False

@asyncio.coroutine
def _wait_and_ping(self):
async def _wait_and_ping(self):
while True:
try:
yield from asyncio.wait_for(self.process.wait(),
self.ping_timer)
await asyncio.wait_for(self.process.wait(),
self.ping_timer)
except asyncio.TimeoutError:
logger.debug("pinging controller %s", self.name)
ok = yield from self._ping()
ok = await self._ping()
if not ok:
logger.warning("Controller %s ping failed", self.name)
yield from self._terminate()
await self._terminate()
return
else:
break

@asyncio.coroutine
def launcher(self):
async def launcher(self):
try:
while True:
logger.info("Starting controller %s with command: %s",
self.name, self.command)
try:
self.process = yield from asyncio.create_subprocess_exec(
self.process = await asyncio.create_subprocess_exec(
*shlex.split(self.command))
yield from self._wait_and_ping()
await self._wait_and_ping()
except FileNotFoundError:
logger.warning("Controller %s failed to start", self.name)
else:
logger.warning("Controller %s exited", self.name)
logger.warning("Restarting in %.1f seconds",
self.retry_timer_cur)
try:
yield from asyncio.wait_for(self.retry_now.wait(),
self.retry_timer_cur)
await asyncio.wait_for(self.retry_now.wait(),
self.retry_timer_cur)
except asyncio.TimeoutError:
pass
self.retry_timer_cur *= self.retry_timer_backoff
except asyncio.CancelledError:
yield from self._terminate()
await self._terminate()

@asyncio.coroutine
def _terminate(self):
async def _terminate(self):
logger.info("Terminating controller %s", self.name)
if self.process is not None and self.process.returncode is None:
try:
yield from asyncio.wait_for(self._call_controller("terminate"),
self.term_timeout)
await asyncio.wait_for(self._call_controller("terminate"),
self.term_timeout)
except:
logger.warning("Controller %s did not respond to terminate "
"command, killing", self.name)
self.process.kill()
try:
yield from asyncio.wait_for(self.process.wait(),
self.term_timeout)
await asyncio.wait_for(self.process.wait(),
self.term_timeout)
except:
logger.warning("Controller %s failed to exit, killing",
self.name)
self.process.kill()
yield from self.process.wait()
await self.process.wait()
logger.debug("Controller %s terminated", self.name)


@@ -163,17 +157,16 @@ def __init__(self):
self.active = dict()
self.process_task = asyncio.Task(self._process())

@asyncio.coroutine
def _process(self):
async def _process(self):
while True:
action, param = yield from self.queue.get()
action, param = await self.queue.get()
if action == "set":
k, ddb_entry = param
if k in self.active:
yield from self.active[k].end()
await self.active[k].end()
self.active[k] = Controller(k, ddb_entry)
elif action == "del":
yield from self.active[param].end()
await self.active[param].end()
del self.active[param]
else:
raise ValueError
@@ -196,11 +189,10 @@ def delete_all(self):
for name in set(self.active_or_queued):
del self[name]

@asyncio.coroutine
def shutdown(self):
async def shutdown(self):
self.process_task.cancel()
for c in self.active.values():
yield from c.end()
await c.end()


class ControllerDB:
@@ -225,8 +217,7 @@ def __init__(self, server, port, retry_master):
self.retry_master = retry_master
self.controller_db = ControllerDB()

@asyncio.coroutine
def _do(self):
async def _do(self):
try:
subscriber = Subscriber("devices",
self.controller_db.sync_struct_init)
@@ -236,24 +227,24 @@ def set_host_filter():
s = subscriber.writer.get_extra_info("socket")
localhost = s.getsockname()[0]
self.controller_db.set_host_filter(localhost)
yield from subscriber.connect(self.server, self.port,
set_host_filter)
await subscriber.connect(self.server, self.port,
set_host_filter)
try:
yield from asyncio.wait_for(subscriber.receive_task, None)
await asyncio.wait_for(subscriber.receive_task, None)
finally:
yield from subscriber.close()
await subscriber.close()
except (ConnectionAbortedError, ConnectionError,
ConnectionRefusedError, ConnectionResetError) as e:
logger.warning("Connection to master failed (%s: %s)",
e.__class__.__name__, str(e))
else:
logger.warning("Connection to master lost")
logger.warning("Retrying in %.1f seconds", self.retry_master)
yield from asyncio.sleep(self.retry_master)
await asyncio.sleep(self.retry_master)
except asyncio.CancelledError:
pass
finally:
yield from self.controller_db.current_controllers.shutdown()
await self.controller_db.current_controllers.shutdown()

def retry_now(self, k):
"""If a controller is disabled and pending retry, perform that retry
20 changes: 9 additions & 11 deletions artiq/frontend/artiq_influxdb.py
View file
Original file line number Diff line number Diff line change
@@ -96,24 +96,23 @@ def update(self, k, v):
logger.warning("failed to update parameter '%s': "
"too many pending updates", k)

@asyncio.coroutine
def _do(self):
async def _do(self):
while True:
k, v = yield from self._queue.get()
k, v = await self._queue.get()
url = self.base_url + "/write"
params = {"u": self.user, "p": self.password, "db": self.database,
"consistency": "any", "precision": "n"}
fmt_ty, fmt_v = format_influxdb(v)
data = "{},parameter={} {}={}".format(self.table, k, fmt_ty, fmt_v)
try:
response = yield from aiohttp.request(
response = await aiohttp.request(
"POST", url, params=params, data=data)
except:
logger.warning("got exception trying to update '%s'",
k, exc_info=True)
else:
if response.status not in (200, 204):
content = (yield from response.content.read()).decode()
content = (await response.content.read()).decode()
if content:
content = content[:-1] # drop \n
logger.warning("got HTTP status %d "
@@ -144,26 +143,25 @@ def __init__(self, server, port, retry, filter_function, writer):
self.filter_function = filter_function
self.writer = writer

@asyncio.coroutine
def _do(self):
async def _do(self):
subscriber = Subscriber(
"parameters",
partial(Parameters, self.filter_function, self.writer))
while True:
try:
yield from subscriber.connect(self.server, self.port)
await subscriber.connect(self.server, self.port)
try:
yield from asyncio.wait_for(subscriber.receive_task, None)
await asyncio.wait_for(subscriber.receive_task, None)
finally:
yield from subscriber.close()
await subscriber.close()
except (ConnectionAbortedError, ConnectionError,
ConnectionRefusedError, ConnectionResetError) as e:
logger.warning("Connection to master failed (%s: %s)",
e.__class__.__name__, str(e))
else:
logger.warning("Connection to master lost")
logger.warning("Retrying in %.1f seconds", self.retry)
yield from asyncio.sleep(self.retry)
await asyncio.sleep(self.retry)


class Filter:
17 changes: 7 additions & 10 deletions artiq/gui/explorer.py
View file
Original file line number Diff line number Diff line change
@@ -300,32 +300,29 @@ def restore_state(self, state):
def enable_duedate(self):
self.datetime_en.setChecked(True)

@asyncio.coroutine
def sub_connect(self, host, port):
async def sub_connect(self, host, port):
self.explist_subscriber = Subscriber("explist",
self.init_explist_model)
yield from self.explist_subscriber.connect(host, port)
await self.explist_subscriber.connect(host, port)

@asyncio.coroutine
def sub_close(self):
yield from self.explist_subscriber.close()
async def sub_close(self):
await self.explist_subscriber.close()

def init_explist_model(self, init):
self.explist_model = _ExplistModel(self, self.el, init)
self.el.setModel(self.explist_model)
return self.explist_model

@asyncio.coroutine
def submit(self, pipeline_name, file, class_name, arguments,
async def submit(self, pipeline_name, file, class_name, arguments,
priority, due_date, flush):
expid = {
"repo_rev": None,
"file": file,
"class_name": class_name,
"arguments": arguments,
}
rid = yield from self.schedule_ctl.submit(pipeline_name, expid,
priority, due_date, flush)
rid = await self.schedule_ctl.submit(pipeline_name, expid,
priority, due_date, flush)
self.status_bar.showMessage("Submitted RID {}".format(rid))

def submit_clicked(self):
10 changes: 4 additions & 6 deletions artiq/gui/log.py
View file
Original file line number Diff line number Diff line change
@@ -41,14 +41,12 @@ def __init__(self):
self.addWidget(self.log)
self.scroll_at_bottom = False

@asyncio.coroutine
def sub_connect(self, host, port):
async def sub_connect(self, host, port):
self.subscriber = Subscriber("log", self.init_log_model)
yield from self.subscriber.connect(host, port)
await self.subscriber.connect(host, port)

@asyncio.coroutine
def sub_close(self):
yield from self.subscriber.close()
async def sub_close(self):
await self.subscriber.close()

def rows_inserted_before(self):
scrollbar = self.log.verticalScrollBar()
21 changes: 9 additions & 12 deletions artiq/gui/moninj.py
View file
Original file line number Diff line number Diff line change
@@ -232,26 +232,24 @@ def __init__(self):
self.dm = _DeviceManager(self.send_to_device, dict())
self.transport = None

@asyncio.coroutine
def start(self, server, port):
async def start(self, server, port):
loop = asyncio.get_event_loop()
yield from loop.create_datagram_endpoint(lambda: self,
await loop.create_datagram_endpoint(lambda: self,
family=socket.AF_INET)
try:
yield from self.subscriber.connect(server, port)
await self.subscriber.connect(server, port)
try:
TaskObject.start(self)
except:
yield from self.subscriber.close()
await self.subscriber.close()
raise
except:
self.transport.close()
raise

@asyncio.coroutine
def stop(self):
yield from TaskObject.stop(self)
yield from self.subscriber.close()
async def stop(self):
await TaskObject.stop(self)
await self.subscriber.close()
if self.transport is not None:
self.transport.close()
self.transport = None
@@ -295,10 +293,9 @@ def send_to_device(self, data):
else:
self.transport.sendto(data, (ca, 3250))

@asyncio.coroutine
def _do(self):
async def _do(self):
while True:
yield from asyncio.sleep(0.2)
await asyncio.sleep(0.2)
# MONINJ_REQ_MONITOR
self.send_to_device(b"\x01")

10 changes: 4 additions & 6 deletions artiq/gui/parameters.py
View file
Original file line number Diff line number Diff line change
@@ -59,14 +59,12 @@ def _search_parameters(self):
else:
self.table.hideRow(row)

@asyncio.coroutine
def sub_connect(self, host, port):
async def sub_connect(self, host, port):
self.subscriber = Subscriber("parameters", self.init_parameters_model)
yield from self.subscriber.connect(host, port)
await self.subscriber.connect(host, port)

@asyncio.coroutine
def sub_close(self):
yield from self.subscriber.close()
async def sub_close(self):
await self.subscriber.close()

def init_parameters_model(self, init):
self.table_model = ParametersModel(self.table, init)
Loading