|
| 1 | +// SPDX-License-Identifier: Apache-2.0 |
| 2 | +// SPDX-FileCopyrightText: Copyright the Vortex contributors |
| 3 | + |
| 4 | +//! CUDA-aware Vortex file opening. |
| 5 | +
|
| 6 | +use std::path::Path; |
| 7 | +use std::sync::Arc; |
| 8 | + |
| 9 | +use vortex::error::VortexResult; |
| 10 | +use vortex::error::vortex_bail; |
| 11 | +use vortex::file::VortexFile; |
| 12 | +use vortex::file::VortexOpenOptions; |
| 13 | +use vortex::io::session::RuntimeSessionExt; |
| 14 | +use vortex::layout::Layout; |
| 15 | +use vortex::layout::layouts::zoned::LegacyStats; |
| 16 | +use vortex::layout::layouts::zoned::Zoned; |
| 17 | +use vortex::layout::segments::SegmentFuture; |
| 18 | +use vortex::layout::segments::SegmentId; |
| 19 | +use vortex::layout::segments::SegmentSource; |
| 20 | + |
| 21 | +use crate::CudaSessionExt; |
| 22 | +use crate::PooledFileReadAt; |
| 23 | +use crate::layout::register_cuda_layout; |
| 24 | + |
| 25 | +/// Extension trait for opening CUDA-readable files from [`VortexOpenOptions`]. |
| 26 | +pub trait CudaOpenOptionsExt { |
| 27 | + /// Configure the opener to keep control-plane segments on the host and read data-plane |
| 28 | + /// segments into CUDA device memory. |
| 29 | + fn with_cuda(self) -> CudaOpenOptions; |
| 30 | +} |
| 31 | + |
| 32 | +impl CudaOpenOptionsExt for VortexOpenOptions { |
| 33 | + fn with_cuda(self) -> CudaOpenOptions { |
| 34 | + CudaOpenOptions { inner: self } |
| 35 | + } |
| 36 | +} |
| 37 | + |
| 38 | +/// CUDA-aware Vortex file open options. |
| 39 | +/// |
| 40 | +/// Construct this adapter with [`CudaOpenOptionsExt::with_cuda`]. Standard file options should be |
| 41 | +/// configured before calling `with_cuda`. |
| 42 | +pub struct CudaOpenOptions { |
| 43 | + inner: VortexOpenOptions, |
| 44 | +} |
| 45 | + |
| 46 | +impl CudaOpenOptions { |
| 47 | + /// Open a local Vortex file for CUDA execution. |
| 48 | + /// |
| 49 | + /// The footer and zone-map segments are read through the ordinary host path. All other file |
| 50 | + /// segments are staged through the session-scoped pinned buffer pool and transferred to the |
| 51 | + /// device. |
| 52 | + pub async fn open_path(self, path: impl AsRef<Path>) -> VortexResult<VortexFile> { |
| 53 | + let path = path.as_ref().to_owned(); |
| 54 | + register_cuda_layout(self.inner.session()); |
| 55 | + |
| 56 | + // A host cache cannot front the data source: cached host buffers would violate the CUDA |
| 57 | + // source's device-placement contract. It remains enabled on the control-plane source. |
| 58 | + let data_options = self.inner.clone().without_segment_cache(); |
| 59 | + let control_file = self.inner.open_path(&path).await?; |
| 60 | + let footer = control_file.footer().clone(); |
| 61 | + |
| 62 | + let session = control_file.session().clone(); |
| 63 | + let cuda_session = session.cuda_session(); |
| 64 | + let stream = cuda_session.stream()?; |
| 65 | + let pool = Arc::clone(cuda_session.pinned_buffer_pool()); |
| 66 | + drop(cuda_session); |
| 67 | + |
| 68 | + let reader = Arc::new(PooledFileReadAt::open( |
| 69 | + &path, |
| 70 | + session.handle(), |
| 71 | + pool, |
| 72 | + stream, |
| 73 | + )?); |
| 74 | + let data_file = data_options |
| 75 | + .with_footer(footer.clone()) |
| 76 | + .open(reader) |
| 77 | + .await?; |
| 78 | + |
| 79 | + let control_segments = |
| 80 | + control_plane_segments(footer.layout().as_ref(), footer.segment_map().len())?; |
| 81 | + let routed = Arc::new(RoutingSegmentSource { |
| 82 | + control_segments: control_segments.into(), |
| 83 | + control: control_file.segment_source(), |
| 84 | + data: data_file.segment_source(), |
| 85 | + }); |
| 86 | + |
| 87 | + Ok(data_file.with_segment_source(routed)) |
| 88 | + } |
| 89 | +} |
| 90 | + |
| 91 | +struct RoutingSegmentSource { |
| 92 | + control_segments: Arc<[bool]>, |
| 93 | + control: Arc<dyn SegmentSource>, |
| 94 | + data: Arc<dyn SegmentSource>, |
| 95 | +} |
| 96 | + |
| 97 | +impl SegmentSource for RoutingSegmentSource { |
| 98 | + fn request(&self, id: SegmentId) -> SegmentFuture { |
| 99 | + let Ok(idx) = usize::try_from(*id) else { |
| 100 | + return self.data.request(id); |
| 101 | + }; |
| 102 | + if self.control_segments.get(idx).copied().unwrap_or(false) { |
| 103 | + self.control.request(id) |
| 104 | + } else { |
| 105 | + self.data.request(id) |
| 106 | + } |
| 107 | + } |
| 108 | +} |
| 109 | + |
| 110 | +fn control_plane_segments(layout: &dyn Layout, segment_count: usize) -> VortexResult<Vec<bool>> { |
| 111 | + let mut control_segments = vec![false; segment_count]; |
| 112 | + mark_zone_map_segments(layout, &mut control_segments)?; |
| 113 | + Ok(control_segments) |
| 114 | +} |
| 115 | + |
| 116 | +fn mark_zone_map_segments(layout: &dyn Layout, control_segments: &mut [bool]) -> VortexResult<()> { |
| 117 | + if layout.is::<Zoned>() || layout.is::<LegacyStats>() { |
| 118 | + mark_layout_segments(layout.child(1)?.as_ref(), control_segments)?; |
| 119 | + mark_zone_map_segments(layout.child(0)?.as_ref(), control_segments)?; |
| 120 | + return Ok(()); |
| 121 | + } |
| 122 | + |
| 123 | + for child in layout.children()? { |
| 124 | + mark_zone_map_segments(child.as_ref(), control_segments)?; |
| 125 | + } |
| 126 | + Ok(()) |
| 127 | +} |
| 128 | + |
| 129 | +fn mark_layout_segments(layout: &dyn Layout, control_segments: &mut [bool]) -> VortexResult<()> { |
| 130 | + for id in layout.segment_ids() { |
| 131 | + let idx = usize::try_from(*id)?; |
| 132 | + let Some(is_control) = control_segments.get_mut(idx) else { |
| 133 | + vortex_bail!("layout references out-of-range segment {id}"); |
| 134 | + }; |
| 135 | + *is_control = true; |
| 136 | + } |
| 137 | + |
| 138 | + for child in layout.children()? { |
| 139 | + mark_layout_segments(child.as_ref(), control_segments)?; |
| 140 | + } |
| 141 | + Ok(()) |
| 142 | +} |
| 143 | + |
| 144 | +#[cfg(test)] |
| 145 | +mod tests { |
| 146 | + use std::num::NonZeroUsize; |
| 147 | + |
| 148 | + use futures::FutureExt; |
| 149 | + use parking_lot::Mutex; |
| 150 | + use vortex::array::buffer::BufferHandle; |
| 151 | + use vortex::array::dtype::DType; |
| 152 | + use vortex::array::dtype::Nullability::NonNullable; |
| 153 | + use vortex::array::dtype::PType; |
| 154 | + use vortex::array::dtype::StructFields; |
| 155 | + use vortex::buffer::ByteBuffer; |
| 156 | + use vortex::layout::IntoLayout; |
| 157 | + use vortex::layout::layouts::flat::FlatLayout; |
| 158 | + use vortex::layout::layouts::zoned::ZonedLayout; |
| 159 | + use vortex::session::registry::ReadContext; |
| 160 | + |
| 161 | + use super::*; |
| 162 | + |
| 163 | + #[derive(Clone)] |
| 164 | + struct RecordingSource { |
| 165 | + requests: Arc<Mutex<Vec<SegmentId>>>, |
| 166 | + value: u8, |
| 167 | + } |
| 168 | + |
| 169 | + impl RecordingSource { |
| 170 | + fn new(value: u8) -> Self { |
| 171 | + Self { |
| 172 | + requests: Arc::new(Mutex::new(Vec::new())), |
| 173 | + value, |
| 174 | + } |
| 175 | + } |
| 176 | + } |
| 177 | + |
| 178 | + impl SegmentSource for RecordingSource { |
| 179 | + fn request(&self, id: SegmentId) -> SegmentFuture { |
| 180 | + self.requests.lock().push(id); |
| 181 | + let value = self.value; |
| 182 | + async move { Ok(BufferHandle::new_host(ByteBuffer::from(vec![value]))) }.boxed() |
| 183 | + } |
| 184 | + } |
| 185 | + |
| 186 | + #[tokio::test] |
| 187 | + async fn routes_control_and_data_segments() { |
| 188 | + let control = RecordingSource::new(1); |
| 189 | + let data = RecordingSource::new(2); |
| 190 | + let source = RoutingSegmentSource { |
| 191 | + control_segments: Arc::from([false, true, false]), |
| 192 | + control: Arc::new(control.clone()), |
| 193 | + data: Arc::new(data.clone()), |
| 194 | + }; |
| 195 | + |
| 196 | + let control_buffer = source |
| 197 | + .request(SegmentId::from(1)) |
| 198 | + .await |
| 199 | + .unwrap() |
| 200 | + .unwrap_host(); |
| 201 | + let data_buffer = source |
| 202 | + .request(SegmentId::from(2)) |
| 203 | + .await |
| 204 | + .unwrap() |
| 205 | + .unwrap_host(); |
| 206 | + |
| 207 | + assert_eq!(control_buffer.as_ref(), &[1]); |
| 208 | + assert_eq!(data_buffer.as_ref(), &[2]); |
| 209 | + assert_eq!(*control.requests.lock(), [SegmentId::from(1)]); |
| 210 | + assert_eq!(*data.requests.lock(), [SegmentId::from(2)]); |
| 211 | + } |
| 212 | + |
| 213 | + #[test] |
| 214 | + fn marks_only_zone_map_subtree_segments_as_control_plane() -> VortexResult<()> { |
| 215 | + let read_ctx = ReadContext::new([]); |
| 216 | + let data_dtype = DType::Primitive(PType::I32, NonNullable); |
| 217 | + let data = |
| 218 | + FlatLayout::new(16, data_dtype, SegmentId::from(0), read_ctx.clone()).into_layout(); |
| 219 | + let zones = FlatLayout::new( |
| 220 | + 1, |
| 221 | + DType::Struct(StructFields::empty(), NonNullable), |
| 222 | + SegmentId::from(1), |
| 223 | + read_ctx, |
| 224 | + ) |
| 225 | + .into_layout(); |
| 226 | + let layout = |
| 227 | + ZonedLayout::try_new(data, zones, NonZeroUsize::new(16).unwrap(), Arc::new([]))? |
| 228 | + .into_layout(); |
| 229 | + |
| 230 | + let control_segments = control_plane_segments(layout.as_ref(), 2)?; |
| 231 | + |
| 232 | + assert_eq!(control_segments, [false, true]); |
| 233 | + Ok(()) |
| 234 | + } |
| 235 | +} |
0 commit comments