@@ -3,11 +3,13 @@ use alloc::collections::VecDeque;
33use alloc:: sync:: Arc ;
44use alloc:: task:: Wake ;
55use alloc:: vec:: Vec ;
6+ use core:: any:: Any ;
67use core:: fmt:: Write ;
78use core:: future:: { poll_fn, Future } ;
89use core:: pin:: Pin ;
910use core:: sync:: atomic:: { AtomicBool , AtomicU64 , Ordering } ;
1011use core:: task:: { Context , Poll , Waker } ;
12+ use futures:: channel:: oneshot;
1113
1214use spin:: Mutex ;
1315use uefi:: boot:: { EventType , TimerTrigger , Tpl } ;
@@ -25,6 +27,7 @@ struct Task {
2527 future : Mutex < BoxFuture > ,
2628 micros : AtomicU64 ,
2729 last_micros : AtomicU64 ,
30+ done : AtomicBool ,
2831}
2932
3033impl Task {
@@ -38,13 +41,15 @@ impl Task {
3841 micros : AtomicU64 :: new ( 0 ) ,
3942 last_micros : AtomicU64 :: new ( 0 ) ,
4043 in_queue : AtomicBool :: new ( false ) ,
44+ done : AtomicBool :: new ( false ) ,
4145 } )
4246 }
4347}
4448
4549impl Wake for Task {
4650 fn wake ( self : Arc < Self > ) {
4751 if !self . in_queue . swap ( true , Ordering :: Relaxed ) {
52+ assert ! ( !self . done. load( Ordering :: Relaxed ) ) ;
4853 EXECUTOR . lock ( ) . ready_tasks . push_back ( self ) ;
4954 }
5055 }
@@ -55,6 +60,14 @@ static EXECUTOR: Mutex<Executor> = Mutex::new(Executor {
5560 tasks : vec ! [ ] ,
5661} ) ;
5762
63+ pub struct JoinHandle < T > ( oneshot:: Receiver < T > ) ;
64+
65+ impl < T > JoinHandle < T > {
66+ pub async fn join ( self ) -> T {
67+ self . 0 . await . expect ( "tasks should never be cancelled" )
68+ }
69+ }
70+
5871pub struct Executor {
5972 // TODO(veluca): scheduling.
6073 ready_tasks : VecDeque < Arc < Task > > ,
@@ -145,6 +158,9 @@ impl Executor {
145158 t. last_micros
146159 . store ( t. micros . load ( Ordering :: Relaxed ) , Ordering :: Relaxed ) ;
147160 }
161+
162+ // Clear completed tasks.
163+ tasks. retain ( |t| !t. done . load ( Ordering :: Relaxed ) ) ;
148164 }
149165 last = Timer :: micros ( ) as u64 ;
150166 Self :: sleep_us ( 1_000_000 ) . await ;
@@ -165,10 +181,13 @@ impl Executor {
165181 let mut context = Context :: from_waker ( & waker) ;
166182 let mut fut = task. future . try_lock ( ) . unwrap ( ) ;
167183 let begin = Timer :: micros ( ) ;
168- let _ = fut. 0 . as_mut ( ) . poll ( & mut context) ;
184+ let done = fut. 0 . as_mut ( ) . poll ( & mut context) ;
169185 let end = Timer :: micros ( ) ;
170186 task. micros
171187 . fetch_add ( ( end - begin) as u64 , Ordering :: Relaxed ) ;
188+ if done. is_ready ( ) {
189+ task. done . swap ( true , Ordering :: Relaxed ) ;
190+ }
172191 }
173192 }
174193
@@ -211,13 +230,18 @@ impl Executor {
211230 }
212231
213232 /// Spawn a new task.
214- pub fn spawn < Fut > ( name : & ' static str , f : Fut )
233+ pub fn spawn < Fut , T : ' static > ( name : & ' static str , f : Fut ) -> JoinHandle < T >
215234 where
216- Fut : Future < Output = ( ) > + ' static ,
235+ Fut : Future < Output = T > + ' static ,
217236 {
218- let task = Task :: new ( name, f) ;
237+ let ( send, recv) = oneshot:: channel ( ) ;
238+ let task = Task :: new ( name, async move {
239+ let t = f. await ;
240+ let _ = send. send ( t) ;
241+ } ) ;
219242 let mut executor = EXECUTOR . lock ( ) ;
220243 executor. tasks . push ( task. clone ( ) ) ;
221244 executor. ready_tasks . push_back ( task) ;
245+ JoinHandle ( recv)
222246 }
223247}
0 commit comments