@@ -115,68 +115,88 @@ impl Task {
115115 None => return ,
116116 } ;
117117
118- // Also the same as `singlethread.rs`, flag ourselves as ready to
119- // receive a notification.
120- let prev = self . atomic . state . swap ( SLEEPING , SeqCst ) ;
121- debug_assert_eq ! ( prev, AWAKE ) ;
122-
123- let poll = {
124- let mut cx = Context :: from_waker ( & self . waker ) ;
125- inner. future . as_mut ( ) . poll ( & mut cx)
126- } ;
127-
128- match poll {
129- // Same as `singlethread.rs` (noticing a pattern?) clean up
130- // resources associated with the future ASAP.
131- Poll :: Ready ( ( ) ) => {
132- * borrow = None ;
133- }
134-
135- // Unlike `singlethread.rs` we are responsible for ensuring there's
136- // a closure to handle the notification that a Future is ready. In
137- // the single-threaded case the notification itself enqueues work,
138- // but in the multithreaded case we don't know what thread a
139- // notification comes from so we need to ensure the current running
140- // thread is the one that enqueues the work. To do that we execute
141- // `Atomics.waitAsync`, creating a local Promise on our own thread
142- // which will resolve once `Atomics.notify` is called.
143- //
144- // We could be in one of two states as we execute this:
145- //
146- // * `SLEEPING` - we'll get notified via `Atomics.notify`
147- // and then this Promise will resolve.
148- //
149- // * `AWAKE` - the Promise will immediately be resolved and
150- // we'll execute the work on the next microtask queue.
151- Poll :: Pending => {
152- drop ( wait_async ( & self . atomic . state , SLEEPING ) . then ( & inner. closure ) ) ;
118+ loop {
119+ // Also the same as `singlethread.rs`, flag ourselves as ready to
120+ // receive a notification.
121+ let prev = self . atomic . state . swap ( SLEEPING , SeqCst ) ;
122+ debug_assert_eq ! ( prev, AWAKE ) ;
123+
124+ let poll = {
125+ let mut cx = Context :: from_waker ( & self . waker ) ;
126+ inner. future . as_mut ( ) . poll ( & mut cx)
127+ } ;
128+
129+ match poll {
130+ // Same as `singlethread.rs` (noticing a pattern?) clean up
131+ // resources associated with the future ASAP.
132+ Poll :: Ready ( ( ) ) => {
133+ * borrow = None ;
134+ }
135+
136+ // Unlike `singlethread.rs` we are responsible for ensuring there's
137+ // a closure to handle the notification that a Future is ready. In
138+ // the single-threaded case the notification itself enqueues work,
139+ // but in the multithreaded case we don't know what thread a
140+ // notification comes from so we need to ensure the current running
141+ // thread is the one that enqueues the work. To do that we execute
142+ // `Atomics.waitAsync`, creating a local Promise on our own thread
143+ // which will resolve once `Atomics.notify` is called.
144+ //
145+ // We could be in one of two states as we execute this:
146+ //
147+ // * `SLEEPING` - we'll get notified via `Atomics.notify`
148+ // and then this Promise will resolve.
149+ //
150+ // * `AWAKE` - the Promise will immediately be resolved and
151+ // we'll execute the work on the next microtask queue.
152+ Poll :: Pending => {
153+ match wait_async ( & self . atomic . state , SLEEPING ) {
154+ Some ( promise) => drop ( promise. then ( & inner. closure ) ) ,
155+ // our state has already changed so we can just do the work
156+ // again inline.
157+ None => continue ,
158+ }
159+ }
153160 }
161+ break ;
154162 }
155163 }
156164}
157165
158- fn wait_async ( ptr : & AtomicI32 , current_value : i32 ) -> js_sys:: Promise {
159- // If `Atomics.waitAsync` isn't defined (as it isn't defined anywhere today)
160- // then we use our fallback, otherwise we use the native function.
166+ fn wait_async ( ptr : & AtomicI32 , current_value : i32 ) -> Option < js_sys:: Promise > {
167+ // If `Atomics.waitAsync` isn't defined then we use our fallback, otherwise
168+ // we use the native function.
161169 return if Atomics :: get_wait_async ( ) . is_undefined ( ) {
162- crate :: task:: wait_async_polyfill:: wait_async ( ptr, current_value)
170+ Some ( crate :: task:: wait_async_polyfill:: wait_async (
171+ ptr,
172+ current_value,
173+ ) )
163174 } else {
164175 let mem = wasm_bindgen:: memory ( ) . unchecked_into :: < js_sys:: WebAssembly :: Memory > ( ) ;
165- Atomics :: wait_async (
166- & mem. buffer ( ) ,
167- ptr as * const AtomicI32 as i32 / 4 ,
168- current_value,
169- )
176+ let array = js_sys:: Int32Array :: new ( & mem. buffer ( ) ) ;
177+ let result = Atomics :: wait_async ( & array, ptr as * const AtomicI32 as i32 / 4 , current_value) ;
178+ if result. async_ ( ) {
179+ Some ( result. value ( ) )
180+ } else {
181+ None
182+ }
170183 } ;
171184
172185 #[ wasm_bindgen]
173186 extern "C" {
174187 type Atomics ;
188+ type WaitAsyncResult ;
175189
176190 #[ wasm_bindgen( static_method_of = Atomics , js_name = waitAsync) ]
177- fn wait_async ( buf : & JsValue , index : i32 , value : i32 ) -> js_sys :: Promise ;
191+ fn wait_async ( buf : & js_sys :: Int32Array , index : i32 , value : i32 ) -> WaitAsyncResult ;
178192
179193 #[ wasm_bindgen( static_method_of = Atomics , js_name = waitAsync, getter) ]
180194 fn get_wait_async ( ) -> JsValue ;
195+
196+ #[ wasm_bindgen( method, getter, structural, js_name = async ) ]
197+ fn async_ ( this : & WaitAsyncResult ) -> bool ;
198+
199+ #[ wasm_bindgen( method, getter, structural) ]
200+ fn value ( this : & WaitAsyncResult ) -> js_sys:: Promise ;
181201 }
182202}
0 commit comments