C API 2.0 Checkpointing + unblock dali.h (#5879)

* This commit adds C API for pipeline checkpointing.
* Checkpointing support in dali::Pipeline is also slightly refactored to allow the Pipeline object to serialize checkpoint objects passed as an argument.
* The checkpoint in C API can be obtained from a Pipeline handle by calling daliPipelineGetCheckpoint and then serialized separately. This allows for the checkpoint to be passed between pipeline instances without serialization.
* External checkpoint data (pipeline_data, iterator_data) is stored with the checkpoint object and does not need manual memory management.
* This commit also removes the #error that blocked inclusion of dali.h header, making it more or less official.

---------

Signed-off-by: Michal Zientkiewicz <michalz@nvidia.com>

Author: mzient

dali/benchmark/checkpointing_bench.cc

@@ -49,7 +49,7 @@ class CheckpointingOverhead : public DALIBenchmark {
       if (policy == CheckpointingPolicy::SaveEveryIter) {
         volatile auto cpt = pipe->GetCheckpoint();
       } else if (policy == CheckpointingPolicy::SerializeEveryIter) {
-        volatile auto cpt = pipe->SerializedCheckpoint({});
+        volatile auto cpt = pipe->GetSerializedCheckpoint({});
       }
     }
 

dali/c_api/c_api.cc

@@ -873,7 +873,7 @@ void daliGetSerializedCheckpoint(
       external_context->iterator_data.size
     };
   }
-  std::string cpt = pipeline->SerializedCheckpoint(ctx);
+  std::string cpt = pipeline->GetSerializedCheckpoint(ctx);
   *n = cpt.size();
   *checkpoint = reinterpret_cast<char *>(daliAlloc(cpt.size()));
   DALI_ENFORCE(*checkpoint, "Failed to allocate memory");

dali/c_api_2/c_api_internal_test.cc

@@ -15,7 +15,6 @@
 #include <gtest/gtest.h>
 #include <stdexcept>
 #include <system_error>
-#define DALI_ALLOW_NEW_C_API
 #include "dali/dali.h"
 #include "dali/c_api_2/error_handling.h"
 #include "dali/core/cuda_error.h"

dali/c_api_2/checkpoint.h

@@ -0,0 +1,73 @@
+// Copyright (c) 2025, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#ifndef DALI_C_API_2_CHECKPOINT_H_
+#define DALI_C_API_2_CHECKPOINT_H_
+
+#include <memory>
+#include <string>
+#include <string_view>
+#include <utility>
+#include <vector>
+#include "dali/dali.h"
+#include "dali/pipeline/operator/checkpointing/checkpoint.h"
+
+// A dummy base that the handle points to
+struct _DALICheckpoint {
+ protected:
+  _DALICheckpoint() = default;
+  ~_DALICheckpoint() = default;
+};
+
+
+namespace dali::c_api {
+
+class PipelineWrapper;
+
+class CheckpointWrapper : public _DALICheckpoint {
+ public:
+  explicit CheckpointWrapper(Checkpoint &&cpt)
+  : cpt_(std::move(cpt)) {}
+
+  const std::string &Serialized() const & {
+    return serialized_.value();
+  }
+
+  void Serialize(const PipelineWrapper &pipeline);
+
+  daliCheckpointExternalData_t ExternalData() const {
+    daliCheckpointExternalData_t ext;
+    ext.iterator_data.data = cpt_.external_ctx_cpt_.iterator_data.data();
+    ext.iterator_data.size = cpt_.external_ctx_cpt_.iterator_data.size();
+    ext.pipeline_data.data = cpt_.external_ctx_cpt_.pipeline_data.data();
+    ext.pipeline_data.size = cpt_.external_ctx_cpt_.pipeline_data.size();
+    return ext;
+  }
+
+  Checkpoint *Unwrap() & {
+    return &cpt_;
+  }
+
+  const Checkpoint *Unwrap() const & {
+    return &cpt_;
+  }
+
+ private:
+  Checkpoint cpt_;
+  std::optional<std::string> serialized_;
+};
+
+}  // namespace dali::c_api
+
+#endif  // DALI_C_API_2_CHECKPOINT_H_

dali/c_api_2/data_objects.h

@@ -21,7 +21,6 @@
 #include <string>
 #include <utility>
 #include <vector>
-#define DALI_ALLOW_NEW_C_API
 #include "dali/dali.h"
 #include "dali/pipeline/data/tensor_list.h"
 #include "dali/c_api_2/ref_counting.h"
@@ -661,9 +660,9 @@ class TensorListWrapper : public ITensorList {
   }
 
   RefCountedPtr<ITensor> ViewAsTensor() const override {
-    if (!tl_->IsContiguous())
+    if (!tl_->IsDenseTensor())
       throw std::runtime_error(
-        "The TensorList is not contiguous and cannot be viewed as a Tensor.");
+        "Only a densely packed list of tensors of uniform shape can be viewed as a Tensor.");
 
     auto t = std::make_shared<Tensor<Backend>>();
     auto buf = unsafe_owner(*tl_);

dali/c_api_2/error_handling.h

@@ -19,7 +19,6 @@
 #include <iostream>
 #include <string>
 #include <sstream>
-#define DALI_ALLOW_NEW_C_API
 #include "dali/dali.h"
 #include "dali/core/error_handling.h"
 

dali/c_api_2/init.cc

@@ -13,7 +13,6 @@
 // limitations under the License.
 
 #include <atomic>
-#define DALI_ALLOW_NEW_C_API
 #include "dali/dali.h"
 #include "dali/c_api_2/error_handling.h"
 #include "dali/pipeline/init.h"

dali/c_api_2/managed_handle.h

@@ -18,7 +18,6 @@
 #include <cassert>
 #include <stdexcept>
 #include <utility>
-#define DALI_ALLOW_NEW_C_API
 #include "dali/dali.h"
 #include "dali/core/unique_handle.h"
 
@@ -122,6 +121,7 @@ class Resource##Handle \
 
 DALI_C_UNIQUE_HANDLE(Pipeline);
 DALI_C_UNIQUE_HANDLE(PipelineOutputs);
+DALI_C_UNIQUE_HANDLE(Checkpoint);
 DALI_C_REF_HANDLE(TensorList);
 DALI_C_REF_HANDLE(Tensor);
 

dali/c_api_2/op_test/complex_pipeline_test.cc

@@ -24,10 +24,17 @@
 namespace dali::c_api::test {
 
 std::unique_ptr<Pipeline>
-ReaderDecoderPipe(std::string_view decoder_device, StorageDevice output_device) {
+ReaderDecoderPipe(
+      std::string_view decoder_device,
+      StorageDevice output_device,
+      PipelineParams params = {}) {
   std::string file_root = testing::dali_extra_path() + "/db/single/jpeg/";
   std::string file_list = file_root + "image_list.txt";
-  auto pipe = std::make_unique<Pipeline>(4, 1, 0, 12345, true, 2, true, true);
+  if (!params.max_batch_size) params.max_batch_size = 4;
+  if (!params.num_threads) params.num_threads = 1;
+  if (!params.seed) params.seed = 12345;
+  if (!params.executor_type) params.executor_type = ExecutorType::Dynamic;
+  auto pipe = std::make_unique<Pipeline>(params);
   pipe->AddOperator(OpSpec("FileReader")
     .AddArg("device", "cpu")
     .AddArg("file_root", file_root)
@@ -81,4 +88,79 @@ TEST(CAPI2_PipelineTest, ReaderDecoderMixed2CPU) {
   TestReaderDecoder("mixed", StorageDevice::CPU);
 }
 
+TEST(CAPI2_PipelineTest, Checkpointing) {
+  // This test creates three pipelines - a C++ pipeline (ref) and two C pipelines (pipe1, pipe2),
+  // created by deserializing the serialized representation of the C++ pipeline.
+  //
+  // (pipe1) advances 3 iterations and then a checkpoint is taken and restored in (ref),
+  // after which 5 iterations of outputs are compared.
+  // Then a checkpoint is taken in (ref) and restored in (pipe2), after which another 5 iterations
+  // are compared.
+  PipelineParams params{};
+  params.enable_checkpointing = true;
+  params.seed = 1234;
+  auto ref = ReaderDecoderPipe("cpu", StorageDevice::GPU, params);
+  ref->Build();
+  auto pipe_str = ref->SerializeToProtobuf();  // serialize the ref...
+  auto pipe1 = Deserialize(pipe_str, {});  // ...and create pipe1
+  auto pipe2 = Deserialize(pipe_str, {});  // ...and pipe2 from serialized ref
+  CHECK_DALI(daliPipelineBuild(pipe1));
+  CHECK_DALI(daliPipelineBuild(pipe2));
+
+  // Advance a few iterations...
+  CHECK_DALI(daliPipelinePrefetch(pipe1));
+  daliPipelineOutputs_h out1_h{};
+  CHECK_DALI(daliPipelinePopOutputs(pipe1, &out1_h));
+  CHECK_DALI(daliPipelineOutputsDestroy(out1_h));
+  CHECK_DALI(daliPipelineRun(pipe1));
+  CHECK_DALI(daliPipelinePopOutputs(pipe1, &out1_h));
+  CHECK_DALI(daliPipelineOutputsDestroy(out1_h));
+  CHECK_DALI(daliPipelineRun(pipe1));
+  CHECK_DALI(daliPipelinePopOutputs(pipe1, &out1_h));
+  CHECK_DALI(daliPipelineOutputsDestroy(out1_h));
+
+  const char pipeline_data[] = "A rose by any other name would smell as sweet";
+  size_t pipeline_data_size = strlen(pipeline_data);
+
+  daliCheckpointExternalData_t ext{};
+  ext.iterator_data.data = "ITER";
+  ext.iterator_data.size = 4;
+  ext.pipeline_data.data = pipeline_data;
+  ext.pipeline_data.size = strlen(ext.pipeline_data.data);
+
+  daliCheckpoint_h checkpoint_h{};
+  // Take a checkpoint...
+  CHECK_DALI(daliPipelineGetCheckpoint(pipe1, &checkpoint_h, &ext));
+  CheckpointHandle checkpoint(checkpoint_h);
+
+  const char *data = nullptr;
+  size_t size = 0;
+  CHECK_DALI(daliPipelineSerializeCheckpoint(pipe1, checkpoint, &data, &size));
+  ASSERT_NE(data, nullptr);
+
+  // ...restore...
+  ref->RestoreFromSerializedCheckpoint(std::string(data, size));
+  // ...run and compare.
+  ComparePipelineOutputs(*ref, pipe1, 5, false);
+
+  // Now take another checkpoint...
+  auto chk_str = ref->GetSerializedCheckpoint({ ext.pipeline_data.data, ext.iterator_data.data });
+
+  // ...deserialize...
+  CHECK_DALI(daliPipelineDeserializeCheckpoint(
+        pipe2, &checkpoint_h, chk_str.data(), chk_str.length()));
+  CheckpointHandle checkpoint2(checkpoint_h);
+
+  daliCheckpointExternalData_t ext2{};
+  CHECK_DALI(daliCheckpointGetExternalData(checkpoint2, &ext2));
+  EXPECT_EQ(ext2.iterator_data.size, 4);
+  EXPECT_STREQ(ext2.iterator_data.data, "ITER");
+  EXPECT_EQ(ext2.pipeline_data.size, pipeline_data_size);
+  EXPECT_STREQ(ext2.pipeline_data.data, pipeline_data);
+  // ...restore...
+  CHECK_DALI(daliPipelineRestoreCheckpoint(pipe2, checkpoint2));
+  // ...run and compare.
+  ComparePipelineOutputs(*ref, pipe2, 5, false);
+}
+
 }  // namespace dali::c_api::test

dali/c_api_2/pipeline.cc

@@ -14,6 +14,7 @@
 
 #include "dali/c_api_2/pipeline.h"
 #include "dali/c_api_2/pipeline_outputs.h"
+#include "dali/c_api_2/checkpoint.h"
 #include "dali/c_api_2/error_handling.h"
 #include "dali/pipeline/pipeline.h"
 #include "dali/c_api_2/utils.h"
@@ -27,6 +28,12 @@ PipelineWrapper *ToPointer(daliPipeline_h handle) {
   return static_cast<PipelineWrapper *>(handle);
 }
 
+CheckpointWrapper *ToPointer(daliCheckpoint_h handle) {
+  if (!handle)
+    throw NullHandle("Checkpoint");
+  return static_cast<CheckpointWrapper *>(handle);
+}
+
 PipelineParams ToCppParams(const daliPipelineParams_t &params) {
   PipelineParams cpp_params = {};
 
@@ -237,6 +244,39 @@ void PipelineWrapper::FeedInputImpl(
     data_id ? std::optional<std::string>(std::in_place, *data_id) : std::nullopt);
 }
 
+std::unique_ptr<CheckpointWrapper> PipelineWrapper::GetCheckpoint(
+      const daliCheckpointExternalData_t *ext) const {
+  auto cpt = std::make_unique<CheckpointWrapper>(pipeline_->GetCheckpoint());
+  if (ext) {
+    cpt->Unwrap()->external_ctx_cpt_.pipeline_data =
+        std::string(ext->pipeline_data.data, ext->pipeline_data.size);
+    cpt->Unwrap()->external_ctx_cpt_.iterator_data =
+        std::string(ext->iterator_data.data, ext->iterator_data.size);
+  }
+  return cpt;
+}
+
+std::string_view PipelineWrapper::SerializeCheckpoint(CheckpointWrapper &chk) const {
+  chk.Serialize(*this);
+  return chk.Serialized();
+}
+
+void CheckpointWrapper::Serialize(const PipelineWrapper &pipeline) {
+  if (!serialized_)
+    serialized_ = pipeline.Unwrap()->SerializeCheckpoint(cpt_);
+}
+
+std::unique_ptr<CheckpointWrapper>
+PipelineWrapper::DeserializeCheckpoint(std::string_view serialized) {
+  return std::make_unique<CheckpointWrapper>(pipeline_->DeserializeCheckpoint(serialized));
+}
+
+void PipelineWrapper::RestoreFromCheckpoint(CheckpointWrapper &chk) {
+  pipeline_->RestoreFromCheckpoint(*chk.Unwrap());
+}
+
+
+
 }  // namespace dali::c_api
 
 using namespace dali::c_api;  // NOLINT
@@ -393,3 +433,83 @@ daliResult_t daliPipelinePopOutputsAsync(
   *out = pipe->PopOutputs(stream).release();
   DALI_EPILOG();
 }
+
+
+daliResult_t daliPipelineGetCheckpoint(
+      daliPipeline_h pipeline,
+      daliCheckpoint_h *out_checkpoint,
+      const daliCheckpointExternalData_t *checkpoint_ext) {
+  DALI_PROLOG();
+  auto pipe = ToPointer(pipeline);
+  CHECK_OUTPUT(out_checkpoint);
+  auto chk = pipe->GetCheckpoint(checkpoint_ext);
+  *out_checkpoint = chk.release();  // No throwing beyond this point!
+  DALI_EPILOG();
+}
+
+daliResult_t daliPipelineRestoreCheckpoint(
+      daliPipeline_h pipeline,
+      daliCheckpoint_h checkpoint) {
+  DALI_PROLOG();
+  auto pipe = ToPointer(pipeline);
+  auto chk = ToPointer(checkpoint);
+  pipe->RestoreFromCheckpoint(*chk);
+  DALI_EPILOG();
+}
+
+daliResult_t daliPipelineDeserializeCheckpoint(
+      daliPipeline_h pipeline,
+      daliCheckpoint_h  *out_checkpoint,
+      const char *serialized_checkpoint,
+      size_t serialized_checkpoint_size) {
+  DALI_PROLOG();
+  auto pipe = ToPointer(pipeline);
+  CHECK_OUTPUT(out_checkpoint);
+  if (!serialized_checkpoint_size) {
+    *out_checkpoint = nullptr;
+    return DALI_NO_DATA;
+  }
+  if (!serialized_checkpoint) {
+    throw std::invalid_argument("The parameter `serialized_checkpoint` must not be NULL if "
+                                "`serialize_checkpoint_size` is nonzero.");
+  }
+
+  auto cpt = pipe->DeserializeCheckpoint(
+    std::string_view(serialized_checkpoint, serialized_checkpoint_size));
+
+  *out_checkpoint = cpt.release();  // No throwing beyond this point!
+  DALI_EPILOG();
+}
+
+daliResult_t daliCheckpointGetExternalData(
+      daliCheckpoint_h checkpoint,
+      daliCheckpointExternalData_t *out_ext_data) {
+  DALI_PROLOG();
+  auto cpt = ToPointer(checkpoint);
+  CHECK_OUTPUT(out_ext_data);
+  *out_ext_data = cpt->ExternalData();
+  DALI_EPILOG();
+}
+
+daliResult_t daliPipelineSerializeCheckpoint(
+      daliPipeline_h pipeline,
+      daliCheckpoint_h checkpoint,
+      const char **out_data,
+      size_t *out_size) {
+  DALI_PROLOG();
+  auto pipe = ToPointer(pipeline);
+  auto cpt = ToPointer(checkpoint);
+  CHECK_OUTPUT(out_data);
+  CHECK_OUTPUT(out_size);
+  auto serialized = pipe->SerializeCheckpoint(*cpt);
+  *out_data = serialized.data();
+  *out_size = serialized.size();
+  DALI_EPILOG();
+}
+
+/** Destroys a checkpoint object */
+daliResult_t daliCheckpointDestroy(daliCheckpoint_h checkpoint) {
+  DALI_PROLOG();
+  delete ToPointer(checkpoint);
+  DALI_EPILOG();
+}