Update operations that run system commands to use trio.to_thread.run_…

…sync

CHANGELOG.md

@@ -2,6 +2,12 @@
 
 This is the changelog for RedPepper.
 
+## [Unreleased]
+
+### Changed
+
+- Update operations that run system commands to use `trio.to_thread.run_sync` for better concurrency.
+
 ## [0.3.0]
 
 ### Changed

src/operations/redpepper/operations/apt.py

@@ -1,6 +1,9 @@
+import functools
 import logging
 import subprocess
 
+import trio
+
 from redpepper.operations import Operation, Result
 
 logger = logging.getLogger(__name__)
@@ -29,7 +32,9 @@ async def test(self, agent):
         logger.debug(
             "Testing if apt package %s is installed with command %r", self.name, cmd
         )
-        p = subprocess.run(cmd, capture_output=True, text=True)
+        p = await trio.to_thread.run_sync(
+            functools.partial(subprocess.run, cmd, capture_output=True, text=True)
+        )
         if p.returncode == 1:
             logger.debug("dpkg-query returned 1")
             return False
@@ -63,7 +68,11 @@ async def run(self, agent):
         env["APT_LISTCHANGES_FRONTEND"] = "none"  # don't show changelogs
         env["APT_LISTBUGS_FRONTEND"] = "none"  # don't show bug reports
         env["UCF_FORCE_CONFFOLD"] = "1"  # keep old config files
-        p = subprocess.run(cmd, capture_output=True, text=True, env=env)
+        p = await trio.to_thread.run_sync(
+            functools.partial(
+                subprocess.run, cmd, capture_output=True, text=True, env=env
+            )
+        )
         if result.check_completed_process(p).succeeded:
             result.changed = "Setting up" in p.stdout
         return result
@@ -79,7 +88,11 @@ async def run(self, agent):
             await Installed("unattended-upgrades").ensure(agent)
         ).succeeded:
             return result
-        p = subprocess.run(["unattended-upgrades"], capture_output=True, text=True)
+        p = await trio.to_thread.run_sync(
+            functools.partial(
+                subprocess.run, ["unattended-upgrades"], capture_output=True, text=True
+            )
+        )
         if result.check_completed_process(p).succeeded:
             result.changed = result.changed or p.stdout != ""
         return result

src/operations/redpepper/operations/command.py

@@ -2,6 +2,8 @@
 import pwd
 import subprocess
 
+import trio
+
 from redpepper.operations import Operation, Result
 
 
@@ -55,6 +57,9 @@ def __str__(self):
         return f'Run("{self.command}{"" if self.wait else " &"}"{" as " + self.user if self.user else ""})'
 
     async def run(self, agent):
+        return await trio.to_thread.run_sync(self._run)
+
+    def _run(self):
         result = Result(self)
         kw = {}
         kw["env"] = os.environ.copy()

src/operations/redpepper/operations/git.py

@@ -1,7 +1,10 @@
+import functools
 import os
 import pwd
 import subprocess
 
+import trio
+
 from redpepper.operations import Operation, Result
 
 
@@ -72,8 +75,15 @@ async def ensure(self, agent):
         if self.identity:
             kw["env"]["GIT_SSH_COMMAND"] = f'ssh -i "{self.identity}"'
 
-        p = subprocess.run(
-            git_cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, **kw
+        p = await trio.to_thread.run_sync(
+            functools.partial(
+                subprocess.run,
+                git_cmd,
+                stdout=subprocess.PIPE,
+                stderr=subprocess.PIPE,
+                text=True,
+                **kw,
+            )
         )
         if not result.check_completed_process(p).succeeded:
             return result

src/operations/redpepper/operations/py.py

@@ -1,3 +1,5 @@
+import trio
+
 from redpepper.operations import Operation, Result
 
 
@@ -20,7 +22,7 @@ def __str__(self):
     async def ensure(self, agent):
         result = Result(self)
         try:
-            exec(self.script, self.env, locals())
+            await trio.to_thread.run_sync(exec, self.script, self.env, locals())
         except Exception:
             result.exception()
         return result

src/operations/redpepper/operations/sysctl.py

@@ -1,7 +1,10 @@
+import functools
 import os
 import subprocess
 import sys
 
+import trio
+
 from redpepper.operations import Operation, Result
 
 if sys.platform != "linux":
@@ -53,6 +56,10 @@ async def run(self, agent):
             f'Added sysctl parameter "{self.name} = {self.value}" to {SYSCTL_CONF_PATH}'
         )
         result.changed = True
-        p = subprocess.run(["sysctl", "--system"], capture_output=True, text=True)
+        p = await trio.to_thread.run_sync(
+            functools.partial(
+                subprocess.run, ["sysctl", "--system"], capture_output=True, text=True
+            )
+        )
         result.check_completed_process(p)
         return result

src/operations/redpepper/operations/systemd.py

@@ -1,5 +1,8 @@
+import functools
 import subprocess
 
+import trio
+
 from redpepper.operations import Operation, Result
 
 
@@ -14,13 +17,17 @@ def __str__(self):
 
     async def test(self, agent):
         cmd = ["systemctl", "is-active", self.name]
-        p = subprocess.run(cmd, stdout=subprocess.DEVNULL)
+        p = await trio.to_thread.run_sync(
+            functools.partial(subprocess.run, cmd, stdout=subprocess.DEVNULL)
+        )
         return p.returncode == 0
 
     async def run(self, agent):
         result = Result(self)
         cmd = ["systemctl", "start", self.name]
-        p = subprocess.run(cmd, capture_output=True, text=True)
+        p = await trio.to_thread.run_sync(
+            functools.partial(subprocess.run, cmd, capture_output=True, text=True)
+        )
         if result.check_completed_process(p).succeeded:
             result += f"Service {self.name} started."
             result.changed = True
@@ -38,13 +45,17 @@ def __str__(self):
 
     async def test(self, agent):
         cmd = ["systemctl", "is-enabled", self.name]
-        p = subprocess.run(cmd, stdout=subprocess.DEVNULL)
+        p = await trio.to_thread.run_sync(
+            functools.partial(subprocess.run, cmd, stdout=subprocess.DEVNULL)
+        )
         return p.returncode == 0
 
     async def run(self, agent):
         result = Result(self)
         cmd = ["systemctl", "enable", self.name]
-        p = subprocess.run(cmd, capture_output=True, text=True)
+        p = await trio.to_thread.run_sync(
+            functools.partial(subprocess.run, cmd, capture_output=True, text=True)
+        )
         if result.check_completed_process(p).succeeded:
             result += f"Service {self.name} enabled."
             result.changed = True
@@ -61,7 +72,9 @@ def __str__(self):
     async def run(self, agent):
         result = Result(self)
         cmd = ["systemctl", "restart", self.name]
-        p = subprocess.run(cmd, capture_output=True, text=True)
+        p = await trio.to_thread.run_sync(
+            functools.partial(subprocess.run, cmd, capture_output=True, text=True)
+        )
         if result.check_completed_process(p).succeeded:
             result += f"Service {self.name} restarted."
             result.changed = True
@@ -78,7 +91,9 @@ def __str__(self):
     async def run(self, agent):
         result = Result(self)
         cmd = ["systemctl", "reload", self.name]
-        p = subprocess.run(cmd, capture_output=True, text=True)
+        p = await trio.to_thread.run_sync(
+            functools.partial(subprocess.run, cmd, capture_output=True, text=True)
+        )
         if result.check_completed_process(p).succeeded:
             result += f"Service {self.name} reloaded."
             result.changed = True