mlua/
thread.rs

1//! Lua thread (coroutine) handling.
2//!
3//! This module provides types for creating and working with Lua coroutines from Rust.
4//! Coroutines allow cooperative multitasking within a single Lua state by suspending and
5//! resuming execution at well-defined yield points.
6//!
7//! # Basic Usage
8//!
9//! Threads are created via [`Lua::create_thread`] and driven by calling [`Thread::resume`]:
10//!
11//! ```rust
12//! # use mlua::{Lua, Result, Thread};
13//! # fn main() -> Result<()> {
14//! let lua = Lua::new();
15//! let thread: Thread = lua.load(r#"
16//!     coroutine.create(function(a, b)
17//!         coroutine.yield(a + b)
18//!         return a * b
19//!     end)
20//! "#).eval()?;
21//!
22//! assert_eq!(thread.resume::<i32>((3, 4))?, 7);
23//! assert_eq!(thread.resume::<i32>(())?,    12);
24//! # Ok(())
25//! # }
26//! ```
27//!
28//! # Async Support
29//!
30//! When the `async` feature is enabled, a [`Thread`] can be converted into an [`AsyncThread`]
31//! via [`Thread::into_async`], which implements both [`Future`] and [`Stream`].
32//! This integrates Lua coroutines naturally with Rust async runtimes such as Tokio.
33//!
34//! [`Lua::create_thread`]: crate::Lua::create_thread
35//! [`Future`]: std::future::Future
36//! [`Stream`]: futures_util::stream::Stream
37
38use std::fmt;
39use std::os::raw::{c_int, c_void};
40
41use crate::error::{Error, Result};
42use crate::function::Function;
43use crate::state::RawLua;
44use crate::traits::{FromLuaMulti, IntoLuaMulti};
45use crate::types::{LuaType, ValueRef};
46use crate::util::{StackGuard, check_stack, error_traceback_thread, pop_error};
47
48#[cfg(not(feature = "luau"))]
49use crate::{
50    debug::{Debug, HookTriggers},
51    types::HookKind,
52};
53
54#[cfg(feature = "async")]
55use {
56    futures_util::stream::Stream,
57    std::{
58        future::Future,
59        marker::PhantomData,
60        pin::Pin,
61        ptr::NonNull,
62        task::{Context, Poll, Waker},
63    },
64};
65
66/// Status of a Lua thread (coroutine).
67#[derive(Debug, Copy, Clone, Eq, PartialEq)]
68pub enum ThreadStatus {
69    /// The thread was just created or is suspended (yielded).
70    ///
71    /// If a thread is in this state, it can be resumed by calling [`Thread::resume`].
72    Resumable,
73    /// The thread is currently running.
74    Running,
75    /// The thread has finished executing.
76    Finished,
77    /// The thread has raised a Lua error during execution.
78    Error,
79}
80
81/// Internal representation of a Lua thread status.
82///
83/// The number in `New` and `Yielded` variants is the number of arguments pushed
84/// to the thread stack.
85#[derive(Clone, Copy)]
86enum ThreadStatusInner {
87    New(c_int),
88    Running,
89    Yielded(c_int),
90    Finished,
91    Error,
92}
93
94impl ThreadStatusInner {
95    #[cfg(feature = "async")]
96    #[inline(always)]
97    fn is_resumable(self) -> bool {
98        matches!(self, ThreadStatusInner::New(_) | ThreadStatusInner::Yielded(_))
99    }
100
101    #[cfg(feature = "async")]
102    #[inline(always)]
103    fn is_yielded(self) -> bool {
104        matches!(self, ThreadStatusInner::Yielded(_))
105    }
106}
107
108/// Handle to an internal Lua thread (coroutine).
109#[derive(Clone, PartialEq)]
110pub struct Thread(pub(crate) ValueRef, pub(crate) *mut ffi::lua_State);
111
112#[cfg(feature = "send")]
113unsafe impl Send for Thread {}
114#[cfg(feature = "send")]
115unsafe impl Sync for Thread {}
116
117/// Thread (coroutine) representation as an async [`Future`] or [`Stream`].
118///
119/// [`Future`]: std::future::Future
120/// [`Stream`]: futures_util::stream::Stream
121#[cfg(feature = "async")]
122#[cfg_attr(docsrs, doc(cfg(feature = "async")))]
123#[must_use = "futures do nothing unless you `.await` or poll them"]
124pub struct AsyncThread<R> {
125    thread: Thread,
126    ret: PhantomData<fn() -> R>,
127    recycle: bool,
128}
129
130impl Thread {
131    /// Returns reference to the Lua state that this thread is associated with.
132    #[inline(always)]
133    pub fn state(&self) -> *mut ffi::lua_State {
134        self.1
135    }
136
137    /// Resumes execution of this thread.
138    ///
139    /// Equivalent to [`coroutine.resume`].
140    ///
141    /// Passes `args` as arguments to the thread. If the coroutine has called [`coroutine.yield`],
142    /// it will return these arguments. Otherwise, the coroutine wasn't yet started, so the
143    /// arguments are passed to its main function.
144    ///
145    /// If the thread is no longer resumable (meaning it has finished execution or encountered an
146    /// error), this will return [`Error::CoroutineUnresumable`], otherwise will return `Ok` as
147    /// follows:
148    ///
149    /// If the thread calls [`coroutine.yield`], returns the values passed to `yield`. If the thread
150    /// `return`s values from its main function, returns those.
151    ///
152    /// # Examples
153    ///
154    /// ```
155    /// # use mlua::{Error, Lua, Result, Thread};
156    /// # fn main() -> Result<()> {
157    /// # let lua = Lua::new();
158    /// let thread: Thread = lua.load(r#"
159    ///     coroutine.create(function(arg)
160    ///         assert(arg == 42)
161    ///         local yieldarg = coroutine.yield(123)
162    ///         assert(yieldarg == 43)
163    ///         return 987
164    ///     end)
165    /// "#).eval()?;
166    ///
167    /// assert_eq!(thread.resume::<u32>(42)?, 123);
168    /// assert_eq!(thread.resume::<u32>(43)?, 987);
169    ///
170    /// // The coroutine has now returned, so `resume` will fail
171    /// match thread.resume::<u32>(()) {
172    ///     Err(Error::CoroutineUnresumable) => {},
173    ///     unexpected => panic!("unexpected result {:?}", unexpected),
174    /// }
175    /// # Ok(())
176    /// # }
177    /// ```
178    ///
179    /// [`coroutine.resume`]: https://www.lua.org/manual/5.4/manual.html#pdf-coroutine.resume
180    /// [`coroutine.yield`]: https://www.lua.org/manual/5.4/manual.html#pdf-coroutine.yield
181    pub fn resume<R>(&self, args: impl IntoLuaMulti) -> Result<R>
182    where
183        R: FromLuaMulti,
184    {
185        let lua = self.0.lua.lock();
186        let mut pushed_nargs = match self.status_inner(&lua) {
187            ThreadStatusInner::New(nargs) | ThreadStatusInner::Yielded(nargs) => nargs,
188            _ => return Err(Error::CoroutineUnresumable),
189        };
190
191        let state = lua.state();
192        let thread_state = self.state();
193        unsafe {
194            let _sg = StackGuard::new(state);
195
196            let nargs = args.push_into_stack_multi(&lua)?;
197            if nargs > 0 {
198                check_stack(thread_state, nargs)?;
199                ffi::lua_xmove(state, thread_state, nargs);
200                pushed_nargs += nargs;
201            }
202
203            let _thread_sg = StackGuard::with_top(thread_state, 0);
204            let (_, nresults) = self.resume_inner(&lua, pushed_nargs)?;
205            check_stack(state, nresults + 1)?;
206            ffi::lua_xmove(thread_state, state, nresults);
207
208            R::from_stack_multi(nresults, &lua)
209        }
210    }
211
212    /// Resumes execution of this thread, immediately raising an error.
213    ///
214    /// This is a Luau specific extension.
215    #[cfg(feature = "luau")]
216    #[cfg_attr(docsrs, doc(cfg(feature = "luau")))]
217    pub fn resume_error<R>(&self, error: impl crate::IntoLua) -> Result<R>
218    where
219        R: FromLuaMulti,
220    {
221        let lua = self.0.lua.lock();
222        match self.status_inner(&lua) {
223            ThreadStatusInner::New(_) | ThreadStatusInner::Yielded(_) => {}
224            _ => return Err(Error::CoroutineUnresumable),
225        };
226
227        let state = lua.state();
228        let thread_state = self.state();
229        unsafe {
230            let _sg = StackGuard::new(state);
231
232            check_stack(state, 1)?;
233            error.push_into_stack(&lua)?;
234            ffi::lua_xmove(state, thread_state, 1);
235
236            let _thread_sg = StackGuard::with_top(thread_state, 0);
237            let (_, nresults) = self.resume_inner(&lua, ffi::LUA_RESUMEERROR)?;
238            check_stack(state, nresults + 1)?;
239            ffi::lua_xmove(thread_state, state, nresults);
240
241            R::from_stack_multi(nresults, &lua)
242        }
243    }
244
245    /// Resumes execution of this thread.
246    ///
247    /// It's similar to `resume()` but leaves `nresults` values on the thread stack.
248    unsafe fn resume_inner(&self, lua: &RawLua, nargs: c_int) -> Result<(ThreadStatusInner, c_int)> {
249        let state = lua.state();
250        let thread_state = self.state();
251        let mut nresults = 0;
252        #[cfg(not(feature = "luau"))]
253        let ret = ffi::lua_resume(thread_state, state, nargs, &mut nresults as *mut c_int);
254        #[cfg(feature = "luau")]
255        let ret = ffi::lua_resumex(thread_state, state, nargs, &mut nresults as *mut c_int);
256        match ret {
257            ffi::LUA_OK => Ok((ThreadStatusInner::Finished, nresults)),
258            ffi::LUA_YIELD => Ok((ThreadStatusInner::Yielded(0), nresults)),
259            ffi::LUA_ERRMEM => {
260                // Don't call error handler for memory errors
261                Err(pop_error(thread_state, ret))
262            }
263            _ => {
264                check_stack(state, 3)?;
265                protect_lua!(state, 0, 1, |state| error_traceback_thread(state, thread_state))?;
266                Err(pop_error(state, ret))
267            }
268        }
269    }
270
271    /// Gets the status of the thread.
272    pub fn status(&self) -> ThreadStatus {
273        match self.status_inner(&self.0.lua.lock()) {
274            ThreadStatusInner::New(_) | ThreadStatusInner::Yielded(_) => ThreadStatus::Resumable,
275            ThreadStatusInner::Running => ThreadStatus::Running,
276            ThreadStatusInner::Finished => ThreadStatus::Finished,
277            ThreadStatusInner::Error => ThreadStatus::Error,
278        }
279    }
280
281    /// Gets the status of the thread (internal implementation).
282    fn status_inner(&self, lua: &RawLua) -> ThreadStatusInner {
283        let thread_state = self.state();
284        if thread_state == lua.state() {
285            // The thread is currently running
286            return ThreadStatusInner::Running;
287        }
288        let status = unsafe { ffi::lua_status(thread_state) };
289        let top = unsafe { ffi::lua_gettop(thread_state) };
290        match status {
291            ffi::LUA_YIELD => ThreadStatusInner::Yielded(top),
292            ffi::LUA_OK if top > 0 => ThreadStatusInner::New(top - 1),
293            ffi::LUA_OK => ThreadStatusInner::Finished,
294            _ => ThreadStatusInner::Error,
295        }
296    }
297
298    /// Returns `true` if this thread is resumable (meaning it can be resumed by calling
299    /// [`Thread::resume`]).
300    #[inline(always)]
301    pub fn is_resumable(&self) -> bool {
302        self.status() == ThreadStatus::Resumable
303    }
304
305    /// Returns `true` if this thread is currently running.
306    #[inline(always)]
307    pub fn is_running(&self) -> bool {
308        self.status() == ThreadStatus::Running
309    }
310
311    /// Returns `true` if this thread has finished executing.
312    #[inline(always)]
313    pub fn is_finished(&self) -> bool {
314        self.status() == ThreadStatus::Finished
315    }
316
317    /// Returns `true` if this thread has raised a Lua error during execution.
318    #[inline(always)]
319    pub fn is_error(&self) -> bool {
320        self.status() == ThreadStatus::Error
321    }
322
323    /// Sets a hook function that will periodically be called as Lua code executes.
324    ///
325    /// This function is similar or [`Lua::set_hook`] except that it sets for the thread.
326    /// You can have multiple hooks for different threads.
327    ///
328    /// To remove a hook call [`Thread::remove_hook`].
329    ///
330    /// [`Lua::set_hook`]: crate::Lua::set_hook
331    #[cfg(not(feature = "luau"))]
332    #[cfg_attr(docsrs, doc(cfg(not(feature = "luau"))))]
333    pub fn set_hook<F>(&self, triggers: HookTriggers, callback: F) -> Result<()>
334    where
335        F: Fn(&crate::Lua, &Debug) -> Result<crate::VmState> + crate::MaybeSend + 'static,
336    {
337        let lua = self.0.lua.lock();
338        unsafe {
339            lua.set_thread_hook(
340                self.state(),
341                HookKind::Thread(triggers, crate::types::XRc::new(callback)),
342            )
343        }
344    }
345
346    /// Removes any hook function from this thread.
347    #[cfg(not(feature = "luau"))]
348    #[cfg_attr(docsrs, doc(cfg(not(feature = "luau"))))]
349    pub fn remove_hook(&self) {
350        let _lua = self.0.lua.lock();
351        unsafe {
352            ffi::lua_sethook(self.state(), None, 0, 0);
353        }
354    }
355
356    /// Resets a thread
357    ///
358    /// In [Lua 5.4]: cleans its call stack and closes all pending to-be-closed variables.
359    /// Returns an error in case of either the original error that stopped the thread or errors
360    /// in closing methods.
361    ///
362    /// In Luau: resets to the initial state of a newly created Lua thread.
363    /// Lua threads in arbitrary states (like yielded or errored) can be reset properly.
364    ///
365    /// Other Lua versions can reset only new or finished threads.
366    ///
367    /// Sets a Lua function for the thread afterwards.
368    ///
369    /// [Lua 5.4]: https://www.lua.org/manual/5.4/manual.html#lua_closethread
370    pub fn reset(&self, func: Function) -> Result<()> {
371        let lua = self.0.lua.lock();
372        let thread_state = self.state();
373        unsafe {
374            let status = self.status_inner(&lua);
375            self.reset_inner(status)?;
376
377            // Push function to the top of the thread stack
378            ffi::lua_xpush(lua.ref_thread(), thread_state, func.0.index);
379
380            #[cfg(feature = "luau")]
381            {
382                // Inherit `LUA_GLOBALSINDEX` from the main thread
383                ffi::lua_xpush(lua.main_state(), thread_state, ffi::LUA_GLOBALSINDEX);
384                ffi::lua_replace(thread_state, ffi::LUA_GLOBALSINDEX);
385            }
386
387            Ok(())
388        }
389    }
390
391    unsafe fn reset_inner(&self, status: ThreadStatusInner) -> Result<()> {
392        match status {
393            ThreadStatusInner::New(_) => {
394                // The thread is new, so we can just set the top to 0
395                ffi::lua_settop(self.state(), 0);
396                Ok(())
397            }
398            ThreadStatusInner::Running => Err(Error::runtime("cannot reset a running thread")),
399            ThreadStatusInner::Finished => Ok(()),
400            #[cfg(not(any(feature = "lua55", feature = "lua54", feature = "luau")))]
401            ThreadStatusInner::Yielded(_) | ThreadStatusInner::Error => {
402                Err(Error::runtime("cannot reset non-finished thread"))
403            }
404            #[cfg(any(feature = "lua55", feature = "lua54", feature = "luau"))]
405            ThreadStatusInner::Yielded(_) | ThreadStatusInner::Error => {
406                let thread_state = self.state();
407
408                #[cfg(all(feature = "lua54", not(feature = "vendored")))]
409                let status = ffi::lua_resetthread(thread_state);
410                #[cfg(any(feature = "lua55", all(feature = "lua54", feature = "vendored")))]
411                let status = {
412                    let lua = self.0.lua.lock();
413                    ffi::lua_closethread(thread_state, lua.state())
414                };
415                #[cfg(any(feature = "lua55", feature = "lua54"))]
416                if status != ffi::LUA_OK {
417                    return Err(pop_error(thread_state, status));
418                }
419                #[cfg(feature = "luau")]
420                ffi::lua_resetthread(thread_state);
421
422                Ok(())
423            }
424        }
425    }
426
427    /// Converts [`Thread`] to an [`AsyncThread`] which implements [`Future`] and [`Stream`] traits.
428    ///
429    /// Only resumable threads can be converted to [`AsyncThread`].
430    ///
431    /// `args` are pushed to the thread stack and will be used when the thread is resumed.
432    /// The object calls [`resume`] while polling and also allow to run Rust futures
433    /// to completion using an executor.
434    ///
435    /// Using [`AsyncThread`] as a [`Stream`] allow to iterate through [`coroutine.yield`]
436    /// values whereas [`Future`] version discards that values and poll until the final
437    /// one (returned from the thread function).
438    ///
439    /// [`Future`]: std::future::Future
440    /// [`Stream`]: futures_util::stream::Stream
441    /// [`resume`]: https://www.lua.org/manual/5.4/manual.html#lua_resume
442    /// [`coroutine.yield`]: https://www.lua.org/manual/5.4/manual.html#pdf-coroutine.yield
443    ///
444    /// # Examples
445    ///
446    /// ```
447    /// # use mlua::{Lua, Result, Thread};
448    /// use futures_util::stream::TryStreamExt;
449    /// # #[tokio::main]
450    /// # async fn main() -> Result<()> {
451    /// # let lua = Lua::new();
452    /// let thread: Thread = lua.load(r#"
453    ///     coroutine.create(function (sum)
454    ///         for i = 1,10 do
455    ///             sum = sum + i
456    ///             coroutine.yield(sum)
457    ///         end
458    ///         return sum
459    ///     end)
460    /// "#).eval()?;
461    ///
462    /// let mut stream = thread.into_async::<i64>(1)?;
463    /// let mut sum = 0;
464    /// while let Some(n) = stream.try_next().await? {
465    ///     sum += n;
466    /// }
467    ///
468    /// assert_eq!(sum, 286);
469    ///
470    /// # Ok(())
471    /// # }
472    /// ```
473    #[cfg(feature = "async")]
474    #[cfg_attr(docsrs, doc(cfg(feature = "async")))]
475    pub fn into_async<R>(self, args: impl IntoLuaMulti) -> Result<AsyncThread<R>>
476    where
477        R: FromLuaMulti,
478    {
479        let lua = self.0.lua.lock();
480        if !self.status_inner(&lua).is_resumable() {
481            return Err(Error::CoroutineUnresumable);
482        }
483
484        let state = lua.state();
485        let thread_state = self.state();
486        unsafe {
487            let _sg = StackGuard::new(state);
488
489            let nargs = args.push_into_stack_multi(&lua)?;
490            if nargs > 0 {
491                check_stack(thread_state, nargs)?;
492                ffi::lua_xmove(state, thread_state, nargs);
493            }
494
495            Ok(AsyncThread {
496                thread: self,
497                ret: PhantomData,
498                recycle: false,
499            })
500        }
501    }
502
503    /// Enables sandbox mode on this thread.
504    ///
505    /// Under the hood replaces the global environment table with a new table,
506    /// that performs writes locally and proxies reads to caller's global environment.
507    ///
508    /// This mode ideally should be used together with the global sandbox mode [`Lua::sandbox`].
509    ///
510    /// Please note that Luau links environment table with chunk when loading it into Lua state.
511    /// Therefore you need to load chunks into a thread to link with the thread environment.
512    ///
513    /// [`Lua::sandbox`]: crate::Lua::sandbox
514    ///
515    /// # Examples
516    ///
517    /// ```
518    /// # use mlua::{Lua, Result};
519    /// # #[cfg(feature = "luau")]
520    /// # fn main() -> Result<()> {
521    /// let lua = Lua::new();
522    /// let thread = lua.create_thread(lua.create_function(|lua2, ()| {
523    ///     lua2.load("var = 123").exec()?;
524    ///     assert_eq!(lua2.globals().get::<u32>("var")?, 123);
525    ///     Ok(())
526    /// })?)?;
527    /// thread.sandbox()?;
528    /// thread.resume::<()>(())?;
529    ///
530    /// // The global environment should be unchanged
531    /// assert_eq!(lua.globals().get::<Option<u32>>("var")?, None);
532    /// # Ok(())
533    /// # }
534    ///
535    /// # #[cfg(not(feature = "luau"))]
536    /// # fn main() { }
537    /// ```
538    #[cfg(any(feature = "luau", doc))]
539    #[cfg_attr(docsrs, doc(cfg(feature = "luau")))]
540    pub fn sandbox(&self) -> Result<()> {
541        let lua = self.0.lua.lock();
542        let state = lua.state();
543        let thread_state = self.state();
544        unsafe {
545            check_stack(thread_state, 3)?;
546            check_stack(state, 3)?;
547            protect_lua!(state, 0, 0, |_| ffi::luaL_sandboxthread(thread_state))
548        }
549    }
550
551    /// Converts this thread to a generic C pointer.
552    ///
553    /// There is no way to convert the pointer back to its original value.
554    ///
555    /// Typically this function is used only for hashing and debug information.
556    #[inline]
557    pub fn to_pointer(&self) -> *const c_void {
558        self.0.to_pointer()
559    }
560}
561
562impl fmt::Debug for Thread {
563    fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result {
564        fmt.debug_tuple("Thread").field(&self.0).finish()
565    }
566}
567
568impl LuaType for Thread {
569    const TYPE_ID: c_int = ffi::LUA_TTHREAD;
570}
571
572#[cfg(feature = "async")]
573impl<R> AsyncThread<R> {
574    #[inline(always)]
575    pub(crate) fn set_recyclable(&mut self, recyclable: bool) {
576        self.recycle = recyclable;
577    }
578}
579
580#[cfg(feature = "async")]
581impl<R> Drop for AsyncThread<R> {
582    fn drop(&mut self) {
583        #[allow(clippy::collapsible_if)]
584        if self.recycle {
585            if let Some(lua) = self.thread.0.lua.try_lock() {
586                unsafe {
587                    let mut status = self.thread.status_inner(&lua);
588                    if matches!(status, ThreadStatusInner::Yielded(0)) {
589                        // The thread is dropped while yielded, resume it with the "terminate" signal
590                        ffi::lua_pushlightuserdata(self.thread.1, crate::Lua::poll_terminate().0);
591                        if let Ok((new_status, _)) = self.thread.resume_inner(&lua, 1) {
592                            // `new_status` should always be `ThreadStatusInner::Yielded(0)`
593                            status = new_status;
594                        }
595                    }
596
597                    // For Lua 5.4 this also closes all pending to-be-closed variables
598                    if self.thread.reset_inner(status).is_ok() {
599                        lua.recycle_thread(&mut self.thread);
600                    }
601                }
602            }
603        }
604    }
605}
606
607#[cfg(feature = "async")]
608impl<R: FromLuaMulti> Stream for AsyncThread<R> {
609    type Item = Result<R>;
610
611    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
612        let lua = self.thread.0.lua.lock();
613        let nargs = match self.thread.status_inner(&lua) {
614            ThreadStatusInner::New(nargs) | ThreadStatusInner::Yielded(nargs) => nargs,
615            _ => return Poll::Ready(None),
616        };
617
618        let state = lua.state();
619        let thread_state = self.thread.state();
620        unsafe {
621            let _sg = StackGuard::new(state);
622            let _thread_sg = StackGuard::with_top(thread_state, 0);
623            let _wg = WakerGuard::new(&lua, cx.waker());
624
625            let (status, nresults) = (self.thread).resume_inner(&lua, nargs)?;
626
627            if status.is_yielded() {
628                if nresults == 1 && is_poll_pending(thread_state) {
629                    return Poll::Pending;
630                }
631                // Continue polling
632                cx.waker().wake_by_ref();
633            }
634
635            check_stack(state, nresults + 1)?;
636            ffi::lua_xmove(thread_state, state, nresults);
637
638            Poll::Ready(Some(R::from_stack_multi(nresults, &lua)))
639        }
640    }
641}
642
643#[cfg(feature = "async")]
644impl<R: FromLuaMulti> Future for AsyncThread<R> {
645    type Output = Result<R>;
646
647    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
648        let lua = self.thread.0.lua.lock();
649        let nargs = match self.thread.status_inner(&lua) {
650            ThreadStatusInner::New(nargs) | ThreadStatusInner::Yielded(nargs) => nargs,
651            _ => return Poll::Ready(Err(Error::CoroutineUnresumable)),
652        };
653
654        let state = lua.state();
655        let thread_state = self.thread.state();
656        unsafe {
657            let _sg = StackGuard::new(state);
658            let _thread_sg = StackGuard::with_top(thread_state, 0);
659            let _wg = WakerGuard::new(&lua, cx.waker());
660
661            let (status, nresults) = self.thread.resume_inner(&lua, nargs)?;
662
663            if status.is_yielded() {
664                if !(nresults == 1 && is_poll_pending(thread_state)) {
665                    // Ignore values returned via yield()
666                    cx.waker().wake_by_ref();
667                }
668                return Poll::Pending;
669            }
670
671            check_stack(state, nresults + 1)?;
672            ffi::lua_xmove(thread_state, state, nresults);
673
674            Poll::Ready(R::from_stack_multi(nresults, &lua))
675        }
676    }
677}
678
679#[cfg(feature = "async")]
680#[inline(always)]
681unsafe fn is_poll_pending(state: *mut ffi::lua_State) -> bool {
682    ffi::lua_tolightuserdata(state, -1) == crate::Lua::poll_pending().0
683}
684
685#[cfg(feature = "async")]
686struct WakerGuard<'lua, 'a> {
687    lua: &'lua RawLua,
688    prev: NonNull<Waker>,
689    _phantom: PhantomData<&'a ()>,
690}
691
692#[cfg(feature = "async")]
693impl<'lua, 'a> WakerGuard<'lua, 'a> {
694    #[inline]
695    pub fn new(lua: &'lua RawLua, waker: &'a Waker) -> Result<WakerGuard<'lua, 'a>> {
696        let prev = lua.set_waker(NonNull::from(waker));
697        Ok(WakerGuard {
698            lua,
699            prev,
700            _phantom: PhantomData,
701        })
702    }
703}
704
705#[cfg(feature = "async")]
706impl Drop for WakerGuard<'_, '_> {
707    fn drop(&mut self) {
708        self.lua.set_waker(self.prev);
709    }
710}
711
712#[cfg(test)]
713mod assertions {
714    use super::*;
715
716    #[cfg(not(feature = "send"))]
717    static_assertions::assert_not_impl_any!(Thread: Send);
718    #[cfg(feature = "send")]
719    static_assertions::assert_impl_all!(Thread: Send, Sync);
720    #[cfg(all(feature = "async", not(feature = "send")))]
721    static_assertions::assert_not_impl_any!(AsyncThread<()>: Send);
722    #[cfg(all(feature = "async", feature = "send"))]
723    static_assertions::assert_impl_all!(AsyncThread<()>: Send, Sync);
724}
mlua/thread.rs

mlua/
thread.rs
