1use std::ffi::c_void;
8use std::sync::Arc;
9
10use cudarc::driver::sys;
11use xlog_core::{Result, ScalarType, Schema, XlogError};
12
13use crate::cuda_compat::{AsKernelParam, DeviceSlice, LaunchAsync, LaunchConfig};
14use crate::launch::LaunchRecorder;
15use crate::memory::{GpuMemoryReservation, TrackedCudaSlice};
16use crate::{CudaBuffer, CudaColumn, CudaStream};
17
18use super::CudaKernelProvider;
19
20const MODULE: &str = "xlog_resident_relational";
21const BLOCK_SIZE: u32 = 256;
22
23pub const RESIDENT_RELATIONAL_MAX_ARITY: usize = 17;
25
26#[repr(u32)]
28#[derive(Debug, Clone, Copy, PartialEq, Eq)]
29pub enum ResidentTerminalCode {
30 Running = 0,
31 Success = 1,
32 IterationLimit = 2,
33 CapacityOverflow = 3,
34 ResourceExhausted = 4,
35}
36
37#[repr(u32)]
39#[derive(Debug, Clone, Copy, PartialEq, Eq)]
40pub enum ResidentResourceCode {
41 None = 0,
42 SetHashSlots = 1,
43 JoinBuckets = 2,
44 JoinChains = 3,
45 InputRows = 4,
46 OutputRows = 5,
47}
48
49#[repr(C, align(8))]
55#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
56pub struct ResidentTerminalStatus {
57 pub code: u32,
58 pub op_id: u32,
59 pub resource_code: u32,
60 pub iterations: u32,
61 pub limit: u32,
62 pub reserved: u32,
63 pub required: u64,
64 pub capacity: u64,
65}
66
67unsafe impl cudarc::driver::DeviceRepr for ResidentTerminalStatus {}
70
71#[repr(C, align(8))]
72#[derive(Clone, Copy)]
73struct ResidentRelationView {
74 columns: [u64; RESIDENT_RELATIONAL_MAX_ARITY],
75 widths: [u32; RESIDENT_RELATIONAL_MAX_ARITY],
76 arity: u32,
77 capacity: u32,
78 reserved: u32,
79 num_rows: u64,
80}
81
82impl AsKernelParam for ResidentRelationView {
83 fn as_kernel_param(&self) -> *mut c_void {
84 (self as *const Self).cast_mut().cast()
85 }
86}
87
88pub struct ResidentRelation {
90 buffer: CudaBuffer,
91}
92
93impl ResidentRelation {
94 pub fn from_buffer(buffer: CudaBuffer) -> Self {
96 Self { buffer }
97 }
98
99 pub fn buffer(&self) -> &CudaBuffer {
100 &self.buffer
101 }
102
103 pub fn invalidate_observed_row_count(&self) {
105 self.buffer.invalidate_cached_row_count();
106 }
107
108 pub fn into_buffer(self) -> CudaBuffer {
112 self.buffer
113 }
114
115 pub fn into_buffer_with_observed_schema(mut self, schema: Schema) -> Result<CudaBuffer> {
120 if schema.arity() != self.buffer.arity() {
121 return Err(XlogError::Kernel(format!(
122 "resident observed schema arity {} does not match allocation arity {}",
123 schema.arity(),
124 self.buffer.arity()
125 )));
126 }
127 for column in 0..schema.arity() {
128 let old_width = width(
129 self.buffer
130 .schema()
131 .column_type(column)
132 .expect("matching arity"),
133 )?;
134 let new_width = width(schema.column_type(column).expect("matching arity"))?;
135 if old_width != new_width {
136 return Err(XlogError::Kernel(format!(
137 "resident observed schema column {column} changes physical width from \
138 {old_width} to {new_width} bytes"
139 )));
140 }
141 }
142 self.buffer.set_schema(schema);
143 Ok(self.buffer)
144 }
145
146 pub fn retag_schema_for_capture(&mut self, schema: Schema) -> Result<()> {
153 if schema.arity() != self.buffer.arity() {
154 return Err(XlogError::Kernel(format!(
155 "resident scratch retag arity {} does not match allocation arity {}",
156 schema.arity(),
157 self.buffer.arity()
158 )));
159 }
160 for column in 0..schema.arity() {
161 let old_width = width(
162 self.buffer
163 .schema()
164 .column_type(column)
165 .expect("matching arity"),
166 )?;
167 let new_width = width(schema.column_type(column).expect("matching arity"))?;
168 if old_width != new_width {
169 return Err(XlogError::Kernel(format!(
170 "resident scratch retag column {column} changes physical width from \
171 {old_width} to {new_width} bytes"
172 )));
173 }
174 }
175 self.buffer.set_schema(schema);
176 Ok(())
177 }
178
179 pub fn capacity(&self) -> u32 {
180 self.buffer
181 .num_rows()
182 .try_into()
183 .expect("resident capacity")
184 }
185
186 pub fn num_rows_device(&self) -> &TrackedCudaSlice<u32> {
187 self.buffer.num_rows_device()
188 }
189
190 pub fn record_uses(&self, recorder: &mut LaunchRecorder) {
192 for column in self.buffer.columns() {
193 recorder.write_column(column);
194 }
195 recorder.read_write(self.buffer.num_rows_device());
196 }
197}
198
199pub struct ResidentSetWorkspace {
201 slots: TrackedCudaSlice<u64>,
202 required: TrackedCudaSlice<u64>,
203 candidate_capacity: u32,
204 slot_mask: u32,
205}
206
207impl ResidentSetWorkspace {
208 pub fn record_uses(&self, recorder: &mut LaunchRecorder) {
209 recorder.read_write(&self.slots);
210 recorder.read_write(&self.required);
211 }
212
213 pub(crate) fn schedule_parts(&self) -> (u64, u64, u32, u32) {
214 (
215 self.slots.device_ptr_value(),
216 self.required.device_ptr_value(),
217 self.slot_mask,
218 self.candidate_capacity,
219 )
220 }
221
222 pub(crate) fn schedule_owner_snapshots(
223 &self,
224 ) -> Result<[Option<crate::memory::RuntimeAllocationIdentity>; 2]> {
225 Ok([
226 self.slots.runtime_allocation_identity()?,
227 self.required.runtime_allocation_identity()?,
228 ])
229 }
230}
231
232pub struct ResidentJoinWorkspace {
234 bucket_heads: TrackedCudaSlice<u32>,
235 next: TrackedCudaSlice<u32>,
236 required: TrackedCudaSlice<u64>,
237 right_capacity: u32,
238 bucket_mask: u32,
239}
240
241impl ResidentJoinWorkspace {
242 pub fn record_uses(&self, recorder: &mut LaunchRecorder) {
243 recorder.read_write(&self.bucket_heads);
244 recorder.read_write(&self.next);
245 recorder.read_write(&self.required);
246 }
247
248 pub(crate) fn schedule_parts(&self) -> (u64, u64, u64, u32, u32) {
249 (
250 self.bucket_heads.device_ptr_value(),
251 self.next.device_ptr_value(),
252 self.required.device_ptr_value(),
253 self.bucket_mask,
254 self.right_capacity,
255 )
256 }
257
258 pub(crate) fn schedule_owner_snapshots(
259 &self,
260 ) -> Result<[Option<crate::memory::RuntimeAllocationIdentity>; 3]> {
261 Ok([
262 self.bucket_heads.runtime_allocation_identity()?,
263 self.next.runtime_allocation_identity()?,
264 self.required.runtime_allocation_identity()?,
265 ])
266 }
267}
268
269pub struct ResidentConvergenceControl {
271 status: TrackedCudaSlice<ResidentTerminalStatus>,
272 changed: TrackedCudaSlice<u32>,
273 loop_iterations: TrackedCudaSlice<u32>,
274}
275
276pub struct ResidentDeviceTrace {
278 scan_invocations: TrackedCudaSlice<u32>,
279 filter_invocations: TrackedCudaSlice<u32>,
280 semantic_scan_invocations: TrackedCudaSlice<u32>,
281 semantic_filter_invocations: TrackedCudaSlice<u32>,
282}
283
284pub struct ResidentSchemaWinners {
286 seen_nonempty: TrackedCudaSlice<u32>,
287 winner_schema_ids: TrackedCudaSlice<u32>,
288 default_schema_ids: Vec<u32>,
289 len: u32,
290}
291
292#[derive(Debug, Clone, Copy, PartialEq, Eq)]
293pub(crate) enum ResidentReceiptPointeeRole {
294 RelationCount(u32),
295 ScanTrace,
296 FilterTrace,
297 SemanticScanTrace,
298 SemanticFilterTrace,
299 SchemaWinner(u32),
300}
301
302#[derive(Debug, Clone, Copy, PartialEq, Eq)]
303pub(crate) struct ResidentReceiptPointee {
304 pub(crate) role: ResidentReceiptPointeeRole,
305 pub(crate) ptr: u64,
306 pub(crate) range_end: u64,
307 pub(crate) manager_id: usize,
308 pub(crate) block: Option<(crate::device_runtime::BlockId, usize)>,
309}
310
311fn checked_receipt_pointee(
312 role: ResidentReceiptPointeeRole,
313 ptr: u64,
314 manager_id: usize,
315 runtime_block: Option<(crate::device_runtime::BlockId, usize)>,
316) -> Result<ResidentReceiptPointee> {
317 let range_end = ptr.checked_add(4).ok_or_else(|| {
318 XlogError::Kernel(
319 "resident receipt pointee range overflows the device address space".into(),
320 )
321 })?;
322 let block = match runtime_block {
323 Some((block, bytes)) => {
324 let block_bytes = u64::try_from(bytes).map_err(|_| {
325 XlogError::Kernel("resident receipt runtime block size is not representable".into())
326 })?;
327 let block_end = block.ptr.checked_add(block_bytes).ok_or_else(|| {
328 XlogError::Kernel("resident receipt runtime block range overflows".into())
329 })?;
330 if ptr < block.ptr || range_end > block_end {
331 return Err(XlogError::Kernel(
332 "resident receipt pointee is outside its runtime block".into(),
333 ));
334 }
335 Some((block, bytes))
336 }
337 None => None,
338 };
339 Ok(ResidentReceiptPointee {
340 role,
341 ptr,
342 range_end,
343 manager_id,
344 block,
345 })
346}
347
348fn validate_receipt_pointee_ranges(pointees: &[ResidentReceiptPointee]) -> Result<()> {
349 for (index, pointee) in pointees.iter().enumerate() {
350 for previous in &pointees[..index] {
351 if pointee.ptr < previous.range_end && previous.ptr < pointee.range_end {
352 return Err(XlogError::Kernel(
353 "resident receipt pointee ranges overlap".into(),
354 ));
355 }
356 }
357 }
358 Ok(())
359}
360
361fn validate_receipt_pointee_owners(
362 pointees: &[ResidentReceiptPointee],
363 manager_id: usize,
364 device_ordinal: u32,
365) -> Result<()> {
366 for pointee in pointees {
367 if pointee.manager_id != manager_id {
368 return Err(XlogError::Kernel(
369 "resident receipt pointee belongs to a foreign memory manager".into(),
370 ));
371 }
372 let block = pointee.block.ok_or_else(|| {
373 XlogError::Kernel("resident receipt pointee has no runtime block identity".into())
374 })?;
375 if block.0.device_ordinal != device_ordinal {
376 return Err(XlogError::Kernel(
377 "resident receipt pointee belongs to a foreign CUDA device".into(),
378 ));
379 }
380 }
381 Ok(())
382}
383
384fn validate_receipt_schedule_block_mapping(
385 pointees: &[ResidentReceiptPointee],
386 expected_blocks: &[crate::device_runtime::BlockId],
387) -> Result<()> {
388 if pointees.len() != expected_blocks.len()
389 || pointees
390 .iter()
391 .zip(expected_blocks)
392 .any(|(pointee, expected)| pointee.block.map(|(block, _)| block) != Some(*expected))
393 {
394 return Err(XlogError::Kernel(
395 "resident receipt runtime-block mapping differs from the schedule".into(),
396 ));
397 }
398 Ok(())
399}
400
401fn record_receipt_pointee_uses(pointees: &[ResidentReceiptPointee], recorder: &mut LaunchRecorder) {
402 for pointee in pointees {
403 recorder.read_optional_block_identity(pointee.block);
404 }
405}
406
407fn validate_receipt_schedule_mapping(
408 pointees: &[ResidentReceiptPointee],
409 relation_counts: &[u64],
410 trace_counts: [u64; 4],
411 schema_winners: &[u64],
412) -> Result<()> {
413 if relation_counts.len() != schema_winners.len() {
414 return Err(XlogError::Kernel(
415 "resident receipt relation and schema-winner mappings differ in length".into(),
416 ));
417 }
418 let expected_len = relation_counts
419 .len()
420 .checked_mul(2)
421 .and_then(|count| count.checked_add(4))
422 .ok_or_else(|| XlogError::Kernel("resident receipt mapping length overflow".into()))?;
423 if pointees.len() != expected_len {
424 return Err(XlogError::Kernel(
425 "resident receipt pointee manifest has the wrong length".into(),
426 ));
427 }
428 for (index, ptr) in relation_counts.iter().copied().enumerate() {
429 let index = u32::try_from(index)
430 .map_err(|_| XlogError::Kernel("resident receipt head index overflow".into()))?;
431 if pointees[index as usize].role != ResidentReceiptPointeeRole::RelationCount(index)
432 || pointees[index as usize].ptr != ptr
433 {
434 return Err(XlogError::Kernel(
435 "resident receipt relation-count mapping differs from the schedule".into(),
436 ));
437 }
438 }
439 let trace_offset = relation_counts.len();
440 if pointees[trace_offset].role != ResidentReceiptPointeeRole::ScanTrace
441 || pointees[trace_offset].ptr != trace_counts[0]
442 || pointees[trace_offset + 1].role != ResidentReceiptPointeeRole::FilterTrace
443 || pointees[trace_offset + 1].ptr != trace_counts[1]
444 || pointees[trace_offset + 2].role != ResidentReceiptPointeeRole::SemanticScanTrace
445 || pointees[trace_offset + 2].ptr != trace_counts[2]
446 || pointees[trace_offset + 3].role != ResidentReceiptPointeeRole::SemanticFilterTrace
447 || pointees[trace_offset + 3].ptr != trace_counts[3]
448 {
449 return Err(XlogError::Kernel(
450 "resident receipt trace mapping differs from the schedule".into(),
451 ));
452 }
453 let winner_offset = trace_offset + 4;
454 for (index, ptr) in schema_winners.iter().copied().enumerate() {
455 let index_u32 = u32::try_from(index)
456 .map_err(|_| XlogError::Kernel("resident receipt head index overflow".into()))?;
457 let pointee = &pointees[winner_offset + index];
458 if pointee.role != ResidentReceiptPointeeRole::SchemaWinner(index_u32) || pointee.ptr != ptr
459 {
460 return Err(XlogError::Kernel(
461 "resident receipt schema-winner mapping differs from the schedule".into(),
462 ));
463 }
464 }
465 Ok(())
466}
467
468impl ResidentSchemaWinners {
469 pub fn record_uses(&self, recorder: &mut LaunchRecorder) {
471 recorder.read_write(&self.seen_nonempty);
472 recorder.read_write(&self.winner_schema_ids);
473 }
474
475 pub(crate) fn schedule_parts(&self) -> (u64, u64, u32) {
476 (
477 self.seen_nonempty.device_ptr_value(),
478 self.winner_schema_ids.device_ptr_value(),
479 self.len,
480 )
481 }
482
483 pub(crate) fn default_schema_ids(&self) -> &[u32] {
484 &self.default_schema_ids
485 }
486
487 pub fn len(&self) -> u32 {
488 self.len
489 }
490
491 pub fn is_empty(&self) -> bool {
493 self.len() == 0
494 }
495
496 pub(crate) fn schedule_owner_snapshots(
497 &self,
498 ) -> Result<[Option<crate::memory::RuntimeAllocationIdentity>; 2]> {
499 Ok([
500 self.seen_nonempty.runtime_allocation_identity()?,
501 self.winner_schema_ids.runtime_allocation_identity()?,
502 ])
503 }
504}
505
506impl ResidentDeviceTrace {
507 pub fn record_uses(&self, recorder: &mut LaunchRecorder) {
509 recorder.read_write(&self.scan_invocations);
510 recorder.read_write(&self.filter_invocations);
511 recorder.read_write(&self.semantic_scan_invocations);
512 recorder.read_write(&self.semantic_filter_invocations);
513 }
514
515 pub(crate) fn schedule_parts(&self) -> (u64, u64, u64, u64) {
516 (
517 self.scan_invocations.device_ptr_value(),
518 self.filter_invocations.device_ptr_value(),
519 self.semantic_scan_invocations.device_ptr_value(),
520 self.semantic_filter_invocations.device_ptr_value(),
521 )
522 }
523
524 pub(crate) fn schedule_owner_snapshots(
525 &self,
526 ) -> Result<[Option<crate::memory::RuntimeAllocationIdentity>; 4]> {
527 Ok([
528 self.scan_invocations.runtime_allocation_identity()?,
529 self.filter_invocations.runtime_allocation_identity()?,
530 self.semantic_scan_invocations
531 .runtime_allocation_identity()?,
532 self.semantic_filter_invocations
533 .runtime_allocation_identity()?,
534 ])
535 }
536}
537
538pub struct ResidentPackedReceipt {
540 count_ptrs: TrackedCudaSlice<u64>,
541 bytes: TrackedCudaSlice<u8>,
542 pointees: Vec<ResidentReceiptPointee>,
543 count_len: u32,
544 relation_count_len: u32,
545 device_trace_field_count: u32,
546 schema_winner_count: u32,
547}
548
549#[derive(Debug)]
551pub struct ResidentPinnedReceipt {
552 buffer: crate::device::PinnedHostBuffer,
553}
554
555impl ResidentPinnedReceipt {
556 pub fn len_bytes(&self) -> usize {
557 self.buffer.len()
558 }
559}
560
561impl ResidentPackedReceipt {
562 #[cfg(test)]
563 pub(crate) fn pointee_manifest(&self) -> &[ResidentReceiptPointee] {
564 &self.pointees
565 }
566
567 pub fn device_bytes(&self) -> &TrackedCudaSlice<u8> {
568 &self.bytes
569 }
570
571 pub fn len_bytes(&self) -> usize {
572 self.bytes.len()
573 }
574
575 pub fn relation_count_len(&self) -> u32 {
576 self.relation_count_len
577 }
578
579 pub fn device_trace_field_count(&self) -> u32 {
580 self.device_trace_field_count
581 }
582
583 pub fn schema_winner_count(&self) -> u32 {
584 self.schema_winner_count
585 }
586
587 pub fn total_count_field_len(&self) -> u32 {
588 self.count_len
589 }
590
591 pub fn record_uses(&self, recorder: &mut LaunchRecorder) {
592 recorder.read(&self.count_ptrs);
593 recorder.write(&self.bytes);
594 record_receipt_pointee_uses(&self.pointees, recorder);
595 }
596
597 pub(crate) fn schedule_parts(&self) -> (u64, u64, u32, u32) {
598 (
599 self.count_ptrs.device_ptr_value(),
600 self.bytes.device_ptr_value(),
601 self.count_len,
602 u32::try_from(self.bytes.len()).unwrap_or(u32::MAX),
603 )
604 }
605
606 pub(crate) fn validate_schedule_pointees(
607 &self,
608 manager_id: usize,
609 device_ordinal: u32,
610 relation_counts: &[u64],
611 trace_counts: [u64; 4],
612 schema_winners: &[u64],
613 expected_blocks: &[crate::device_runtime::BlockId],
614 ) -> Result<()> {
615 validate_receipt_pointee_owners(&self.pointees, manager_id, device_ordinal)?;
616 validate_receipt_schedule_mapping(
617 &self.pointees,
618 relation_counts,
619 trace_counts,
620 schema_winners,
621 )?;
622 validate_receipt_schedule_block_mapping(&self.pointees, expected_blocks)
623 }
624
625 pub(crate) fn schedule_owner_snapshots(
626 &self,
627 ) -> Result<[Option<crate::memory::RuntimeAllocationIdentity>; 2]> {
628 Ok([
629 self.count_ptrs.runtime_allocation_identity()?,
630 self.bytes.runtime_allocation_identity()?,
631 ])
632 }
633}
634
635impl ResidentConvergenceControl {
636 pub fn status_device(&self) -> &TrackedCudaSlice<ResidentTerminalStatus> {
637 &self.status
638 }
639
640 pub fn status_device_ptr(&self) -> u64 {
641 self.status.device_ptr_value()
642 }
643
644 pub fn iterations_device_ptr(&self) -> u64 {
645 self.status.device_ptr_value() + 12
646 }
647
648 pub(crate) fn schedule_owner_snapshots(
649 &self,
650 ) -> Result<[Option<crate::memory::RuntimeAllocationIdentity>; 3]> {
651 Ok([
652 self.status.runtime_allocation_identity()?,
653 self.changed.runtime_allocation_identity()?,
654 self.loop_iterations.runtime_allocation_identity()?,
655 ])
656 }
657
658 pub fn changed_device(&self) -> &TrackedCudaSlice<u32> {
659 &self.changed
660 }
661
662 pub fn changed_device_ptr(&self) -> u64 {
663 self.changed.device_ptr_value()
664 }
665
666 pub fn loop_iterations_device(&self) -> &TrackedCudaSlice<u32> {
667 &self.loop_iterations
668 }
669
670 pub fn record_uses(&self, recorder: &mut LaunchRecorder) {
671 recorder.read_write(&self.status);
672 recorder.read_write(&self.changed);
673 recorder.read_write(&self.loop_iterations);
674 }
675}
676
677#[derive(Debug, Clone, Copy, PartialEq, Eq)]
679pub enum ResidentJoinKind {
680 Inner,
681 Semi,
682}
683
684pub(crate) fn checked_hash_slot_capacity(candidate_capacity: u64) -> Result<u64> {
685 let doubled = candidate_capacity
686 .checked_mul(2)
687 .ok_or_else(|| XlogError::Kernel("resident hash capacity overflow".to_string()))?;
688 let slots = doubled
689 .max(1)
690 .checked_next_power_of_two()
691 .ok_or_else(|| XlogError::Kernel("resident hash capacity overflow".to_string()))?;
692 if slots > u64::from(u32::MAX) + 1 {
693 return Err(XlogError::Kernel(format!(
694 "resident hash requires {slots} slots, exceeding the u32 index space"
695 )));
696 }
697 Ok(slots)
698}
699
700fn checked_capacity(capacity: u64, label: &str) -> Result<u32> {
701 u32::try_from(capacity).map_err(|_| {
702 XlogError::Kernel(format!(
703 "resident {label} capacity {capacity} exceeds u32::MAX"
704 ))
705 })
706}
707
708#[derive(Debug)]
709struct ResidentRelationLayout {
710 columns: Vec<std::ops::Range<u64>>,
711 row_count: std::ops::Range<u64>,
712 bytes: u64,
713}
714
715fn resident_relation_align(cursor: u64, alignment: u64) -> Result<u64> {
716 debug_assert!(alignment.is_power_of_two());
717 cursor
718 .checked_add(alignment - 1)
719 .map(|end| end & !(alignment - 1))
720 .ok_or_else(|| XlogError::Kernel("resident relation layout byte overflow".into()))
721}
722
723fn resident_relation_layout(schema: &Schema, capacity: u64) -> Result<ResidentRelationLayout> {
724 checked_capacity(capacity, "relation")?;
725 validate_schema(schema)?;
726 let mut cursor = 0u64;
727 let mut columns = Vec::with_capacity(schema.arity());
728 for column in 0..schema.arity() {
729 let column_width = u64::from(width(
730 schema
731 .column_type(column)
732 .expect("resident schema arity checked"),
733 )?);
734 cursor = resident_relation_align(cursor, column_width)?;
735 let column_bytes = capacity
736 .checked_mul(column_width)
737 .ok_or_else(|| XlogError::Kernel("resident column byte overflow".into()))?;
738 let end = cursor
739 .checked_add(column_bytes)
740 .ok_or_else(|| XlogError::Kernel("resident relation layout byte overflow".into()))?;
741 columns.push(cursor..end);
742 cursor = end;
743 }
744 cursor = resident_relation_align(cursor, std::mem::align_of::<u32>() as u64)?;
745 let count_end = cursor
746 .checked_add(std::mem::size_of::<u32>() as u64)
747 .ok_or_else(|| XlogError::Kernel("resident relation layout byte overflow".into()))?;
748 let row_count = cursor..count_end;
749 let bytes = resident_relation_align(count_end, std::mem::align_of::<u64>() as u64)?;
752 Ok(ResidentRelationLayout {
753 columns,
754 row_count,
755 bytes,
756 })
757}
758
759pub fn resident_relation_device_bytes(schema: &Schema, capacity: u64) -> Result<u64> {
761 Ok(resident_relation_layout(schema, capacity)?.bytes)
762}
763
764pub fn resident_set_workspace_device_bytes(candidate_capacity: u64) -> Result<u64> {
766 let slots = checked_hash_slot_capacity(candidate_capacity)?;
767 slots
768 .checked_mul(std::mem::size_of::<u64>() as u64)
769 .and_then(|bytes| bytes.checked_add(std::mem::size_of::<u64>() as u64))
770 .ok_or_else(|| XlogError::Kernel("resident set workspace byte overflow".into()))
771}
772
773pub fn resident_join_workspace_device_bytes(right_capacity: u64) -> Result<u64> {
775 let buckets = checked_hash_slot_capacity(right_capacity)?;
776 buckets
777 .checked_mul(std::mem::size_of::<u32>() as u64)
778 .and_then(|bytes| {
779 right_capacity
780 .checked_mul(std::mem::size_of::<u32>() as u64)
781 .and_then(|next| bytes.checked_add(next))
782 })
783 .and_then(|bytes| bytes.checked_add(std::mem::size_of::<u64>() as u64))
784 .ok_or_else(|| XlogError::Kernel("resident join workspace byte overflow".into()))
785}
786
787pub const fn resident_control_device_bytes() -> u64 {
789 (std::mem::size_of::<ResidentTerminalStatus>() + 2 * std::mem::size_of::<u32>()) as u64
790}
791
792pub const fn resident_device_trace_bytes() -> u64 {
794 (4 * std::mem::size_of::<u32>()) as u64
795}
796
797pub fn resident_schema_winners_device_bytes(head_count: usize) -> Result<u64> {
799 u64::try_from(head_count.max(1))
800 .ok()
801 .and_then(|count| count.checked_mul(2 * std::mem::size_of::<u32>() as u64))
802 .ok_or_else(|| XlogError::Kernel("resident schema winner byte overflow".into()))
803}
804
805fn resident_packed_receipt_device_bytes_for_fields(
806 head_count: usize,
807 schema_winner_count: usize,
808) -> Result<u64> {
809 let count_fields = head_count
810 .checked_add(4)
811 .and_then(|fields| fields.checked_add(schema_winner_count))
812 .ok_or_else(|| XlogError::Kernel("resident receipt count overflow".into()))?;
813 let pointer_bytes = u64::try_from(count_fields.max(1))
814 .ok()
815 .and_then(|count| count.checked_mul(std::mem::size_of::<u64>() as u64))
816 .ok_or_else(|| XlogError::Kernel("resident receipt pointer byte overflow".into()))?;
817 let packed_bytes = std::mem::size_of::<ResidentTerminalStatus>()
818 .checked_add(
819 count_fields
820 .checked_add(1)
821 .and_then(|fields| fields.checked_mul(std::mem::size_of::<u32>()))
822 .ok_or_else(|| XlogError::Kernel("resident receipt byte overflow".into()))?,
823 )
824 .ok_or_else(|| XlogError::Kernel("resident receipt byte overflow".into()))?;
825 pointer_bytes
826 .checked_add(packed_bytes as u64)
827 .ok_or_else(|| XlogError::Kernel("resident receipt total byte overflow".into()))
828}
829
830pub fn resident_packed_receipt_device_bytes(head_count: usize) -> Result<u64> {
832 resident_packed_receipt_device_bytes_for_fields(head_count, 0)
833}
834
835pub fn resident_packed_receipt_with_schema_winners_device_bytes(head_count: usize) -> Result<u64> {
837 resident_packed_receipt_device_bytes_for_fields(head_count, head_count)
838}
839
840fn width(ty: ScalarType) -> Result<u32> {
841 match ty {
842 ScalarType::Symbol | ScalarType::U32 => Ok(4),
843 ScalarType::U64 => Ok(8),
844 other => Err(XlogError::Kernel(format!(
845 "resident relational type {other:?} is unsupported; expected Symbol, U32, or U64"
846 ))),
847 }
848}
849
850fn validate_schema(schema: &Schema) -> Result<()> {
851 if schema.arity() > RESIDENT_RELATIONAL_MAX_ARITY {
852 return Err(XlogError::Kernel(format!(
853 "resident relational arity {} exceeds {RESIDENT_RELATIONAL_MAX_ARITY}",
854 schema.arity()
855 )));
856 }
857 for column in 0..schema.arity() {
858 width(schema.column_type(column).expect("schema arity checked"))?;
859 }
860 Ok(())
861}
862
863fn relation_view(buffer: &CudaBuffer) -> Result<ResidentRelationView> {
864 validate_schema(buffer.schema())?;
865 let capacity = checked_capacity(buffer.num_rows(), "relation")?;
866 let mut columns = [0; RESIDENT_RELATIONAL_MAX_ARITY];
867 let mut widths = [0; RESIDENT_RELATIONAL_MAX_ARITY];
868 for column in 0..buffer.arity() {
869 columns[column] = *buffer.column(column).expect("arity checked").device_ptr();
870 widths[column] = width(
871 buffer
872 .schema()
873 .column_type(column)
874 .expect("schema arity checked"),
875 )?;
876 }
877 Ok(ResidentRelationView {
878 columns,
879 widths,
880 arity: buffer.arity() as u32,
881 capacity,
882 reserved: 0,
883 num_rows: buffer.num_rows_device().device_ptr_value(),
884 })
885}
886
887fn ensure_same_schema(a: &CudaBuffer, b: &CudaBuffer) -> Result<()> {
888 if !same_physical_layout(a.schema(), b.schema()) {
889 return Err(XlogError::Kernel(
890 "resident set operands must have identical physical layouts".to_string(),
891 ));
892 }
893 Ok(())
894}
895
896fn same_physical_layout(left: &Schema, right: &Schema) -> bool {
897 left.arity() == right.arity()
898 && (0..left.arity()).all(|column| left.column_type(column) == right.column_type(column))
899}
900
901fn launch_config(capacity: u32) -> LaunchConfig {
902 LaunchConfig {
903 grid_dim: (capacity.max(1).div_ceil(BLOCK_SIZE), 1, 1),
904 block_dim: (BLOCK_SIZE, 1, 1),
905 shared_mem_bytes: 0,
906 }
907}
908
909fn memset_u8_async(
910 ptr: sys::CUdeviceptr,
911 value: u8,
912 bytes: usize,
913 stream: &CudaStream,
914 label: &str,
915) -> Result<()> {
916 if bytes == 0 {
917 return Ok(());
918 }
919 let code = unsafe { sys::cuMemsetD8Async(ptr, value, bytes, stream.cu_stream()) };
921 if code != sys::cudaError_enum::CUDA_SUCCESS {
922 return Err(XlogError::Kernel(format!(
923 "resident {label} memset failed: {code:?}"
924 )));
925 }
926 Ok(())
927}
928
929impl CudaKernelProvider {
930 pub fn prepare_resident_relation(
932 &self,
933 schema: Schema,
934 capacity: u64,
935 ) -> Result<ResidentRelation> {
936 validate_schema(&schema)?;
937 let capacity_u32 = checked_capacity(capacity, "output")?;
938 let mut columns = Vec::with_capacity(schema.arity());
939 for column in 0..schema.arity() {
940 let bytes = (capacity as usize)
941 .checked_mul(width(schema.column_type(column).expect("schema"))? as usize)
942 .ok_or_else(|| XlogError::Kernel("resident column byte overflow".to_string()))?;
943 columns.push(CudaColumn::Owned(self.memory().alloc::<u8>(bytes)?));
944 }
945 let d_num_rows = self.memory().alloc::<u32>(1)?;
946 let buffer = CudaBuffer::from_columns(columns, capacity, d_num_rows, schema);
947 debug_assert_eq!(capacity_u32 as u64, capacity);
948 Ok(ResidentRelation { buffer })
949 }
950
951 pub fn prepare_resident_relation_in_reservation(
953 &self,
954 schema: Schema,
955 capacity: u64,
956 reservation: &mut GpuMemoryReservation,
957 ) -> Result<ResidentRelation> {
958 self.prepare_resident_relations_in_reservation(
959 std::iter::once(&schema),
960 capacity,
961 reservation,
962 )?
963 .pop()
964 .ok_or_else(|| XlogError::Kernel("resident relation allocation was empty".into()))
965 }
966
967 pub fn prepare_resident_relations_in_reservation<'a>(
972 &self,
973 schemas: impl IntoIterator<Item = &'a Schema>,
974 capacity: u64,
975 reservation: &mut GpuMemoryReservation,
976 ) -> Result<Vec<ResidentRelation>> {
977 let schemas: Vec<_> = schemas.into_iter().collect();
978 if schemas.is_empty() {
979 return Ok(Vec::new());
980 }
981 let layouts = schemas
982 .iter()
983 .map(|schema| resident_relation_layout(schema, capacity))
984 .collect::<Result<Vec<_>>>()?;
985 let arena_bytes = layouts.iter().try_fold(0u64, |bytes, layout| {
986 bytes
987 .checked_add(layout.bytes)
988 .ok_or_else(|| XlogError::Kernel("resident relation arena byte overflow".into()))
989 })?;
990 let arena_bytes = usize::try_from(arena_bytes).map_err(|_| {
991 XlogError::Kernel("resident relation arena exceeds platform usize".into())
992 })?;
993
994 let arena = reservation.alloc::<u8>(arena_bytes)?;
995 let mut cursor = 0usize;
996 let mut relations = Vec::with_capacity(schemas.len());
997 for (schema, layout) in schemas.into_iter().zip(layouts) {
998 let mut columns = Vec::with_capacity(schema.arity());
999 for range in layout.columns {
1000 let start = cursor
1001 .checked_add(
1002 usize::try_from(range.start)
1003 .expect("resident relation layout fits its arena"),
1004 )
1005 .expect("resident relation arena sizing already checked");
1006 let end = cursor
1007 .checked_add(
1008 usize::try_from(range.end)
1009 .expect("resident relation layout fits its arena"),
1010 )
1011 .expect("resident relation arena sizing already checked");
1012 let column = arena.try_owned_subslice::<u8>(start..end).ok_or_else(|| {
1013 XlogError::Kernel("resident column is outside its allocation arena".into())
1014 })?;
1015 columns.push(CudaColumn::Owned(column));
1016 }
1017 let count_start = cursor
1018 .checked_add(
1019 usize::try_from(layout.row_count.start)
1020 .expect("resident relation layout fits its arena"),
1021 )
1022 .expect("resident relation arena sizing already checked");
1023 let count_end = cursor
1024 .checked_add(
1025 usize::try_from(layout.row_count.end)
1026 .expect("resident relation layout fits its arena"),
1027 )
1028 .expect("resident relation arena sizing already checked");
1029 let d_num_rows = arena
1030 .try_owned_subslice::<u32>(count_start..count_end)
1031 .ok_or_else(|| {
1032 XlogError::Kernel("resident row count is outside its allocation arena".into())
1033 })?;
1034 cursor = cursor
1035 .checked_add(
1036 usize::try_from(layout.bytes).expect("resident relation layout fits its arena"),
1037 )
1038 .expect("resident relation arena sizing already checked");
1039 let buffer = CudaBuffer::from_columns(columns, capacity, d_num_rows, schema.clone());
1040 relations.push(ResidentRelation { buffer });
1041 }
1042 debug_assert_eq!(cursor, arena_bytes);
1043 Ok(relations)
1044 }
1045
1046 pub fn record_resident_relation_count_initialize_on_stream(
1049 &self,
1050 relation: &ResidentRelation,
1051 initial_count: u32,
1052 enqueue: &crate::launch::CudaEnqueue<'_>,
1053 ) -> Result<()> {
1054 if initial_count > 1 {
1055 return Err(XlogError::Kernel(format!(
1056 "resident relation initial count {initial_count} is invalid; expected 0 or 1"
1057 )));
1058 }
1059 self.initialize_launch_metadata_u32(
1060 initial_count,
1061 relation.num_rows_device(),
1062 enqueue,
1063 "resident relation count initialization",
1064 )
1065 }
1066
1067 #[cfg(test)]
1068 pub fn initialize_resident_relation_count(
1069 &self,
1070 relation: &mut ResidentRelation,
1071 initial_count: u32,
1072 ) -> Result<()> {
1073 if initial_count > 1 {
1074 return Err(XlogError::Kernel(format!(
1075 "resident relation initial count {initial_count} is invalid; expected 0 or 1"
1076 )));
1077 }
1078 let code = unsafe {
1079 sys::cuMemsetD32_v2(
1080 relation.num_rows_device().device_ptr_value(),
1081 initial_count,
1082 1,
1083 )
1084 };
1085 if code != sys::cudaError_enum::CUDA_SUCCESS {
1086 return Err(XlogError::Kernel(format!(
1087 "resident relation count initialization failed: {code:?}"
1088 )));
1089 }
1090 Ok(())
1091 }
1092
1093 pub fn record_resident_relation_clear_on_stream(
1095 &self,
1096 relation: &ResidentRelation,
1097 stream: &CudaStream,
1098 ) -> Result<()> {
1099 memset_u8_async(
1100 relation.num_rows_device().device_ptr_value(),
1101 0,
1102 std::mem::size_of::<u32>(),
1103 stream,
1104 "relation count clear",
1105 )
1106 }
1107
1108 pub fn prepare_resident_set_workspace(
1109 &self,
1110 candidate_capacity: u64,
1111 ) -> Result<ResidentSetWorkspace> {
1112 self.prepare_resident_set_workspace_with_reservation(candidate_capacity, None)
1113 }
1114
1115 pub fn prepare_resident_set_workspace_in_reservation(
1117 &self,
1118 candidate_capacity: u64,
1119 reservation: &mut GpuMemoryReservation,
1120 ) -> Result<ResidentSetWorkspace> {
1121 self.prepare_resident_set_workspace_with_reservation(candidate_capacity, Some(reservation))
1122 }
1123
1124 fn prepare_resident_set_workspace_with_reservation(
1125 &self,
1126 candidate_capacity: u64,
1127 mut reservation: Option<&mut GpuMemoryReservation>,
1128 ) -> Result<ResidentSetWorkspace> {
1129 let candidate_capacity_u32 = checked_capacity(candidate_capacity, "set candidate")?;
1130 let slots = checked_hash_slot_capacity(candidate_capacity)?;
1131 let slot_storage = match reservation.as_deref_mut() {
1132 Some(reservation) => reservation.alloc::<u64>(slots as usize)?,
1133 None => self.memory().alloc::<u64>(slots as usize)?,
1134 };
1135 let required = match reservation.as_mut() {
1136 Some(reservation) => reservation.alloc::<u64>(1)?,
1137 None => self.memory().alloc::<u64>(1)?,
1138 };
1139 Ok(ResidentSetWorkspace {
1140 slots: slot_storage,
1141 required,
1142 candidate_capacity: candidate_capacity_u32,
1143 slot_mask: (slots - 1) as u32,
1144 })
1145 }
1146
1147 pub fn prepare_resident_join_workspace(
1148 &self,
1149 right_capacity: u64,
1150 ) -> Result<ResidentJoinWorkspace> {
1151 self.prepare_resident_join_workspace_with_reservation(right_capacity, None)
1152 }
1153
1154 pub fn prepare_resident_join_workspace_in_reservation(
1156 &self,
1157 right_capacity: u64,
1158 reservation: &mut GpuMemoryReservation,
1159 ) -> Result<ResidentJoinWorkspace> {
1160 self.prepare_resident_join_workspace_with_reservation(right_capacity, Some(reservation))
1161 }
1162
1163 fn prepare_resident_join_workspace_with_reservation(
1164 &self,
1165 right_capacity: u64,
1166 mut reservation: Option<&mut GpuMemoryReservation>,
1167 ) -> Result<ResidentJoinWorkspace> {
1168 let right_capacity_u32 = checked_capacity(right_capacity, "join right")?;
1169 let buckets = checked_hash_slot_capacity(right_capacity)?;
1170 let bucket_heads = match reservation.as_deref_mut() {
1171 Some(reservation) => reservation.alloc::<u32>(buckets as usize)?,
1172 None => self.memory().alloc::<u32>(buckets as usize)?,
1173 };
1174 let next = match reservation.as_deref_mut() {
1175 Some(reservation) => reservation.alloc::<u32>(right_capacity as usize)?,
1176 None => self.memory().alloc::<u32>(right_capacity as usize)?,
1177 };
1178 let required = match reservation.as_mut() {
1179 Some(reservation) => reservation.alloc::<u64>(1)?,
1180 None => self.memory().alloc::<u64>(1)?,
1181 };
1182 Ok(ResidentJoinWorkspace {
1183 bucket_heads,
1184 next,
1185 required,
1186 right_capacity: right_capacity_u32,
1187 bucket_mask: (buckets - 1) as u32,
1188 })
1189 }
1190
1191 pub fn prepare_resident_convergence_control(&self) -> Result<ResidentConvergenceControl> {
1192 self.prepare_resident_convergence_control_with_reservation(None)
1193 }
1194
1195 pub fn prepare_resident_convergence_control_in_reservation(
1197 &self,
1198 reservation: &mut GpuMemoryReservation,
1199 ) -> Result<ResidentConvergenceControl> {
1200 self.prepare_resident_convergence_control_with_reservation(Some(reservation))
1201 }
1202
1203 fn prepare_resident_convergence_control_with_reservation(
1204 &self,
1205 mut reservation: Option<&mut GpuMemoryReservation>,
1206 ) -> Result<ResidentConvergenceControl> {
1207 let status = match reservation.as_deref_mut() {
1208 Some(reservation) => reservation.alloc::<ResidentTerminalStatus>(1)?,
1209 None => self.memory().alloc::<ResidentTerminalStatus>(1)?,
1210 };
1211 let changed = match reservation.as_deref_mut() {
1212 Some(reservation) => reservation.alloc::<u32>(1)?,
1213 None => self.memory().alloc::<u32>(1)?,
1214 };
1215 let loop_iterations = match reservation.as_mut() {
1216 Some(reservation) => reservation.alloc::<u32>(1)?,
1217 None => self.memory().alloc::<u32>(1)?,
1218 };
1219 Ok(ResidentConvergenceControl {
1220 status,
1221 changed,
1222 loop_iterations,
1223 })
1224 }
1225
1226 pub fn prepare_resident_device_trace(&self) -> Result<ResidentDeviceTrace> {
1228 self.prepare_resident_device_trace_with_reservation(None)
1229 }
1230
1231 pub fn prepare_resident_device_trace_in_reservation(
1233 &self,
1234 reservation: &mut GpuMemoryReservation,
1235 ) -> Result<ResidentDeviceTrace> {
1236 self.prepare_resident_device_trace_with_reservation(Some(reservation))
1237 }
1238
1239 fn prepare_resident_device_trace_with_reservation(
1240 &self,
1241 mut reservation: Option<&mut GpuMemoryReservation>,
1242 ) -> Result<ResidentDeviceTrace> {
1243 let scan_invocations = match reservation.as_deref_mut() {
1244 Some(reservation) => reservation.alloc::<u32>(1)?,
1245 None => self.memory().alloc::<u32>(1)?,
1246 };
1247 let filter_invocations = match reservation.as_deref_mut() {
1248 Some(reservation) => reservation.alloc::<u32>(1)?,
1249 None => self.memory().alloc::<u32>(1)?,
1250 };
1251 let semantic_scan_invocations = match reservation.as_deref_mut() {
1252 Some(reservation) => reservation.alloc::<u32>(1)?,
1253 None => self.memory().alloc::<u32>(1)?,
1254 };
1255 let semantic_filter_invocations = match reservation.as_mut() {
1256 Some(reservation) => reservation.alloc::<u32>(1)?,
1257 None => self.memory().alloc::<u32>(1)?,
1258 };
1259 Ok(ResidentDeviceTrace {
1260 scan_invocations,
1261 filter_invocations,
1262 semantic_scan_invocations,
1263 semantic_filter_invocations,
1264 })
1265 }
1266
1267 pub fn prepare_resident_schema_winners(
1269 &self,
1270 default_schema_ids: &[u32],
1271 ) -> Result<ResidentSchemaWinners> {
1272 self.prepare_resident_schema_winners_with_reservation(default_schema_ids, None)
1273 }
1274
1275 pub fn prepare_resident_schema_winners_in_reservation(
1277 &self,
1278 default_schema_ids: &[u32],
1279 reservation: &mut GpuMemoryReservation,
1280 ) -> Result<ResidentSchemaWinners> {
1281 self.prepare_resident_schema_winners_with_reservation(default_schema_ids, Some(reservation))
1282 }
1283
1284 fn prepare_resident_schema_winners_with_reservation(
1285 &self,
1286 default_schema_ids: &[u32],
1287 mut reservation: Option<&mut GpuMemoryReservation>,
1288 ) -> Result<ResidentSchemaWinners> {
1289 let len = u32::try_from(default_schema_ids.len())
1290 .map_err(|_| XlogError::Kernel("too many resident schema winners".into()))?;
1291 let allocation_len = default_schema_ids.len().max(1);
1292 let seen_nonempty = match reservation.as_deref_mut() {
1293 Some(reservation) => reservation.alloc::<u32>(allocation_len)?,
1294 None => self.memory().alloc::<u32>(allocation_len)?,
1295 };
1296 let mut winner_schema_ids = match reservation.as_mut() {
1297 Some(reservation) => reservation.alloc::<u32>(allocation_len)?,
1298 None => self.memory().alloc::<u32>(allocation_len)?,
1299 };
1300 if !default_schema_ids.is_empty() {
1301 self.device()
1302 .inner()
1303 .htod_sync_copy_into(default_schema_ids, &mut winner_schema_ids)
1304 .map_err(|error| {
1305 XlogError::Kernel(format!("resident schema winner upload failed: {error}"))
1306 })?;
1307 }
1308 Ok(ResidentSchemaWinners {
1309 seen_nonempty,
1310 winner_schema_ids,
1311 default_schema_ids: default_schema_ids.to_vec(),
1312 len,
1313 })
1314 }
1315
1316 fn resident_receipt_pointee<T: cudarc::driver::DeviceRepr>(
1317 &self,
1318 role: ResidentReceiptPointeeRole,
1319 slice: &TrackedCudaSlice<T>,
1320 byte_offset: usize,
1321 ) -> Result<ResidentReceiptPointee> {
1322 let manager_id = slice.memory_manager_ptr_value();
1323 if manager_id != Arc::as_ptr(self.memory()) as usize {
1324 return Err(XlogError::Kernel(
1325 "resident receipt pointee belongs to a foreign memory manager".into(),
1326 ));
1327 }
1328 let provider_context = self.device().inner().stream().context();
1329 let slice_context = DeviceSlice::stream(slice).context();
1330 if !Arc::ptr_eq(slice_context, provider_context)
1331 || slice_context.cu_ctx() != provider_context.cu_ctx()
1332 {
1333 return Err(XlogError::Kernel(
1334 "resident receipt pointee belongs to a foreign CUDA context".into(),
1335 ));
1336 }
1337 let slice_bytes = slice
1338 .len()
1339 .checked_mul(std::mem::size_of::<T>())
1340 .ok_or_else(|| XlogError::Kernel("resident receipt slice byte size overflow".into()))?;
1341 let field_end = byte_offset
1342 .checked_add(4)
1343 .ok_or_else(|| XlogError::Kernel("resident receipt pointee offset overflow".into()))?;
1344 if field_end > slice_bytes {
1345 return Err(XlogError::Kernel(
1346 "resident receipt pointee is outside its source slice".into(),
1347 ));
1348 }
1349 let byte_offset = u64::try_from(byte_offset)
1350 .map_err(|_| XlogError::Kernel("resident receipt pointee offset overflow".into()))?;
1351 let ptr = slice
1352 .device_ptr_value()
1353 .checked_add(byte_offset)
1354 .ok_or_else(|| XlogError::Kernel("resident receipt pointee address overflow".into()))?;
1355 let runtime_block = match slice.runtime_block() {
1356 Some(block) => {
1357 if block.state != crate::device_runtime::BlockState::Live {
1358 return Err(XlogError::Kernel(
1359 "resident receipt pointee runtime block is not live".into(),
1360 ));
1361 }
1362 Some((
1363 crate::device_runtime::BlockId::from_block(block),
1364 block.bytes,
1365 ))
1366 }
1367 None => None,
1368 };
1369 checked_receipt_pointee(role, ptr, manager_id, runtime_block)
1370 }
1371
1372 pub fn prepare_resident_packed_receipt(
1374 &self,
1375 relations: &[&ResidentRelation],
1376 ) -> Result<ResidentPackedReceipt> {
1377 let relation_count_len = u32::try_from(relations.len())
1378 .map_err(|_| XlogError::Kernel("resident receipt has too many count fields".into()))?;
1379 let pointees = relations
1380 .iter()
1381 .enumerate()
1382 .map(|(index, relation)| {
1383 let index = u32::try_from(index).map_err(|_| {
1384 XlogError::Kernel("resident receipt head index overflow".into())
1385 })?;
1386 self.resident_receipt_pointee(
1387 ResidentReceiptPointeeRole::RelationCount(index),
1388 relation.num_rows_device(),
1389 0,
1390 )
1391 })
1392 .collect::<Result<Vec<_>>>()?;
1393 self.prepare_resident_packed_receipt_from_pointees(pointees, relation_count_len, 0, 0, None)
1394 }
1395
1396 pub fn prepare_resident_packed_receipt_with_trace(
1398 &self,
1399 relations: &[&ResidentRelation],
1400 trace: &ResidentDeviceTrace,
1401 ) -> Result<ResidentPackedReceipt> {
1402 self.prepare_resident_packed_receipt_with_trace_and_reservation(relations, trace, None)
1403 }
1404
1405 pub fn prepare_resident_packed_receipt_with_trace_in_reservation(
1407 &self,
1408 relations: &[&ResidentRelation],
1409 trace: &ResidentDeviceTrace,
1410 reservation: &mut GpuMemoryReservation,
1411 ) -> Result<ResidentPackedReceipt> {
1412 self.prepare_resident_packed_receipt_with_trace_and_reservation(
1413 relations,
1414 trace,
1415 Some(reservation),
1416 )
1417 }
1418
1419 fn prepare_resident_packed_receipt_with_trace_and_reservation(
1420 &self,
1421 relations: &[&ResidentRelation],
1422 trace: &ResidentDeviceTrace,
1423 reservation: Option<&mut GpuMemoryReservation>,
1424 ) -> Result<ResidentPackedReceipt> {
1425 let relation_count_len = u32::try_from(relations.len())
1426 .map_err(|_| XlogError::Kernel("resident receipt has too many count fields".into()))?;
1427 let mut pointees = Vec::with_capacity(relations.len().saturating_add(4));
1428 for (index, relation) in relations.iter().enumerate() {
1429 let index = u32::try_from(index)
1430 .map_err(|_| XlogError::Kernel("resident receipt head index overflow".into()))?;
1431 pointees.push(self.resident_receipt_pointee(
1432 ResidentReceiptPointeeRole::RelationCount(index),
1433 relation.num_rows_device(),
1434 0,
1435 )?);
1436 }
1437 pointees.push(self.resident_receipt_pointee(
1438 ResidentReceiptPointeeRole::ScanTrace,
1439 &trace.scan_invocations,
1440 0,
1441 )?);
1442 pointees.push(self.resident_receipt_pointee(
1443 ResidentReceiptPointeeRole::FilterTrace,
1444 &trace.filter_invocations,
1445 0,
1446 )?);
1447 pointees.push(self.resident_receipt_pointee(
1448 ResidentReceiptPointeeRole::SemanticScanTrace,
1449 &trace.semantic_scan_invocations,
1450 0,
1451 )?);
1452 pointees.push(self.resident_receipt_pointee(
1453 ResidentReceiptPointeeRole::SemanticFilterTrace,
1454 &trace.semantic_filter_invocations,
1455 0,
1456 )?);
1457 self.prepare_resident_packed_receipt_from_pointees(
1458 pointees,
1459 relation_count_len,
1460 4,
1461 0,
1462 reservation,
1463 )
1464 }
1465
1466 pub fn prepare_resident_packed_receipt_with_trace_and_schema_winners(
1468 &self,
1469 relations: &[&ResidentRelation],
1470 trace: &ResidentDeviceTrace,
1471 winners: &ResidentSchemaWinners,
1472 ) -> Result<ResidentPackedReceipt> {
1473 self.prepare_resident_packed_receipt_with_trace_and_schema_winners_and_reservation(
1474 relations, trace, winners, None,
1475 )
1476 }
1477
1478 pub fn prepare_resident_packed_receipt_with_trace_and_schema_winners_in_reservation(
1480 &self,
1481 relations: &[&ResidentRelation],
1482 trace: &ResidentDeviceTrace,
1483 winners: &ResidentSchemaWinners,
1484 reservation: &mut GpuMemoryReservation,
1485 ) -> Result<ResidentPackedReceipt> {
1486 self.prepare_resident_packed_receipt_with_trace_and_schema_winners_and_reservation(
1487 relations,
1488 trace,
1489 winners,
1490 Some(reservation),
1491 )
1492 }
1493
1494 fn prepare_resident_packed_receipt_with_trace_and_schema_winners_and_reservation(
1495 &self,
1496 relations: &[&ResidentRelation],
1497 trace: &ResidentDeviceTrace,
1498 winners: &ResidentSchemaWinners,
1499 reservation: Option<&mut GpuMemoryReservation>,
1500 ) -> Result<ResidentPackedReceipt> {
1501 if winners.len() as usize != relations.len() {
1502 return Err(XlogError::Kernel(format!(
1503 "resident schema winner count {} does not match relation count {}",
1504 winners.len(),
1505 relations.len()
1506 )));
1507 }
1508 let relation_count_len = u32::try_from(relations.len())
1509 .map_err(|_| XlogError::Kernel("resident receipt has too many count fields".into()))?;
1510 let mut pointees = Vec::with_capacity(relations.len().saturating_mul(2).saturating_add(4));
1511 for (index, relation) in relations.iter().enumerate() {
1512 let index = u32::try_from(index)
1513 .map_err(|_| XlogError::Kernel("resident receipt head index overflow".into()))?;
1514 pointees.push(self.resident_receipt_pointee(
1515 ResidentReceiptPointeeRole::RelationCount(index),
1516 relation.num_rows_device(),
1517 0,
1518 )?);
1519 }
1520 pointees.push(self.resident_receipt_pointee(
1521 ResidentReceiptPointeeRole::ScanTrace,
1522 &trace.scan_invocations,
1523 0,
1524 )?);
1525 pointees.push(self.resident_receipt_pointee(
1526 ResidentReceiptPointeeRole::FilterTrace,
1527 &trace.filter_invocations,
1528 0,
1529 )?);
1530 pointees.push(self.resident_receipt_pointee(
1531 ResidentReceiptPointeeRole::SemanticScanTrace,
1532 &trace.semantic_scan_invocations,
1533 0,
1534 )?);
1535 pointees.push(self.resident_receipt_pointee(
1536 ResidentReceiptPointeeRole::SemanticFilterTrace,
1537 &trace.semantic_filter_invocations,
1538 0,
1539 )?);
1540 for index in 0..relations.len() {
1541 let index_u32 = u32::try_from(index)
1542 .map_err(|_| XlogError::Kernel("resident receipt head index overflow".into()))?;
1543 let byte_offset = index
1544 .checked_mul(std::mem::size_of::<u32>())
1545 .ok_or_else(|| {
1546 XlogError::Kernel("resident receipt winner offset overflow".into())
1547 })?;
1548 pointees.push(self.resident_receipt_pointee(
1549 ResidentReceiptPointeeRole::SchemaWinner(index_u32),
1550 &winners.winner_schema_ids,
1551 byte_offset,
1552 )?);
1553 }
1554 self.prepare_resident_packed_receipt_from_pointees(
1555 pointees,
1556 relation_count_len,
1557 4,
1558 winners.len(),
1559 reservation,
1560 )
1561 }
1562
1563 fn prepare_resident_packed_receipt_from_pointees(
1564 &self,
1565 pointees: Vec<ResidentReceiptPointee>,
1566 relation_count_len: u32,
1567 device_trace_field_count: u32,
1568 schema_winner_count: u32,
1569 mut reservation: Option<&mut GpuMemoryReservation>,
1570 ) -> Result<ResidentPackedReceipt> {
1571 validate_receipt_pointee_ranges(&pointees)?;
1572 let expected_count_len = relation_count_len
1573 .checked_add(device_trace_field_count)
1574 .and_then(|count| count.checked_add(schema_winner_count))
1575 .ok_or_else(|| XlogError::Kernel("resident receipt field count overflow".into()))?;
1576 if usize::try_from(expected_count_len).ok() != Some(pointees.len()) {
1577 return Err(XlogError::Kernel(
1578 "resident receipt pointee manifest count does not match its shape".into(),
1579 ));
1580 }
1581 let count_ptrs: Vec<u64> = pointees.iter().map(|pointee| pointee.ptr).collect();
1582 let count_len = u32::try_from(count_ptrs.len())
1583 .map_err(|_| XlogError::Kernel("resident receipt has too many count fields".into()))?;
1584 let mut d_count_ptrs = match reservation.as_deref_mut() {
1585 Some(reservation) => reservation.alloc::<u64>(count_ptrs.len().max(1))?,
1586 None => self.memory().alloc::<u64>(count_ptrs.len().max(1))?,
1587 };
1588 if !count_ptrs.is_empty() {
1589 self.device()
1590 .inner()
1591 .htod_sync_copy_into(&count_ptrs, &mut d_count_ptrs)
1592 .map_err(|error| {
1593 XlogError::Kernel(format!("resident receipt pointer upload failed: {error}"))
1594 })?;
1595 }
1596 let u32_fields = count_ptrs
1597 .len()
1598 .checked_add(1)
1599 .ok_or_else(|| XlogError::Kernel("resident receipt field count overflow".into()))?;
1600 let bytes =
1601 std::mem::size_of::<ResidentTerminalStatus>()
1602 .checked_add(u32_fields.checked_mul(4).ok_or_else(|| {
1603 XlogError::Kernel("resident receipt byte size overflow".into())
1604 })?)
1605 .ok_or_else(|| XlogError::Kernel("resident receipt byte size overflow".into()))?;
1606 let bytes = match reservation.as_mut() {
1607 Some(reservation) => reservation.alloc::<u8>(bytes)?,
1608 None => self.memory().alloc::<u8>(bytes)?,
1609 };
1610 Ok(ResidentPackedReceipt {
1611 count_ptrs: d_count_ptrs,
1612 bytes,
1613 pointees,
1614 count_len,
1615 relation_count_len,
1616 device_trace_field_count,
1617 schema_winner_count,
1618 })
1619 }
1620
1621 pub fn prepare_resident_pinned_receipt(
1623 &self,
1624 receipt: &ResidentPackedReceipt,
1625 ) -> Result<ResidentPinnedReceipt> {
1626 Ok(ResidentPinnedReceipt {
1627 buffer: crate::device::PinnedHostBuffer::new(
1628 self.device().inner().stream(),
1629 receipt.len_bytes(),
1630 )
1631 .map_err(|error| {
1632 XlogError::Kernel(format!(
1633 "resident pinned receipt allocation failed: {error}"
1634 ))
1635 })?,
1636 })
1637 }
1638
1639 pub fn observe_resident_packed_receipt(
1644 &self,
1645 receipt: &ResidentPackedReceipt,
1646 pinned: &mut ResidentPinnedReceipt,
1647 stream: &Arc<CudaStream>,
1648 ) -> Result<Vec<u8>> {
1649 if pinned.len_bytes() != receipt.len_bytes() {
1650 return Err(XlogError::Kernel(format!(
1651 "resident pinned receipt size {} does not match device receipt size {}",
1652 pinned.len_bytes(),
1653 receipt.len_bytes()
1654 )));
1655 }
1656 let bytes = pinned
1657 .buffer
1658 .copy_from_device(&receipt.device_bytes().view(), stream)
1659 .map_err(|error| {
1660 XlogError::Kernel(format!("resident final receipt copy failed: {error}"))
1661 })?;
1662 self.record_final_observation_transfer(bytes.len() as u64);
1663 Ok(bytes)
1664 }
1665
1666 pub fn record_resident_receipt_pack_on_stream(
1668 &self,
1669 control: &ResidentConvergenceControl,
1670 receipt: &ResidentPackedReceipt,
1671 stream: &CudaStream,
1672 ) -> Result<()> {
1673 let function = self
1674 .device()
1675 .inner()
1676 .get_func(MODULE, "resident_receipt_pack")
1677 .ok_or_else(|| XlogError::Kernel("resident_receipt_pack kernel missing".into()))?;
1678 let status = control.status.device_ptr_value();
1679 let changed = control.changed.device_ptr_value();
1680 let count_ptrs = receipt.count_ptrs.device_ptr_value();
1681 let output = receipt.bytes.device_ptr_value();
1682 let mut params = vec![
1683 status.as_kernel_param(),
1684 changed.as_kernel_param(),
1685 count_ptrs.as_kernel_param(),
1686 receipt.count_len.as_kernel_param(),
1687 output.as_kernel_param(),
1688 ];
1689 unsafe {
1691 function
1692 .clone()
1693 .launch_on_stream(stream, LaunchConfig::for_num_elems(1), &mut params)
1694 }
1695 .map_err(|error| XlogError::Kernel(format!("resident receipt pack launch: {error}")))
1696 }
1697
1698 pub fn record_resident_schema_winners_initialize_on_stream(
1700 &self,
1701 winners: &ResidentSchemaWinners,
1702 receipt: &ResidentPackedReceipt,
1703 stream: &CudaStream,
1704 ) -> Result<()> {
1705 if receipt.relation_count_len != winners.len {
1706 return Err(XlogError::Kernel(format!(
1707 "resident schema winner count {} does not match receipt relation count {}",
1708 winners.len, receipt.relation_count_len
1709 )));
1710 }
1711 if winners.len == 0 {
1712 return Ok(());
1713 }
1714 let function = self
1715 .device()
1716 .inner()
1717 .get_func(MODULE, "resident_schema_winners_initialize")
1718 .ok_or_else(|| {
1719 XlogError::Kernel("resident_schema_winners_initialize kernel missing".into())
1720 })?;
1721 let count_ptrs = receipt.count_ptrs.device_ptr_value();
1722 let seen_nonempty = winners.seen_nonempty.device_ptr_value();
1723 let mut params = vec![
1724 count_ptrs.as_kernel_param(),
1725 winners.len.as_kernel_param(),
1726 seen_nonempty.as_kernel_param(),
1727 ];
1728 unsafe {
1730 function.clone().launch_on_stream(
1731 stream,
1732 LaunchConfig::for_num_elems(winners.len),
1733 &mut params,
1734 )
1735 }
1736 .map_err(|error| XlogError::Kernel(format!("resident schema winner init launch: {error}")))
1737 }
1738
1739 pub fn record_resident_schema_winner_mark_on_stream(
1741 &self,
1742 contribution_count: &TrackedCudaSlice<u32>,
1743 winners: &ResidentSchemaWinners,
1744 head_index: u32,
1745 schema_id: u32,
1746 stream: &CudaStream,
1747 ) -> Result<()> {
1748 if head_index >= winners.len {
1749 return Err(XlogError::Kernel(format!(
1750 "resident schema winner index {head_index} exceeds {} heads",
1751 winners.len
1752 )));
1753 }
1754 let function = self
1755 .device()
1756 .inner()
1757 .get_func(MODULE, "resident_schema_winner_mark")
1758 .ok_or_else(|| {
1759 XlogError::Kernel("resident_schema_winner_mark kernel missing".into())
1760 })?;
1761 let contribution_count = contribution_count.device_ptr_value();
1762 let seen_nonempty = winners.seen_nonempty.device_ptr_value();
1763 let winner_schema_ids = winners.winner_schema_ids.device_ptr_value();
1764 let mut params = vec![
1765 contribution_count.as_kernel_param(),
1766 seen_nonempty.as_kernel_param(),
1767 winner_schema_ids.as_kernel_param(),
1768 head_index.as_kernel_param(),
1769 schema_id.as_kernel_param(),
1770 ];
1771 unsafe {
1773 function
1774 .clone()
1775 .launch_on_stream(stream, LaunchConfig::for_num_elems(1), &mut params)
1776 }
1777 .map_err(|error| XlogError::Kernel(format!("resident schema winner mark launch: {error}")))
1778 }
1779
1780 pub fn record_resident_device_trace_initialize_on_stream(
1782 &self,
1783 trace: &ResidentDeviceTrace,
1784 stream: &CudaStream,
1785 ) -> Result<()> {
1786 let function = self
1787 .device()
1788 .inner()
1789 .get_func(MODULE, "resident_trace_initialize")
1790 .ok_or_else(|| XlogError::Kernel("resident_trace_initialize kernel missing".into()))?;
1791 let scan_invocations = trace.scan_invocations.device_ptr_value();
1792 let filter_invocations = trace.filter_invocations.device_ptr_value();
1793 let semantic_scan_invocations = trace.semantic_scan_invocations.device_ptr_value();
1794 let semantic_filter_invocations = trace.semantic_filter_invocations.device_ptr_value();
1795 let mut params = vec![
1796 scan_invocations.as_kernel_param(),
1797 filter_invocations.as_kernel_param(),
1798 semantic_scan_invocations.as_kernel_param(),
1799 semantic_filter_invocations.as_kernel_param(),
1800 ];
1801 unsafe {
1803 function
1804 .clone()
1805 .launch_on_stream(stream, LaunchConfig::for_num_elems(1), &mut params)
1806 }
1807 .map_err(|error| XlogError::Kernel(format!("resident trace init launch: {error}")))
1808 }
1809
1810 pub fn record_resident_scan_trace_on_stream(
1812 &self,
1813 trace: &ResidentDeviceTrace,
1814 stream: &CudaStream,
1815 ) -> Result<()> {
1816 self.record_resident_trace_increment_on_stream(
1817 trace.scan_invocations.device_ptr_value(),
1818 trace.semantic_scan_invocations.device_ptr_value(),
1819 "scan",
1820 stream,
1821 )
1822 }
1823
1824 pub fn record_resident_filter_trace_on_stream(
1826 &self,
1827 trace: &ResidentDeviceTrace,
1828 stream: &CudaStream,
1829 ) -> Result<()> {
1830 self.record_resident_trace_increment_on_stream(
1831 trace.filter_invocations.device_ptr_value(),
1832 trace.semantic_filter_invocations.device_ptr_value(),
1833 "filter",
1834 stream,
1835 )
1836 }
1837
1838 fn record_resident_trace_increment_on_stream(
1839 &self,
1840 counter: u64,
1841 semantic_counter: u64,
1842 label: &str,
1843 stream: &CudaStream,
1844 ) -> Result<()> {
1845 let function = self
1846 .device()
1847 .inner()
1848 .get_func(MODULE, "resident_trace_increment")
1849 .ok_or_else(|| XlogError::Kernel("resident_trace_increment kernel missing".into()))?;
1850 let mut params = vec![
1851 counter.as_kernel_param(),
1852 semantic_counter.as_kernel_param(),
1853 ];
1854 unsafe {
1856 function
1857 .clone()
1858 .launch_on_stream(stream, LaunchConfig::for_num_elems(1), &mut params)
1859 }
1860 .map_err(|error| {
1861 XlogError::Kernel(format!("resident {label} trace increment launch: {error}"))
1862 })
1863 }
1864
1865 pub fn record_resident_control_initialize_on_stream(
1867 &self,
1868 control: &ResidentConvergenceControl,
1869 stream: &CudaStream,
1870 ) -> Result<()> {
1871 let function = self
1872 .device()
1873 .inner()
1874 .get_func(MODULE, "resident_control_initialize")
1875 .ok_or_else(|| {
1876 XlogError::Kernel("resident_control_initialize kernel missing".into())
1877 })?;
1878 let status = control.status.device_ptr_value();
1879 let changed = control.changed.device_ptr_value();
1880 let loop_iterations = control.loop_iterations.device_ptr_value();
1881 let mut params = vec![
1882 status.as_kernel_param(),
1883 changed.as_kernel_param(),
1884 loop_iterations.as_kernel_param(),
1885 ];
1886 unsafe {
1888 function
1889 .clone()
1890 .launch_on_stream(stream, LaunchConfig::for_num_elems(1), &mut params)
1891 }
1892 .map_err(|error| XlogError::Kernel(format!("resident control init launch: {error}")))
1893 }
1894
1895 pub fn record_resident_scc_begin_on_stream(
1898 &self,
1899 iteration_limit: u32,
1900 op_id: u32,
1901 control: &ResidentConvergenceControl,
1902 stream: &CudaStream,
1903 ) -> Result<()> {
1904 let function = self
1905 .device()
1906 .inner()
1907 .get_func(MODULE, "resident_scc_begin")
1908 .ok_or_else(|| XlogError::Kernel("resident_scc_begin kernel missing".into()))?;
1909 let status = control.status.device_ptr_value();
1910 let changed = control.changed.device_ptr_value();
1911 let loop_iterations = control.loop_iterations.device_ptr_value();
1912 let mut params = vec![
1913 iteration_limit.as_kernel_param(),
1914 op_id.as_kernel_param(),
1915 status.as_kernel_param(),
1916 changed.as_kernel_param(),
1917 loop_iterations.as_kernel_param(),
1918 ];
1919 unsafe {
1921 function
1922 .clone()
1923 .launch_on_stream(stream, LaunchConfig::for_num_elems(1), &mut params)
1924 }
1925 .map_err(|error| XlogError::Kernel(format!("resident SCC begin launch: {error}")))
1926 }
1927
1928 pub fn record_resident_dedup_on_stream(
1929 &self,
1930 input: &CudaBuffer,
1931 output: &ResidentRelation,
1932 workspace: &ResidentSetWorkspace,
1933 control: &ResidentConvergenceControl,
1934 op_id: u32,
1935 stream: &CudaStream,
1936 ) -> Result<()> {
1937 ensure_same_schema(input, output.buffer())?;
1938 self.record_resident_set_on_stream(
1939 input, input, output, workspace, control, op_id, stream, 0,
1940 )
1941 }
1942
1943 #[expect(
1944 clippy::too_many_arguments,
1945 reason = "resident union recording keeps both inputs, output, row widths, capacity, and selected stream explicit"
1946 )]
1947 pub fn record_resident_union_on_stream(
1948 &self,
1949 left: &CudaBuffer,
1950 right: &CudaBuffer,
1951 output: &ResidentRelation,
1952 workspace: &ResidentSetWorkspace,
1953 control: &ResidentConvergenceControl,
1954 op_id: u32,
1955 stream: &CudaStream,
1956 ) -> Result<()> {
1957 ensure_same_schema(left, right)?;
1958 ensure_same_schema(left, output.buffer())?;
1959 self.record_resident_set_on_stream(
1960 left, right, output, workspace, control, op_id, stream, 1,
1961 )
1962 }
1963
1964 #[expect(
1965 clippy::too_many_arguments,
1966 reason = "resident difference recording keeps both inputs, output, row widths, capacity, and selected stream explicit"
1967 )]
1968 pub fn record_resident_diff_on_stream(
1969 &self,
1970 left: &CudaBuffer,
1971 right: &CudaBuffer,
1972 output: &ResidentRelation,
1973 workspace: &ResidentSetWorkspace,
1974 control: &ResidentConvergenceControl,
1975 op_id: u32,
1976 stream: &CudaStream,
1977 ) -> Result<()> {
1978 ensure_same_schema(left, right)?;
1979 ensure_same_schema(left, output.buffer())?;
1980 self.record_resident_set_on_stream(
1981 left, right, output, workspace, control, op_id, stream, 2,
1982 )
1983 }
1984
1985 #[expect(
1986 clippy::too_many_arguments,
1987 reason = "the shared resident set-operation launch core mirrors the fixed kernel ABI and stream binding"
1988 )]
1989 fn record_resident_set_on_stream(
1990 &self,
1991 left: &CudaBuffer,
1992 right: &CudaBuffer,
1993 output: &ResidentRelation,
1994 workspace: &ResidentSetWorkspace,
1995 control: &ResidentConvergenceControl,
1996 op_id: u32,
1997 stream: &CudaStream,
1998 mode: u32,
1999 ) -> Result<()> {
2000 let total = left
2001 .num_rows()
2002 .checked_add(if mode == 0 { 0 } else { right.num_rows() })
2003 .ok_or_else(|| XlogError::Kernel("resident set capacity overflow".into()))?;
2004 if total > u64::from(workspace.candidate_capacity) {
2005 return Err(XlogError::Kernel(format!(
2006 "resident set workspace capacity {} is below required candidate capacity {total}",
2007 workspace.candidate_capacity
2008 )));
2009 }
2010 memset_u8_async(
2011 workspace.slots.device_ptr_value(),
2012 0,
2013 workspace.slots.len() * 8,
2014 stream,
2015 "set slots",
2016 )?;
2017 memset_u8_async(
2018 workspace.required.device_ptr_value(),
2019 0,
2020 8,
2021 stream,
2022 "set required",
2023 )?;
2024 memset_u8_async(
2025 output.num_rows_device().device_ptr_value(),
2026 0,
2027 4,
2028 stream,
2029 "set output count",
2030 )?;
2031 let left_view = relation_view(left)?;
2032 let right_view = relation_view(right)?;
2033 let output_view = relation_view(output.buffer())?;
2034 let insert = self
2035 .device()
2036 .inner()
2037 .get_func(MODULE, "resident_set_insert")
2038 .ok_or_else(|| XlogError::Kernel("resident_set_insert kernel missing".into()))?;
2039 let slots = workspace.slots.device_ptr_value();
2040 let required = workspace.required.device_ptr_value();
2041 let status = control.status.device_ptr_value();
2042 let launch_insert = |candidate: ResidentRelationView,
2043 source_tag: u32,
2044 emit_rows: u32,
2045 materialize: u32|
2046 -> Result<()> {
2047 let mut params = vec![
2048 candidate.as_kernel_param(),
2049 left_view.as_kernel_param(),
2050 right_view.as_kernel_param(),
2051 source_tag.as_kernel_param(),
2052 emit_rows.as_kernel_param(),
2053 materialize.as_kernel_param(),
2054 slots.as_kernel_param(),
2055 workspace.slot_mask.as_kernel_param(),
2056 output_view.as_kernel_param(),
2057 required.as_kernel_param(),
2058 status.as_kernel_param(),
2059 op_id.as_kernel_param(),
2060 ];
2061 unsafe {
2063 insert.clone().launch_on_stream(
2064 stream,
2065 launch_config(candidate.capacity),
2066 &mut params,
2067 )
2068 }
2069 .map_err(|error| XlogError::Kernel(format!("resident set insert launch: {error}")))
2070 };
2071 let launch_pass = |materialize: u32| -> Result<()> {
2072 match mode {
2073 0 => launch_insert(left_view, 0, 1, materialize)?,
2074 1 => {
2075 launch_insert(left_view, 0, 1, materialize)?;
2076 launch_insert(right_view, 1, 1, materialize)?;
2077 }
2078 2 => {
2079 launch_insert(right_view, 1, 0, materialize)?;
2080 launch_insert(left_view, 0, 1, materialize)?;
2081 }
2082 _ => unreachable!(),
2083 }
2084 Ok(())
2085 };
2086 launch_pass(0)?;
2087 self.record_resident_finalize_on_stream(
2088 "resident_set_finalize",
2089 required,
2090 output_view,
2091 status,
2092 op_id,
2093 stream,
2094 )?;
2095 memset_u8_async(
2096 workspace.slots.device_ptr_value(),
2097 0,
2098 workspace.slots.len() * 8,
2099 stream,
2100 "set materialization slots",
2101 )?;
2102 memset_u8_async(
2103 workspace.required.device_ptr_value(),
2104 0,
2105 8,
2106 stream,
2107 "set materialization required",
2108 )?;
2109 launch_pass(1)
2110 }
2111
2112 #[expect(
2113 clippy::too_many_arguments,
2114 reason = "resident join recording keeps both relation views, key sets, output, scratch, capacity, and stream explicit"
2115 )]
2116 pub fn record_resident_join_on_stream(
2117 &self,
2118 kind: ResidentJoinKind,
2119 left: &CudaBuffer,
2120 left_key: usize,
2121 right: &CudaBuffer,
2122 right_key: usize,
2123 output: &ResidentRelation,
2124 workspace: &ResidentJoinWorkspace,
2125 control: &ResidentConvergenceControl,
2126 op_id: u32,
2127 stream: &CudaStream,
2128 ) -> Result<()> {
2129 let left_view = relation_view(left)?;
2130 let right_view = relation_view(right)?;
2131 if right_view.capacity > workspace.right_capacity {
2132 return Err(XlogError::Kernel(format!(
2133 "resident join workspace right capacity {} is below relation capacity {}",
2134 workspace.right_capacity, right_view.capacity
2135 )));
2136 }
2137 if left_key >= left.arity() || right_key >= right.arity() {
2138 return Err(XlogError::Kernel(
2139 "resident join key index out of bounds".into(),
2140 ));
2141 }
2142 if width(left.schema().column_type(left_key).expect("key checked"))?
2143 != width(right.schema().column_type(right_key).expect("key checked"))?
2144 {
2145 return Err(XlogError::Kernel(
2146 "resident join key widths must match".into(),
2147 ));
2148 }
2149 let expected_schema = match kind {
2150 ResidentJoinKind::Semi => left.schema().clone(),
2151 ResidentJoinKind::Inner => {
2152 let mut columns = left.schema().columns.clone();
2153 columns.extend(right.schema().columns.iter().cloned());
2154 Schema::new(columns)
2155 }
2156 };
2157 if !same_physical_layout(output.buffer().schema(), &expected_schema) {
2158 return Err(XlogError::Kernel(
2159 "resident join output schema does not match join kind".into(),
2160 ));
2161 }
2162 let output_view = relation_view(output.buffer())?;
2163 memset_u8_async(
2164 workspace.bucket_heads.device_ptr_value(),
2165 0xff,
2166 workspace.bucket_heads.len() * 4,
2167 stream,
2168 "join buckets",
2169 )?;
2170 memset_u8_async(
2171 workspace.required.device_ptr_value(),
2172 0,
2173 8,
2174 stream,
2175 "join required",
2176 )?;
2177 memset_u8_async(
2178 output.num_rows_device().device_ptr_value(),
2179 0,
2180 4,
2181 stream,
2182 "join output count",
2183 )?;
2184 let buckets = workspace.bucket_heads.device_ptr_value();
2185 let next = workspace.next.device_ptr_value();
2186 let required = workspace.required.device_ptr_value();
2187 let status = control.status.device_ptr_value();
2188 let left_key = left_key as u32;
2189 let right_key = right_key as u32;
2190 let build = self
2191 .device()
2192 .inner()
2193 .get_func(MODULE, "resident_join_build")
2194 .ok_or_else(|| XlogError::Kernel("resident_join_build kernel missing".into()))?;
2195 let mut build_params = vec![
2196 right_view.as_kernel_param(),
2197 right_key.as_kernel_param(),
2198 buckets.as_kernel_param(),
2199 workspace.bucket_mask.as_kernel_param(),
2200 next.as_kernel_param(),
2201 status.as_kernel_param(),
2202 op_id.as_kernel_param(),
2203 ];
2204 unsafe {
2206 build.clone().launch_on_stream(
2207 stream,
2208 launch_config(right_view.capacity),
2209 &mut build_params,
2210 )
2211 }
2212 .map_err(|error| XlogError::Kernel(format!("resident join build launch: {error}")))?;
2213 let probe_name = match kind {
2214 ResidentJoinKind::Inner => "resident_join_probe_inner",
2215 ResidentJoinKind::Semi => "resident_join_probe_semi",
2216 };
2217 let probe = self
2218 .device()
2219 .inner()
2220 .get_func(MODULE, probe_name)
2221 .ok_or_else(|| XlogError::Kernel(format!("{probe_name} kernel missing")))?;
2222 let launch_probe = |materialize: u32| -> Result<()> {
2223 let mut probe_params = vec![
2224 left_view.as_kernel_param(),
2225 left_key.as_kernel_param(),
2226 right_view.as_kernel_param(),
2227 right_key.as_kernel_param(),
2228 buckets.as_kernel_param(),
2229 workspace.bucket_mask.as_kernel_param(),
2230 next.as_kernel_param(),
2231 output_view.as_kernel_param(),
2232 required.as_kernel_param(),
2233 materialize.as_kernel_param(),
2234 status.as_kernel_param(),
2235 op_id.as_kernel_param(),
2236 ];
2237 unsafe {
2239 probe.clone().launch_on_stream(
2240 stream,
2241 launch_config(left_view.capacity),
2242 &mut probe_params,
2243 )
2244 }
2245 .map_err(|error| XlogError::Kernel(format!("resident join probe launch: {error}")))
2246 };
2247 launch_probe(0)?;
2248 self.record_resident_finalize_on_stream(
2249 "resident_join_finalize",
2250 required,
2251 output_view,
2252 status,
2253 op_id,
2254 stream,
2255 )?;
2256 memset_u8_async(
2257 workspace.required.device_ptr_value(),
2258 0,
2259 8,
2260 stream,
2261 "join materialization required",
2262 )?;
2263 launch_probe(1)
2264 }
2265
2266 fn record_resident_finalize_on_stream(
2267 &self,
2268 kernel: &str,
2269 required: u64,
2270 output: ResidentRelationView,
2271 status: u64,
2272 op_id: u32,
2273 stream: &CudaStream,
2274 ) -> Result<()> {
2275 let function = self
2276 .device()
2277 .inner()
2278 .get_func(MODULE, kernel)
2279 .ok_or_else(|| XlogError::Kernel(format!("{kernel} kernel missing")))?;
2280 let mut params = vec![
2281 required.as_kernel_param(),
2282 output.as_kernel_param(),
2283 status.as_kernel_param(),
2284 op_id.as_kernel_param(),
2285 ];
2286 unsafe {
2288 function
2289 .clone()
2290 .launch_on_stream(stream, LaunchConfig::for_num_elems(1), &mut params)
2291 }
2292 .map_err(|error| XlogError::Kernel(format!("resident finalize launch: {error}")))
2293 }
2294
2295 pub fn record_resident_changed_reset_on_stream(
2297 &self,
2298 control: &ResidentConvergenceControl,
2299 stream: &CudaStream,
2300 ) -> Result<()> {
2301 let function = self
2302 .device()
2303 .inner()
2304 .get_func(MODULE, "resident_changed_reset")
2305 .ok_or_else(|| XlogError::Kernel("resident_changed_reset kernel missing".into()))?;
2306 let changed = control.changed.device_ptr_value();
2307 let mut params = vec![changed.as_kernel_param()];
2308 unsafe {
2310 function
2311 .clone()
2312 .launch_on_stream(stream, LaunchConfig::for_num_elems(1), &mut params)
2313 }
2314 .map_err(|error| XlogError::Kernel(format!("resident changed reset launch: {error}")))
2315 }
2316
2317 pub fn record_resident_changed_mark_on_stream(
2319 &self,
2320 novel_count: &TrackedCudaSlice<u32>,
2321 control: &ResidentConvergenceControl,
2322 stream: &CudaStream,
2323 ) -> Result<()> {
2324 if novel_count.is_empty() {
2325 return Err(XlogError::Kernel(
2326 "resident changed mark requires one device count".into(),
2327 ));
2328 }
2329 let function = self
2330 .device()
2331 .inner()
2332 .get_func(MODULE, "resident_changed_mark")
2333 .ok_or_else(|| XlogError::Kernel("resident_changed_mark kernel missing".into()))?;
2334 let novel = novel_count.device_ptr_value();
2335 let changed = control.changed.device_ptr_value();
2336 let mut params = vec![novel.as_kernel_param(), changed.as_kernel_param()];
2337 unsafe {
2339 function
2340 .clone()
2341 .launch_on_stream(stream, LaunchConfig::for_num_elems(1), &mut params)
2342 }
2343 .map_err(|error| XlogError::Kernel(format!("resident changed mark launch: {error}")))
2344 }
2345
2346 pub fn record_resident_convergence_on_stream(
2348 &self,
2349 conditional_handle: u64,
2350 iteration_limit: u32,
2351 op_id: u32,
2352 control: &ResidentConvergenceControl,
2353 stream: &CudaStream,
2354 ) -> Result<()> {
2355 if iteration_limit == 0 {
2356 return Err(XlogError::Kernel(
2357 "resident convergence requires a positive iteration limit".into(),
2358 ));
2359 }
2360 let function = self
2361 .device()
2362 .inner()
2363 .get_func(MODULE, "resident_convergence")
2364 .ok_or_else(|| XlogError::Kernel("resident_convergence kernel missing".into()))?;
2365 let status = control.status.device_ptr_value();
2366 let changed = control.changed.device_ptr_value();
2367 let loop_iterations = control.loop_iterations.device_ptr_value();
2368 let mut params = vec![
2369 conditional_handle.as_kernel_param(),
2370 iteration_limit.as_kernel_param(),
2371 op_id.as_kernel_param(),
2372 status.as_kernel_param(),
2373 changed.as_kernel_param(),
2374 loop_iterations.as_kernel_param(),
2375 ];
2376 unsafe {
2378 function
2379 .clone()
2380 .launch_on_stream(stream, LaunchConfig::for_num_elems(1), &mut params)
2381 }
2382 .map_err(|error| XlogError::Kernel(format!("resident convergence launch: {error}")))
2383 }
2384
2385 pub fn record_resident_terminal_success_on_stream(
2387 &self,
2388 op_id: u32,
2389 control: &ResidentConvergenceControl,
2390 stream: &CudaStream,
2391 ) -> Result<()> {
2392 let function = self
2393 .device()
2394 .inner()
2395 .get_func(MODULE, "resident_terminal_success")
2396 .ok_or_else(|| XlogError::Kernel("resident_terminal_success kernel missing".into()))?;
2397 let status = control.status.device_ptr_value();
2398 let mut params = vec![op_id.as_kernel_param(), status.as_kernel_param()];
2399 unsafe {
2401 function
2402 .clone()
2403 .launch_on_stream(stream, LaunchConfig::for_num_elems(1), &mut params)
2404 }
2405 .map_err(|error| XlogError::Kernel(format!("resident terminal success launch: {error}")))
2406 }
2407
2408 #[doc(hidden)]
2413 pub fn record_resident_test_status_on_stream(
2414 &self,
2415 control: &ResidentConvergenceControl,
2416 injected: ResidentTerminalStatus,
2417 stream: &CudaStream,
2418 ) -> Result<()> {
2419 if !matches!(
2420 injected.code,
2421 code if code == ResidentTerminalCode::Success as u32
2422 || code == ResidentTerminalCode::IterationLimit as u32
2423 || code == ResidentTerminalCode::CapacityOverflow as u32
2424 || code == ResidentTerminalCode::ResourceExhausted as u32
2425 ) {
2426 return Err(XlogError::Kernel(format!(
2427 "resident test status code {} is not terminal",
2428 injected.code
2429 )));
2430 }
2431 let function = self
2432 .device()
2433 .inner()
2434 .get_func(MODULE, "resident_test_status")
2435 .ok_or_else(|| XlogError::Kernel("resident_test_status kernel missing".into()))?;
2436 let code = injected.code;
2437 let op_id = injected.op_id;
2438 let resource_code = injected.resource_code;
2439 let iterations = injected.iterations;
2440 let limit = injected.limit;
2441 let required = injected.required;
2442 let capacity = injected.capacity;
2443 let status = control.status.device_ptr_value();
2444 let mut params = vec![
2445 code.as_kernel_param(),
2446 op_id.as_kernel_param(),
2447 resource_code.as_kernel_param(),
2448 iterations.as_kernel_param(),
2449 limit.as_kernel_param(),
2450 required.as_kernel_param(),
2451 capacity.as_kernel_param(),
2452 status.as_kernel_param(),
2453 ];
2454 unsafe {
2456 function
2457 .clone()
2458 .launch_on_stream(stream, LaunchConfig::for_num_elems(1), &mut params)
2459 }
2460 .map_err(|error| XlogError::Kernel(format!("resident test status launch: {error}")))
2461 }
2462}
2463
2464#[cfg(test)]
2465mod tests {
2466
2467 use xlog_core::{MemoryBudget, ScalarType, Schema};
2468
2469 use crate::cuda_graph::CapturedCudaGraph;
2470
2471 use super::{
2472 checked_hash_slot_capacity, checked_receipt_pointee, record_receipt_pointee_uses,
2473 resident_device_trace_bytes, resident_packed_receipt_with_schema_winners_device_bytes,
2474 resident_schema_winners_device_bytes, validate_receipt_pointee_owners,
2475 validate_receipt_pointee_ranges, validate_receipt_schedule_block_mapping,
2476 validate_receipt_schedule_mapping, ResidentJoinKind, ResidentPackedReceipt,
2477 ResidentReceiptPointee, ResidentReceiptPointeeRole, ResidentResourceCode,
2478 ResidentTerminalCode, ResidentTerminalStatus,
2479 };
2480
2481 #[test]
2482 fn dual_device_trace_and_packed_receipt_have_exact_bytes() {
2483 assert_eq!(resident_device_trace_bytes(), 16);
2484 assert_eq!(
2485 resident_packed_receipt_with_schema_winners_device_bytes(3).unwrap(),
2486 164
2487 );
2488 }
2489
2490 #[test]
2491 fn schema_winners_retain_immutable_host_defaults_for_replay() {
2492 let _: fn(&super::ResidentSchemaWinners) -> &[u32] =
2493 super::ResidentSchemaWinners::default_schema_ids;
2494 }
2495
2496 #[test]
2497 fn receipt_pointee_manifest_rejects_overlap_and_checked_range_overflow() {
2498 let block = crate::device_runtime::BlockId {
2499 ptr: 0x1000,
2500 generation: crate::device_runtime::Generation(3),
2501 alloc_stream: crate::device_runtime::StreamId(4),
2502 device_ordinal: 0,
2503 };
2504 let relation = checked_receipt_pointee(
2505 ResidentReceiptPointeeRole::RelationCount(0),
2506 0x1000,
2507 7,
2508 Some((block, 16)),
2509 )
2510 .expect("valid relation count");
2511 let overlapping = checked_receipt_pointee(
2512 ResidentReceiptPointeeRole::ScanTrace,
2513 0x1002,
2514 7,
2515 Some((block, 16)),
2516 )
2517 .expect("individually valid overlapping counter");
2518 assert!(validate_receipt_pointee_ranges(&[relation, overlapping]).is_err());
2519
2520 assert!(checked_receipt_pointee(
2521 ResidentReceiptPointeeRole::FilterTrace,
2522 u64::MAX - 1,
2523 7,
2524 None,
2525 )
2526 .is_err());
2527 assert!(checked_receipt_pointee(
2528 ResidentReceiptPointeeRole::FilterTrace,
2529 0x100e,
2530 7,
2531 Some((block, 16)),
2532 )
2533 .is_err());
2534 }
2535
2536 #[test]
2537 fn receipt_manifest_mapping_requires_exact_role_and_pointer_order() {
2538 let zero_heads = [
2539 checked_receipt_pointee(ResidentReceiptPointeeRole::ScanTrace, 0x800, 7, None).unwrap(),
2540 checked_receipt_pointee(ResidentReceiptPointeeRole::FilterTrace, 0x810, 7, None)
2541 .unwrap(),
2542 checked_receipt_pointee(
2543 ResidentReceiptPointeeRole::SemanticScanTrace,
2544 0x820,
2545 7,
2546 None,
2547 )
2548 .unwrap(),
2549 checked_receipt_pointee(
2550 ResidentReceiptPointeeRole::SemanticFilterTrace,
2551 0x830,
2552 7,
2553 None,
2554 )
2555 .unwrap(),
2556 ];
2557 validate_receipt_schedule_mapping(&zero_heads, &[], [0x800, 0x810, 0x820, 0x830], &[])
2558 .expect("zero-head receipt mapping");
2559
2560 let one_head = [
2561 checked_receipt_pointee(ResidentReceiptPointeeRole::RelationCount(0), 0x900, 7, None)
2562 .unwrap(),
2563 checked_receipt_pointee(ResidentReceiptPointeeRole::ScanTrace, 0x910, 7, None).unwrap(),
2564 checked_receipt_pointee(ResidentReceiptPointeeRole::FilterTrace, 0x920, 7, None)
2565 .unwrap(),
2566 checked_receipt_pointee(
2567 ResidentReceiptPointeeRole::SemanticScanTrace,
2568 0x930,
2569 7,
2570 None,
2571 )
2572 .unwrap(),
2573 checked_receipt_pointee(
2574 ResidentReceiptPointeeRole::SemanticFilterTrace,
2575 0x940,
2576 7,
2577 None,
2578 )
2579 .unwrap(),
2580 checked_receipt_pointee(ResidentReceiptPointeeRole::SchemaWinner(0), 0x950, 7, None)
2581 .unwrap(),
2582 ];
2583 validate_receipt_schedule_mapping(
2584 &one_head,
2585 &[0x900],
2586 [0x910, 0x920, 0x930, 0x940],
2587 &[0x950],
2588 )
2589 .expect("one-head receipt mapping");
2590
2591 let entries = [
2592 (ResidentReceiptPointeeRole::RelationCount(0), 0x1000),
2593 (ResidentReceiptPointeeRole::RelationCount(1), 0x1010),
2594 (ResidentReceiptPointeeRole::ScanTrace, 0x1020),
2595 (ResidentReceiptPointeeRole::FilterTrace, 0x1030),
2596 (ResidentReceiptPointeeRole::SemanticScanTrace, 0x1040),
2597 (ResidentReceiptPointeeRole::SemanticFilterTrace, 0x1050),
2598 (ResidentReceiptPointeeRole::SchemaWinner(0), 0x1060),
2599 (ResidentReceiptPointeeRole::SchemaWinner(1), 0x1070),
2600 ]
2601 .map(|(role, ptr)| checked_receipt_pointee(role, ptr, 7, None).unwrap());
2602
2603 validate_receipt_schedule_mapping(
2604 &entries,
2605 &[0x1000, 0x1010],
2606 [0x1020, 0x1030, 0x1040, 0x1050],
2607 &[0x1060, 0x1070],
2608 )
2609 .expect("exact receipt mapping");
2610 assert!(validate_receipt_schedule_mapping(
2611 &entries,
2612 &[0x1010, 0x1000],
2613 [0x1020, 0x1030, 0x1040, 0x1050],
2614 &[0x1060, 0x1070],
2615 )
2616 .is_err());
2617 assert!(validate_receipt_schedule_mapping(
2618 &entries,
2619 &[0x1000, 0x1010],
2620 [0x1020, 0x1030, 0x1040, 0x1050],
2621 &[0x1060, 0x9999],
2622 )
2623 .is_err());
2624 }
2625
2626 #[test]
2627 fn additive_receipt_manifest_requires_exact_manager_and_runtime_block_owner() {
2628 let block = crate::device_runtime::BlockId {
2629 ptr: 0x2000,
2630 generation: crate::device_runtime::Generation(1),
2631 alloc_stream: crate::device_runtime::StreamId(2),
2632 device_ordinal: 3,
2633 };
2634 let owned = checked_receipt_pointee(
2635 ResidentReceiptPointeeRole::RelationCount(0),
2636 0x2000,
2637 17,
2638 Some((block, 8)),
2639 )
2640 .unwrap();
2641 validate_receipt_pointee_owners(&[owned], 17, 3).expect("exact owner");
2642
2643 let foreign_manager = checked_receipt_pointee(
2644 ResidentReceiptPointeeRole::RelationCount(0),
2645 0x2000,
2646 99,
2647 Some((block, 8)),
2648 )
2649 .unwrap();
2650 assert!(validate_receipt_pointee_owners(&[foreign_manager], 17, 3).is_err());
2651 let untracked = checked_receipt_pointee(
2652 ResidentReceiptPointeeRole::RelationCount(0),
2653 0x2000,
2654 17,
2655 None,
2656 )
2657 .unwrap();
2658 assert!(validate_receipt_pointee_owners(&[untracked], 17, 3).is_err());
2659 assert!(validate_receipt_pointee_owners(&[owned], 17, 4).is_err());
2660 }
2661
2662 #[test]
2663 fn receipt_schedule_mapping_requires_full_block_identity() {
2664 let owned = crate::device_runtime::BlockId {
2665 ptr: 0x2000,
2666 generation: crate::device_runtime::Generation(1),
2667 alloc_stream: crate::device_runtime::StreamId(2),
2668 device_ordinal: 3,
2669 };
2670 let different_stream = crate::device_runtime::BlockId {
2671 alloc_stream: crate::device_runtime::StreamId(9),
2672 ..owned
2673 };
2674 let pointee = checked_receipt_pointee(
2675 ResidentReceiptPointeeRole::RelationCount(0),
2676 0x2000,
2677 17,
2678 Some((owned, 8)),
2679 )
2680 .unwrap();
2681
2682 validate_receipt_schedule_block_mapping(&[pointee], &[owned])
2683 .expect("exact allocation generation and stream");
2684 assert!(validate_receipt_schedule_block_mapping(&[pointee], &[different_stream]).is_err());
2685 }
2686
2687 #[test]
2688 fn packed_receipt_retains_immutable_pointee_manifest() {
2689 let _: fn(&ResidentPackedReceipt) -> &[ResidentReceiptPointee] =
2690 ResidentPackedReceipt::pointee_manifest;
2691 }
2692
2693 #[test]
2694 fn receipt_manifest_records_every_runtime_pointee_as_read() {
2695 let first = crate::device_runtime::BlockId {
2696 ptr: 0x3000,
2697 generation: crate::device_runtime::Generation(1),
2698 alloc_stream: crate::device_runtime::StreamId(2),
2699 device_ordinal: 0,
2700 };
2701 let second = crate::device_runtime::BlockId {
2702 ptr: 0x4000,
2703 ..first
2704 };
2705 let pointees = [
2706 checked_receipt_pointee(
2707 ResidentReceiptPointeeRole::ScanTrace,
2708 first.ptr,
2709 17,
2710 Some((first, 4)),
2711 )
2712 .unwrap(),
2713 checked_receipt_pointee(
2714 ResidentReceiptPointeeRole::FilterTrace,
2715 second.ptr,
2716 17,
2717 Some((second, 4)),
2718 )
2719 .unwrap(),
2720 ];
2721 let mut recorder =
2722 crate::launch::LaunchRecorder::new_strict(crate::device_runtime::StreamId(8));
2723
2724 record_receipt_pointee_uses(&pointees, &mut recorder);
2725
2726 assert_eq!(recorder.recorded_count(), 2);
2727 }
2728
2729 #[test]
2730 fn relational_external_owners_expose_complete_allocation_snapshots() {
2731 type Owner = Option<crate::memory::RuntimeAllocationIdentity>;
2732 let _: fn(&super::ResidentSetWorkspace) -> xlog_core::Result<[Owner; 2]> =
2733 super::ResidentSetWorkspace::schedule_owner_snapshots;
2734 let _: fn(&super::ResidentJoinWorkspace) -> xlog_core::Result<[Owner; 3]> =
2735 super::ResidentJoinWorkspace::schedule_owner_snapshots;
2736 let _: fn(&super::ResidentConvergenceControl) -> xlog_core::Result<[Owner; 3]> =
2737 super::ResidentConvergenceControl::schedule_owner_snapshots;
2738 let _: fn(&super::ResidentDeviceTrace) -> xlog_core::Result<[Owner; 4]> =
2739 super::ResidentDeviceTrace::schedule_owner_snapshots;
2740 let _: fn(&super::ResidentSchemaWinners) -> xlog_core::Result<[Owner; 2]> =
2741 super::ResidentSchemaWinners::schedule_owner_snapshots;
2742 let _: fn(&ResidentPackedReceipt) -> xlog_core::Result<[Owner; 2]> =
2743 ResidentPackedReceipt::schedule_owner_snapshots;
2744 }
2745
2746 fn provider() -> Option<super::CudaKernelProvider> {
2747 match crate::CudaProviderBuilder::new(0, MemoryBudget::with_limit(512 * 1024 * 1024))
2748 .build()
2749 {
2750 Ok(provider) => Some(provider),
2751 Err(error) if std::env::var("XLOG_REQUIRE_CUDA").as_deref() == Ok("1") => {
2752 panic!("XLOG_REQUIRE_CUDA=1 but resident provider setup failed: {error}")
2753 }
2754 Err(error) => {
2755 eprintln!("Skipping resident CUDA test: {error}");
2756 None
2757 }
2758 }
2759 }
2760
2761 fn pair_schema() -> Schema {
2762 Schema::new(vec![
2763 ("key".to_string(), ScalarType::Symbol),
2764 ("value".to_string(), ScalarType::U64),
2765 ])
2766 }
2767
2768 fn pair_buffer(
2769 provider: &super::CudaKernelProvider,
2770 keys: &[u32],
2771 values: &[u64],
2772 ) -> crate::CudaBuffer {
2773 let key_bytes: Vec<u8> = keys.iter().flat_map(|value| value.to_le_bytes()).collect();
2774 let value_bytes: Vec<u8> = values
2775 .iter()
2776 .flat_map(|value| value.to_le_bytes())
2777 .collect();
2778 provider
2779 .create_buffer_from_slices(&[&key_bytes, &value_bytes], pair_schema())
2780 .expect("pair buffer")
2781 }
2782
2783 #[test]
2784 fn terminal_status_wire_layout_is_stable() {
2785 assert_eq!(std::mem::size_of::<ResidentTerminalStatus>(), 40);
2786 assert_eq!(std::mem::align_of::<ResidentTerminalStatus>(), 8);
2787 assert_eq!(ResidentTerminalCode::Running as u32, 0);
2788 assert_eq!(ResidentTerminalCode::Success as u32, 1);
2789 assert_eq!(ResidentTerminalCode::IterationLimit as u32, 2);
2790 assert_eq!(ResidentTerminalCode::CapacityOverflow as u32, 3);
2791 assert_eq!(ResidentTerminalCode::ResourceExhausted as u32, 4);
2792 assert_eq!(ResidentResourceCode::SetHashSlots as u32, 1);
2793 assert_eq!(ResidentResourceCode::JoinBuckets as u32, 2);
2794 assert_eq!(ResidentResourceCode::JoinChains as u32, 3);
2795 assert_eq!(ResidentResourceCode::InputRows as u32, 4);
2796 assert_eq!(ResidentResourceCode::OutputRows as u32, 5);
2797 }
2798
2799 #[test]
2800 fn hash_capacity_is_checked_and_keeps_load_at_most_one_half() {
2801 assert_eq!(checked_hash_slot_capacity(0).unwrap(), 1);
2802 assert_eq!(checked_hash_slot_capacity(1).unwrap(), 2);
2803 assert_eq!(checked_hash_slot_capacity(4_994).unwrap(), 16_384);
2804 assert_eq!(checked_hash_slot_capacity(9_988).unwrap(), 32_768);
2805 assert!(checked_hash_slot_capacity(u64::from(u32::MAX)).is_err());
2806 }
2807
2808 #[test]
2809 fn schema_winner_workspace_bytes_are_exact_and_checked() {
2810 assert_eq!(resident_schema_winners_device_bytes(0).unwrap(), 8);
2811 assert_eq!(resident_schema_winners_device_bytes(1).unwrap(), 8);
2812 assert_eq!(resident_schema_winners_device_bytes(3).unwrap(), 24);
2813 assert!(resident_schema_winners_device_bytes(usize::MAX).is_err());
2814 }
2815
2816 #[test]
2817 fn real_cuda_schema_winner_uses_first_nonempty_and_packs_one_receipt() {
2818 let Some(provider) = provider() else { return };
2819 let mut empty = provider
2820 .prepare_resident_relation(pair_schema(), 1)
2821 .expect("empty relation");
2822 provider
2823 .initialize_resident_relation_count(&mut empty, 0)
2824 .expect("empty count");
2825 let mut nonempty = provider
2826 .prepare_resident_relation(pair_schema(), 1)
2827 .expect("nonempty relation");
2828 provider
2829 .initialize_resident_relation_count(&mut nonempty, 1)
2830 .expect("nonempty count");
2831 let mut all_empty = provider
2832 .prepare_resident_relation(pair_schema(), 1)
2833 .expect("all-empty relation");
2834 provider
2835 .initialize_resident_relation_count(&mut all_empty, 0)
2836 .expect("all-empty count");
2837 let control = provider
2838 .prepare_resident_convergence_control()
2839 .expect("control");
2840 let trace = provider.prepare_resident_device_trace().expect("trace");
2841 let winners = provider
2842 .prepare_resident_schema_winners(&[11, 22, 33])
2843 .expect("schema winners");
2844 let receipt = provider
2845 .prepare_resident_packed_receipt_with_trace_and_schema_winners(
2846 &[&empty, &nonempty, &all_empty],
2847 &trace,
2848 &winners,
2849 )
2850 .expect("receipt");
2851 let mut pinned = provider
2852 .prepare_resident_pinned_receipt(&receipt)
2853 .expect("pinned receipt");
2854 let stream = provider
2855 .device()
2856 .inner()
2857 .stream()
2858 .context()
2859 .new_stream()
2860 .expect("stream");
2861 let graph = CapturedCudaGraph::capture_on_stream(&stream, || {
2862 provider.record_resident_control_initialize_on_stream(&control, &stream)?;
2863 provider.record_resident_device_trace_initialize_on_stream(&trace, &stream)?;
2864 provider
2865 .record_resident_schema_winners_initialize_on_stream(&winners, &receipt, &stream)?;
2866 provider.record_resident_schema_winner_mark_on_stream(
2867 empty.num_rows_device(),
2868 &winners,
2869 0,
2870 33,
2871 &stream,
2872 )?;
2873 provider.record_resident_schema_winner_mark_on_stream(
2874 nonempty.num_rows_device(),
2875 &winners,
2876 0,
2877 44,
2878 &stream,
2879 )?;
2880 provider.record_resident_schema_winner_mark_on_stream(
2881 all_empty.num_rows_device(),
2882 &winners,
2883 2,
2884 88,
2885 &stream,
2886 )?;
2887 provider.record_resident_schema_winner_mark_on_stream(
2888 nonempty.num_rows_device(),
2889 &winners,
2890 0,
2891 55,
2892 &stream,
2893 )?;
2894 provider.record_resident_schema_winner_mark_on_stream(
2895 empty.num_rows_device(),
2896 &winners,
2897 1,
2898 66,
2899 &stream,
2900 )?;
2901 provider.record_resident_schema_winner_mark_on_stream(
2902 nonempty.num_rows_device(),
2903 &winners,
2904 1,
2905 77,
2906 &stream,
2907 )?;
2908 provider.record_resident_terminal_success_on_stream(91, &control, &stream)?;
2909 provider.record_resident_receipt_pack_on_stream(&control, &receipt, &stream)
2910 })
2911 .expect("capture schema winners");
2912
2913 provider.reset_host_transfer_stats();
2914 provider.reset_d2h_transfer_count();
2915 provider.reset_untracked_metadata_dtoh_count();
2916 provider.reset_final_observation_transfer_stats();
2917 graph.launch(&stream).expect("launch schema winners");
2918 stream.synchronize().expect("schema winner core sync");
2919 let ordinary = provider.host_transfer_stats();
2920 let launch_metadata = provider.host_launch_metadata_transfer_stats();
2921 assert_eq!(ordinary.htod_calls, 0);
2922 assert_eq!(ordinary.htod_bytes, 0);
2923 assert_eq!(ordinary.dtoh_calls, 0);
2924 assert_eq!(ordinary.dtoh_bytes, 0);
2925 assert_eq!(launch_metadata.htod_calls, 0);
2926 assert_eq!(launch_metadata.htod_bytes, 0);
2927 assert_eq!(provider.d2h_transfer_count(), 0);
2928 assert_eq!(provider.untracked_metadata_dtoh_count(), 0);
2929
2930 let bytes = provider
2931 .observe_resident_packed_receipt(&receipt, &mut pinned, &stream)
2932 .expect("observe schema winners");
2933 assert_eq!(receipt.relation_count_len(), 3);
2934 assert_eq!(receipt.device_trace_field_count(), 4);
2935 assert_eq!(receipt.schema_winner_count(), 3);
2936 assert_eq!(receipt.total_count_field_len(), 10);
2937 assert_eq!(receipt.len_bytes(), 84);
2938 assert_eq!(u32::from_ne_bytes(bytes[72..76].try_into().unwrap()), 44);
2939 assert_eq!(u32::from_ne_bytes(bytes[76..80].try_into().unwrap()), 22);
2940 assert_eq!(u32::from_ne_bytes(bytes[80..84].try_into().unwrap()), 33);
2941 let final_observation = provider.final_observation_transfer_stats();
2942 assert_eq!(final_observation.dtoh_calls, 1);
2943 assert_eq!(final_observation.dtoh_bytes, 84);
2944 assert_eq!(final_observation.pinned_receipts, 1);
2945 }
2946
2947 #[test]
2948 fn real_cuda_relation_count_initialization_is_checked_and_cache_free() {
2949 let Some(provider) = provider() else { return };
2950 let mut relation = provider
2951 .prepare_resident_relation(pair_schema(), 1)
2952 .expect("relation");
2953
2954 provider
2955 .initialize_resident_relation_count(&mut relation, 1)
2956 .expect("singleton initialization");
2957 assert_eq!(relation.buffer().cached_row_count(), None);
2958 assert_eq!(
2959 provider
2960 .device()
2961 .inner()
2962 .dtoh_sync_copy(relation.num_rows_device())
2963 .expect("singleton count"),
2964 vec![1]
2965 );
2966
2967 provider
2968 .initialize_resident_relation_count(&mut relation, 0)
2969 .expect("empty initialization");
2970 assert_eq!(relation.buffer().cached_row_count(), None);
2971 assert_eq!(
2972 provider
2973 .device()
2974 .inner()
2975 .dtoh_sync_copy(relation.num_rows_device())
2976 .expect("empty count"),
2977 vec![0]
2978 );
2979
2980 let error = provider
2981 .initialize_resident_relation_count(&mut relation, 2)
2982 .expect_err("non-set count must be rejected");
2983 assert!(error.to_string().contains("expected 0 or 1"));
2984 assert_eq!(relation.buffer().cached_row_count(), None);
2985 }
2986
2987 #[test]
2988 fn real_cuda_relation_count_clear_is_graph_capturable_on_supplied_stream() {
2989 let Some(provider) = provider() else { return };
2990 let mut relation = provider
2991 .prepare_resident_relation(pair_schema(), 1)
2992 .expect("relation");
2993 provider
2994 .initialize_resident_relation_count(&mut relation, 1)
2995 .expect("singleton initialization");
2996 let stream = provider
2997 .device()
2998 .inner()
2999 .stream()
3000 .context()
3001 .new_stream()
3002 .expect("non-default stream");
3003
3004 let graph = CapturedCudaGraph::capture_on_stream(&stream, || {
3005 provider.record_resident_relation_clear_on_stream(&relation, &stream)
3006 })
3007 .expect("capture relation clear");
3008 graph.launch(&stream).expect("launch relation clear");
3009 stream.synchronize().expect("relation clear sync");
3010
3011 assert_eq!(
3012 provider
3013 .device()
3014 .inner()
3015 .dtoh_sync_copy(relation.num_rows_device())
3016 .expect("cleared count"),
3017 vec![0]
3018 );
3019 assert_eq!(relation.buffer().cached_row_count(), None);
3020 }
3021
3022 #[test]
3023 fn real_cuda_final_receipt_uses_one_pinned_dtoh_without_ordinary_counters() {
3024 let Some(provider) = provider() else { return };
3025 let mut relation = provider
3026 .prepare_resident_relation(pair_schema(), 1)
3027 .expect("relation");
3028 provider
3029 .initialize_resident_relation_count(&mut relation, 1)
3030 .expect("singleton initialization");
3031 let control = provider
3032 .prepare_resident_convergence_control()
3033 .expect("control");
3034 let receipt = provider
3035 .prepare_resident_packed_receipt(&[&relation])
3036 .expect("device receipt");
3037 let mut pinned = provider
3038 .prepare_resident_pinned_receipt(&receipt)
3039 .expect("pinned receipt");
3040 assert_eq!(pinned.len_bytes(), receipt.len_bytes());
3041 let stream = provider
3042 .device()
3043 .inner()
3044 .stream()
3045 .context()
3046 .new_stream()
3047 .expect("non-default stream");
3048 let graph = CapturedCudaGraph::capture_on_stream(&stream, || {
3049 provider.record_resident_control_initialize_on_stream(&control, &stream)?;
3050 provider.record_resident_terminal_success_on_stream(77, &control, &stream)?;
3051 provider.record_resident_receipt_pack_on_stream(&control, &receipt, &stream)
3052 })
3053 .expect("capture receipt production");
3054
3055 provider.reset_host_transfer_stats();
3056 provider.reset_d2h_transfer_count();
3057 provider.reset_untracked_metadata_dtoh_count();
3058 provider.reset_final_observation_transfer_stats();
3059 graph.launch(&stream).expect("launch receipt production");
3060 stream.synchronize().expect("core graph sync");
3061 let bytes = provider
3062 .observe_resident_packed_receipt(&receipt, &mut pinned, &stream)
3063 .expect("final receipt observation");
3064
3065 assert_eq!(bytes.len(), receipt.len_bytes());
3066 assert_eq!(u32::from_ne_bytes(bytes[0..4].try_into().unwrap()), 1);
3067 assert_eq!(u32::from_ne_bytes(bytes[4..8].try_into().unwrap()), 77);
3068 assert_eq!(u32::from_ne_bytes(bytes[40..44].try_into().unwrap()), 0);
3069 assert_eq!(u32::from_ne_bytes(bytes[44..48].try_into().unwrap()), 1);
3070 let ordinary = provider.host_transfer_stats();
3071 let launch_metadata = provider.host_launch_metadata_transfer_stats();
3072 assert_eq!(ordinary.htod_calls, 0);
3073 assert_eq!(ordinary.htod_bytes, 0);
3074 assert_eq!(ordinary.dtoh_calls, 0);
3075 assert_eq!(ordinary.dtoh_bytes, 0);
3076 assert_eq!(launch_metadata.htod_calls, 0);
3077 assert_eq!(launch_metadata.htod_bytes, 0);
3078 assert_eq!(provider.d2h_transfer_count(), 0);
3079 assert_eq!(provider.untracked_metadata_dtoh_count(), 0);
3080 let final_observation = provider.final_observation_transfer_stats();
3081 assert_eq!(final_observation.dtoh_calls, 1);
3082 assert_eq!(final_observation.dtoh_bytes, receipt.len_bytes() as u64);
3083 assert_eq!(final_observation.pinned_receipts, 1);
3084 }
3085
3086 #[test]
3087 fn real_cuda_device_trace_packs_exact_invocation_counts_in_final_receipt() {
3088 let Some(provider) = provider() else { return };
3089 let mut relation = provider
3090 .prepare_resident_relation(pair_schema(), 1)
3091 .expect("relation");
3092 provider
3093 .initialize_resident_relation_count(&mut relation, 1)
3094 .expect("singleton initialization");
3095 let control = provider
3096 .prepare_resident_convergence_control()
3097 .expect("control");
3098 let trace = provider
3099 .prepare_resident_device_trace()
3100 .expect("device trace");
3101 let receipt = provider
3102 .prepare_resident_packed_receipt_with_trace(&[&relation], &trace)
3103 .expect("traced device receipt");
3104 let mut pinned = provider
3105 .prepare_resident_pinned_receipt(&receipt)
3106 .expect("pinned receipt");
3107 assert_eq!(receipt.relation_count_len(), 1);
3108 assert_eq!(receipt.device_trace_field_count(), 4);
3109 assert_eq!(receipt.total_count_field_len(), 5);
3110 assert_eq!(receipt.len_bytes(), 64);
3111 assert_eq!(pinned.len_bytes(), receipt.len_bytes());
3112 let stream = provider
3113 .device()
3114 .inner()
3115 .stream()
3116 .context()
3117 .new_stream()
3118 .expect("non-default stream");
3119 let graph = CapturedCudaGraph::capture_on_stream(&stream, || {
3120 provider.record_resident_control_initialize_on_stream(&control, &stream)?;
3121 provider.record_resident_device_trace_initialize_on_stream(&trace, &stream)?;
3122 provider.record_resident_scan_trace_on_stream(&trace, &stream)?;
3123 provider.record_resident_scan_trace_on_stream(&trace, &stream)?;
3124 provider.record_resident_filter_trace_on_stream(&trace, &stream)?;
3125 provider.record_resident_scan_trace_on_stream(&trace, &stream)?;
3126 provider.record_resident_filter_trace_on_stream(&trace, &stream)?;
3127 provider.record_resident_terminal_success_on_stream(88, &control, &stream)?;
3128 provider.record_resident_receipt_pack_on_stream(&control, &receipt, &stream)
3129 })
3130 .expect("capture traced receipt production");
3131
3132 provider.reset_host_transfer_stats();
3133 provider.reset_d2h_transfer_count();
3134 provider.reset_untracked_metadata_dtoh_count();
3135 provider.reset_final_observation_transfer_stats();
3136 graph
3137 .launch(&stream)
3138 .expect("launch traced receipt production");
3139 stream.synchronize().expect("core graph sync");
3140 let ordinary_before_observation = provider.host_transfer_stats();
3141 let launch_metadata_before_observation = provider.host_launch_metadata_transfer_stats();
3142 assert_eq!(ordinary_before_observation.htod_calls, 0);
3143 assert_eq!(ordinary_before_observation.htod_bytes, 0);
3144 assert_eq!(ordinary_before_observation.dtoh_calls, 0);
3145 assert_eq!(ordinary_before_observation.dtoh_bytes, 0);
3146 assert_eq!(launch_metadata_before_observation.htod_calls, 0);
3147 assert_eq!(launch_metadata_before_observation.htod_bytes, 0);
3148 assert_eq!(provider.d2h_transfer_count(), 0);
3149 assert_eq!(provider.untracked_metadata_dtoh_count(), 0);
3150
3151 let bytes = provider
3152 .observe_resident_packed_receipt(&receipt, &mut pinned, &stream)
3153 .expect("final traced receipt observation");
3154 assert_eq!(bytes.len(), receipt.len_bytes());
3155 assert_eq!(u32::from_ne_bytes(bytes[0..4].try_into().unwrap()), 1);
3156 assert_eq!(u32::from_ne_bytes(bytes[4..8].try_into().unwrap()), 88);
3157 assert_eq!(u32::from_ne_bytes(bytes[40..44].try_into().unwrap()), 0);
3158 assert_eq!(u32::from_ne_bytes(bytes[44..48].try_into().unwrap()), 1);
3159 assert_eq!(u32::from_ne_bytes(bytes[48..52].try_into().unwrap()), 3);
3160 assert_eq!(u32::from_ne_bytes(bytes[52..56].try_into().unwrap()), 2);
3161 assert_eq!(u32::from_ne_bytes(bytes[56..60].try_into().unwrap()), 3);
3162 assert_eq!(u32::from_ne_bytes(bytes[60..64].try_into().unwrap()), 2);
3163 let ordinary_after_observation = provider.host_transfer_stats();
3164 assert_eq!(ordinary_after_observation.htod_calls, 0);
3165 assert_eq!(ordinary_after_observation.dtoh_calls, 0);
3166 assert_eq!(provider.d2h_transfer_count(), 0);
3167 assert_eq!(provider.untracked_metadata_dtoh_count(), 0);
3168 let final_observation = provider.final_observation_transfer_stats();
3169 assert_eq!(final_observation.dtoh_calls, 1);
3170 assert_eq!(final_observation.dtoh_bytes, receipt.len_bytes() as u64);
3171 assert_eq!(final_observation.pinned_receipts, 1);
3172 }
3173
3174 #[test]
3175 fn real_cuda_test_status_kernel_publishes_first_error_after_indexed_op() {
3176 let Some(provider) = provider() else { return };
3177 let input = pair_buffer(&provider, &[1], &[10]);
3178 let output = provider
3179 .prepare_resident_relation(pair_schema(), 1)
3180 .expect("output");
3181 let workspace = provider
3182 .prepare_resident_set_workspace(1)
3183 .expect("workspace");
3184 let control = provider
3185 .prepare_resident_convergence_control()
3186 .expect("control");
3187 let first = ResidentTerminalStatus {
3188 code: ResidentTerminalCode::ResourceExhausted as u32,
3189 op_id: 61,
3190 resource_code: ResidentResourceCode::SetHashSlots as u32,
3191 iterations: 2,
3192 limit: 9,
3193 reserved: u32::MAX,
3194 required: 33,
3195 capacity: 16,
3196 };
3197 let second = ResidentTerminalStatus {
3198 code: ResidentTerminalCode::CapacityOverflow as u32,
3199 op_id: 62,
3200 resource_code: ResidentResourceCode::OutputRows as u32,
3201 iterations: 3,
3202 limit: 10,
3203 reserved: 0,
3204 required: 65,
3205 capacity: 32,
3206 };
3207 let stream = provider
3208 .device()
3209 .inner()
3210 .stream()
3211 .context()
3212 .new_stream()
3213 .expect("non-default stream");
3214 let graph = CapturedCudaGraph::capture_on_stream(&stream, || {
3215 provider.record_resident_control_initialize_on_stream(&control, &stream)?;
3216 provider.record_resident_dedup_on_stream(
3217 &input, &output, &workspace, &control, 61, &stream,
3218 )?;
3219 provider.record_resident_test_status_on_stream(&control, first, &stream)?;
3220 provider.record_resident_test_status_on_stream(&control, second, &stream)?;
3221 provider.record_resident_terminal_success_on_stream(63, &control, &stream)
3222 })
3223 .expect("capture injected device status");
3224
3225 provider.reset_host_transfer_stats();
3226 graph
3227 .launch(&stream)
3228 .expect("launch injected device status");
3229 stream.synchronize().expect("injected status sync");
3230 let observed = provider
3231 .device()
3232 .inner()
3233 .dtoh_sync_copy(control.status_device())
3234 .expect("status observation")[0];
3235
3236 assert_eq!(
3237 observed,
3238 ResidentTerminalStatus {
3239 reserved: 0,
3240 ..first
3241 }
3242 );
3243 let transfers = provider.host_transfer_stats();
3244 let launch_metadata = provider.host_launch_metadata_transfer_stats();
3245 assert_eq!(transfers.htod_calls, 0);
3246 assert_eq!(transfers.htod_bytes, 0);
3247 assert_eq!(launch_metadata.htod_calls, 0);
3248 assert_eq!(launch_metadata.htod_bytes, 0);
3249 }
3250
3251 #[test]
3252 fn real_cuda_graph_union_deduplicates_full_mixed_width_rows() {
3253 let Some(provider) = provider() else { return };
3254 let left = pair_buffer(&provider, &[1, 1, 2], &[10, 10, 20]);
3255 let right = pair_buffer(&provider, &[2, 3], &[20, 30]);
3256 let output = provider
3257 .prepare_resident_relation(pair_schema(), 3)
3258 .expect("output");
3259 let workspace = provider
3260 .prepare_resident_set_workspace(5)
3261 .expect("workspace");
3262 let control = provider
3263 .prepare_resident_convergence_control()
3264 .expect("control");
3265 let stream = provider
3266 .device()
3267 .inner()
3268 .stream()
3269 .context()
3270 .new_stream()
3271 .expect("non-default stream");
3272 let graph = CapturedCudaGraph::capture_on_stream(&stream, || {
3273 provider.record_resident_control_initialize_on_stream(&control, &stream)?;
3274 provider.record_resident_union_on_stream(
3275 &left, &right, &output, &workspace, &control, 7, &stream,
3276 )
3277 })
3278 .expect("capture resident union");
3279 graph.launch(&stream).expect("launch resident union");
3280 stream.synchronize().expect("resident union sync");
3281 let keys = provider
3282 .download_column::<u32>(output.buffer(), 0)
3283 .expect("keys");
3284 let values = provider
3285 .download_column::<u64>(output.buffer(), 1)
3286 .expect("values");
3287 let mut rows: Vec<_> = keys.into_iter().zip(values).collect();
3288 rows.sort_unstable();
3289 assert_eq!(rows, vec![(1, 10), (2, 20), (3, 30)]);
3290 }
3291
3292 #[test]
3293 fn real_cuda_graph_diff_compares_and_deduplicates_full_rows() {
3294 let Some(provider) = provider() else { return };
3295 let left = pair_buffer(&provider, &[1, 1, 1, 2], &[10, 10, 11, 20]);
3296 let right = pair_buffer(&provider, &[1, 9], &[10, 90]);
3297 let output = provider
3298 .prepare_resident_relation(pair_schema(), 2)
3299 .expect("output");
3300 let workspace = provider
3301 .prepare_resident_set_workspace(6)
3302 .expect("workspace");
3303 let control = provider
3304 .prepare_resident_convergence_control()
3305 .expect("control");
3306 let stream = provider
3307 .device()
3308 .inner()
3309 .stream()
3310 .context()
3311 .new_stream()
3312 .expect("non-default stream");
3313 let graph = CapturedCudaGraph::capture_on_stream(&stream, || {
3314 provider.record_resident_control_initialize_on_stream(&control, &stream)?;
3315 provider.record_resident_diff_on_stream(
3316 &left, &right, &output, &workspace, &control, 10, &stream,
3317 )
3318 })
3319 .expect("capture resident diff");
3320 graph.launch(&stream).expect("launch resident diff");
3321 stream.synchronize().expect("resident diff sync");
3322 let keys = provider
3323 .download_column::<u32>(output.buffer(), 0)
3324 .expect("keys");
3325 let values = provider
3326 .download_column::<u64>(output.buffer(), 1)
3327 .expect("values");
3328 let mut rows: Vec<_> = keys.into_iter().zip(values).collect();
3329 rows.sort_unstable();
3330 assert_eq!(rows, vec![(1, 11), (2, 20)]);
3331 }
3332
3333 #[test]
3334 fn real_cuda_nullary_union_implements_unit_set_semantics() {
3335 let Some(provider) = provider() else { return };
3336 let unit_schema = Schema::new(Vec::new());
3337 let left = provider
3338 .create_zero_arity_buffer(unit_schema.clone(), 1)
3339 .expect("left unit");
3340 let right = provider
3341 .create_zero_arity_buffer(unit_schema.clone(), 1)
3342 .expect("right unit");
3343 let output = provider
3344 .prepare_resident_relation(unit_schema, 1)
3345 .expect("unit output");
3346 let workspace = provider
3347 .prepare_resident_set_workspace(2)
3348 .expect("workspace");
3349 let control = provider
3350 .prepare_resident_convergence_control()
3351 .expect("control");
3352 let stream = provider
3353 .device()
3354 .inner()
3355 .stream()
3356 .context()
3357 .new_stream()
3358 .expect("non-default stream");
3359 let graph = CapturedCudaGraph::capture_on_stream(&stream, || {
3360 provider.record_resident_control_initialize_on_stream(&control, &stream)?;
3361 provider.record_resident_union_on_stream(
3362 &left, &right, &output, &workspace, &control, 11, &stream,
3363 )
3364 })
3365 .expect("capture nullary union");
3366 graph.launch(&stream).expect("launch nullary union");
3367 stream.synchronize().expect("nullary union sync");
3368 assert_eq!(
3369 provider
3370 .device()
3371 .inner()
3372 .dtoh_sync_copy(output.num_rows_device())
3373 .expect("unit count"),
3374 vec![1]
3375 );
3376 }
3377
3378 #[test]
3379 fn real_cuda_inner_and_semi_join_preserve_required_multiplicity() {
3380 let Some(provider) = provider() else { return };
3381 let left = pair_buffer(&provider, &[1, 1, 2], &[10, 11, 12]);
3382 let right = pair_buffer(&provider, &[1, 1, 3], &[20, 21, 22]);
3383 let mut inner_columns = pair_schema().columns;
3384 inner_columns.extend(pair_schema().columns);
3385 let inner = provider
3386 .prepare_resident_relation(Schema::new(inner_columns), 4)
3387 .expect("inner output");
3388 let semi = provider
3389 .prepare_resident_relation(pair_schema(), 2)
3390 .expect("semi output");
3391 let inner_workspace = provider
3392 .prepare_resident_join_workspace(3)
3393 .expect("inner workspace");
3394 let semi_workspace = provider
3395 .prepare_resident_join_workspace(3)
3396 .expect("semi workspace");
3397 let control = provider
3398 .prepare_resident_convergence_control()
3399 .expect("control");
3400 let stream = provider
3401 .device()
3402 .inner()
3403 .stream()
3404 .context()
3405 .new_stream()
3406 .expect("non-default stream");
3407 let graph = CapturedCudaGraph::capture_on_stream(&stream, || {
3408 provider.record_resident_control_initialize_on_stream(&control, &stream)?;
3409 provider.record_resident_join_on_stream(
3410 ResidentJoinKind::Inner,
3411 &left,
3412 0,
3413 &right,
3414 0,
3415 &inner,
3416 &inner_workspace,
3417 &control,
3418 8,
3419 &stream,
3420 )?;
3421 provider.record_resident_join_on_stream(
3422 ResidentJoinKind::Semi,
3423 &left,
3424 0,
3425 &right,
3426 0,
3427 &semi,
3428 &semi_workspace,
3429 &control,
3430 9,
3431 &stream,
3432 )
3433 })
3434 .expect("capture resident joins");
3435 graph.launch(&stream).expect("launch resident joins");
3436 stream.synchronize().expect("resident joins sync");
3437 let inner_left_values = provider
3438 .download_column::<u64>(inner.buffer(), 1)
3439 .expect("inner left values");
3440 let inner_right_values = provider
3441 .download_column::<u64>(inner.buffer(), 3)
3442 .expect("inner right values");
3443 let mut pairs: Vec<_> = inner_left_values
3444 .into_iter()
3445 .zip(inner_right_values)
3446 .collect();
3447 pairs.sort_unstable();
3448 assert_eq!(pairs, vec![(10, 20), (10, 21), (11, 20), (11, 21)]);
3449 let mut semi_values = provider
3450 .download_column::<u64>(semi.buffer(), 1)
3451 .expect("semi values");
3452 semi_values.sort_unstable();
3453 assert_eq!(semi_values, vec![10, 11]);
3454 }
3455
3456 #[test]
3457 fn real_cuda_capacity_failure_reports_exact_required_and_clamps_output() {
3458 let Some(provider) = provider() else { return };
3459 let input = pair_buffer(&provider, &[1, 2, 3], &[10, 20, 30]);
3460 let output = provider
3461 .prepare_resident_relation(pair_schema(), 1)
3462 .expect("output");
3463 let workspace = provider
3464 .prepare_resident_set_workspace(3)
3465 .expect("workspace");
3466 let control = provider
3467 .prepare_resident_convergence_control()
3468 .expect("control");
3469 let stream = provider.device().inner().stream();
3470 provider
3471 .record_resident_control_initialize_on_stream(&control, stream)
3472 .expect("init");
3473 provider
3474 .record_resident_dedup_on_stream(&input, &output, &workspace, &control, 23, stream)
3475 .expect("dedup");
3476 stream.synchronize().expect("capacity sync");
3477 let receipt = provider
3478 .device()
3479 .inner()
3480 .dtoh_sync_copy(control.status_device())
3481 .expect("receipt")[0];
3482 assert_eq!(receipt.code, ResidentTerminalCode::CapacityOverflow as u32);
3483 assert_eq!(receipt.op_id, 23);
3484 assert_eq!(
3485 receipt.resource_code,
3486 ResidentResourceCode::OutputRows as u32
3487 );
3488 assert_eq!(receipt.required, 3);
3489 assert_eq!(receipt.capacity, 1);
3490 assert_eq!(
3491 provider
3492 .device()
3493 .inner()
3494 .dtoh_sync_copy(output.num_rows_device())
3495 .unwrap(),
3496 vec![1]
3497 );
3498 }
3499
3500 #[test]
3501 fn real_cuda_set_overflow_reports_exact_required_without_writing_output_storage() {
3502 let Some(provider) = provider() else { return };
3503 let left = pair_buffer(&provider, &[1, 2], &[10, 20]);
3504 let right = pair_buffer(&provider, &[3, 4], &[30, 40]);
3505 let sentinel_keys = [0xdead_beef_u32, 0xcafe_babe];
3506 let sentinel_values = [0x1111_2222_3333_4444_u64, 0x5555_6666_7777_8888];
3507 let output = super::ResidentRelation {
3508 buffer: pair_buffer(&provider, &sentinel_keys, &sentinel_values),
3509 };
3510 let workspace = provider
3511 .prepare_resident_set_workspace(4)
3512 .expect("set workspace");
3513 let control = provider
3514 .prepare_resident_convergence_control()
3515 .expect("set control");
3516 let stream = provider.device().inner().stream();
3517 provider
3518 .record_resident_control_initialize_on_stream(&control, stream)
3519 .expect("set control initialize");
3520 provider
3521 .record_resident_union_on_stream(
3522 &left, &right, &output, &workspace, &control, 31, stream,
3523 )
3524 .expect("overflowing union");
3525 stream.synchronize().expect("overflowing union sync");
3526 let status = provider
3527 .device()
3528 .inner()
3529 .dtoh_sync_copy(control.status_device())
3530 .expect("set overflow status")[0];
3531 assert_eq!(status.code, ResidentTerminalCode::CapacityOverflow as u32);
3532 assert_eq!(status.op_id, 31);
3533 assert_eq!(status.required, 4);
3534 assert_eq!(status.capacity, 2);
3535 let key_bytes: Vec<u8> = sentinel_keys
3536 .iter()
3537 .flat_map(|value| value.to_le_bytes())
3538 .collect();
3539 let value_bytes: Vec<u8> = sentinel_values
3540 .iter()
3541 .flat_map(|value| value.to_le_bytes())
3542 .collect();
3543 assert_eq!(
3544 provider
3545 .device()
3546 .inner()
3547 .dtoh_sync_copy(output.buffer().column(0).expect("set key column"))
3548 .expect("set key storage"),
3549 key_bytes
3550 );
3551 assert_eq!(
3552 provider
3553 .device()
3554 .inner()
3555 .dtoh_sync_copy(output.buffer().column(1).expect("set value column"))
3556 .expect("set value storage"),
3557 value_bytes
3558 );
3559 }
3560
3561 #[test]
3562 fn real_cuda_join_overflow_reports_exact_required_without_writing_output_storage() {
3563 let Some(provider) = provider() else { return };
3564 let left = pair_buffer(&provider, &[1, 1], &[10, 11]);
3565 let right = pair_buffer(&provider, &[1, 1], &[20, 21]);
3566 let mut output_columns = pair_schema().columns;
3567 output_columns.extend(pair_schema().columns);
3568 let output_schema = Schema::new(output_columns);
3569 let sentinel_columns = [
3570 [0xaaaa_aaaa_u32.to_le_bytes(), 0xbbbb_bbbb_u32.to_le_bytes()].concat(),
3571 [
3572 0x1111_2222_3333_4444_u64.to_le_bytes(),
3573 0x5555_6666_7777_8888_u64.to_le_bytes(),
3574 ]
3575 .concat(),
3576 [0xcccc_cccc_u32.to_le_bytes(), 0xdddd_dddd_u32.to_le_bytes()].concat(),
3577 [
3578 0x9999_aaaa_bbbb_cccc_u64.to_le_bytes(),
3579 0xdddd_eeee_ffff_0000_u64.to_le_bytes(),
3580 ]
3581 .concat(),
3582 ];
3583 let slices: Vec<&[u8]> = sentinel_columns.iter().map(Vec::as_slice).collect();
3584 let output = super::ResidentRelation {
3585 buffer: provider
3586 .create_buffer_from_slices(&slices, output_schema)
3587 .expect("sentinel join output"),
3588 };
3589 let workspace = provider
3590 .prepare_resident_join_workspace(2)
3591 .expect("join workspace");
3592 let control = provider
3593 .prepare_resident_convergence_control()
3594 .expect("join control");
3595 let stream = provider.device().inner().stream();
3596 provider
3597 .record_resident_control_initialize_on_stream(&control, stream)
3598 .expect("join control initialize");
3599 provider
3600 .record_resident_join_on_stream(
3601 ResidentJoinKind::Inner,
3602 &left,
3603 0,
3604 &right,
3605 0,
3606 &output,
3607 &workspace,
3608 &control,
3609 32,
3610 stream,
3611 )
3612 .expect("overflowing join");
3613 stream.synchronize().expect("overflowing join sync");
3614 let status = provider
3615 .device()
3616 .inner()
3617 .dtoh_sync_copy(control.status_device())
3618 .expect("join overflow status")[0];
3619 assert_eq!(status.code, ResidentTerminalCode::CapacityOverflow as u32);
3620 assert_eq!(status.op_id, 32);
3621 assert_eq!(status.required, 4);
3622 assert_eq!(status.capacity, 2);
3623 for (column, expected) in sentinel_columns.iter().enumerate() {
3624 let actual = provider
3625 .device()
3626 .inner()
3627 .dtoh_sync_copy(output.buffer().column(column).expect("join output column"))
3628 .expect("join output storage");
3629 assert_eq!(actual.as_slice(), expected.as_slice());
3630 }
3631 }
3632
3633 #[test]
3634 fn real_cuda_conditional_while_stops_from_device_convergence() {
3635 let Some(provider) = provider() else { return };
3636 let empty = provider
3637 .create_empty_buffer(Schema::new(vec![("novel".into(), ScalarType::U32)]))
3638 .expect("empty count owner");
3639 let control = provider
3640 .prepare_resident_convergence_control()
3641 .expect("control");
3642 let stream = provider
3643 .device()
3644 .inner()
3645 .stream()
3646 .context()
3647 .new_stream()
3648 .expect("non-default stream");
3649 provider
3650 .record_resident_control_initialize_on_stream(&control, &stream)
3651 .expect("control init");
3652 provider
3653 .record_resident_scc_begin_on_stream(8, 31, &control, &stream)
3654 .expect("SCC begin");
3655 stream.synchronize().expect("cold control init sync");
3656 let graph = CapturedCudaGraph::conditional_while_on_stream(&stream, 1, true, |body| {
3657 body.capture_on_stream(&stream, || {
3658 provider.record_resident_changed_reset_on_stream(&control, &stream)?;
3659 provider.record_resident_changed_mark_on_stream(
3660 empty.num_rows_device(),
3661 &control,
3662 &stream,
3663 )?;
3664 provider.record_resident_convergence_on_stream(
3665 body.handle(),
3666 8,
3667 31,
3668 &control,
3669 &stream,
3670 )
3671 })
3672 })
3673 .expect("conditional convergence graph");
3674 graph.launch(&stream).expect("launch conditional graph");
3675 provider
3676 .record_resident_terminal_success_on_stream(99, &control, &stream)
3677 .expect("terminal success");
3678 stream.synchronize().expect("conditional convergence sync");
3679 let receipt = provider
3680 .device()
3681 .inner()
3682 .dtoh_sync_copy(control.status_device())
3683 .expect("convergence receipt")[0];
3684 assert_eq!(receipt.code, ResidentTerminalCode::Success as u32);
3685 assert_eq!(receipt.op_id, 99);
3686 assert_eq!(receipt.iterations, 1);
3687 assert_eq!(
3688 provider
3689 .device()
3690 .inner()
3691 .dtoh_sync_copy(control.changed_device())
3692 .expect("changed flag"),
3693 vec![0]
3694 );
3695 }
3696
3697 #[test]
3698 fn real_cuda_multi_head_convergence_ors_every_recursive_head() {
3699 let Some(provider) = provider() else { return };
3700 let empty = provider
3701 .create_empty_buffer(Schema::new(vec![("empty".into(), ScalarType::U32)]))
3702 .expect("empty head");
3703 let changed = provider
3704 .create_buffer_from_slice(
3705 &[7_u32],
3706 Schema::new(vec![("changed".into(), ScalarType::U32)]),
3707 )
3708 .expect("changed head");
3709 let control = provider
3710 .prepare_resident_convergence_control()
3711 .expect("control");
3712 let stream = provider
3713 .device()
3714 .inner()
3715 .stream()
3716 .context()
3717 .new_stream()
3718 .expect("non-default stream");
3719 provider
3720 .record_resident_control_initialize_on_stream(&control, &stream)
3721 .expect("control init");
3722 provider
3723 .record_resident_scc_begin_on_stream(2, 41, &control, &stream)
3724 .expect("SCC begin");
3725 stream.synchronize().expect("control init sync");
3726 let graph = CapturedCudaGraph::conditional_while_on_stream(&stream, 1, true, |body| {
3727 body.capture_on_stream(&stream, || {
3728 provider.record_resident_changed_reset_on_stream(&control, &stream)?;
3729 provider.record_resident_changed_mark_on_stream(
3730 empty.num_rows_device(),
3731 &control,
3732 &stream,
3733 )?;
3734 provider.record_resident_changed_mark_on_stream(
3735 changed.num_rows_device(),
3736 &control,
3737 &stream,
3738 )?;
3739 provider.record_resident_convergence_on_stream(
3740 body.handle(),
3741 2,
3742 41,
3743 &control,
3744 &stream,
3745 )
3746 })
3747 })
3748 .expect("multi-head conditional graph");
3749 graph.launch(&stream).expect("multi-head launch");
3750 stream.synchronize().expect("multi-head sync");
3751 let receipt = provider
3752 .device()
3753 .inner()
3754 .dtoh_sync_copy(control.status_device())
3755 .expect("multi-head receipt")[0];
3756 assert_eq!(receipt.code, ResidentTerminalCode::IterationLimit as u32);
3757 assert_eq!(receipt.op_id, 41);
3758 assert_eq!(receipt.iterations, 2);
3759 assert_eq!(receipt.limit, 2);
3760 }
3761
3762 #[test]
3763 fn real_cuda_zero_iteration_limit_fails_before_body_replay() {
3764 let Some(provider) = provider() else { return };
3765 let control = provider
3766 .prepare_resident_convergence_control()
3767 .expect("control");
3768 let stream = provider.device().inner().stream();
3769 provider
3770 .record_resident_control_initialize_on_stream(&control, stream)
3771 .expect("control init");
3772 provider
3773 .record_resident_scc_begin_on_stream(0, 51, &control, stream)
3774 .expect("zero-limit SCC begin");
3775 stream.synchronize().expect("zero-limit sync");
3776 let receipt = provider
3777 .device()
3778 .inner()
3779 .dtoh_sync_copy(control.status_device())
3780 .expect("zero-limit receipt")[0];
3781 assert_eq!(receipt.code, ResidentTerminalCode::IterationLimit as u32);
3782 assert_eq!(receipt.op_id, 51);
3783 assert_eq!(receipt.iterations, 0);
3784 assert_eq!(receipt.limit, 0);
3785 }
3786}