Improve DLPack support for external tensor consumption (#6261)

- Adds bulk DLPack TensorList constructors for lists of external tensors. The C++ path builds
  CPU and GPU TensorLists in one pass after preflighting devices and peeking all capsule dtypes,
  so dtype or device mismatches can fall back without partially consuming single-use capsules.
- Adds `ndd.Batch` fast paths for:
  - lists of already evaluated `ndd.Tensor` objects, preserving storage, layout, dtype, enum
    metadata, laziness behavior where applicable, and CUDA stream order;
  - lists of external GPU tensors exposing DLPack, including stream-keyword handshakes and
    same-device validation;
  - lists of CPU and CUDA-host/pinned CPU DLPack tensors.
- Keeps requested dtype conversion on the existing slow path, so external DLPack capsules are not
  consumed before conversion fallback.
- Improves `ndd.Tensor` DLPack ingestion:
  - CPU read-only arrays that raise `BufferError` fall back to an explicit copy through the array
    interface;
  - GPU objects retry `__dlpack__()` without the `stream` keyword only when the producer rejects
    the keyword, and fall back to `__cuda_array_interface__` only when DLPack is unavailable.
- Fixes mixed pinned/non-pinned non-contiguous GPU TensorLists. GPU sample sharing no longer
  requires uniform `is_pinned()` state, which unblocks mixed-GPU batches where cross-device copies
  use pinned staging memory. CPU TensorLists still enforce pinned-state compatibility.
- Makes `UserStream::GetStream(size_t dev)` public so the new bulk DLPack path can derive a
  concrete per-device consumer stream.

Signed-off-by: Janusz Lisiecki <jlisiecki@nvidia.com>

Author: JanuszL

dali/pipeline/data/tensor_list.cc

@@ -257,9 +257,11 @@ void TensorList<Backend>::VerifySampleShareCompatibility(DALIDataType type, int
                            this->GetLayout(), ", new: ", layout, " or come with empty layout ",
                            error_suffix));
 
-  DALI_ENFORCE(this->is_pinned() == pinned,
-               make_string("Sample must have the same pinned status as target batch, current: ",
-                           this->is_pinned(), ", new: ", pinned, error_suffix));
+  if constexpr (std::is_same_v<Backend, CPUBackend>) {
+    DALI_ENFORCE(this->is_pinned() == pinned,
+                 make_string("Sample must have the same pinned status as target batch, current: ",
+                             this->is_pinned(), ", new: ", pinned, error_suffix));
+  }
 
   DALI_ENFORCE(this->device_id() == device_id,
                make_string("Sample must have the same device id as target batch, current: ",

dali/pipeline/data/tensor_list_test.cc

@@ -983,13 +983,18 @@ TYPED_TEST(TensorListSuite, SetupLikeMultiGPU) {
 template <typename Backend>
 std::vector<std::pair<std::string, std::function<void(TensorList<Backend> &)>>> SetRequiredSetters(
     int sample_dim, DALIDataType type, TensorLayout layout, bool pinned, int device_id) {
-  return {
+  std::vector<std::pair<std::string, std::function<void(TensorList<Backend> &)>>> result = {
       {"sample dim", [sample_dim](TensorList<Backend> &t) { t.set_sample_dim(sample_dim); }},
       {"type", [type](TensorList<Backend> &t) { t.set_type(type); }},
       {"layout", [layout](TensorList<Backend> &t) { t.SetLayout(layout); }},
       {"device id", [device_id](TensorList<Backend> &t) { t.set_device_id(device_id); }},
-      {"pinned", [pinned](TensorList<Backend> &t) { t.set_pinned(pinned); }},
   };
+  // GPU TensorList allows mixed pinned/non-pinned samples in non-contiguous mode
+  // (pinned tensors are valid GPU-accessible staging buffers).
+  if constexpr (std::is_same_v<Backend, CPUBackend>) {
+    result.push_back({"pinned", [pinned](TensorList<Backend> &t) { t.set_pinned(pinned); }});
+  }
+  return result;
 }
 
 TYPED_TEST(TensorListSuite, PartialSetupSetMultiGPU) {