Move command completion to command queue

The base frontend had a somewhat complicated mechanism for
tracking outstanding commands in order to process completions
and allow for the callers to wait for command completions.

This required managing multiple lists with their associated
locks. In order to avoid any completions triggering prior to
the frontend being able to insert the command into the
appropriate list.

However, it is much simpler to move all of that logic to the
command queue. The queue already knows about all submitted
commands since frontends submit all commands through it. It
can also managed completions using a single since there is
no possible way for the command to complete prior to it
being de-queued. This makes the entire management and frontend
simpler.

Author: voidtrance

emulator/__init__.py

@@ -46,8 +46,9 @@ def __init__(self, config, frontend, sequential=False):
         self._frontend = frontends.create_frontend(frontend)
         if self._frontend is None:
             raise EmulatorError(f"Failed to create frontend '{frontend}'")
+        command_queue = self._frontend.get_queue()
         mcu = load_mcu(machine.controller)
-        self._controller = mcu(config, self._frontend.complete_command)
+        self._controller = mcu(config, command_queue.complete_command)
         if self._controller is None:
             raise EmulatorError(f"Controller creation failure")
         self._frontend.set_sequential_mode(sequential)

frontends/__init__.py

@@ -17,13 +17,12 @@
 import importlib
 import select
 import os
-import time
 import pickle
 import threading
 import vortex.core.lib.logging as logging
 from vortex.core import ObjectKlass
 from vortex.frontends.lib import create_pty
-from vortex.frontends.queues import CommandQueue, Command
+from vortex.frontends.queues import CommandQueue
 from vortex.frontends.proto import *
 
 class BaseFrontend:
@@ -36,8 +35,6 @@ def __init__(self, queue_size=0):
         self._obj_id_2_name = {x: {} for x in ObjectKlass}
         self._command_completion = {}
         self._command_completion_lock = threading.Lock()
-        self._command_results = {}
-        self._command_results_lock = threading.Lock()
         self._stop = threading.Event()
         self._run_sequential = False
         self._queue = CommandQueue(queue_size)
@@ -65,8 +62,6 @@ def set_controller(self, controller):
     # return the base class method instead of the overriden
     # one.
     def reset(self, *args, **kwargs):
-        with self._command_completion_lock:
-            self._command_results.clear()
         self._queue.clear()
         return self._controller.reset(*args, **kwargs)
 
@@ -150,9 +145,6 @@ def stop(self):
         if self._thread:
             self._stop.set()
             self._thread.join()
-        cmds = list(self._command_completion.keys())
-        for cmd in cmds:
-            self.complete_command(cmd, -1)
 
     def set_sequential_mode(self, mode):
         self._run_sequential = mode
@@ -205,36 +197,13 @@ def queue_command(self, klass, object, cmd, opts, callback=None):
             opts = {_o:_v for _o, _v in (s.split('=') for s in opts.split(','))} if opts else {}
         elif not isinstance(opts, dict):
             return False
-        self.log.debug(f"Submitting command: {self.get_object_name(klass, obj_id)} {cmd_id} {opts}")
-        with self._command_completion_lock:
-            cmd_id, cmd = self._queue.queue_command(obj_id, cmd_id, opts)
-            self.log.debug(f"Command ID:{cmd_id}")
-            self._command_completion[cmd_id] = (cmd, callback)
+        cmd_id, cmd = self._queue.queue_command(obj_id, cmd_id, opts, callback)
         if self._run_sequential:
             self.wait_for_command(cmd_id)
         return cmd_id
 
-    def complete_command(self, id, result, data=None):
-        self.log.debug(f"Completing command: {id} {result}")
-        with self._command_completion_lock:
-            _, callback = self._command_completion.pop(id)
-        if callback is not None:
-            callback(id, result, data)
-        else:
-            with self._command_results_lock:
-                self._command_results[id] = (result, data)
-
     def wait_for_command(self, cmd_set):
-        if isinstance(cmd_set, int):
-            cmd_set = [cmd_set]
-        if not isinstance(cmd_set, (list, tuple, set)):
-            cmd_set = list(cmd_set)
-        cmd_set = set(cmd_set)
-        completed = set()
-        while not self._stop.is_set() and not cmd_set & completed:
-            with self._command_results_lock:
-                completed = set(self._command_results.keys())
-        return [self._command_results.pop(i) for i in cmd_set & completed]
+        return self._queue.wait_for_command(cmd_set)
 
     def respond(self, code, data):
         response = Response(code, data)

frontends/direct/__init__.py

@@ -45,7 +45,7 @@ def _process_command(self, data):
             return
 
         klass = ObjectKlass[klass.upper()]
-        cmd_id = self.queue_command(klass, object, cmd, opts)
+        cmd_id = self.queue_command(klass, object, cmd, opts, self.complete_command)
         if cmd_id is False:
             self.log.error("Failed to queue command")
             super().respond(CommandStatus.FAIL, False)

frontends/gcode/__init__.py

@@ -275,7 +275,7 @@ def M204(self, cmd):
                     self.log.error("Failed to queue command")
                     return CommandStatus.FAIL
                 cmds.append(cmd_id)
-        self.wait_for_commands(cmds)
+        self.wait_for_command(cmds)
         return CommandStatus.SUCCESS
     def M400(self, cmd):
         while self._command_completion:

frontends/klipper/helpers.py

@@ -228,9 +228,9 @@ def __init__(self, frontend, queue, oid, obj_id, klass, name):
                                              self.name, "use_pins",
                                              {"enable": True})
         result = self.frontend.wait_for_command(cmd_id)
-        if not result or result[0][0] != 0:
+        if not result or result[0].id != cmd_id or result[0].result != 0:
             raise ValueError(f"Heater {self.name} pin enable error")
-        self.pin_word = ctypes.cast(result[0][1]["pin_addr"], ctypes.POINTER(ctypes.c_uint8))
+        self.pin_word = ctypes.cast(result[0].data["pin_addr"], ctypes.POINTER(ctypes.c_uint8))
     def _set_pin(self, value):
         self.pin_word.contents.value = int(not (not (Flags.ON & value)))
     def shutdown(self):
@@ -293,9 +293,9 @@ def __init__(self, frontend, queue, oid, obj_id, klass, name):
                                              self.name, "use_pins",
                                              {"enable": True})
         result = self.frontend.wait_for_command(cmd_id)
-        if not result or result[0][0] != 0:
+        if not result or result[0].id != cmd_id or result[0].result != 0:
             raise ValueError(f"Stepper {self.name} pin enable error")
-        self.pin_word = atomics.Atomic(32, var=result[0][1]["pin_addr"])
+        self.pin_word = atomics.Atomic(32, var=result[0].data["pin_addr"])
     def owns_pin(self, pin):
         return pin in self.pins.values()
     def configure_pin(self, oid, pin):
@@ -605,7 +605,7 @@ def reset(self, value):
         result = self.frontend.wait_for_command(cmd_id)
         if not result:
             return -1
-        return result[0][0]
+        return result[0].result
 
 class ButtonsState(enum.IntEnum):
     NONE = 0
@@ -721,8 +721,8 @@ def finish_config(self):
         if cmd_id == -1:
             raise HelperException("Failed to set PWM object id.")
         result = self.frontend.wait_for_command(cmd_id)
-        if not result or result[0][0] != 0:
-            raise HelperException(f"Failed command to set PWM object id ({result[0]})")
+        if not result or result[0].id != cmd_id or result[0].result != 0:
+            raise HelperException(f"Failed command to set PWM object id ({result[0].id})")
         self.pin = DigitalPin(self.frontend, -1, pin_id, ObjectKlass.DIGITAL_PIN, obj_name)
         self.pin.set_max_duration(self.pin_max_duration)
         self.pin.set_initial_value(self.pin_value, self.pin_default)
@@ -756,7 +756,7 @@ def event(self, ticks):
             self.frontend.shutdown("Failed to set PWM duty cycle")
             return 0
         result = self.frontend.wait_for_command(cmd_id)
-        if not result or result[0][0] != 0:
+        if not result or result[0].id != cmd_id or result[0].result != 0:
             self.frontend.shutdown("Failed to set PWM duty cycle.")
             return 0
         if self.move_queue.empty(self.oid):
@@ -796,4 +796,4 @@ def send(self):
                                                  {"index": int(i / self.unit_len),
                                                   "color":  self.data[i:i+self.unit_len]})
             completion = self.frontend.wait_for_command(cmd_id)
-        return completion[0][0]
+        return completion[0].result

frontends/queues.py

@@ -14,39 +14,98 @@
 # You should have received a copy of the GNU General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 import queue
+from threading import Lock
+from collections import namedtuple
+import vortex.core.lib.logging as logging
+
 class Command:
-    def __init__(self, obj_id, cmd_id, opts):
-        self.obj_id, self.cmd_id, self.opts = \
-            obj_id, cmd_id, opts
+    def __init__(self, obj_id, cmd_id, opts, callback):
+        self.obj_id = obj_id
+        self.cmd_id = cmd_id
+        self.opts = opts
+        self.callback = callback
         self.id = id(self)
+
     def __str__(self):
         return f"Command({self.id}, {self.obj_id}:{self.cmd_id})"
-        
+
+QueueCompletion = namedtuple("QueueCompletion", ["command", "result", "data"])
+Completion = namedtuple("Completion", ["id", "result", "data"])
+
+log = logging.getLogger("vortex.queue")
+
 class CommandQueue(queue.Queue):
     def __init__(self, maxsize=0):
         super().__init__(maxsize)
         self.__cmd_count = 0
         self.__max_size = maxsize
+        self.__lock = Lock()
+        self.__cmd_queue = {}
+        self.__comp_queue = {}
+
     def put(self, command):
         if not isinstance(command, Command):
             raise ValueError("'command' is not a Command instance")
-        super().put(command)
+        with self.__lock:
+            super().put(command)
+            self.__cmd_queue[command.id] = command
         self.__cmd_count += 1
         return command.id
+
     def get(self, block=True, timeout=None):
         cmd = super().get(block, timeout)
         self.__cmd_count -= 1
         return cmd
-    def queue_command(self, obj_id, cmd_id, opts):
-        cmd = Command(obj_id, cmd_id, opts)
-        return (self.put(cmd), cmd)
+
+    def queue_command(self, obj_id, cmd_id, opts, callback=None):
+        log.debug(f"Submitting command: {obj_id}: {cmd_id} {opts}")
+        cmd = Command(obj_id, cmd_id, opts, callback)
+        self.put(cmd)
+        log.debug(f"Command queued as {cmd.id}")
+        return cmd.id, cmd
+
+    def complete_command(self, cmd_id, result, data=None):
+        log.debug(f"Command {cmd_id} competed with {result}")
+        with self.__lock:
+            cmd = self.__cmd_queue.pop(cmd_id)
+
+        if cmd.callback:
+            cmd.callback(cmd_id, result, data)
+        else:
+            with self.__lock:
+                self.__comp_queue[cmd_id] = QueueCompletion(cmd, result, data)
+
+    def wait_for_command(self, cmd_ids):
+        if isinstance(cmd_ids, int):
+            cmd_set = [cmd_ids]
+        if not isinstance(cmd_ids, (list, tuple, set)):
+            cmd_set = list(cmd_set)
+        cmd_set = set(cmd_set)
+        comp_set = set()
+        completed = []
+        while not cmd_set & comp_set:
+            with self.__lock:
+                comp_set = set(self.__comp_queue.keys())
+                for cmd_id in (cmd_set & comp_set):
+                    completion = self.__comp_queue.pop(cmd_id)
+                    completed.append(Completion(completion.command.id,
+                                                completion.result,
+                                                completion.data))
+        return completed
+
     def clear(self):
         self.shutdown(True)
+        with self.__lock:
+            for cmd in self.get():
+                if cmd.callback:
+                    cmd.callback(cmd.id, -1, None)
         self.__cmd_count = 0
         self.is_shutdown = False
+
     @property
     def size(self):
         return self.__cmd_count
+
     @property
     def max_size(self):
         return self.__max_size