NamedTemporaryFile usage

gprofiler/profilers/perf.py

@@ -10,6 +10,7 @@
 from collections import Counter, defaultdict
 from pathlib import Path
 from subprocess import Popen
+from tempfile import NamedTemporaryFile
 from threading import Event
 from typing import Any, Dict, Iterable, List, Optional
 
@@ -322,27 +323,26 @@ def wait_and_script(self) -> str:
             # (unlike Popen.communicate())
             assert self._process is not None and self._process.stderr is not None
             logger.debug(f"{self._log_name} run output", perf_stderr=self._process.stderr.read1())  # type: ignore
-
         try:
-            inject_data = Path(f"{str(perf_data)}.inject")
-            if self._inject_jit:
-                run_process(
-                    [perf_path(), "inject", "--jit", "-o", str(inject_data), "-i", str(perf_data)],
+            with NamedTemporaryFile(dir=os.path.dirname(self._output_path), suffix=".inject") as inject_data:
+                if self._inject_jit:
+                    run_process(
+                        [perf_path(), "inject", "--jit", "-o", str(inject_data), "-i", str(perf_data)],
+                    )
+                    perf_data = Path(inject_data.name)
+
+                perf_script_proc = run_process(
+                    [perf_path(), "script", "-F", "+pid", "-i", str(perf_data)],
+                    suppress_log=True,
                 )
-                perf_data.unlink()
-                perf_data = inject_data
-
-            perf_script_proc = run_process(
-                [perf_path(), "script", "-F", "+pid", "-i", str(perf_data)],
-                suppress_log=True,
-            )
-            return perf_script_proc.stdout.decode("utf8")
+                return perf_script_proc.stdout.decode("utf8")
         finally:
             perf_data.unlink()
-            if self._inject_jit:
-                # might be missing if it's already removed.
-                # might be existing if "perf inject" itself fails
-                remove_path(inject_data, missing_ok=True)
+            # always read its stderr
+            # using read1() which performs just a single read() call and doesn't read until EOF
+            # (unlike Popen.communicate())
+            assert self._process is not None and self._process.stderr is not None
+            logger.debug(f"{self._log_name} run output", perf_stderr=self._process.stderr.read1())  # type: ignore
 
 
 @register_profiler(

gprofiler/utils/fs.py

@@ -6,11 +6,10 @@
 import errno
 import os
 import shutil
-from pathlib import Path
-from secrets import token_hex
+from tempfile import NamedTemporaryFile
 
 from gprofiler.platform import is_windows
-from gprofiler.utils import remove_path, run_process
+from gprofiler.utils import run_process
 
 
 def safe_copy(src: str, dst: str) -> None:
@@ -27,29 +26,27 @@ def is_rw_exec_dir(path: str) -> bool:
     """
     Is 'path' rw and exec?
     """
-    # randomize the name - this function runs concurrently on paths of in same mnt namespace.
-    test_script = Path(path) / f"t-{token_hex(10)}.sh"
 
     # try creating & writing
     try:
         os.makedirs(path, 0o755, exist_ok=True)
-        test_script.write_text("#!/bin/sh\nexit 0")
-        test_script.chmod(0o755)
+        test_script = NamedTemporaryFile(dir=path, suffix=".sh")
+        with open(test_script.name, "w") as f:
+            f.write("#!/bin/sh\nexit 0")
+        os.chmod(test_script.name, 0o755)
+        test_script.file.close()
     except OSError as e:
         if e.errno == errno.EROFS:
             # ro
             return False
-        remove_path(test_script)
         raise
 
     # try executing
     try:
-        run_process([str(test_script)], suppress_log=True)
+        run_process([str(test_script.name)], suppress_log=True)
     except PermissionError:
         # noexec
         return False
-    finally:
-        test_script.unlink()
 
     return True