From 4a35f93b8c0c9a04b6720fbaf3ee53c4c6b761a1 Mon Sep 17 00:00:00 2001 From: Gareth Farrington Date: Fri, 26 Jun 2026 14:08:27 -0700 Subject: [PATCH 1/8] Create exceptions.py Move CommandError and WaitInterruption to a file that can be imported anywhere in the codebase without creating circular dependencies Signed-off-by: Gareth Farrington --- klippy/exceptions.py | 13 +++++++++++++ klippy/gcode.py | 26 ++++++++++++++------------ klippy/printer.py | 15 ++++++++------- 3 files changed, 35 insertions(+), 19 deletions(-) create mode 100644 klippy/exceptions.py diff --git a/klippy/exceptions.py b/klippy/exceptions.py new file mode 100644 index 0000000000..7b43ab2a86 --- /dev/null +++ b/klippy/exceptions.py @@ -0,0 +1,13 @@ +# Exceptions that are broadly used in the klippy codebase +# +# Copyright (C) 2026 Gareth Farrington +# +# This file may be distributed under the terms of the GNU GPLv3 license. + + +class CommandError(Exception): + pass + + +class WaitInterruption(CommandError): + pass diff --git a/klippy/gcode.py b/klippy/gcode.py index d3590c2dc5..9664b758b4 100644 --- a/klippy/gcode.py +++ b/klippy/gcode.py @@ -9,18 +9,20 @@ import re import shlex -from . import mathutil +import klippy.exceptions as klippy_ex +from . import mathutil -class CommandError(Exception): - pass +# legacy export, use exceptions.CommandError +CommandError = klippy_ex.CommandError Coord = collections.namedtuple("Coord", ("x", "y", "z", "e")) class GCodeCommand: - error = CommandError + # legacy export, use exceptions.CommandError + error = klippy_ex.CommandError def __init__(self, gcode, command, commandline, params, need_ack): self._command = command @@ -84,33 +86,33 @@ def get( value = self._params.get(name) if value is None: if default is self.sentinel: - raise self.error( + raise klippy_ex.CommandError( "Error on '%s': missing %s" % (self._commandline, name) ) return default try: value = parser(value) except: - raise self.error( + raise klippy_ex.CommandError( "Error on '%s': unable to parse %s" % (self._commandline, value) ) if minval is not None and value < minval: - raise self.error( + raise klippy_ex.CommandError( "Error on '%s': %s must have minimum of %s" % (self._commandline, name, minval) ) if maxval is not None and value > maxval: - raise self.error( + raise klippy_ex.CommandError( "Error on '%s': %s must have maximum of %s" % (self._commandline, name, maxval) ) if above is not None and value <= above: - raise self.error( + raise klippy_ex.CommandError( "Error on '%s': %s must be above %s" % (self._commandline, name, above) ) if below is not None and value >= below: - raise self.error( + raise klippy_ex.CommandError( "Error on '%s': %s must be below %s" % (self._commandline, name, below) ) @@ -316,7 +318,7 @@ def _process_commands(self, commands, need_ack=True): handler = self.gcode_handlers.get(cmd, self.cmd_default) try: handler(gcmd) - except self.error as e: + except klippy_ex.CommandError as e: self._respond_error(str(e)) self.printer.send_event("gcode:command_error") if not need_ack: @@ -384,7 +386,7 @@ def _get_extended_params(self, gcmd): eparams = [earg.split("=", 1) for earg in s] eparams = {k.upper(): v for k, v in eparams} except ValueError as e: - raise self.error( + raise klippy_ex.CommandError( "Malformed command '%s'" % (gcmd.get_commandline(),) ) # Update gcmd with new parameters diff --git a/klippy/printer.py b/klippy/printer.py index b9724d2510..739e6b5f1d 100644 --- a/klippy/printer.py +++ b/klippy/printer.py @@ -20,6 +20,7 @@ from types import ModuleType from typing import Any, Callable, Generator, Optional, Union +import klippy.exceptions as klippy_ex from klippy.configfile import ConfigWrapper from . import ( @@ -79,9 +80,8 @@ Printer is shutdown """ - -class WaitInterruption(gcode.CommandError): - pass +# legacy export, use exceptions.WaitInterruption +WaitInterruption = klippy_ex.WaitInterruption class SubsystemComponentCollection: @@ -156,7 +156,10 @@ def get_method(self, function_name): class Printer: config_error = configfile.error - command_error = gcode.CommandError + # legacy export, use exceptions.CommandError + command_error = klippy_ex.CommandError + # legacy export, use exceptions.WaitInterruption + wait_interrupted = WaitInterruption def __init__(self, main_reactor, bglogger, start_args): if sys.version_info[0] < 3: @@ -499,8 +502,6 @@ def request_exit(self, result): self.run_result = result self.reactor.end() - wait_interrupted = WaitInterruption - def wait_while(self, condition_cb, error_on_cancel=True, interval=1.0): """ receives a callback @@ -513,7 +514,7 @@ def wait_while(self, condition_cb, error_on_cancel=True, interval=1.0): while condition_cb(eventtime): if self.is_shutdown() or counter != gcode.get_interrupt_counter(): if error_on_cancel: - raise WaitInterruption("Command interrupted") + raise klippy_ex.WaitInterruption("Command interrupted") else: return eventtime = self.reactor.pause(eventtime + interval) From 22e9a689f9dcd7207e5319aebaaac436e88ea8a5 Mon Sep 17 00:00:00 2001 From: Gareth Farrington Date: Mon, 22 Jun 2026 16:13:09 -0700 Subject: [PATCH 2/8] Make interruptions immediate Use RactorCompletion to trigger an interrupt of a waiting process as soon as the HEATER_INTERRUPT gcode is processed. GcodeDispatch now implements the same contract as VirtualSD: interruption is checked between each gcode command being executed. This means at most the user will have to wait for 1 gcode command to complete before their INTERUPT command is actued on. This also means gcode macros can now be interrupted. The counter mechanisim was replaced with a ReactorCompletion based InterruptToken that supports waking from wait() immediatly after trigger_interrupt(). Signed-off-by: Gareth Farrington --- klippy/gcode.py | 187 ++++++++++++++++++++++++++++++-- klippy/printer.py | 13 +-- test/klippy_testing/__init__.py | 3 +- test/klippy_testing/shims.py | 26 ++++- test/test_pid_profile.py | 25 ++++- 5 files changed, 230 insertions(+), 24 deletions(-) diff --git a/klippy/gcode.py b/klippy/gcode.py index 9664b758b4..0132e77204 100644 --- a/klippy/gcode.py +++ b/klippy/gcode.py @@ -3,15 +3,22 @@ # Copyright (C) 2016-2024 Kevin O'Connor # # This file may be distributed under the terms of the GNU GPLv3 license. +from __future__ import annotations + import collections import logging import os import re import shlex +from typing import Optional +from weakref import WeakKeyDictionary + +import greenlet import klippy.exceptions as klippy_ex from . import mathutil +from .reactor import ReactorCompletion, SelectReactor # legacy export, use exceptions.CommandError CommandError = klippy_ex.CommandError @@ -24,11 +31,21 @@ class GCodeCommand: # legacy export, use exceptions.CommandError error = klippy_ex.CommandError - def __init__(self, gcode, command, commandline, params, need_ack): + def __init__( + self, + gcode: GCodeDispatch, + command, + commandline, + params, + need_ack, + interrupt_token: Optional[InterruptToken], + ): + self._printer = gcode.printer self._command = command self._commandline = commandline self._params = params self._need_ack = need_ack + self._interrupt_token: InterruptToken = interrupt_token # Method wrappers self.respond_info = gcode.respond_info self.respond_raw = gcode.respond_raw @@ -140,6 +157,136 @@ def get_float( below=below, ) + def was_interrupted(self): + self._interrupt_token.test() + + # if the command was interrupted, throws WaitInterruption + def check_interrupted(self): + self._interrupt_token.check_interrupted() + + def get_interrupt_token(self) -> InterruptToken: + return self._interrupt_token + + def wait_while(self, condition_cb, interval=1.0): + """ + receives a callback + waits until callback returns False + (or is interrupted, or printer shuts down) + """ + eventtime = self._printer.get_reactor().monotonic() + while condition_cb(eventtime): + self._interrupt_token.wait_until(eventtime + interval) + eventtime = self._printer.get_reactor().monotonic() + + def delay(self, delay_time): + """ + waits for delay_time seconds + (unless interrupted, or printer shuts down) + """ + waketime = self._printer.get_reactor().monotonic() + delay_time + self._interrupt_token.wait_until(waketime) + + +class InterruptedSentinel: + pass + + +# wrap a ReactorCompletion so callers cannot call complete() +class InterruptToken: + def __init__(self, printer, completion: ReactorCompletion): + self._printer = printer + self._completion = completion + + def test(self) -> bool: + """ + Test for interruption, returns True if interrupted + """ + return self._completion.test() or self._printer.is_shutdown() + + def check_interrupted(self) -> bool: + """ + Test for interruption and throws WaitInterruption if interrupted + or returns False + """ + if self.test(): + raise klippy_ex.WaitInterruption("Command Interrupted") + return False + + def wait(self, waketime=SelectReactor.NEVER) -> bool: + """ + Wait until waketime, return the value of test() + """ + if not self.test(): + self._completion.wait(waketime, None) + return self.test() + + def wait_until( + self, waketime=SelectReactor.NEVER, error_on_interrupt=True + ) -> bool: + """ + Wait until waketime, the value of test(), raise WaitInterruption if interrupted + """ + self.wait(waketime) + if error_on_interrupt: + self.check_interrupted() + return self.test() + + +class GCodeCommandStacks: + def __init__(self, printer): + self._printer = printer + # weak keys, so greenlets can be garbage collected without coordination + self._gcmd_stacks: WeakKeyDictionary[ + greenlet.greenlet, collections.deque[GCodeCommand] + ] = WeakKeyDictionary() + self._interrupt_completion = printer.get_reactor().completion() + + def _get_stack(self) -> Optional[collections.deque[GCodeCommand]]: + return self._gcmd_stacks.get(greenlet.getcurrent(), None) + + def _get_default_stack(self) -> collections.deque[GCodeCommand]: + current_greenlet = greenlet.getcurrent() + if current_greenlet not in self._gcmd_stacks: + self._gcmd_stacks[current_greenlet] = collections.deque() + return self._gcmd_stacks.get(current_greenlet) + + def create_interrupt_token(self): + """ + Create an interrupt token based directly on the interrupt_completion + """ + return InterruptToken(self._printer, self._interrupt_completion) + + def get_interrupt_token(self) -> InterruptToken: + """ + Get the InterruptToken for the currently executing GCodeCommand on + the active greenlet. + """ + gcmd_stack = self._get_stack() + if gcmd_stack: + return gcmd_stack[-1].get_interrupt_token() + return self.create_interrupt_token() + + def push(self, gcmd: GCodeCommand): + self._get_default_stack().append(gcmd) + + def pop(self, gcmd: GCodeCommand): + current_greenlet = greenlet.getcurrent() + gcmd_stack = self._get_stack() + if not gcmd_stack: + raise RuntimeError("GCode command stack corrupted") + popped_gcmd = gcmd_stack.pop() + if popped_gcmd is not gcmd: + raise RuntimeError("GCode command stack corrupted") + if not gcmd_stack: + del self._gcmd_stacks[current_greenlet] + + def trigger_interrupt(self): + self._interrupt_completion.complete(InterruptedSentinel) + self._interrupt_completion = self._printer.get_reactor().completion() + + def async_trigger_interrupt(self): + self._printer.get_reactor().register_callback(self.trigger_interrupt) + # Parse and dispatch G-Code commands class GCodeDispatch: @@ -163,7 +310,7 @@ def __init__(self, printer): self.mux_commands = {} self.gcode_help = {} self.status_commands = {} - self._interrupt_counter = 0 + self._gcmd_stacks = GCodeCommandStacks(printer) self._script_context = 0 # Register commands needed before config file is loaded handlers = [ @@ -183,11 +330,13 @@ def __init__(self, printer): desc = getattr(self, "cmd_" + cmd + "_help", None) self.register_command(cmd, func, True, desc) - def get_interrupt_counter(self): - return self._interrupt_counter + # get the token for the active GCodeCommand context + def get_interrupt_token(self) -> InterruptToken: + return self._gcmd_stacks.get_interrupt_token() - def increment_interrupt_counter(self): - self._interrupt_counter += 1 + # trigger interrupt from another OS thread safely + def async_trigger_interrupt(self): + self._gcmd_stacks.async_trigger_interrupt() def is_traditional_gcode(self, cmd): # A "traditional" g-code command is a letter and followed by a number @@ -274,6 +423,7 @@ def register_output_handler(self, cb): self.output_callbacks.append(cb) def _handle_shutdown(self): + self._gcmd_stacks.trigger_interrupt() if not self.is_printer_ready: return self.is_printer_ready = False @@ -311,12 +461,22 @@ def _process_commands(self, commands, need_ack=True): params = { parts[i]: parts[i + 1].strip() for i in range(1, len(parts), 2) } - gcmd = GCodeCommand(self, cmd, origline, params, need_ack) + gcmd = GCodeCommand( + self, + cmd, + origline, + params, + need_ack, + self._gcmd_stacks.create_interrupt_token(), + ) # Invoke handler for command if self._script_context > 0 and cmd == "RETURN": return handler = self.gcode_handlers.get(cmd, self.cmd_default) + self._gcmd_stacks.push(gcmd) try: + # raise interrupted exception if interrupted + gcmd.check_interrupted() handler(gcmd) except klippy_ex.CommandError as e: self._respond_error(str(e)) @@ -330,6 +490,8 @@ def _process_commands(self, commands, need_ack=True): self._respond_error(msg) if not need_ack: raise + finally: + self._gcmd_stacks.pop(gcmd) gcmd.ack() def run_script_from_command(self, script): @@ -350,7 +512,14 @@ def get_mutex(self): return self.mutex def create_gcode_command(self, command, commandline, params): - return GCodeCommand(self, command, commandline, params, False) + return GCodeCommand( + self, + command, + commandline, + params, + False, + self.get_interrupt_token(), + ) # Response handling def respond_raw(self, msg): @@ -510,7 +679,7 @@ def cmd_HELP(self, gcmd): gcmd.respond_info("\n".join(cmdhelp), log=False) def cmd_HEATER_INTERRUPT(self, gcmd): - self.increment_interrupt_counter() + self._gcmd_stacks.trigger_interrupt() def cmd_LOG_ROLLOVER(self, gcmd): self.printer.bglogger.doRollover() diff --git a/klippy/printer.py b/klippy/printer.py index 739e6b5f1d..8ae5dd6dc1 100644 --- a/klippy/printer.py +++ b/klippy/printer.py @@ -509,15 +509,14 @@ def wait_while(self, condition_cb, error_on_cancel=True, interval=1.0): (or is interrupted, or printer shuts down) """ gcode = self.lookup_object("gcode") - counter = gcode.get_interrupt_counter() + interrupt_token = gcode.get_interrupt_token() eventtime = self.reactor.monotonic() while condition_cb(eventtime): - if self.is_shutdown() or counter != gcode.get_interrupt_counter(): - if error_on_cancel: - raise klippy_ex.WaitInterruption("Command interrupted") - else: - return - eventtime = self.reactor.pause(eventtime + interval) + if interrupt_token.wait_until( + eventtime + interval, error_on_interrupt=error_on_cancel + ): + return + eventtime = self.reactor.monotonic() ###################################################################### diff --git a/test/klippy_testing/__init__.py b/test/klippy_testing/__init__.py index 8c0b65c709..4c901cc4d7 100644 --- a/test/klippy_testing/__init__.py +++ b/test/klippy_testing/__init__.py @@ -1,9 +1,10 @@ -from .shims import PrinterShim, Restart +from .shims import InterruptTokenShim, PrinterShim, Restart from .utils import configwrapper_to_dict __all__ = ( "ConfigRoot", "configwrapper_to_dict", + "InterruptTokenShim", "PrinterShim", "Restart", ) diff --git a/test/klippy_testing/shims.py b/test/klippy_testing/shims.py index f3b4be21c2..bca3c3d419 100644 --- a/test/klippy_testing/shims.py +++ b/test/klippy_testing/shims.py @@ -8,11 +8,26 @@ class Restart(Exception): ... +class InterruptTokenShim: + def test(self): + return False + + def check_interrupted(self): + return False + + def wait(self, waketime=None): + return False + + def wait_until(self, waketime=None, error_on_interrupt=True): + return False + + class PrinterShim: class GCode: error = Exception - def __init__(self): + def __init__(self, printer): + self.printer = printer self.ready_gcode_handlers = {} def register_command(self, cmd, func, *_, **__): @@ -32,7 +47,12 @@ def call(self, cmdline): func = self.ready_gcode_handlers[command.upper()] params = dict(param.split("=", 1) for param in paramlist) gcmd = klippy.gcode.GCodeCommand( - self, command, cmdline, params, False + self, + command, + cmdline, + params, + False, + InterruptTokenShim(), ) print("Calling", func, "with", params) func(gcmd) @@ -40,7 +60,7 @@ def call(self, cmdline): def __init__(self, start_args): self.start_args = start_args self.objects = {} - self.add_object("gcode", self.GCode()) + self.add_object("gcode", self.GCode(self)) self.add_object("configfile", klippy.configfile.PrinterConfig(self)) self.call = self.lookup_object("gcode").call diff --git a/test/test_pid_profile.py b/test/test_pid_profile.py index 131769a4e2..6a0d5cb32c 100644 --- a/test/test_pid_profile.py +++ b/test/test_pid_profile.py @@ -1,7 +1,7 @@ import pathlib import typing -from klippy_testing import PrinterShim +from klippy_testing import InterruptTokenShim, PrinterShim import klippy.extras.heaters as heaters import klippy.gcode @@ -35,7 +35,12 @@ def _make_pmgr(heater): def _gcmd(heater, params): return klippy.gcode.GCodeCommand( - heater.gcode, "PID_PROFILE", "PID_PROFILE", params, False + heater.gcode, + "PID_PROFILE", + "PID_PROFILE", + params, + False, + InterruptTokenShim(), ) @@ -103,6 +108,17 @@ def monotonic(self): return 0.0 +class _FakePrinter: + def __init__(self, reactor): + self.reactor = reactor + + def get_reactor(self): + return self.reactor + + def is_shutdown(self): + return False + + class _FakeConfig: def getfloat(self, key, default=None, **kw): return {"inner_target_temp": 135.0}.get(key, default) @@ -114,7 +130,8 @@ def error(self, msg): class _FakeGCode: error = Exception - def __init__(self): + def __init__(self, printer): + self.printer = printer self.messages = [] def respond_info(self, msg): @@ -140,7 +157,7 @@ class _FakeDualLoopHeater: def __init__(self): self.reactor = _FakeReactor() self.config = _FakeConfig() - self.gcode = _FakeGCode() + self.gcode = _FakeGCode(_FakePrinter(self.reactor)) self.configfile = _FakeConfigFile() self.smooth_time = 1.0 self.target_temp = 0.0 From 31d1d4dea85b5d61f18f0508e164b69f281ddf54 Mon Sep 17 00:00:00 2001 From: Gareth Farrington Date: Thu, 25 Jun 2026 22:08:03 -0700 Subject: [PATCH 3/8] Use interruption token in Toolhead Toolhead looks up the Interrupt token in move() to absort moves if the token is in the interrupted state. _check_pause() uses the Interrupt token for pauses so it can be instantly woken on interrupt. This stops the toolhead from blocking and from executing the last move it recieved before it blocked. `drip_move` is gracefully aborted by the InterruptToken. A new completion_any() method in reactor combines the existing drip move complation with the interrupt token and complectes when either one is completed. Signed-off-by: Gareth Farrington --- klippy/extras/trad_rack.py | 1 + klippy/reactor.py | 16 +++++ klippy/toolhead.py | 137 +++++++++++++++++++++++++------------ 3 files changed, 110 insertions(+), 44 deletions(-) diff --git a/klippy/extras/trad_rack.py b/klippy/extras/trad_rack.py index e96419a1d8..260b7fde8e 100644 --- a/klippy/extras/trad_rack.py +++ b/klippy/extras/trad_rack.py @@ -2317,6 +2317,7 @@ class TradRackToolHead(toolhead.ToolHead, object): def __init__(self, config, buffer_pull_speed, is_extruder_synced): self.printer = config.get_printer() self.danger_options = self.printer.lookup_object("danger_options") + self.gcode = self.printer.lookup_object("gcode") self.reactor = self.printer.get_reactor() self.all_mcus = [ m for n, m in self.printer.lookup_objects(module="mcu") diff --git a/klippy/reactor.py b/klippy/reactor.py index 005f8fa4ba..9d5fd10484 100644 --- a/klippy/reactor.py +++ b/klippy/reactor.py @@ -203,6 +203,22 @@ def _check_timers(self, eventtime, busy): def completion(self): return ReactorCompletion(self) + def completion_any(self, child_completions): + if len(child_completions) == 1: + return child_completions[0] + any_completion = self.completion() + + def complete_once(child_completion): + if any_completion.test(): + return + result = child_completion.wait() + if not any_completion.test(): + any_completion.complete(result) + + for child in child_completions: + self.register_callback(lambda e, c=child: complete_once(c)) + return any_completion + def register_callback(self, callback, waketime=NOW): rcb = ReactorCallback(self, callback, waketime) return rcb.completion diff --git a/klippy/toolhead.py b/klippy/toolhead.py index 1b4b0d90bb..1a4f03c079 100644 --- a/klippy/toolhead.py +++ b/klippy/toolhead.py @@ -3,13 +3,18 @@ # Copyright (C) 2016-2025 Kevin O'Connor # # This file may be distributed under the terms of the GNU GPLv3 license. +from __future__ import annotations + import importlib import logging import math from . import chelper +from .exceptions import WaitInterruption from .extras.danger_options import get_danger_options +from .gcode import GCodeDispatch, InterruptToken from .kinematics import extruder as kinematics_extruder +from .reactor import ReactorCompletion, SelectReactor # Common suffixes: _d is distance (in mm), _v is velocity (in # mm/second), _v2 is velocity squared (mm^2/s^2), _t is time (in @@ -253,11 +258,38 @@ def add_move(self, move): DRIP_TIME = 0.100 +class DripMoveCompletion: + def __init__( + self, + reactor: SelectReactor, + drip_completion: ReactorCompletion, + interrupt_token: InterruptToken, + ): + self.drip_completion = drip_completion + self.interrupt_token = interrupt_token + self.wait_completion = drip_completion + if interrupt_token: + self.wait_completion = reactor.completion_any( + [drip_completion, interrupt_token] + ) + + def check(self): + return ( + self.interrupt_token.check_interrupted() + or self.drip_completion.test() + ) + + def wait(self, waketime): + self.wait_completion.wait(waketime) + self.check() + + # Main code to track events (and their timing) on the printer toolhead class ToolHead: def __init__(self, config): self.printer = config.get_printer() self.reactor = self.printer.get_reactor() + self.gcode: GCodeDispatch = self.printer.lookup_object("gcode") self.all_mcus = [ m for n, m in self.printer.lookup_objects(module="mcu") ] @@ -532,7 +564,9 @@ def _check_pause(self): if not self.can_pause: self.need_check_pause = self.reactor.NEVER return - eventtime = self.reactor.pause(eventtime + min(1.0, pause_time)) + interrupt_token = self.gcode.get_interrupt_token() + interrupt_token.wait_until(eventtime + min(1.0, pause_time)) + eventtime = self.reactor.monotonic() est_print_time = self.mcu.estimated_print_time(eventtime) buffer_time = self.print_time - est_print_time if not self.special_queuing_state: @@ -607,6 +641,7 @@ def move(self, newpos, speed): move = Move(self, self.commanded_pos, newpos, speed) if not move.move_d: return + self.gcode.get_interrupt_token().check_interrupted() if move.is_kinematic_move: self.kin.check_move(move) for e_index, ea in enumerate(self.extra_axes): @@ -641,7 +676,9 @@ def wait_moves(self): ): if not self.can_pause: break - eventtime = self.reactor.pause(eventtime + 0.100) + interrupt_token = self.gcode.get_interrupt_token() + interrupt_token.wait_until(eventtime + 0.100) + eventtime = self.reactor.monotonic() def set_extruder(self, extruder, extrude_pos): # XXX - should use add_extra_axis @@ -693,7 +730,7 @@ def drip_update_time(self, next_print_time, drip_completion, addstepper=()): # Update print_time in segments until drip_completion signal flush_delay = DRIP_TIME + STEPCOMPRESS_FLUSH_TIME + self.kin_flush_delay while self.print_time < next_print_time: - if drip_completion.test(): + if drip_completion.check(): break curtime = self.reactor.monotonic() est_print_time = self.mcu.estimated_print_time(curtime) @@ -714,50 +751,62 @@ def drip_update_time(self, next_print_time, drip_completion, addstepper=()): self.flush_step_generation() def _drip_load_trapq(self, submit_move): - # Queue move into trapezoid motion queue (trapq) - if submit_move.move_d: - self.commanded_pos[:] = submit_move.end_pos - self.lookahead.add_move(submit_move) - moves = self.lookahead.flush() - self._calc_print_time() - next_move_time = self.print_time - for move in moves: - self.trapq_append( - self.trapq, - next_move_time, - move.accel_t, - move.cruise_t, - move.decel_t, - move.start_pos[0], - move.start_pos[1], - move.start_pos[2], - move.axes_r[0], - move.axes_r[1], - move.axes_r[2], - move.start_v, - move.cruise_v, - move.accel, - ) - next_move_time = ( - next_move_time + move.accel_t + move.cruise_t + move.decel_t - ) - self.lookahead.reset() + try: + # Queue move into trapezoid motion queue (trapq) + if submit_move.move_d: + self.commanded_pos[:] = submit_move.end_pos + self.lookahead.add_move(submit_move) + moves = self.lookahead.flush() + self._calc_print_time() + next_move_time = self.print_time + for move in moves: + self.trapq_append( + self.trapq, + next_move_time, + move.accel_t, + move.cruise_t, + move.decel_t, + move.start_pos[0], + move.start_pos[1], + move.start_pos[2], + move.axes_r[0], + move.axes_r[1], + move.axes_r[2], + move.start_v, + move.cruise_v, + move.accel, + ) + next_move_time = ( + next_move_time + move.accel_t + move.cruise_t + move.decel_t + ) + except WaitInterruption: + logging.info("caught WaitInterruption in _drip_load_trapq") + raise + finally: + self.lookahead.reset() return next_move_time def drip_move(self, newpos, speed, drip_completion): - # Create and verify move is valid - newpos = newpos[:3] + self.commanded_pos[3:] - move = Move(self, self.commanded_pos, newpos, speed) - if move.move_d: - self.kin.check_move(move) - # Make sure stepper movement doesn't start before nominal start time - self.dwell(self.kin_flush_delay) - # Transmit move in "drip" mode - self._process_lookahead() - next_move_time = self._drip_load_trapq(move) - self.drip_update_time(next_move_time, drip_completion) - # Move finished; cleanup any remnants on trapq - self.trapq_finalize_moves(self.trapq, self.reactor.NEVER, 0) + try: + # Create and verify move is valid + newpos = newpos[:3] + self.commanded_pos[3:] + move = Move(self, self.commanded_pos, newpos, speed) + if move.move_d: + self.kin.check_move(move) + # Make sure stepper movement doesn't start before nominal start time + self.dwell(self.kin_flush_delay) + # Transmit move in "drip" mode + drip_completion = DripMoveCompletion( + self.reactor, drip_completion, self.gcode.get_interrupt_token() + ) + self._process_lookahead() + next_move_time = self._drip_load_trapq(move) + self.drip_update_time(next_move_time, drip_completion) + except WaitInterruption: + raise + finally: + # Move finished; cleanup any remnants on trapq + self.trapq_finalize_moves(self.trapq, self.reactor.NEVER, 0) # Misc commands def stats(self, eventtime): From 2b944baba0d40200cf9ad0ed2571db9c34389114 Mon Sep 17 00:00:00 2001 From: Gareth Farrington Date: Wed, 1 Jul 2026 12:41:31 -0700 Subject: [PATCH 4/8] Add ReactorMutex.throw(Exception) and ReactorMutex.owns_lock() Add a method that will throw an exception into all of the greenlets currently waiting on a mutex. If an exception is raised on one of the waiting greeenlets and the lock was not claimed, the next greenlet in the queue is woken up to claim the lock. owns_lock() reportes true if the current greenlet own the lock. Since we are getting rid of some `with` blocks, it simplifies bookeeping and makes expressing intent more clear. Signed-off-by: Gareth Farrington --- klippy/reactor.py | 27 +++++++++++++++++++++++++++ 1 file changed, 27 insertions(+) diff --git a/klippy/reactor.py b/klippy/reactor.py index 9d5fd10484..5cfc526937 100644 --- a/klippy/reactor.py +++ b/klippy/reactor.py @@ -88,34 +88,61 @@ class ReactorMutex: def __init__(self, reactor, is_locked): self.reactor = reactor self.is_locked = is_locked + self.owner = None self.next_pending = False self.queue = [] + self.pending_exceptions = {} self.lock = self.__enter__ self.unlock = self.__exit__ def test(self): return self.is_locked + def owns_lock(self): + return self.owner is greenlet.getcurrent() + def __enter__(self): if not self.is_locked: self.is_locked = True + self.owner = greenlet.getcurrent() return g = greenlet.getcurrent() self.queue.append(g) while True: self.reactor.pause(self.reactor.NEVER) + exc = self.pending_exceptions.pop(g, None) + if exc is not None: + self.queue.remove(g) + if self.next_pending: + # wake next pending. __exit__ won't run for this greenlet + # because it never successfully completes __enter__ + if self.queue: + self.reactor.update_timer( + self.queue[0].timer, self.reactor.NOW + ) + else: + self.next_pending = False + self.is_locked = False + raise exc if self.next_pending and self.queue[0] is g: self.next_pending = False self.queue.pop(0) + self.owner = g return def __exit__(self, type=None, value=None, tb=None): + self.owner = None if not self.queue: self.is_locked = False return self.next_pending = True self.reactor.update_timer(self.queue[0].timer, self.reactor.NOW) + def throw(self, exception): + for g in list(self.queue): + self.pending_exceptions[g] = exception + self.reactor.update_timer(g.timer, self.reactor.NOW) + class SelectReactor: NOW = _NOW From 8d2e0a9e476d2b63fb6da0a6766ae8d33a6db60d Mon Sep 17 00:00:00 2001 From: Gareth Farrington Date: Wed, 1 Jul 2026 13:55:26 -0700 Subject: [PATCH 5/8] Raise WaitInterruption on mutex in trigger_interrupt This causes all waiting threads to recieve an interrupt and never get the lock. Signed-off-by: Gareth Farrington --- klippy/gcode.py | 10 +++++++--- 1 file changed, 7 insertions(+), 3 deletions(-) diff --git a/klippy/gcode.py b/klippy/gcode.py index 0132e77204..a21939fe3f 100644 --- a/klippy/gcode.py +++ b/klippy/gcode.py @@ -334,9 +334,13 @@ def __init__(self, printer): def get_interrupt_token(self) -> InterruptToken: return self._gcmd_stacks.get_interrupt_token() + def _trigger_interrupt(self): + self._gcmd_stacks.trigger_interrupt() + self.mutex.throw(klippy_ex.WaitInterruption("Command Interrupted")) + # trigger interrupt from another OS thread safely def async_trigger_interrupt(self): - self._gcmd_stacks.async_trigger_interrupt() + self.printer.get_reactor().register_callback(self._trigger_interrupt) def is_traditional_gcode(self, cmd): # A "traditional" g-code command is a letter and followed by a number @@ -423,7 +427,7 @@ def register_output_handler(self, cb): self.output_callbacks.append(cb) def _handle_shutdown(self): - self._gcmd_stacks.trigger_interrupt() + self._trigger_interrupt() if not self.is_printer_ready: return self.is_printer_ready = False @@ -679,7 +683,7 @@ def cmd_HELP(self, gcmd): gcmd.respond_info("\n".join(cmdhelp), log=False) def cmd_HEATER_INTERRUPT(self, gcmd): - self._gcmd_stacks.trigger_interrupt() + self._trigger_interrupt() def cmd_LOG_ROLLOVER(self, gcmd): self.printer.bglogger.doRollover() From 5a44d2b530d7d7946ef02dfd3f8478b1e8f9f821 Mon Sep 17 00:00:00 2001 From: Gareth Farrington Date: Mon, 6 Jul 2026 12:01:45 -0700 Subject: [PATCH 6/8] Refactor trigger_interrupt to be a property of a gcode command Add a property to gcode commands so the interrupt can be cleanly executed before the command requiring the interrupt runs. Signed-off-by: Gareth Farrington --- klippy/gcode.py | 51 +++++++++++++++++++++++++++++++++---------------- 1 file changed, 35 insertions(+), 16 deletions(-) diff --git a/klippy/gcode.py b/klippy/gcode.py index a21939fe3f..3df8e8cd59 100644 --- a/klippy/gcode.py +++ b/klippy/gcode.py @@ -309,26 +309,34 @@ def __init__(self, printer): self.ready_gcode_handlers = {} self.mux_commands = {} self.gcode_help = {} + # Name-bound policy: intentionally survives unregister/rename_existing. + self.trigger_interrupt_commands = set() self.status_commands = {} self._gcmd_stacks = GCodeCommandStacks(printer) self._script_context = 0 # Register commands needed before config file is loaded - handlers = [ - "M110", - "M112", - "M115", - "RESTART", - "FIRMWARE_RESTART", - "ECHO", - "STATUS", - "HELP", - "HEATER_INTERRUPT", - "LOG_ROLLOVER", - ] - for cmd in handlers: + handlers: dict[str, bool] = { + "M110": False, + "M112": False, + "M115": False, + "RESTART": False, + "FIRMWARE_RESTART": False, + "ECHO": False, + "STATUS": False, + "HELP": False, + "LOG_ROLLOVER": False, + "HEATER_INTERRUPT": True, + } + for cmd, interrupt in handlers.items(): func = getattr(self, "cmd_" + cmd) desc = getattr(self, "cmd_" + cmd + "_help", None) - self.register_command(cmd, func, True, desc) + self.register_command( + cmd, + func, + when_not_ready=True, + desc=desc, + trigger_interrupt=interrupt, + ) # get the token for the active GCodeCommand context def get_interrupt_token(self) -> InterruptToken: @@ -351,7 +359,14 @@ def is_traditional_gcode(self, cmd): except: return False - def register_command(self, cmd, func, when_not_ready=False, desc=None): + def register_command( + self, + cmd, + func, + when_not_ready=False, + desc=None, + trigger_interrupt=False, + ): if func is None: old_cmd = self.ready_gcode_handlers.get(cmd) if cmd in self.ready_gcode_handlers: @@ -381,6 +396,8 @@ def register_command(self, cmd, func, when_not_ready=False, desc=None): def func(params): return origfunc(self._get_extended_params(params)) + if trigger_interrupt: + self.trigger_interrupt_commands.add(cmd) self.ready_gcode_handlers[cmd] = func if when_not_ready: self.base_gcode_handlers[cmd] = func @@ -465,6 +482,8 @@ def _process_commands(self, commands, need_ack=True): params = { parts[i]: parts[i + 1].strip() for i in range(1, len(parts), 2) } + if cmd in self.trigger_interrupt_commands: + self._trigger_interrupt() gcmd = GCodeCommand( self, cmd, @@ -683,7 +702,7 @@ def cmd_HELP(self, gcmd): gcmd.respond_info("\n".join(cmdhelp), log=False) def cmd_HEATER_INTERRUPT(self, gcmd): - self._trigger_interrupt() + pass def cmd_LOG_ROLLOVER(self, gcmd): self.printer.bglogger.doRollover() From 1470bb3fdd0ab1dd7834790a5a28fd99473709d0 Mon Sep 17 00:00:00 2001 From: Gareth Farrington Date: Mon, 6 Jul 2026 12:56:25 -0700 Subject: [PATCH 7/8] Move mutex acquisition inside _process_commands Move the mutex acquision inside the _process_commands loop. Mutex is now acquired after we understand what the command is and if it will require an interrupt before it runs. In the case of an interrupt the mutex is released to allow the other waiting threads to run any exception handlers or finally blocks before this thread runs. Signed-off-by: Gareth Farrington --- klippy/gcode.py | 136 ++++++++++++++++++++++++++++++------------------ 1 file changed, 84 insertions(+), 52 deletions(-) diff --git a/klippy/gcode.py b/klippy/gcode.py index 3df8e8cd59..b06edfcdc4 100644 --- a/klippy/gcode.py +++ b/klippy/gcode.py @@ -74,6 +74,9 @@ def get_raw_command_parameters(self): param_start += 1 return origline[param_start:param_end] + def get_need_ack(self): + return self._need_ack + def ack(self, msg=None): if not self._need_ack: return False @@ -464,72 +467,103 @@ def _handle_ready(self): # Parse input into commands args_r = re.compile("([A-Z_]+|[A-Z*])") - def _process_commands(self, commands, need_ack=True): + def _parse_command_line(self, line: str): + # Ignore comments and leading/trailing spaces + line = stripped_line = line.strip() + cpos = line.find(";") + if cpos >= 0: + line = line[:cpos] + # Break line into parts and determine command + parts = self.args_r.split(line.upper()) + if "".join(parts[:2]) == "N": + # Skip line number at start of command + cmd = "".join(parts[3:5]).strip() + else: + cmd = "".join(parts[:3]).strip() + # Build gcode "params" dictionary + params = { + parts[i]: parts[i + 1].strip() for i in range(1, len(parts), 2) + } + return cmd, params, stripped_line + + def _process_command(self, gcmd: GCodeCommand): + handler = self.gcode_handlers.get(gcmd.get_command(), self.cmd_default) + self._gcmd_stacks.push(gcmd) + try: + # raise interrupted exception if interrupted + gcmd.check_interrupted() + handler(gcmd) + except klippy_ex.CommandError as e: + self._respond_error(str(e)) + self.printer.send_event("gcode:command_error") + if not gcmd.get_need_ack(): + raise + except: + msg = 'Internal error on command:"%s"' % (gcmd.get_command(),) + logging.exception(msg) + self.printer.invoke_shutdown(msg) + self._respond_error(msg) + if not gcmd.get_need_ack(): + raise + finally: + self._gcmd_stacks.pop(gcmd) + gcmd.ack() + + def _process_commands_loop(self, commands, need_ack=True): for line in commands: - # Ignore comments and leading/trailing spaces - line = origline = line.strip() - cpos = line.find(";") - if cpos >= 0: - line = line[:cpos] - # Break line into parts and determine command - parts = self.args_r.split(line.upper()) - if "".join(parts[:2]) == "N": - # Skip line number at start of command - cmd = "".join(parts[3:5]).strip() - else: - cmd = "".join(parts[:3]).strip() - # Build gcode "params" dictionary - params = { - parts[i]: parts[i + 1].strip() for i in range(1, len(parts), 2) - } + cmd, params, line = self._parse_command_line(line) + # RETURN from gcode macro + if self._script_context > 0 and cmd == "RETURN": + return if cmd in self.trigger_interrupt_commands: + # release the mutex so this thread will suspend on mutex.lock() + # to allow all interrupted threads to finish first when + # it waits to re-acquire the lock + if self.mutex.owns_lock(): + self.mutex.unlock() + # causes all waiters of the mutex to get an exception self._trigger_interrupt() - gcmd = GCodeCommand( + gcmd: GCodeCommand = GCodeCommand( self, cmd, - origline, + line, params, need_ack, + # acquires the token after the interrupt, this command + # belongs to the next generation of commands self._gcmd_stacks.create_interrupt_token(), ) - # Invoke handler for command - if self._script_context > 0 and cmd == "RETURN": - return - handler = self.gcode_handlers.get(cmd, self.cmd_default) - self._gcmd_stacks.push(gcmd) - try: - # raise interrupted exception if interrupted - gcmd.check_interrupted() - handler(gcmd) - except klippy_ex.CommandError as e: - self._respond_error(str(e)) - self.printer.send_event("gcode:command_error") - if not need_ack: - raise - except: - msg = 'Internal error on command:"%s"' % (cmd,) - logging.exception(msg) - self.printer.invoke_shutdown(msg) - self._respond_error(msg) - if not need_ack: - raise - finally: - self._gcmd_stacks.pop(gcmd) - gcmd.ack() + # take lock once per loop + if not self.mutex.owns_lock(): + self.mutex.lock() + self._process_command(gcmd) + + def _process_commands(self, commands, need_ack=True): + if self.mutex.owns_lock(): + raise RuntimeError( + "The current greenlet already owns the lock! Did " + "you mean to call run_script_from_command()?" + ) + try: + self._process_commands_loop(commands, need_ack) + finally: + if self.mutex.owns_lock(): + self.mutex.unlock() def run_script_from_command(self, script): + if not self.mutex.owns_lock(): + raise RuntimeError( + "The current greenlet does not own the lock! " + "Did you mean to call run_script()?" + ) self._script_context += 1 try: - self._process_commands(script.split("\n"), need_ack=False) + self._process_commands_loop(script.split("\n"), need_ack=False) finally: self._script_context -= 1 def run_script(self, script): - if "INTERRUPT" in script: - self._process_commands(script.split("\n"), need_ack=False) - else: - with self.mutex: - self._process_commands(script.split("\n"), need_ack=False) + self._process_commands(script.split("\n"), need_ack=False) def get_mutex(self): return self.mutex @@ -715,7 +749,6 @@ def __init__(self, printer): printer.register_event_handler("klippy:ready", self._handle_ready) printer.register_event_handler("klippy:shutdown", self._handle_shutdown) self.gcode = printer.lookup_object("gcode") - self.gcode_mutex = self.gcode.get_mutex() self.fd = printer.get_start_args().get("gcode_fd") self.reactor = printer.get_reactor() self.is_printer_ready = False @@ -796,8 +829,7 @@ def _process_data(self, eventtime): self.is_processing_data = True while pending_commands: self.pending_commands = [] - with self.gcode_mutex: - self.gcode._process_commands(pending_commands) + self.gcode._process_commands(pending_commands) pending_commands = self.pending_commands self.is_processing_data = False if self.fd_handle is None: From 2ad0255ddfc81eec77aa44b1234879f88a0f5fb4 Mon Sep 17 00:00:00 2001 From: Gareth Farrington Date: Mon, 6 Jul 2026 13:53:04 -0700 Subject: [PATCH 8/8] CANCEL_PRINT interrupts the printer Now when the CANCEL_PRINT command is submitted, all activve and buffered commands are interrupted and then CANCEL_PRINT runs. Signed-off-by: Gareth Farrington --- klippy/extras/pause_resume.py | 1 + 1 file changed, 1 insertion(+) diff --git a/klippy/extras/pause_resume.py b/klippy/extras/pause_resume.py index 29482d0160..41d3cd346f 100644 --- a/klippy/extras/pause_resume.py +++ b/klippy/extras/pause_resume.py @@ -30,6 +30,7 @@ def __init__(self, config): "CANCEL_PRINT", self.cmd_CANCEL_PRINT, desc=self.cmd_CANCEL_PRINT_help, + trigger_interrupt=True, ) webhooks = self.printer.lookup_object("webhooks") webhooks.register_endpoint(