diff --git a/devlib/bin/scripts/devlib-signal-target b/devlib/bin/scripts/devlib-signal-target
new file mode 100644
index 000000000..26ac6fe84
--- /dev/null
+++ b/devlib/bin/scripts/devlib-signal-target
@@ -0,0 +1,20 @@
+(
+    # If there is no data dir, it means we are not running as a background
+    # command so we just do nothing
+    if [ -e "$_DEVLIB_BG_CMD_DATA_DIR" ]; then
+        pid_file="$_DEVLIB_BG_CMD_DATA_DIR/pid"
+        # Atomically check if the PID file already exist and make the write
+        # fail if it already does. This way we don't have any race condition
+        # with the Python API, as there is either no PID or the same PID for
+        # the duration of the command
+        set -o noclobber
+        if ! printf "%u\n" $$ > "$pid_file"; then
+            echo "$0 was already called for this command" >&2
+            exit 1
+        fi
+    fi
+) || exit $?
+
+# Use exec so that the PID of the command we run is the same as the current $$
+# PID that we just registered
+exec "$@"
diff --git a/devlib/collector/ftrace.py b/devlib/collector/ftrace.py
index 9a887efdf..319cd36cc 100644
--- a/devlib/collector/ftrace.py
+++ b/devlib/collector/ftrace.py
@@ -21,6 +21,7 @@
 import sys
 import contextlib
 from shlex import quote
+import signal
 
 from devlib.collector import (CollectorBase, CollectorOutput,
                               CollectorOutputEntry)
@@ -71,6 +72,7 @@ def __init__(self, target,
                  report_on_target=False,
                  trace_clock='local',
                  saved_cmdlines_nr=4096,
+                 mode='write-to-memory',
                  ):
         super(FtraceCollector, self).__init__(target)
         self.events = events if events is not None else DEFAULT_EVENTS
@@ -98,6 +100,8 @@ def __init__(self, target,
         self.trace_clock = trace_clock
         self.saved_cmdlines_nr = saved_cmdlines_nr
         self._reset_needed = True
+        self.mode = mode
+        self._bg_cmd = None
 
         # pylint: disable=bad-whitespace
         # Setup tracing paths
@@ -288,18 +292,33 @@ async def start(self):
         with contextlib.suppress(TargetStableError):
             self.target.write_value('/proc/sys/kernel/kptr_restrict', 0)
 
-        self.target.execute(
-            '{} start -B devlib {buffer_size} {cmdlines_size} {clock} {events} {tracer} {functions}'.format(
-                self.target_binary,
-                events=self.event_string,
-                tracer=tracer_string,
-                functions=tracecmd_functions,
-                buffer_size='-b {}'.format(self.buffer_size) if self.buffer_size is not None else '',
-                clock='-C {}'.format(self.trace_clock) if self.trace_clock else '',
-                cmdlines_size='--cmdlines-size {}'.format(self.saved_cmdlines_nr) if self.saved_cmdlines_nr is not None else '',
-            ),
-            as_root=True,
+        params = '-B devlib {buffer_size} {cmdlines_size} {clock} {events} {tracer} {functions}'.format(
+            events=self.event_string,
+            tracer=tracer_string,
+            functions=tracecmd_functions,
+            buffer_size='-b {}'.format(self.buffer_size) if self.buffer_size is not None else '',
+            clock='-C {}'.format(self.trace_clock) if self.trace_clock else '',
+            cmdlines_size='--cmdlines-size {}'.format(self.saved_cmdlines_nr) if self.saved_cmdlines_nr is not None else '',
         )
+
+        mode = self.mode
+        if mode == 'write-to-disk':
+            bg_cmd = self.target.background(
+                # cd into the working_directory first to workaround this issue:
+                # https://lore.kernel.org/linux-trace-devel/20240119162743.1a107fa9@gandalf.local.home/
+                f'cd {self.target.working_directory} && devlib-signal-target {self.target_binary} record -o {quote(self.target_output_file)} {params}',
+                as_root=True,
+            )
+            assert self._bg_cmd is None
+            self._bg_cmd = bg_cmd.__enter__()
+        elif mode == 'write-to-memory':
+            self.target.execute(
+                f'{self.target_binary} start {params}',
+                as_root=True,
+            )
+        else:
+            raise ValueError(f'Unknown mode {mode}')
+
         if self.automark:
             self.mark_start()
         if 'cpufreq' in self.target.modules:
@@ -334,8 +353,21 @@ def stop(self):
         self.stop_time = time.time()
         if self.automark:
             self.mark_stop()
-        self.target.execute('{} stop -B devlib'.format(self.target_binary),
-                            timeout=TIMEOUT, as_root=True)
+
+        mode = self.mode
+        if mode == 'write-to-disk':
+            bg_cmd = self._bg_cmd
+            self._bg_cmd = None
+            assert bg_cmd is not None
+            bg_cmd.send_signal(signal.SIGINT)
+            bg_cmd.communicate()
+            bg_cmd.__exit__(None, None, None)
+        elif mode == 'write-to-memory':
+            self.target.execute('{} stop -B devlib'.format(self.target_binary),
+                                timeout=TIMEOUT, as_root=True)
+        else:
+            raise ValueError(f'Unknown mode {mode}')
+
         self._reset_needed = True
 
     def set_output(self, output_path):
@@ -346,9 +378,18 @@ def set_output(self, output_path):
     def get_data(self):
         if self.output_path is None:
             raise RuntimeError("Output path was not set.")
-        self.target.execute('{0} extract -B devlib -o {1}; chmod 666 {1}'.format(self.target_binary,
-                                                                       self.target_output_file),
-                            timeout=TIMEOUT, as_root=True)
+
+        busybox = quote(self.target.busybox)
+
+        mode = self.mode
+        if mode == 'write-to-disk':
+            # Interrupting trace-cmd record will make it create the file
+            pass
+        elif mode == 'write-to-memory':
+            cmd = f'{self.target_binary} extract -B devlib -o {self.target_output_file} && {busybox} chmod 666 {self.target_output_file}'
+            self.target.execute(cmd, timeout=TIMEOUT, as_root=True)
+        else:
+            raise ValueError(f'Unknown mode {mode}')
 
         # The size of trace.dat will depend on how long trace-cmd was running.
         # Therefore timout for the pull command must also be adjusted
diff --git a/devlib/connection.py b/devlib/connection.py
index 460997580..99055a3c1 100644
--- a/devlib/connection.py
+++ b/devlib/connection.py
@@ -17,6 +17,7 @@
 from contextlib import contextmanager, nullcontext
 from shlex import quote
 import os
+from pathlib import Path
 import signal
 import subprocess
 import threading
@@ -25,14 +26,11 @@
 import select
 import fcntl
 
-from devlib.utils.misc import InitCheckpoint
+from devlib.utils.misc import InitCheckpoint, memoized
 
 _KILL_TIMEOUT = 3
 
 
-def _kill_pgid_cmd(pgid, sig, busybox):
-    return '{} kill -{} -{}'.format(busybox, sig.value, pgid)
-
 def _popen_communicate(bg, popen, input, timeout):
     try:
         stdout, stderr = popen.communicate(input=input, timeout=timeout)
@@ -130,8 +128,11 @@ class BackgroundCommand(ABC):
     semantic as :class:`subprocess.Popen`.
     """
 
-    def __init__(self, conn):
+    def __init__(self, conn, data_dir, cmd, as_root):
         self.conn = conn
+        self._data_dir = data_dir
+        self.as_root = as_root
+        self.cmd = cmd
 
         # Poll currently opened background commands on that connection to make
         # them deregister themselves if they are completed. This avoids
@@ -147,15 +148,65 @@ def __init__(self, conn):
 
         conn._current_bg_cmds.add(self)
 
+    @classmethod
+    def from_factory(cls, conn, cmd, as_root, make_init_kwargs):
+        cmd, data_dir = cls._with_data_dir(conn, cmd)
+        return cls(
+            conn=conn,
+            data_dir=data_dir,
+            cmd=cmd,
+            as_root=as_root,
+            **make_init_kwargs(cmd),
+        )
+
     def _deregister(self):
         try:
             self.conn._current_bg_cmds.remove(self)
         except KeyError:
             pass
 
-    @abstractmethod
-    def _send_signal(self, sig):
-        pass
+    @property
+    def _pid_file(self):
+        return str(Path(self._data_dir, 'pid'))
+
+    @property
+    @memoized
+    def _targeted_pid(self):
+        """
+        PID of the process pointed at by ``devlib-signal-target`` command.
+        """
+        path = quote(self._pid_file)
+        busybox = quote(self.conn.busybox)
+
+        def execute(cmd):
+            return self.conn.execute(cmd, as_root=self.as_root)
+
+        while self.poll() is None:
+            try:
+                pid = execute(f'{busybox} cat {path}')
+            except subprocess.CalledProcessError:
+                time.sleep(0.01)
+            else:
+                if pid.endswith('\n'):
+                    return int(pid.strip())
+                else:
+                    # We got a partial write in the PID file
+                    continue
+
+        raise ValueError(f'The background commmand did not use devlib-signal-target wrapper to designate which command should be the target of signals')
+
+    @classmethod
+    def _with_data_dir(cls, conn, cmd):
+        busybox = quote(conn.busybox)
+        data_dir = conn.execute(f'{busybox} mktemp -d').strip()
+        cmd = f'_DEVLIB_BG_CMD_DATA_DIR={data_dir} exec {busybox} sh -c {quote(cmd)}'
+        return cmd, data_dir
+
+    def _cleanup_data_dir(self):
+        path = quote(self._data_dir)
+        busybox = quote(self.conn.busybox)
+        cmd = f'{busybox} rm -r {path} || true'
+        self.conn.execute(cmd, as_root=self.as_root)
 
     def send_signal(self, sig):
         """
@@ -165,8 +216,29 @@ def send_signal(self, sig):
         :param signal: Signal to send.
         :type signal: signal.Signals
         """
+
+        def execute(cmd):
+            return self.conn.execute(cmd, as_root=self.as_root)
+
+        def send(sig):
+            busybox = quote(self.conn.busybox)
+            # If the command has already completed, we don't want to send a
+            # signal to another process that might have gotten that PID in the
+            # meantime.
+            if self.poll() is None:
+                if sig in (signal.SIGTERM, signal.SIGQUIT, signal.SIGKILL):
+                    # Use -PGID to target a process group rather than just the
+                    # process itself. This will work in any condition and will
+                    # not require cooperation from the command.
+                    execute(f'{busybox} kill -{sig.value} -{self.pid}')
+                else:
+                    # Other signals require cooperation from the shell command
+                    # so that it points to a specific process using
+                    # devlib-signal-target
+                    pid = self._targeted_pid
+                    execute(f'{busybox} kill -{sig.value} {pid}')
         try:
-            return self._send_signal(sig)
+            return send(sig)
         finally:
             # Deregister if the command has finished
             self.poll()
@@ -287,6 +359,7 @@ def close(self):
             return self._close()
         finally:
             self._deregister()
+            self._cleanup_data_dir()
 
     def __enter__(self):
         return self
@@ -300,13 +373,15 @@ class PopenBackgroundCommand(BackgroundCommand):
     :class:`subprocess.Popen`-based background command.
     """
 
-    def __init__(self, conn, popen):
-        super().__init__(conn=conn)
+    def __init__(self, conn, data_dir, cmd, as_root, popen):
+        super().__init__(
+            conn=conn,
+            data_dir=data_dir,
+            cmd=cmd,
+            as_root=as_root,
+        )
         self.popen = popen
 
-    def _send_signal(self, sig):
-        return os.killpg(self.popen.pid, sig)
-
     @property
     def stdin(self):
         return self.popen.stdin
@@ -354,26 +429,20 @@ class ParamikoBackgroundCommand(BackgroundCommand):
     """
     :mod:`paramiko`-based background command.
     """
-    def __init__(self, conn, chan, pid, as_root, cmd, stdin, stdout, stderr, redirect_thread):
-        super().__init__(conn=conn)
+    def __init__(self, conn, data_dir, cmd, as_root, chan, pid, stdin, stdout, stderr, redirect_thread):
+        super().__init__(
+            conn=conn,
+            data_dir=data_dir,
+            cmd=cmd,
+            as_root=as_root,
+        )
+
         self.chan = chan
-        self.as_root = as_root
         self._pid = pid
         self._stdin = stdin
         self._stdout = stdout
         self._stderr = stderr
         self.redirect_thread = redirect_thread
-        self.cmd = cmd
-
-    def _send_signal(self, sig):
-        # If the command has already completed, we don't want to send a signal
-        # to another process that might have gotten that PID in the meantime.
-        if self.poll() is not None:
-            return
-        # Use -PGID to target a process group rather than just the process
-        # itself
-        cmd = _kill_pgid_cmd(self.pid, sig, self.conn.busybox)
-        self.conn.execute(cmd, as_root=self.as_root)
 
     @property
     def pid(self):
@@ -517,18 +586,16 @@ class AdbBackgroundCommand(BackgroundCommand):
     ``adb``-based background command.
     """
 
-    def __init__(self, conn, adb_popen, pid, as_root):
-        super().__init__(conn=conn)
-        self.as_root = as_root
+    def __init__(self, conn, data_dir, cmd, as_root, adb_popen, pid):
+        super().__init__(
+            conn=conn,
+            data_dir=data_dir,
+            cmd=cmd,
+            as_root=as_root,
+        )
         self.adb_popen = adb_popen
         self._pid = pid
 
-    def _send_signal(self, sig):
-        self.conn.execute(
-            _kill_pgid_cmd(self.pid, sig, self.conn.busybox),
-            as_root=self.as_root,
-        )
-
     @property
     def stdin(self):
         return self.adb_popen.stdin
@@ -638,7 +705,7 @@ def cancel(self):
 
 
 class PopenTransferHandle(TransferHandleBase):
-    def __init__(self, bg_cmd, dest, direction, *args, **kwargs):
+    def __init__(self, popen, dest, direction, *args, **kwargs):
         super().__init__(*args, **kwargs)
 
         if direction == 'push':
@@ -650,7 +717,7 @@ def __init__(self, bg_cmd, dest, direction, *args, **kwargs):
 
         self.sample_size = lambda: sample_size(dest)
 
-        self.bg_cmd = bg_cmd
+        self.popen = popen
         self.last_sample = 0
 
     @staticmethod
@@ -671,7 +738,7 @@ def _push_dest_size(self, dest):
         return int(out.split()[0])
 
     def cancel(self):
-        self.bg_cmd.cancel()
+        self.popen.terminate()
 
     def isactive(self):
         try:
diff --git a/devlib/host.py b/devlib/host.py
index a9958a349..a65b00f9f 100644
--- a/devlib/host.py
+++ b/devlib/host.py
@@ -147,16 +147,25 @@ def background(self, command, stdout=subprocess.PIPE, stderr=subprocess.PIPE, as
         def preexec_fn():
             os.setpgrp()
 
-        popen = subprocess.Popen(
-            command,
-            stdout=stdout,
-            stderr=stderr,
-            stdin=subprocess.PIPE,
-            shell=True,
-            preexec_fn=preexec_fn,
+        def make_init_kwargs(command):
+            popen = subprocess.Popen(
+                command,
+                stdout=stdout,
+                stderr=stderr,
+                stdin=subprocess.PIPE,
+                shell=True,
+                preexec_fn=preexec_fn,
+            )
+            return dict(
+                popen=popen,
+            )
+
+        return PopenBackgroundCommand.from_factory(
+            conn=self,
+            cmd=command,
+            as_root=as_root,
+            make_init_kwargs=make_init_kwargs,
         )
-        bg_cmd = PopenBackgroundCommand(self, popen)
-        return bg_cmd
 
     def _close(self):
         pass
diff --git a/devlib/module/cgroups2.py b/devlib/module/cgroups2.py
index 83cbf3948..a632bbee6 100644
--- a/devlib/module/cgroups2.py
+++ b/devlib/module/cgroups2.py
@@ -335,35 +335,28 @@ def _mount_v2_controllers(target: LinuxTarget):
 
     :yield: The path to the root of the mounted V2 controller hierarchy.
     :rtype: str
-    
-    :raises TargetStableError: Occurs in the case where the root directory of the requested CGroup V2 Controller hierarchy 
+
+    :raises TargetStableError: Occurs in the case where the root directory of the requested CGroup V2 Controller hierarchy
         is unable to be created up on the target system.
     """
 
-    path = target.tempfile()
-    
-    try:
-        target.makedirs(path, as_root=True)
-    except TargetStableCalledProcessError:
-        raise TargetStableError("Un-able to create the root directory of the requested CGroup V2 hierarchy")
-        
-        
-    try:
-        target.execute(
-            "{busybox} mount -t cgroup2 none {path}".format(
-                busybox=quote(target.busybox), path=quote(path)
-            ),
-            as_root=True,
-        )
-        yield path
-    finally:
-        target.execute(
-            "{busybox} umount {path} && {busybox} rmdir -- {path}".format(
-                busybox=quote(target.busybox),
-                path=quote(path),
-            ),
-            as_root=True,
-        )
+    with target.make_temp() as path:
+        try:
+            target.execute(
+                "{busybox} mount -t cgroup2 none {path}".format(
+                    busybox=quote(target.busybox), path=quote(path)
+                ),
+                as_root=True,
+            )
+            yield path
+        finally:
+            target.execute(
+                "{busybox} umount {path}".format(
+                    busybox=quote(target.busybox),
+                    path=quote(path),
+                ),
+                as_root=True,
+            )
 
 
 @contextmanager
@@ -379,8 +372,8 @@ def _mount_v1_controllers(target: LinuxTarget, controllers: Set[str]):
 
     :yield: A dictionary mapping CGroup controller names to the paths that they're currently mounted at.
     :rtype: Dict[str,str]
-    
-    :raises TargetStableError: Occurs in the case where the root directory of a requested CGroup V1 Controller hierarchy 
+
+    :raises TargetStableError: Occurs in the case where the root directory of a requested CGroup V1 Controller hierarchy
         is unable to be created up on the target system.
     """
 
@@ -388,33 +381,25 @@ def _mount_v1_controllers(target: LinuxTarget, controllers: Set[str]):
     # its mount path.
     @contextmanager
     def _mount_controller(controller):
+        with target.make_temp() as path:
+            try:
+                target.execute(
+                    "{busybox} mount -t cgroup -o {controller} none {path}".format(
+                        busybox=quote(target.busybox),
+                        controller=quote(controller),
+                        path=quote(path),
+                    ),
+                )
 
-        path = target.tempfile()
-        
-        try:
-            target.makedirs(path, as_root=True)
-        except TargetStableCalledProcessError as err:
-            raise TargetStableError("Un-able to create the root directory of the {controller} CGroup V1 hierarchy".format(controller = controller))
-
-        try:
-            target.execute(
-                "{busybox} mount -t cgroup -o {controller} none {path}".format(
-                    busybox=quote(target.busybox),
-                    controller=quote(controller),
-                    path=quote(path),
-                ),
-            )
-
-            yield path
-
-        finally:
-            target.execute(
-                "{busybox} umount {path} && {busybox} rmdir -- {path}".format(
-                    busybox=quote(target.busybox),
-                    path=quote(path),
-                ),
-                as_root=True,
-            )
+                yield path
+            finally:
+                target.execute(
+                    "{busybox} umount {path}".format(
+                        busybox=quote(target.busybox),
+                        path=quote(path),
+                    ),
+                    as_root=True,
+                )
 
     with ExitStack() as stack:
         yield {
@@ -569,7 +554,7 @@ def _add_process(self, pid: Union[str, int]):
             )
         except TargetStableError:
             self._set_controller_attribute("cgroup", "procs", pid)
-        
+
         else:
             if str(pid) not in member_processes:
                 self._set_controller_attribute("cgroup", "procs", pid)
diff --git a/devlib/target.py b/devlib/target.py
index 44e7fc685..fee70e333 100644
--- a/devlib/target.py
+++ b/devlib/target.py
@@ -281,7 +281,7 @@ def page_size_kb(self):
     @property
     def shutils(self):
         if self._shutils is None:
-            self._setup_shutils()
+            self._setup_scripts()
         return self._shutils
 
     def is_running(self, comm):
@@ -320,6 +320,7 @@ def __init__(self,
                  conn_cls=None,
                  is_container=False,
                  max_async=50,
+                 tmp_directory=None,
                  ):
 
         self._lock = threading.RLock()
@@ -347,6 +348,7 @@ def __init__(self,
         self.connection_settings['platform'] = self.platform
         self.working_directory = working_directory
         self.executables_directory = executables_directory
+        self.tmp_directory = tmp_directory
         self.load_default_modules = load_default_modules
         self.shell_prompt = bytes_regex(shell_prompt)
         self.conn_cls = conn_cls
@@ -356,7 +358,6 @@ def __init__(self,
         self._installed_modules = {}
         self._cache = {}
         self._shutils = None
-        self._file_transfer_cache = None
         self._max_async = max_async
         self.busybox = None
 
@@ -477,10 +478,35 @@ async def connect(self, timeout=None, check_boot_completed=True, max_async=None)
             self.wait_boot_complete(timeout)
         self.check_connection()
         self._resolve_paths()
-        self.execute('mkdir -p {}'.format(quote(self.working_directory)))
-        self.execute('mkdir -p {}'.format(quote(self.executables_directory)))
+        assert self.working_directory
+        if self.executables_directory is None:
+            self.executables_directory = self.path.join(
+                self.working_directory,
+                'bin'
+            )
+
+        for path in (self.working_directory, self.executables_directory):
+            self.makedirs(path)
+
         self.busybox = self.install(os.path.join(PACKAGE_BIN_DIRECTORY, self.abi, 'busybox'), timeout=30)
         self.conn.busybox = self.busybox
+
+        # If neither the mktemp call nor _resolve_paths() managed to get a
+        # temporary directory, we just make one in the working directory.
+        if self.tmp_directory is None:
+            assert self.busybox
+            try:
+                tmp = await self.execute.asyn(f'{quote(self.busybox)} mktemp -d')
+            except Exception:
+                # Some Android platforms don't have a working mktemp unless
+                # TMPDIR is set, so we let AndroidTarget._resolve_paths() deal
+                # with finding a suitable location.
+                tmp = self.path.join(self.working_directory, 'tmp')
+            else:
+                tmp = tmp.strip()
+            self.tmp_directory = tmp
+        self.makedirs(self.tmp_directory)
+
         self._detect_max_async(max_async or self._max_async)
         self.platform.update_from_target(self)
         self._update_modules('connected')
@@ -591,7 +617,7 @@ def wait_boot_complete(self, timeout=10):
 
     @asyn.asyncf
     async def setup(self, executables=None):
-        await self._setup_shutils.asyn()
+        await self._setup_scripts.asyn()
 
         for host_exe in (executables or []):  # pylint: disable=superfluous-parens
             await self.install.asyn(host_exe)
@@ -602,8 +628,6 @@ async def setup(self, executables=None):
         # Initialize modules which requires Busybox (e.g. shutil dependent tasks)
         self._update_modules('setup')
 
-        await self.execute.asyn('mkdir -p {}'.format(quote(self._file_transfer_cache)))
-
     def reboot(self, hard=False, connect=True, timeout=180):
         if hard:
             if not self.has('hard_reset'):
@@ -637,24 +661,11 @@ async def _xfer_cache_path(self, name):
         Context manager to provide a unique path in the transfer cache with the
         basename of the given name.
         """
-        # Use a UUID to avoid race conditions on the target side
-        xfer_uuid = uuid.uuid4().hex
-        folder = self.path.join(self._file_transfer_cache, xfer_uuid)
         # Make sure basename will work on folders too
         name = os.path.normpath(name)
-        # Ensure the name is relative so that os.path.join() will actually
-        # join the paths rather than ignoring the first one.
-        name = './{}'.format(os.path.basename(name))
-
-        check_rm = False
-        try:
-            await self.makedirs.asyn(folder)
-            # Don't check the exit code as the folder might not even exist
-            # before this point, if creating it failed
-            check_rm = True
-            yield self.path.join(folder, name)
-        finally:
-            await self.execute.asyn('rm -rf -- {}'.format(quote(folder)), check_exit_code=check_rm)
+        name = os.path.basename(name)
+        async with self.make_temp() as tmp:
+            yield self.path.join(tmp, name)
 
     @asyn.asyncf
     async def _prepare_xfer(self, action, sources, dest, pattern=None, as_root=False):
@@ -944,15 +955,17 @@ async def get_directory(self, source_dir, dest, as_root=False):
     # execution
 
     def _prepare_cmd(self, command, force_locale):
+        tmpdir = f'TMPDIR={quote(self.tmp_directory)}' if self.tmp_directory else ''
+
         # Force the locale if necessary for more predictable output
         if force_locale:
             # Use an explicit export so that the command is allowed to be any
             # shell statement, rather than just a command invocation
-            command = 'export LC_ALL={} && {}'.format(quote(force_locale), command)
+            command = f'export LC_ALL={quote(force_locale)} {tmpdir} && {command}'
 
         # Ensure to use deployed command when availables
         if self.executables_directory:
-            command = "export PATH={}:$PATH && {}".format(quote(self.executables_directory), command)
+            command = f"export PATH={quote(self.executables_directory)}:$PATH && {command}"
 
         return command
 
@@ -1178,7 +1191,7 @@ async def write_value(self, path, value, verify=True, as_root=True):
                 raise
 
     @asyn.asynccontextmanager
-    async def make_temp(self, is_directory=True, directory='', prefix='devlib-test'):
+    async def make_temp(self, is_directory=True, directory=None, prefix=None):
         """
         Creates temporary file/folder on target and deletes it once it's done.
 
@@ -1196,10 +1209,11 @@ async def make_temp(self, is_directory=True, directory='', prefix='devlib-test')
         :rtype: str
         """
 
-        directory = directory or self.working_directory
+        directory = directory or self.tmp_directory
+        prefix = f'{prefix}-' if prefix else ''
         temp_obj = None
         try:
-            cmd = f'mktemp -p {quote(directory)} {quote(prefix)}-XXXXXX'
+            cmd = f'mktemp -p {quote(directory)} {quote(prefix)}XXXXXX'
             if is_directory:
                 cmd += ' -d'
 
@@ -1296,13 +1310,15 @@ def get_workpath(self, name):
         return self.path.join(self.working_directory, name)
 
     @asyn.asyncf
-    async def tempfile(self, prefix='', suffix=''):
-        name = '{prefix}_{uuid}_{suffix}'.format(
+    async def tempfile(self, prefix=None, suffix=None):
+        prefix = f'{prefix}-' if prefix else ''
+        suffix = f'-{suffix}' if suffix else ''
+        name = '{prefix}{uuid}{suffix}'.format(
             prefix=prefix,
             uuid=uuid.uuid4().hex,
             suffix=suffix,
         )
-        path = self.get_workpath(name)
+        path = self.path.join(self.tmp_directory, name)
         if (await self.file_exists.asyn(path)):
             raise FileExistsError('Path already exists on the target: {}'.format(path))
         else:
@@ -1562,8 +1578,9 @@ def install_module(self, mod, **params):
     # internal methods
 
     @asyn.asyncf
-    async def _setup_shutils(self):
-        shutils_ifile = os.path.join(PACKAGE_BIN_DIRECTORY, 'scripts', 'shutils.in')
+    async def _setup_scripts(self):
+        scripts = os.path.join(PACKAGE_BIN_DIRECTORY, 'scripts')
+        shutils_ifile = os.path.join(scripts, 'shutils.in')
         with open(shutils_ifile) as fh:
             lines = fh.readlines()
         with tempfile.TemporaryDirectory() as folder:
@@ -1574,6 +1591,8 @@ async def _setup_shutils(self):
                     ofile.write(line)
             self._shutils = await self.install.asyn(shutils_ofile)
 
+        await self.install.asyn(os.path.join(scripts, 'devlib-signal-target'))
+
     @asyn.asyncf
     @call_conn
     async def _execute_util(self, command, timeout=None, check_exit_code=True, as_root=False):
@@ -1775,6 +1794,7 @@ def __init__(self,
                  conn_cls=SshConnection,
                  is_container=False,
                  max_async=50,
+                 tmp_directory=None,
                  ):
         super(LinuxTarget, self).__init__(connection_settings=connection_settings,
                                           platform=platform,
@@ -1786,7 +1806,9 @@ def __init__(self,
                                           shell_prompt=shell_prompt,
                                           conn_cls=conn_cls,
                                           is_container=is_container,
-                                          max_async=max_async)
+                                          max_async=max_async,
+                                          tmp_directory=tmp_directory,
+                                          )
 
     def wait_boot_complete(self, timeout=10):
         pass
@@ -1854,12 +1876,11 @@ async def capture_screen(self, filepath):
             return
         try:
 
-            tmpfile = await self.tempfile.asyn()
-            cmd = 'DISPLAY=:0.0 scrot {} && {} date -u -Iseconds'
-            ts = (await self.execute.asyn(cmd.format(quote(tmpfile), quote(self.busybox)))).strip()
-            filepath = filepath.format(ts=ts)
-            await self.pull.asyn(tmpfile, filepath)
-            await self.remove.asyn(tmpfile)
+            async with self.make_temp(is_directory=False) as tmpfile:
+                cmd = 'DISPLAY=:0.0 scrot {} && {} date -u -Iseconds'
+                ts = (await self.execute.asyn(cmd.format(quote(tmpfile), quote(self.busybox)))).strip()
+                filepath = filepath.format(ts=ts)
+                await self.pull.asyn(tmpfile, filepath)
         except TargetStableError as e:
             if "Can't open X dispay." not in e.message:
                 raise e
@@ -1868,10 +1889,8 @@ async def capture_screen(self, filepath):
 
     def _resolve_paths(self):
         if self.working_directory is None:
+            # This usually lands in the home directory
             self.working_directory = self.path.join(self.execute("pwd").strip(), 'devlib-target')
-        self._file_transfer_cache = self.path.join(self.working_directory, '.file-cache')
-        if self.executables_directory is None:
-            self.executables_directory = self.path.join(self.working_directory, 'bin')
 
 
 class AndroidTarget(Target):
@@ -1986,6 +2005,7 @@ def __init__(self,
                  package_data_directory="/data/data",
                  is_container=False,
                  max_async=50,
+                 tmp_directory=None,
                  ):
         super(AndroidTarget, self).__init__(connection_settings=connection_settings,
                                             platform=platform,
@@ -1997,7 +2017,9 @@ def __init__(self,
                                             shell_prompt=shell_prompt,
                                             conn_cls=conn_cls,
                                             is_container=is_container,
-                                            max_async=max_async)
+                                            max_async=max_async,
+                                            tmp_directory=tmp_directory,
+                                            )
         self.package_data_directory = package_data_directory
         self._init_logcat_lock()
 
@@ -2587,9 +2609,16 @@ async def homescreen(self):
     def _resolve_paths(self):
         if self.working_directory is None:
             self.working_directory = self.path.join(self.external_storage, 'devlib-target')
-        self._file_transfer_cache = self.path.join(self.working_directory, '.file-cache')
+        if self.tmp_directory is None:
+            # Do not rely on the generic default here, as we need to provide an
+            # android-specific default in case it fails.
+            try:
+                tmp = self.execute(f'{quote(self.busybox)} mktemp -d')
+            except Exception:
+                tmp = '/data/local/tmp'
+            self.tmp_directory = tmp
         if self.executables_directory is None:
-            self.executables_directory = '/data/local/tmp/bin'
+            self.executables_directory = self.path.join(self.tmp_directory, 'bin')
 
     @asyn.asyncf
     async def _ensure_executables_directory_is_writable(self):
@@ -3064,6 +3093,7 @@ def __init__(self,
                  conn_cls=LocalConnection,
                  is_container=False,
                  max_async=50,
+                 tmp_directory=None,
                  ):
         super(LocalLinuxTarget, self).__init__(connection_settings=connection_settings,
                                                platform=platform,
@@ -3075,14 +3105,13 @@ def __init__(self,
                                                shell_prompt=shell_prompt,
                                                conn_cls=conn_cls,
                                                is_container=is_container,
-                                               max_async=max_async)
+                                               max_async=max_async,
+                                               tmp_directory=tmp_directory,
+                                               )
 
     def _resolve_paths(self):
         if self.working_directory is None:
             self.working_directory = '/tmp/devlib-target'
-        self._file_transfer_cache = self.path.join(self.working_directory, '.file-cache')
-        if self.executables_directory is None:
-            self.executables_directory = '/tmp/devlib-target/bin'
 
 
 def _get_model_name(section):
@@ -3153,6 +3182,7 @@ def __init__(self,
                  package_data_directory="/data/data",
                  is_container=False,
                  max_async=50,
+                 tmp_directory=None,
                  ):
 
         self.supports_android = None
@@ -3181,7 +3211,9 @@ def __init__(self,
             shell_prompt=shell_prompt,
             conn_cls=SshConnection,
             is_container=is_container,
-            max_async=max_async)
+            max_async=max_async,
+            tmp_directory=tmp_directory,
+            )
 
         # We can't determine if the target supports android until connected to the linux host so
         # create unconditionally.
@@ -3243,6 +3275,3 @@ async def connect(self, timeout=30, check_boot_completed=True, max_async=None):
     def _resolve_paths(self):
         if self.working_directory is None:
             self.working_directory = '/mnt/stateful_partition/devlib-target'
-        self._file_transfer_cache = self.path.join(self.working_directory, '.file-cache')
-        if self.executables_directory is None:
-            self.executables_directory = self.path.join(self.working_directory, 'bin')
diff --git a/devlib/utils/android.py b/devlib/utils/android.py
index c77a86446..001cb93be 100755
--- a/devlib/utils/android.py
+++ b/devlib/utils/android.py
@@ -40,7 +40,7 @@
 
 from devlib.exception import TargetTransientError, TargetStableError, HostError, TargetTransientCalledProcessError, TargetStableCalledProcessError, AdbRootError
 from devlib.utils.misc import check_output, which, ABI_MAP, redirect_streams, get_subprocess
-from devlib.connection import ConnectionBase, AdbBackgroundCommand, PopenBackgroundCommand, PopenTransferHandle
+from devlib.connection import ConnectionBase, AdbBackgroundCommand, PopenTransferHandle
 
 
 logger = logging.getLogger('android')
@@ -341,7 +341,7 @@ def _push_pull(self, action, sources, dest, timeout):
         if timeout:
             adb_command(self.device, command, timeout=timeout, adb_server=self.adb_server, adb_port=self.adb_port)
         else:
-            bg_cmd = adb_command_background(
+            popen = adb_command_popen(
                 device=self.device,
                 conn=self,
                 command=command,
@@ -351,12 +351,12 @@ def _push_pull(self, action, sources, dest, timeout):
 
             handle = PopenTransferHandle(
                 manager=self.transfer_manager,
-                bg_cmd=bg_cmd,
+                popen=popen,
                 dest=dest,
                 direction=action
             )
-            with bg_cmd, self.transfer_manager.manage(sources, dest, action, handle):
-                bg_cmd.communicate()
+            with popen, self.transfer_manager.manage(sources, dest, action, handle):
+                popen.communicate()
 
     # pylint: disable=unused-argument
     def execute(self, command, timeout=None, check_exit_code=False,
@@ -387,12 +387,18 @@ def background(self, command, stdout=subprocess.PIPE, stderr=subprocess.PIPE, as
         return bg_cmd
 
     def _background(self, command, stdout, stderr, as_root):
-        adb_shell, pid = adb_background_shell(self, command, stdout, stderr, as_root)
-        bg_cmd = AdbBackgroundCommand(
+        def make_init_kwargs(command):
+            adb_popen, pid = adb_background_shell(self, command, stdout, stderr, as_root)
+            return dict(
+                adb_popen=adb_popen,
+                pid=pid,
+            )
+
+        bg_cmd = AdbBackgroundCommand.from_factory(
             conn=self,
-            adb_popen=adb_shell,
-            pid=pid,
-            as_root=as_root
+            cmd=command,
+            as_root=as_root,
+            make_init_kwargs=make_init_kwargs,
         )
         return bg_cmd
 
@@ -756,12 +762,11 @@ def adb_command(device, command, timeout=None, adb_server=None, adb_port=None):
     return output
 
 
-def adb_command_background(device, conn, command, adb_server=None, adb_port=None):
-    full_command = get_adb_command(device, command, adb_server, adb_port)
-    logger.debug(full_command)
-    popen = get_subprocess(full_command, shell=True)
-    cmd = PopenBackgroundCommand(conn=conn, popen=popen)
-    return cmd
+def adb_command_popen(device, conn, command, adb_server=None, adb_port=None):
+    command = get_adb_command(device, command, adb_server, adb_port)
+    logger.debug(command)
+    popen = get_subprocess(command, shell=True)
+    return popen
 
 
 def grant_app_permissions(target, package):
diff --git a/devlib/utils/ssh.py b/devlib/utils/ssh.py
index 9fe4c613a..f9bfacdcd 100644
--- a/devlib/utils/ssh.py
+++ b/devlib/utils/ssh.py
@@ -602,138 +602,142 @@ def background(self, command, stdout=subprocess.PIPE, stderr=subprocess.PIPE, as
             return self._background(command, stdout, stderr, as_root)
 
     def _background(self, command, stdout, stderr, as_root):
-        orig_command = command
-        stdout, stderr, command = redirect_streams(stdout, stderr, command)
-
-        command = "printf '%s\n' $$; exec sh -c {}".format(quote(command))
-        channel = self._make_channel()
-
-        def executor(cmd, timeout):
-            channel.exec_command(cmd)
-            # Read are not buffered so we will always get the data as soon as
-            # they arrive
-            return (
-                channel.makefile_stdin('w', 0),
-                channel.makefile(),
-                channel.makefile_stderr(),
-            )
+        def make_init_kwargs(command):
+            _stdout, _stderr, _command = redirect_streams(stdout, stderr, command)
+
+            _command = "printf '%s\n' $$; exec sh -c {}".format(quote(_command))
+            channel = self._make_channel()
+
+            def executor(cmd, timeout):
+                channel.exec_command(cmd)
+                # Read are not buffered so we will always get the data as soon as
+                # they arrive
+                return (
+                    channel.makefile_stdin('w', 0),
+                    channel.makefile(),
+                    channel.makefile_stderr(),
+                )
 
-        stdin, stdout_in, stderr_in = self._execute_command(
-            command,
-            as_root=as_root,
-            log=False,
-            timeout=None,
-            executor=executor,
-        )
-        pid = stdout_in.readline()
-        if not pid:
-            stderr = stderr_in.read()
-            if channel.exit_status_ready():
-                ret = channel.recv_exit_status()
-            else:
-                ret = 126
-            raise subprocess.CalledProcessError(
-                ret,
-                command,
-                b'',
-                stderr,
+            stdin, stdout_in, stderr_in = self._execute_command(
+                _command,
+                as_root=as_root,
+                log=False,
+                timeout=None,
+                executor=executor,
             )
-        pid = int(pid)
-
-        def create_out_stream(stream_in, stream_out):
-            """
-            Create a pair of file-like objects. The first one is used to read
-            data and the second one to write.
-            """
-
-            if stream_out == subprocess.DEVNULL:
-                r, w = None, None
-            # When asked for a pipe, we just give the file-like object as the
-            # reading end and no writing end, since paramiko already writes to
-            # it
-            elif stream_out == subprocess.PIPE:
-                r, w = os.pipe()
-                r = os.fdopen(r, 'rb')
-                w = os.fdopen(w, 'wb')
-            # Turn a file descriptor into a file-like object
-            elif isinstance(stream_out, int) and stream_out >= 0:
-                r = os.fdopen(stream_in, 'rb')
-                w = os.fdopen(stream_out, 'wb')
-            # file-like object
-            else:
-                r = stream_in
-                w = stream_out
+            pid = stdout_in.readline()
+            if not pid:
+                _stderr = stderr_in.read()
+                if channel.exit_status_ready():
+                    ret = channel.recv_exit_status()
+                else:
+                    ret = 126
+                raise subprocess.CalledProcessError(
+                    ret,
+                    _command,
+                    b'',
+                    _stderr,
+                )
+            pid = int(pid)
+
+            def create_out_stream(stream_in, stream_out):
+                """
+                Create a pair of file-like objects. The first one is used to read
+                data and the second one to write.
+                """
+
+                if stream_out == subprocess.DEVNULL:
+                    r, w = None, None
+                # When asked for a pipe, we just give the file-like object as the
+                # reading end and no writing end, since paramiko already writes to
+                # it
+                elif stream_out == subprocess.PIPE:
+                    r, w = os.pipe()
+                    r = os.fdopen(r, 'rb')
+                    w = os.fdopen(w, 'wb')
+                # Turn a file descriptor into a file-like object
+                elif isinstance(stream_out, int) and stream_out >= 0:
+                    r = os.fdopen(stream_in, 'rb')
+                    w = os.fdopen(stream_out, 'wb')
+                # file-like object
+                else:
+                    r = stream_in
+                    w = stream_out
 
-            return (r, w)
+                return (r, w)
 
-        out_streams = {
-            name: create_out_stream(stream_in, stream_out)
-            for stream_in, stream_out, name in (
-                (stdout_in, stdout, 'stdout'),
-                (stderr_in, stderr, 'stderr'),
-            )
-        }
+            out_streams = {
+                name: create_out_stream(stream_in, stream_out)
+                for stream_in, stream_out, name in (
+                    (stdout_in, _stdout, 'stdout'),
+                    (stderr_in, _stderr, 'stderr'),
+                )
+            }
+
+            def redirect_thread_f(stdout_in, stderr_in, out_streams, select_timeout):
+                def callback(out_streams, name, chunk):
+                    try:
+                        r, w = out_streams[name]
+                    except KeyError:
+                        return out_streams
+
+                    try:
+                        w.write(chunk)
+                    # Write failed
+                    except ValueError:
+                        # Since that stream is now closed, stop trying to write to it
+                        del out_streams[name]
+                        # If that was the last open stream, we raise an
+                        # exception so the thread can terminate.
+                        if not out_streams:
+                            raise
 
-        def redirect_thread_f(stdout_in, stderr_in, out_streams, select_timeout):
-            def callback(out_streams, name, chunk):
-                try:
-                    r, w = out_streams[name]
-                except KeyError:
                     return out_streams
 
                 try:
-                    w.write(chunk)
-                # Write failed
+                    _read_paramiko_streams(stdout_in, stderr_in, select_timeout, callback, copy.copy(out_streams))
+                # The streams closed while we were writing to it, the job is done here
                 except ValueError:
-                    # Since that stream is now closed, stop trying to write to it
-                    del out_streams[name]
-                    # If that was the last open stream, we raise an
-                    # exception so the thread can terminate.
-                    if not out_streams:
-                        raise
-
-                return out_streams
-
-            try:
-                _read_paramiko_streams(stdout_in, stderr_in, select_timeout, callback, copy.copy(out_streams))
-            # The streams closed while we were writing to it, the job is done here
-            except ValueError:
-                pass
-
-            # Make sure the writing end are closed proper since we are not
-            # going to write anything anymore
-            for r, w in out_streams.values():
-                w.flush()
-                if r is not w and w is not None:
-                    w.close()
-
-        # If there is anything we need to redirect to, spawn a thread taking
-        # care of that
-        select_timeout = 1
-        thread_out_streams = {
-            name: (r, w)
-            for name, (r, w) in out_streams.items()
-            if w is not None
-        }
-        redirect_thread = threading.Thread(
-            target=redirect_thread_f,
-            args=(stdout_in, stderr_in, thread_out_streams, select_timeout),
-            # The thread will die when the main thread dies
-            daemon=True,
-        )
-        redirect_thread.start()
+                    pass
+
+                # Make sure the writing end are closed proper since we are not
+                # going to write anything anymore
+                for r, w in out_streams.values():
+                    w.flush()
+                    if r is not w and w is not None:
+                        w.close()
+
+            # If there is anything we need to redirect to, spawn a thread taking
+            # care of that
+            select_timeout = 1
+            thread_out_streams = {
+                name: (r, w)
+                for name, (r, w) in out_streams.items()
+                if w is not None
+            }
+            redirect_thread = threading.Thread(
+                target=redirect_thread_f,
+                args=(stdout_in, stderr_in, thread_out_streams, select_timeout),
+                # The thread will die when the main thread dies
+                daemon=True,
+            )
+            redirect_thread.start()
+
+            return dict(
+                chan=channel,
+                pid=pid,
+                stdin=stdin,
+                # We give the reading end to the consumer of the data
+                stdout=out_streams['stdout'][0],
+                stderr=out_streams['stderr'][0],
+                redirect_thread=redirect_thread,
+            )
 
-        return ParamikoBackgroundCommand(
+        return ParamikoBackgroundCommand.from_factory(
             conn=self,
+            cmd=command,
             as_root=as_root,
-            chan=channel,
-            pid=pid,
-            stdin=stdin,
-            # We give the reading end to the consumer of the data
-            stdout=out_streams['stdout'][0],
-            stderr=out_streams['stderr'][0],
-            redirect_thread=redirect_thread,
-            cmd=orig_command,
+            make_init_kwargs=make_init_kwargs,
         )
 
     def _close(self):