Dont block executor when syncing files or mutexes (#32)

* Make Mutex.set async

* Make folder sync async

Author: ftilde

livesync/folder.py

@@ -79,18 +79,18 @@ async def watch(self) -> None:
                                                    watch_filter=lambda _, filepath: not self._ignore_spec.match_file(filepath)):
                 for change, filepath in changes:
                     print('?+U-'[change], filepath)
-                self.sync()
+                await self.sync()
         except RuntimeError as e:
             if 'Already borrowed' not in str(e):
                 raise
 
-    def sync(self) -> None:
+    async def sync(self) -> None:
         args = ' '.join(self._rsync_args)
         args += ''.join(f' --exclude="{e}"' for e in self._get_ignores())
         args += f' -e "ssh -p {self.ssh_port}"'  # NOTE: use SSH with custom port
         args += f' --rsync-path="mkdir -p {self.target_path} && rsync"'  # NOTE: create target folder if not exists
-        run_subprocess(f'rsync {args} "{self.source_path}/" "{self.target}/"', quiet=True)
+        await run_subprocess(f'rsync {args} "{self.source_path}/" "{self.target}/"', quiet=True)
         if isinstance(self.on_change, str):
-            run_subprocess(f'ssh {self.host} -p {self.ssh_port} "cd {self.target_path}; {self.on_change}"')
+            await run_subprocess(f'ssh {self.host} -p {self.ssh_port} "cd {self.target_path}; {self.on_change}"')
         if callable(self.on_change):
             self.on_change()

livesync/mutex.py

@@ -1,6 +1,6 @@
+import asyncio
 import logging
 import socket
-import subprocess
 from datetime import datetime, timedelta
 from typing import Optional
 
@@ -14,10 +14,10 @@ def __init__(self, host: str, port: int) -> None:
         self.occupant: Optional[str] = None
         self.user_id = socket.gethostname()
 
-    def is_free(self) -> bool:
+    async def is_free(self) -> bool:
         try:
             command = f'[ -f {self.DEFAULT_FILEPATH} ] && cat {self.DEFAULT_FILEPATH} || echo'
-            output = self._run_ssh_command(command).strip()
+            output = (await self._run_ssh_command(command)).strip()
             if not output:
                 return True
             words = output.splitlines()[0].strip().split()
@@ -30,20 +30,27 @@ def is_free(self) -> bool:
             logging.exception('Could not access target system')
             return False
 
-    def set(self, info: str) -> bool:
-        if not self.is_free():
+    async def set(self, info: str) -> bool:
+        if not await self.is_free():
             return False
         try:
-            self._run_ssh_command(f'echo "{self.tag}\n{info}" > {self.DEFAULT_FILEPATH}')
+            await self._run_ssh_command(f'echo "{self.tag}\n{info}" > {self.DEFAULT_FILEPATH}')
             return True
-        except subprocess.CalledProcessError:
+        except RuntimeError:
             print('Could not write mutex file')
             return False
 
     @property
     def tag(self) -> str:
         return f'{self.user_id} {datetime.now().isoformat()}'
 
-    def _run_ssh_command(self, command: str) -> str:
-        ssh_command = ['ssh', self.host, '-p', str(self.port), command]
-        return subprocess.check_output(ssh_command, stderr=subprocess.DEVNULL).decode()
+    async def _run_ssh_command(self, command: str) -> str:
+        process = await asyncio.create_subprocess_exec(
+            'ssh', self.host, '-p', str(self.port), command,
+            stdout=asyncio.subprocess.PIPE,
+            stderr=asyncio.subprocess.DEVNULL,
+        )
+        stdout, _ = await process.communicate()
+        if process.returncode != 0:
+            raise RuntimeError(f'SSH command failed with return code {process.returncode}')
+        return stdout.decode()

livesync/run_subprocess.py

@@ -1,11 +1,16 @@
+import asyncio
 import subprocess
 
 
-def run_subprocess(command: str, *, quiet: bool = False) -> None:
-    try:
-        result = subprocess.run(command, shell=True, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, check=True)
-        if not quiet:
-            print(result.stdout.decode())
-    except subprocess.CalledProcessError as e:
-        print(e.stdout.decode())
-        raise
+async def run_subprocess(command: str, *, quiet: bool = False) -> None:
+    process = await asyncio.create_subprocess_shell(
+        command,
+        stdout=asyncio.subprocess.PIPE,
+        stderr=asyncio.subprocess.STDOUT,
+    )
+    stdout, _ = await process.communicate()
+    if process.returncode != 0:
+        print(stdout.decode())
+        raise subprocess.CalledProcessError(process.returncode, command, stdout)
+    if not quiet:
+        print(stdout.decode())

livesync/sync.py

@@ -19,13 +19,13 @@ async def run_folder_tasks(
             mutexes = {folder.host: Mutex(folder.host, folder.ssh_port) for folder in folders}
             for mutex in mutexes.values():
                 print(f'Checking mutex on {mutex.host}', flush=True)
-                if not mutex.set(summary):
+                if not await mutex.set(summary):
                     print(f'Target is in use by {mutex.occupant}')
                     sys.exit(1)
 
         for folder in folders:
             print(f'  {folder.source_path} --> {folder.target}', flush=True)
-            folder.sync()
+            await folder.sync()
 
         if watch:
             for folder in folders:
@@ -36,7 +36,7 @@ async def run_folder_tasks(
                 if not ignore_mutex:
                     summary = get_summary(folders)
                     for mutex in mutexes.values():
-                        if not mutex.set(summary):
+                        if not await mutex.set(summary):
                             break
                 await asyncio.sleep(mutex_interval)
     except Exception as e: