From: Ondrej Lichtner <olichtne(a)redhat.com>
The new wait_for_condition method is based on the older wait_for_finish,
that passively waited for socket messages and looked for a job_finish
message. The new method instead waits for socket messages for one second
in a loop, on each iteration we call a provided condition method that
MUST return a boolean and stop the wait when the return value is True
(condition matched).
The old wait_for_finish method was renamed to wait_for_job_finish that
now just calls the wait_for_condition method with a condition method
that checks the Job.finished property.
Finally this patch merges the wait_for_result method with the
send_message method, as every rpc message is expected to receive a
result from the remote message call.
Commit also includes some updates to the Machine class replacing calls
the old methods with the new ones.
Signed-off-by: Ondrej Lichtner <olichtne(a)redhat.com>
---
lnst/Controller/Machine.py | 17 +++++++++-------
lnst/Controller/MessageDispatcher.py | 39 +++++++++++++++++++-----------------
2 files changed, 31 insertions(+), 25 deletions(-)
diff --git a/lnst/Controller/Machine.py b/lnst/Controller/Machine.py
index 7730d79..cd14f78 100644
--- a/lnst/Controller/Machine.py
+++ b/lnst/Controller/Machine.py
@@ -189,10 +189,7 @@ class Machine(object):
"args": args,
"kwargs": kwargs}
- self._msg_dispatcher.send_message(self, msg)
- result = self._msg_dispatcher.wait_for_result(self)
-
- return result
+ return self._msg_dispatcher.send_message(self, msg)
def _init_connection(self):
""" Initialize the slave connection
@@ -382,7 +379,11 @@ class Machine(object):
elif timeout == 0:
logging.info("Waiting for Job %d on Host %s." %
(job.id, self._id))
- result = self._msg_dispatcher.wait_for_finish(self, job.id)
+
+ def condition():
+ return job.finished
+
+ self._msg_dispatcher.wait_for_condition(condition)
except MachineError as exc:
logging.error(str(exc))
res = False
@@ -405,8 +406,10 @@ class Machine(object):
logging.info("Waiting for Device creation on Host %s." %
(self._id))
- while len(self._tmp_device_database) > 0:
- result = self._msg_dispatcher.handle_messages()
+ def condition():
+ return len(self._tmp_device_database) <= 0
+
+ self._msg_dispatcher.wait_for_condition(condition)
except MachineError as exc:
logging.error(str(exc))
res = False
diff --git a/lnst/Controller/MessageDispatcher.py b/lnst/Controller/MessageDispatcher.py
index 10b712a..abd3b21 100644
--- a/lnst/Controller/MessageDispatcher.py
+++ b/lnst/Controller/MessageDispatcher.py
@@ -93,18 +93,14 @@ class MessageDispatcher(ConnectionHandler):
def send_message(self, machine, data):
soc = self.get_connection(machine)
-
- if data["type"] == "command":
- data["args"] = remote_device_to_deviceref(data["args"])
- data["kwargs"] =
remote_device_to_deviceref(data["kwargs"])
+ data = remote_device_to_deviceref(data)
if send_data(soc, data) == False:
msg = "Connection error from slave %s" % machine.get_id()
raise ConnectionError(msg)
- def wait_for_result(self, machine):
- wait = True
- while wait:
+ result = None
+ while True:
connected_slaves = self._connection_mapping.keys()
messages = self.check_connections()
@@ -113,12 +109,11 @@ class MessageDispatcher(ConnectionHandler):
for msg in messages:
if msg[1]["type"] == "result" and msg[0] == machine:
- wait = False
- result = msg[1]["result"]
-
- machine = self._machines[machine]
- result = deviceref_to_remote_device(machine,
- result)
+ if result is not None:
+ msg = ("Multiple result messages from the same slave "
+ "'{}'".format(machine.get_id()))
+ raise ConnectionError(msg)
+ result = msg[1]
else:
self._process_message(msg)
@@ -129,21 +124,29 @@ class MessageDispatcher(ConnectionHandler):
" disconnected from the controller."
raise ConnectionError(msg)
- return result
+ if result is not None:
+ return deviceref_to_remote_device(machine, result["result"])
- def wait_for_finish(self, machine, job_id):
+ def wait_for_job_finish(self, job):
+ def condition_check():
+ return job.finished
+ self.wait_for_condition(condition_check)
+ return True
+
+ def wait_for_condition(self, condition_check):
wait = True
while wait:
connected_slaves = self._connection_mapping.keys()
- messages = self.check_connections()
+ messages = self.check_connections(timeout=1)
remaining_slaves = self._connection_mapping.keys()
for msg in messages:
self._process_message(msg)
- if msg[1]["type"] == "job_finished" and msg[0] ==
machine:
- wait = False
+ wait = wait and not condition_check()
+
+ wait = wait and not condition_check()
if connected_slaves != remaining_slaves:
disconnected_slaves = set(connected_slaves) -\
--
2.15.1