perf: skip reads for empty fd captures

CHANGELOG.md

@@ -7,6 +7,8 @@ releases are available on [PyPI](https://pypi.org/project/pytask) and
 
 ## Unreleased
 
+- [#968](https://github.com/pytask-dev/pytask/pull/968) improves the performance of file
+  descriptor capture for tasks that produce no output.
 - Documents the pytest provenance of the adapted debugging, marker, and warning
   modules.
 - [#889](https://github.com/pytask-dev/pytask/pull/889) improves typing for tree

src/_pytask/capture.py

@@ -549,6 +549,9 @@ class FDCaptureBinary(FDCaptureBase[bytes]):
 
     def snap(self) -> bytes:
         self._assert_state("snap", ("started", "suspended"))
+        # Avoid seek/read/truncate in the common case of no output.
+        if os.fstat(self.tmpfile.fileno()).st_size == 0:
+            return self.EMPTY_BUFFER
         self.tmpfile.seek(0)
         res = self.tmpfile.buffer.read()
         self.tmpfile.seek(0)
@@ -572,6 +575,9 @@ class FDCapture(FDCaptureBase[str]):
 
     def snap(self) -> str:
         self._assert_state("snap", ("started", "suspended"))
+        # Avoid seek/read/truncate in the common case of no output.
+        if os.fstat(self.tmpfile.fileno()).st_size == 0:
+            return self.EMPTY_BUFFER
         self.tmpfile.seek(0)
         res = self.tmpfile.read()
         self.tmpfile.seek(0)

tests/test_capture.py

@@ -11,6 +11,7 @@
 from typing import Any
 from typing import BinaryIO
 from typing import cast
+from unittest.mock import Mock
 
 import pytest
 
@@ -491,6 +492,25 @@ def lsof_check():
 
 
 class TestFDCapture:
+    @pytest.mark.parametrize(
+        ("capture_class", "empty_buffer"),
+        [(capture.FDCapture, ""), (capture.FDCaptureBinary, b"")],
+    )
+    def test_snap_empty_file_avoids_file_io(
+        self, monkeypatch, capture_class, empty_buffer
+    ):
+        cap = object.__new__(capture_class)
+        cap._state = "started"
+        cap.tmpfile = Mock()
+        cap.tmpfile.fileno.return_value = 42
+        monkeypatch.setattr(capture.os, "fstat", lambda _fd: Mock(st_size=0))
+
+        assert cap.snap() == empty_buffer
+        cap.tmpfile.seek.assert_not_called()
+        cap.tmpfile.read.assert_not_called()
+        cap.tmpfile.buffer.read.assert_not_called()
+        cap.tmpfile.truncate.assert_not_called()
+
     def test_simple(self, tmpfile):
         fd = tmpfile.fileno()
         cap = capture.FDCapture(fd)