Async progress_handler call

[viktorkertesz]

Dear @ronf ,
first, thanks for this very useful library! I am developing a new feature for Scrapli, where I am trying to figure out what is the best way to prevent timeouts on main SSH channel during SCP operation. File transfer can take a lot of time and it makes the ongoing Scrapli connection stale and closed by the device.

As scp is blocking the thread while doing the copy, I tried to use progress_handler callback to make a quick and non-interactive send of an empty command to the staling SSH connection. Problem is that I can't await on the progress_handler as asyncssh does not handle that function call in async way :P

I patched scp.py as a POC:

===================================================================
diff --git a/scp.py b/scp.py
--- a/scp.py	(date 1622580973114)
+++ b/scp.py	(date 1622580973114)
@@ -30,6 +30,7 @@
 import stat
 import string
 import sys
+from inspect import iscoroutinefunction
 
 from .constants import DEFAULT_LANG
 
@@ -507,7 +508,10 @@
                 offset += len(data)
 
                 if self._progress_handler:
-                    self._progress_handler(srcpath, dstpath, offset, size)
+                    if iscoroutinefunction(self._progress_handler):
+                        await self._progress_handler(srcpath, dstpath, offset, size)
+                    else:
+                        self._progress_handler(srcpath, dstpath, offset, size)
         finally:
             await file_obj.close()
 

This worked fine actually. Do you find this useful? Or a bad idea?

Regards,
Viktor

[ronf]

This runs the risk of blocking the scp operation if the callback you pass in blocks to do any async operations. That's one of the reasons it is a callback -- it's not supposed to take any significant amount of time.

Since you don't need the return value from the handler here, I think a better solution might be to have your callback schedule a new async task to do whatever you want to do using asyncio.create_task(), or asyncio.ensure_future() if you are running something earlier than Python 3.7. In other words:

async def async_progress_handler(srcpath, dstpath, bytes_copied, total_bytes):
    """Do whatever async operations you need to do"""

def progress_handler(srcpath, dstpath, bytes_copied, total_bytes):
    """Schedule the async progress handler"""

    asyncio.ensure_future(async_progress_handler(srcpath, dstpath, bytes_copied, total_bytes))

I could potentially make this automatic when a coroutine is passed in, as you have here. I would probably actually call the progress handler and see if it returns an awaitable or not, though, and if so then call ensure_future() on that. To avoid code duplication, I'm thinking a small helper function which does that could be created such as:

def _call_handler(handler, *args):
    """Call a handler, creating a task if it returns an awaitable"""

    if handler:
        result = handler(*args)

        if inspect.isawaitable(result):
            asyncio.ensure_future(result)

One caution about this is that the progress handlers may run in parallel if they block, though, since the transfer will continue before the progress handler returns. Depending on how the handler is written, this may make it look like the progress is jumping forward and back, so perhaps it's better to leave the decision to the caller about whether to create a task or not.

[viktorkertesz]

@ronf, thank you so much! I was able to use the first block of code. I understand that in progress_handler I shouldn't do any extraordinary as it may block the actual progress_handler. I only needed the async capability for my own use case, not as a general solution. I just couldn't figure out how to call an async function in a normal one. I learned again and I appreciate it to you!

So finally I wrapped the original progress_handler into mine which injects an ENTER from time to time into the idle ssh session during copy.

        async def _prevent_timeout():
            """Send enter to idle SSH channel to prevent timing out while transfering file"""
            await self.send_command("")

        def timed_progress_handler(srcpath, dstpath, copied, total):
            """Progress handler wrapper which prevents timeout while file transfer"""
            nonlocal start_time

            now = time()
            if 0 < prevent_timeout <= (now - start_time):
                start_time = now
                self.logger.debug("Preventing timeout")
                asyncio.ensure_future(_prevent_timeout())

            # call original handler if specified
            if progress_handler:
                progress_handler(srcpath, dstpath, copied, total)

[ronf]

Great - glad you got it working!

One thing: If you wanted to, you could actually get rid of the _prevent_timeout function here and just directly do an ensure_future() on self.send_command:

                asyncio.ensure_future(self.send_command(""))

(though if you wanted to keep that split for readability, I completely understand)

Basically, any awaitable can be passed into ensure_future, complete with arguments, and it will create a task and schedule it to run.

[viktorkertesz]

Thanks again Ron! I kept it separate as it might be device driver dependent. It's under decision, how to prevent timeout.