1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
//
// Copyright (C) 2026 Intel Corporation
//
// Under the MIT License or the Apache License v2.0.
// See LICENSE-MIT and LICENSE-APACHE for license information.
// SPDX-License-Identifier: MIT OR Apache-2.0
//
use std::{
pin::Pin,
sync::{Arc, atomic::Ordering::Relaxed},
task::{Context, Poll},
};
use sycl_rs_sys::{event::ffi, types::SharedWaker};
use pin_project::pin_project;
use crate::{Result, info::InfoTarget, private::Sealed, queue::Queue};
pub struct Event(pub(crate) cxx::UniquePtr<ffi::Event>);
impl Event {
/// Performs a blocking wait for the event to complete. Returns an error if a synchronous SYCL
/// exception occurs.
///
/// Dropping the event does not wait for its completion.
pub fn wait(&mut self) -> Result<()> {
ffi::wait(&mut self.0)
}
}
impl Sealed for Event {}
impl InfoTarget for Event {}
impl From<cxx::UniquePtr<ffi::Event>> for Event {
fn from(value: cxx::UniquePtr<ffi::Event>) -> Self {
Self(value)
}
}
impl Clone for Event {
fn clone(&self) -> Self {
ffi::clone(&self.0).into()
}
}
#[pin_project]
pub struct EventFuture {
event: Event,
shared: Arc<SharedWaker>,
set_callback: bool,
queue: Option<Queue>,
}
impl Future for EventFuture {
type Output = Result<()>;
fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
let this = self.project();
// Set the callback on first Future poll (Futures can't be active until polled)
if *this.set_callback == false {
*this.set_callback = true;
let mut queue = Queue::new_immediate();
this.shared.waker.register(cx.waker());
// Safety: registered callback will decrement the Arc strong reference count, which is
// large enough because it was increased by the previous clone().
let ptr = Arc::into_raw(this.shared.clone());
let result = unsafe { ffi::register_callback(&mut queue.0, &this.event.0, ptr) };
match result {
Ok(_) => {
this.queue.replace(queue);
}
Err(_) => {
return Poll::Ready(result);
}
}
} else {
// Quick check before registering to avoid wasting time
if this.shared.done.load(Relaxed) {
// The event finished - waiting for it returns immediately
return Poll::Ready(this.event.wait());
}
this.shared.waker.register(cx.waker());
}
// Check the event again to avoid a race condition
// https://docs.rs/futures/latest/futures/task/struct.AtomicWaker.html#examples
if this.shared.done.load(Relaxed) {
// The event finished - waiting for it returns immediately
Poll::Ready(this.event.wait())
} else {
Poll::Pending
}
}
}
impl IntoFuture for Event {
type Output = Result<()>;
type IntoFuture = EventFuture;
fn into_future(self) -> Self::IntoFuture {
EventFuture {
event: self,
shared: Arc::new(SharedWaker::new()),
set_callback: false,
queue: None,
}
}
}