Stopping an AHK script after Timeout

[Lunatik6572]

I've been trying to stop a script if it executes for too long but I have not had success doing so. Even after it times out and calls the future's done callback, the code keeps executing.

The execution code:

self.ahkExec = AHK(executable_path=os.path.join(self.libPath, 'AutoHotkey64.exe'))
...
future: FutureResult = self.ahkExec.run_script(self.script, blocking=False, timeout=3)
future._fut.add_done_callback(self.handleFuture)

The future handler:

def handleFuture(self, result: Future):
   if result._result == None:
      self.logger.warning('Script reached its timeout')
      # trying to stop the AHK script here either using ahkExec or the Future result
   else:
      self.logger.info('Script completed successfully')
   self.running = False

I have tried using self.ahkExec._transport._proc.kill() as pointed out in a different discussion but I keep getting an error and I am not sure if there is a better way to do this.

[spyoungtech]

Hmm. Yeah, beyond the included timeout parameter (which is only supported in the sync API), there's not currently a great way to do this. The timeout parameter has the same effect as in communicate -- which does not kill the child process when a timeout occurs.

In version 0.x of this library, run_script with blocking=False would return the subprocess object created by Popen directly, which would allow you to interact with it using all the APIs provided in the subprocess module. In v1.0 and onward, you get a FutureResult which doesn't have any public references back to the associated process.

._transport._proc.kill works for the regular AutoHotkey process that handles all blocking calls (not including run_script), but not for processes that are spawned for blocking=False calls.

I will consider how we can handle this case in a future update.

[Lunatik6572]

I got it to work using a non-elegant way. But it works with my limited testing!

Overwriting the Transport's run_script:

from ahk._sync.transport import DaemonProcessTransport, FutureResult, SyncAHKProcess
from typing import NamedTuple
...

class FutureProc(NamedTuple):
    proc: SyncAHKProcess
    future: FutureResult

class CustomTransport(DaemonProcessTransport):
    def run_script(self, script_text_or_path: str, /, *, blocking: bool = True, timeout: int = None) -> str | FutureProc:
        ...
        proc = SyncAHKProcess(runargs)
        proc.start()
        if blocking:
           ...
        else:
            return FutureProc(proc, self._sync_run_nonblocking(proc, script_bytes, timeout=timeout))

Accessing proc and FutureResult when using blocking=False:

def __init__():
    self.ahkExec = AHK(TransportClass=CustomTransport, executable_path=self.libPath)
    ...

def run():
    ...
    futureProc: FutureProc = self.ahkExec.run_script(self.script, blocking=False, timeout=10)
    futureProc.future._fut.add_done_callback(lambda result: self.handleFuture(result, futureProc.proc))

def handleFuture(self, result: Future, proc: SyncAHKProcess):
    if result._result == None:
        self.logger.warning('Script reached its timeout')
        proc.kill()
    else:
        self.logger.info('Script completed successfully')
    self.running = False

[spyoungtech]

I've made a change in #339 which is released in version 1.8.0. Processes called with run_script are cleaned up immediately after the call to communicate, so this should now be addressed as of this version.