DAOS-16362 pydaos: ensure checkpoint path is created

Some use of Checkpoint assumes that path to the checkpoiunt file will be
created with all missing parent directories.

For instance, DLIO benchmark writes checkpoints as `/prefix/global_epochX_stepY/layer-Z.pt`.

This commit adds `ensure_path` parameter to call `mkdirall` before
writing checkpoint file.

Features: pytorch
Signed-off-by: Denis Barakhtanov <dbarahtanov@enakta.com>

Author: Denis Barakhtanov

src/client/pydaos/torch/torch_api.py

@@ -1,6 +1,6 @@
 #
 # (C) Copyright 2024-2025 Google LLC
-# (C) Copyright 2024-2025 Enakta Labs Ltd
+# (C) Copyright 2024-2026 Enakta Labs Ltd
 #
 #  SPDX-License-Identifier: BSD-2-Clause-Patent
 #
@@ -11,11 +11,13 @@
 In addition, it provides Checkpoint class to save and load PyTorch model checkpoints.
 """
 
+import errno
 import io
 import math
 import os
 import stat
 from multiprocessing import Process, Queue
+from pathlib import Path
 
 from torch.utils.data import Dataset as TorchDataset
 from torch.utils.data import IterableDataset as TorchIterableDataset
@@ -619,13 +621,16 @@ def reader(self, file, stream=None):
         stream.seek(0)
         return stream
 
-    def writer(self, file):
+    def writer(self, file, ensure_path=True):
         """ Returns write buffer to save the checkpoint file """
 
         if file is None:
             raise ValueError("file is required")
 
         path = os.path.join(self._prefix, file)
+        if ensure_path:
+            self._dfs.mkdirall(os.path.dirname(path))
+
         return WriteBuffer(self._dfs, path, self._mode, self._oflags,
                            self._class_name, self._file_chunk_size, self._transfer_chunk_size,
                            self._chunks_limit, self._workers)
@@ -810,3 +815,18 @@ def get_file_size(self, path):
         if ret != 0:
             raise OSError(ret, os.strerror(ret), path)
         return size
+
+    def mkdirall(self, path, mode=0o755):
+        """ Creates directory, making parent directories if needed """
+
+        path = os.path.normpath(path)
+        dirs = list(Path(path).parts)
+        if not dirs:
+            raise ValueError(f"invalid path: {path}")
+
+        parent = dirs.pop(0)
+        for name in dirs:
+            parent = os.path.join(parent, name)
+            ret = torch_shim.torch_mkdir(DAOS_MAGIC, self._dfs, parent, mode)
+            if ret not in (0, errno.EEXIST):
+                raise OSError(ret, os.strerror(ret), parent)

src/client/pydaos/torch/torch_shim.c

@@ -1,7 +1,7 @@
 /**
  * (C) Copyright 2019-2024 Intel Corporation.
  * (C) Copyright 2024-2025 Google LLC
- * (C) Copyright 2024-2025 Enakta Labs Ltd
+ * (C) Copyright 2024-2026 Enakta Labs Ltd
  *
  * SPDX-License-Identifier: BSD-2-Clause-Patent
  */
@@ -1061,6 +1061,39 @@ __shim_handle__torch_get_fsize(PyObject *self, PyObject *args)
 	return Py_BuildValue("iK", rc, st.st_size);
 }
 
+static PyObject *
+__shim_handle__torch_mkdir(PyObject *self, PyObject *args)
+{
+	struct dfs_handle *hdl  = NULL;
+	char              *path = NULL;
+	char              *dir  = NULL;
+	char              *name = NULL;
+	mode_t             mode;
+	dfs_obj_t         *parent = NULL;
+
+	RETURN_NULL_IF_FAILED_TO_PARSE(args, "LsI", &hdl, &path, &mode);
+
+	assert(hdl->dfs != NULL);
+
+	int rc = split_path(path, &dir, &name);
+	if (rc) {
+		return PyLong_FromLong(rc);
+	}
+
+	rc = lookup_or_insert_dir_obj(hdl, dir, &parent);
+	if (rc) {
+		D_ERROR("Could not lookup '%s': %s (rc=%d)", dir, strerror(rc), rc);
+		goto out;
+	}
+
+	rc = dfs_mkdir(hdl->dfs, parent, name, mode, 0);
+
+out:
+	D_FREE(dir);
+	D_FREE(name);
+	return PyLong_FromLong(rc);
+}
+
 /**
  * Python shim module
  */
@@ -1080,6 +1113,7 @@ static PyMethodDef torchMethods[] = {
     EXPORT_PYTHON_METHOD(torch_recommended_dir_split),
     EXPORT_PYTHON_METHOD(torch_list_with_anchor),
     EXPORT_PYTHON_METHOD(torch_get_fsize),
+    EXPORT_PYTHON_METHOD(torch_mkdir),
 
     EXPORT_PYTHON_METHOD(module_init),
     EXPORT_PYTHON_METHOD(module_fini),

src/tests/ftest/pytorch/checkpoint.py

@@ -1,9 +1,10 @@
 """
   (C) Copyright 2025 Google LLC
-  (C) Copyright 2025 Enakta Labs Ltd
+  (C) Copyright 2025-2026 Enakta Labs Ltd
 
   SPDX-License-Identifier: BSD-2-Clause-Patent
 """
+import errno
 import os
 import uuid
 
@@ -73,6 +74,39 @@ def test_checkpoint_chunking(self):
                                               chunk_size=chunk_size, chunks_limit=chunks_limit,
                                               workers=worker)
 
+    def test_checkpoint_nested_directories(self):
+        """ Test Pytorch Checkpoint interface with nested directories
+        Test Description: Ensure that parent directories are created for the checkpoint path
+
+        :avocado: tags=all,full_regression
+        :avocado: tags=vm
+        :avocado: tags=pytorch
+        :avocado: tags=PytorchCheckpointTest,test_checkpoint_nested_directories
+        """
+
+        pool = self.get_pool()
+        container = self.get_container(pool)
+        data = os.urandom(4096)
+
+        files = ["/file.pt", "/one/file.pt", "/one/two/file.pt"]
+
+        with Checkpoint(pool, container) as pt:
+            # By default parent should be created
+            for name in files:
+                with pt.writer(name) as w:
+                    w.write(data)
+
+            try:
+                fname = f"/{str(uuid.uuid4())}/file.pt"
+                with pt.writer(fname, ensure_path=False) as w:
+                    w.write(data)
+                    raise RuntimeError("expected OSError with errno.ENOENT")
+            except OSError as e:
+                if e.errno != errno.ENOENT:
+                    raise RuntimeError(f"expected errno.ENOENT, got {os.strerror(e.errno)}") from e
+            except Exception as e:
+                raise RuntimeError(f"unexpected error: {e}") from e
+
     def _test_checkpoint(self, pool, cont, writes, chunk_size=0, chunks_limit=0, workers=0):
         """Creates a checkpoint with the given parameters, writes the given data to it,
         then reads written data back from it and compares it with the expected writes.