fix(io): close aio_fd in FastFileWriter._fini to prevent fd leak (#8005)

Fixes #8003

## Summary

`FastFileWriter._fini()` overwrote `self._aio_fd = INVALID_FD` without
calling `os.close()`, leaking one fd per save. With unlink-based
checkpoint rotation this stranded the unlinked inode in the ext4
orphan list, fs blocks were never reclaimed, and long-running save
loops hit ENOSPC at iter ~60 (60 GB/iter on a 4 TB partition).

This PR adds explicit `os.fsync()` + `os.close()` in `_fini()` and a
regression test that asserts no `/proc/self/fd` entry points at a
deleted file after a save+close+unlink cycle.

## Verification

- 20-iteration repro of `save() / close() / unlink()` leaked 20 fds
  before the fix, 0 after.
- 700-iter / 42 TB / 60 h endurance run on ext4/NVMe: `df_used`
  stable at 736 GB (drift +281 MB / 697 rotations) with the fix;
  same workload hit ENOSPC at iter ~60 without it.
- Performance impact: ~5% wall-time overhead from the added
  `os.fsync()` at ~10 GB/s peak.

## Test plan

- [x] New regression test
  `tests/unit/ops/aio/test_fast_file_writer_fd_close.py` verifies fd
  cleanup after a single save and after 5/20-iter rotation loops via
  `/proc/self/fd` scoped to `tmp_path`.
- [x] Gated on `async_io` compatibility, Linux, and CUDA accelerator
  so unsupported CI matrix entries skip cleanly.
- [x] Confirmed test FAILS without this PR's `_fini()` change and
  PASSES with it.
- [x] `pre-commit run --files <changed files>` clean.

## Notes

- The `__del__` assertion `assert self._aio_fd == INVALID_FD` passes
  even with the bug because it checks the Python attribute that
  `_fini` itself sets. The new test checks OS-level state via
  `/proc/self/fd`.
- `os.fsync()` is included for post-close durability — required for
  correctness on the unaligned-tail path that re-opens the file as
  buffered I/O. If maintainers prefer to drop it for performance,
  removing only the `os.fsync(...)` line still fixes the leak.

Happy to adjust shape, naming, or test placement to fit project
conventions. Thanks for the review.

Signed-off-by: jg-heo <csjg.heo@gmail.com>
Co-authored-by: Masahiro Tanaka <81312776+tohtana@users.noreply.github.com>

Author: jg-heo

deepspeed/io/fast_file_writer.py

@@ -114,7 +114,13 @@ def _fini(self):
         if not self._io_buffer_is_empty():
             self._force_drain()
         self._io_buffer.reset()
+        fd = self._aio_fd
         self._aio_fd = INVALID_FD
+        if fd != INVALID_FD:
+            try:
+                os.fsync(fd)
+            finally:
+                os.close(fd)
 
     def _fill_io_buffer(self, src_tensor, src_offset):
         st = time.time()

tests/unit/ops/aio/test_fast_file_writer_fd_close.py

@@ -0,0 +1,107 @@
+# Copyright (c) Microsoft Corporation.
+# SPDX-License-Identifier: Apache-2.0
+
+# DeepSpeed Team
+"""Regression test for FastFileWriter file-descriptor cleanup.
+
+Without proper os.close() in _fini(), every save through FastFileWriter
+leaks one fd pointing at the just-written file. When the user later
+unlinks the file (e.g. checkpoint rotation), the leaked fd holds the
+inode in the filesystem's orphan list, so blocks are never freed and
+the filesystem eventually reports ENOSPC even though `ls` shows only
+N files on disk.
+
+These tests assert that after FastFileWriter.close() returns, no fd
+in /proc/self/fd points at the (possibly already-unlinked) file.
+"""
+import os
+import sys
+import pytest
+import torch
+
+import deepspeed
+from deepspeed.accelerator import get_accelerator
+from deepspeed.ops.op_builder import AsyncIOBuilder
+from deepspeed.io import FastFileWriter, FastFileWriterConfig
+
+if not deepspeed.ops.__compatible_ops__[AsyncIOBuilder.NAME]:
+    pytest.skip("async_io op is not compatible on this system", allow_module_level=True)
+
+if not sys.platform.startswith("linux"):
+    pytest.skip("test uses /proc/self/fd which is Linux-only", allow_module_level=True)
+
+if get_accelerator().device_name() != 'cuda':
+    pytest.skip("FastFileWriter requires CUDA-pinned tensors", allow_module_level=True)
+
+BLOCK_SIZE = 1 * 1024 * 1024
+QUEUE_DEPTH = 8
+PINNED_BYTES = 8 * 1024 * 1024
+PAYLOAD_BYTES = 1 * 1024 * 1024
+
+
+def _count_deleted_fds(target_dir):
+    """How many fds in /proc/self/fd point at a now-deleted file located
+    under target_dir? Restricting to target_dir avoids false positives
+    from unrelated deleted fds in the test process."""
+    pid = os.getpid()
+    n = 0
+    for entry in os.listdir(f"/proc/{pid}/fd"):
+        try:
+            target = os.readlink(f"/proc/{pid}/fd/{entry}")
+        except OSError:
+            continue
+        if target.startswith(str(target_dir)) and target.endswith("(deleted)"):
+            n += 1
+    return n
+
+
+def _build_writer(file_path):
+    aio = AsyncIOBuilder().load(verbose=False).aio_handle(block_size=BLOCK_SIZE,
+                                                          queue_depth=QUEUE_DEPTH,
+                                                          single_submit=False,
+                                                          overlap_events=False,
+                                                          intra_op_parallelism=1)
+    pinned = torch.zeros(PINNED_BYTES, dtype=torch.uint8).pin_memory()
+    cfg = FastFileWriterConfig(dnvme_handle=aio,
+                               pinned_tensor=pinned,
+                               double_buffer=True,
+                               num_parallel_writers=1,
+                               writer_rank=0)
+    return FastFileWriter(file_path=str(file_path), config=cfg)
+
+
+def test_close_releases_fd_after_unlink(tmp_path):
+    """Single save + unlink must not leave a deleted-fd reference."""
+    target = tmp_path / "ckpt_single.pt"
+    buf = torch.zeros(PAYLOAD_BYTES, dtype=torch.uint8)
+
+    before = _count_deleted_fds(tmp_path)
+    w = _build_writer(target)
+    torch.save(obj=buf, f=w)
+    w.close()
+    os.unlink(target)
+    after = _count_deleted_fds(tmp_path)
+
+    assert after == before, (f"FastFileWriter leaked an fd: deleted-fd count went "
+                             f"from {before} to {after} after a single save+close+unlink. "
+                             f"This indicates _fini() did not os.close(self._aio_fd).")
+
+
+@pytest.mark.parametrize("n_iters", [5, 20])
+def test_rotation_loop_does_not_leak(tmp_path, n_iters):
+    """N iterations of save+close+unlink should leave zero deleted-fds.
+    Mirrors the real checkpoint-rotation workload that originally
+    surfaced this bug."""
+    buf = torch.zeros(PAYLOAD_BYTES, dtype=torch.uint8)
+    before = _count_deleted_fds(tmp_path)
+
+    for i in range(n_iters):
+        path = tmp_path / f"ckpt_{i}.pt"
+        w = _build_writer(path)
+        torch.save(obj=buf, f=w)
+        w.close()
+        os.unlink(path)
+
+    after = _count_deleted_fds(tmp_path)
+    assert after == before, (f"FastFileWriter leaked {after - before} fd(s) over {n_iters} "
+                             f"save+close+unlink iterations (expected 0).")