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/// Controls which thread lifecycle events trigger the callback.
67#[derive(Clone, Copy, Debug, Default)]
68#[non_exhaustive]
69pub struct ThreadTriggers {
70    /// Trigger the callback when a new thread is created.
71    ///
72    /// On Luau this fires for every thread creation, on other Lua versions only for threads created
73    /// via [`Lua::create_thread`](crate::Lua::create_thread).
74    pub on_create: bool,
75    /// Trigger the callback before a thread is resumed via [`Thread::resume`] (or an async resume
76    /// driven by mlua). It does not fire for a `coroutine.resume` performed inside Lua code.
77    pub on_resume: bool,
78    /// Trigger the callback after a thread yields back to a [`Thread::resume`] driven by mlua.
79    /// It does not fire for a yield consumed by a `coroutine.resume` inside Lua code.
80    pub on_yield: bool,
81}
82
83impl ThreadTriggers {
84    /// An instance of [`ThreadTriggers`] with `on_create` trigger set.
85    pub const ON_CREATE: Self = Self::new().on_create();
86
87    /// An instance of [`ThreadTriggers`] with `on_resume` trigger set.
88    pub const ON_RESUME: Self = Self::new().on_resume();
89
90    /// An instance of [`ThreadTriggers`] with `on_yield` trigger set.
91    pub const ON_YIELD: Self = Self::new().on_yield();
92
93    /// Returns a new instance of `ThreadTriggers` with all triggers disabled.
94    pub const fn new() -> Self {
95        Self {
96            on_create: false,
97            on_resume: false,
98            on_yield: false,
99        }
100    }
101
102    /// Returns an instance of `ThreadTriggers` with `on_create` trigger set.
103    #[must_use]
104    pub const fn on_create(mut self) -> Self {
105        self.on_create = true;
106        self
107    }
108
109    /// Returns an instance of `ThreadTriggers` with `on_resume` trigger set.
110    #[must_use]
111    pub const fn on_resume(mut self) -> Self {
112        self.on_resume = true;
113        self
114    }
115
116    /// Returns an instance of `ThreadTriggers` with `on_yield` trigger set.
117    #[must_use]
118    pub const fn on_yield(mut self) -> Self {
119        self.on_yield = true;
120        self
121    }
122}
123
124impl std::ops::BitOr for ThreadTriggers {
125    type Output = Self;
126
127    fn bitor(mut self, rhs: Self) -> Self::Output {
128        self.on_create |= rhs.on_create;
129        self.on_resume |= rhs.on_resume;
130        self.on_yield |= rhs.on_yield;
131        self
132    }
133}
134
135impl std::ops::BitOrAssign for ThreadTriggers {
136    fn bitor_assign(&mut self, rhs: Self) {
137        *self = *self | rhs;
138    }
139}
140
141/// Represents a thread (coroutine) event.
142#[derive(Debug, Clone)]
143#[non_exhaustive]
144pub enum ThreadEvent {
145    /// A new thread was created.
146    Create(Thread),
147    /// A thread is about to be resumed via [`Thread::resume`].
148    Resume(Thread),
149    /// A thread has just yielded.
150    Yield(Thread),
151}
152
153/// Status of a Lua thread (coroutine).
154#[derive(Debug, Copy, Clone, Eq, PartialEq)]
155pub enum ThreadStatus {
156    /// The thread was just created or is suspended (yielded).
157    ///
158    /// If a thread is in this state, it can be resumed by calling [`Thread::resume`].
159    Resumable,
160    /// The thread is currently running.
161    Running,
162    /// The thread is active but not running.
163    ///
164    /// This is the case when the thread has resumed another thread (which has not yet
165    /// returned or yielded).
166    Normal,
167    /// The thread has finished executing.
168    Finished,
169    /// The thread has raised a Lua error during execution.
170    Error,
171}
172
173/// Internal representation of a Lua thread status.
174///
175/// The number in `New` and `Yielded` variants is the number of arguments pushed
176/// to the thread stack.
177#[derive(Clone, Copy)]
178enum ThreadStatusInner {
179    New(c_int),
180    Running,
181    Normal,
182    Yielded(c_int),
183    Finished,
184    Error,
185}
186
187impl ThreadStatusInner {
188    #[inline(always)]
189    fn is_yielded(self) -> bool {
190        matches!(self, ThreadStatusInner::Yielded(_))
191    }
192}
193
194/// Handle to an internal Lua thread (coroutine).
195#[derive(Clone, PartialEq)]
196pub struct Thread(pub(crate) ValueRef, pub(crate) *mut ffi::lua_State);
197
198#[cfg(feature = "send")]
199unsafe impl Send for Thread {}
200#[cfg(feature = "send")]
201unsafe impl Sync for Thread {}
202
203/// Thread (coroutine) representation as an async [`Future`] or [`Stream`].
204///
205/// [`Future`]: std::future::Future
206/// [`Stream`]: futures_util::stream::Stream
207#[cfg(feature = "async")]
208#[cfg_attr(docsrs, doc(cfg(feature = "async")))]
209#[must_use = "futures do nothing unless you `.await` or poll them"]
210pub struct AsyncThread<R> {
211    thread: Thread,
212    ret: PhantomData<fn() -> R>,
213    recycle: bool,
214}
215
216pub(crate) struct ThreadEventGuard<'a> {
217    lua: &'a RawLua,
218    prev_state: *mut ffi::lua_State,
219}
220
221impl<'a> ThreadEventGuard<'a> {
222    #[inline]
223    pub(crate) unsafe fn new(lua: &'a RawLua, thread_state: *mut ffi::lua_State) -> Self {
224        let guard = ThreadEventGuard {
225            lua,
226            prev_state: lua.thread_event_state(),
227        };
228        lua.set_thread_event_state(thread_state);
229        guard
230    }
231}
232
233impl Drop for ThreadEventGuard<'_> {
234    #[inline]
235    fn drop(&mut self) {
236        unsafe { self.lua.set_thread_event_state(self.prev_state) };
237    }
238}
239
240#[inline]
241fn check_thread_reentrancy(thread_state: *mut ffi::lua_State, lua: &RawLua) -> Result<()> {
242    if thread_state == unsafe { lua.thread_event_state() } {
243        let err = "cannot resume or reset a thread from within its own event callback";
244        return Err(Error::runtime(err));
245    }
246    Ok(())
247}
248
249#[inline]
250unsafe fn exec_thread_event(
251    lua: &RawLua,
252    enabled: bool,
253    thread_state: *mut ffi::lua_State,
254    event: impl FnOnce() -> ThreadEvent,
255) -> Result<bool> {
256    if enabled
257        && lua.thread_event_state().is_null()
258        && let Some(cb) = lua.thread_event_callback()
259    {
260        let _guard = ThreadEventGuard::new(lua, thread_state);
261        cb(lua.lua(), event())?;
262        return Ok(true);
263    }
264    Ok(false)
265}
266
267impl Thread {
268    /// Returns the raw pointer to the Lua state that this thread is associated with.
269    ///
270    /// The pointer is valid only while this [`Thread`] is alive.
271    #[inline(always)]
272    pub fn state(&self) -> *mut ffi::lua_State {
273        self.1
274    }
275
276    /// Resumes execution of this thread.
277    ///
278    /// Equivalent to [`coroutine.resume`].
279    ///
280    /// Passes `args` as arguments to the thread. If the coroutine has called [`coroutine.yield`],
281    /// it will return these arguments. Otherwise, the coroutine wasn't yet started, so the
282    /// arguments are passed to its main function.
283    ///
284    /// If the thread is no longer resumable (meaning it has finished execution or encountered an
285    /// error), this will return [`Error::CoroutineUnresumable`], otherwise will return `Ok` as
286    /// follows:
287    ///
288    /// If the thread calls [`coroutine.yield`], returns the values passed to `yield`. If the thread
289    /// `return`s values from its main function, returns those.
290    ///
291    /// # Examples
292    ///
293    /// ```
294    /// # use mlua::{Error, Lua, Result, Thread};
295    /// # fn main() -> Result<()> {
296    /// # let lua = Lua::new();
297    /// let thread: Thread = lua.load(r#"
298    ///     coroutine.create(function(arg)
299    ///         assert(arg == 42)
300    ///         local yieldarg = coroutine.yield(123)
301    ///         assert(yieldarg == 43)
302    ///         return 987
303    ///     end)
304    /// "#).eval()?;
305    ///
306    /// assert_eq!(thread.resume::<u32>(42)?, 123);
307    /// assert_eq!(thread.resume::<u32>(43)?, 987);
308    ///
309    /// // The coroutine has now returned, so `resume` will fail
310    /// match thread.resume::<u32>(()) {
311    ///     Err(Error::CoroutineUnresumable) => {},
312    ///     unexpected => panic!("unexpected result {:?}", unexpected),
313    /// }
314    /// # Ok(())
315    /// # }
316    /// ```
317    ///
318    /// [`coroutine.resume`]: https://www.lua.org/manual/5.4/manual.html#pdf-coroutine.resume
319    /// [`coroutine.yield`]: https://www.lua.org/manual/5.4/manual.html#pdf-coroutine.yield
320    pub fn resume<R>(&self, args: impl IntoLuaMulti) -> Result<R>
321    where
322        R: FromLuaMulti,
323    {
324        let lua = self.0.lua.lock();
325        check_thread_reentrancy(self.state(), &lua)?;
326        let (mut pushed_nargs, mut hook_yielded) = self.resumable_state(&lua)?;
327
328        let state = lua.state();
329        let thread_state = self.state();
330        unsafe {
331            let _sg = StackGuard::new(state);
332
333            // If the resume callback runs, it may touch this thread, so re-read the argument count
334            let on_resume = lua.thread_event_triggers().on_resume;
335            if exec_thread_event(&lua, on_resume, thread_state, || {
336                ThreadEvent::Resume(self.clone())
337            })? {
338                (pushed_nargs, hook_yielded) = self.resumable_state(&lua)?;
339            }
340
341            if !hook_yielded {
342                let nargs = args.push_into_stack_multi(&lua)?;
343                if nargs > 0 {
344                    check_stack(thread_state, nargs)?;
345                    ffi::lua_xmove(state, thread_state, nargs);
346                    pushed_nargs += nargs;
347                }
348            }
349
350            let mut thread_sg = StackGuard::with_top(thread_state, 0);
351            let (status, nresults) = self.resume_inner(&lua, pushed_nargs)?;
352            if status.is_yielded() && self.is_hook_yielded(&lua) {
353                debug_assert_eq!(nresults, 0);
354                thread_sg.keep(ffi::lua_gettop(thread_state));
355            }
356
357            check_stack(state, nresults + 1)?;
358            ffi::lua_xmove(thread_state, state, nresults);
359
360            // Exec thread yield callback
361            let on_yield = lua.thread_event_triggers().on_yield && status.is_yielded();
362            exec_thread_event(&lua, on_yield, thread_state, || ThreadEvent::Yield(self.clone()))?;
363
364            R::from_stack_multi(nresults, &lua)
365        }
366    }
367
368    /// Resumes execution of this thread, immediately raising an error.
369    ///
370    /// This is a Luau specific extension.
371    #[cfg(any(feature = "luau", doc))]
372    #[cfg_attr(docsrs, doc(cfg(feature = "luau")))]
373    pub fn resume_error<R>(&self, error: impl crate::IntoLua) -> Result<R>
374    where
375        R: FromLuaMulti,
376    {
377        let lua = self.0.lua.lock();
378        check_thread_reentrancy(self.state(), &lua)?;
379        match self.status_inner(&lua) {
380            ThreadStatusInner::New(_) | ThreadStatusInner::Yielded(_) => {}
381            _ => return Err(Error::CoroutineUnresumable),
382        };
383
384        let state = lua.state();
385        let thread_state = self.state();
386        unsafe {
387            let _sg = StackGuard::new(state);
388
389            // Exec thread resume callback
390            let on_resume = lua.thread_event_triggers().on_resume;
391            exec_thread_event(&lua, on_resume, thread_state, || {
392                ThreadEvent::Resume(self.clone())
393            })?;
394
395            check_stack(state, 1)?;
396            error.push_into_stack(&lua)?;
397            check_stack(thread_state, 1)?;
398            ffi::lua_xmove(state, thread_state, 1);
399
400            let _thread_sg = StackGuard::with_top(thread_state, 0);
401            let (status, nresults) = self.resume_inner(&lua, ffi::LUA_RESUMEERROR)?;
402
403            check_stack(state, nresults + 1)?;
404            ffi::lua_xmove(thread_state, state, nresults);
405
406            // Exec thread yield callback
407            let on_yield = lua.thread_event_triggers().on_yield && status.is_yielded();
408            exec_thread_event(&lua, on_yield, thread_state, || ThreadEvent::Yield(self.clone()))?;
409
410            R::from_stack_multi(nresults, &lua)
411        }
412    }
413
414    /// Resumes execution of this thread.
415    ///
416    /// It's similar to `resume()` but leaves `nresults` values on the thread stack.
417    unsafe fn resume_inner(&self, lua: &RawLua, nargs: c_int) -> Result<(ThreadStatusInner, c_int)> {
418        let state = lua.state();
419        let thread_state = self.state();
420        let mut nresults = 0;
421        #[cfg(not(feature = "luau"))]
422        let ret = ffi::lua_resume(thread_state, state, nargs, &mut nresults as *mut c_int);
423        #[cfg(feature = "luau")]
424        let ret = ffi::lua_resumex(thread_state, state, nargs, &mut nresults as *mut c_int);
425        match ret {
426            ffi::LUA_OK => Ok((ThreadStatusInner::Finished, nresults)),
427            ffi::LUA_YIELD => Ok((ThreadStatusInner::Yielded(0), nresults)),
428            ffi::LUA_ERRMEM => {
429                // Don't call error handler for memory errors
430                Err(pop_error(thread_state, ret))
431            }
432            _ => {
433                check_stack(state, 3)?;
434                protect_lua!(state, 0, 1, |state| error_traceback_thread(state, thread_state))?;
435                Err(pop_error(state, ret))
436            }
437        }
438    }
439
440    /// Gets the status of the thread.
441    pub fn status(&self) -> ThreadStatus {
442        match self.status_inner(&self.0.lua.lock()) {
443            ThreadStatusInner::New(_) | ThreadStatusInner::Yielded(_) => ThreadStatus::Resumable,
444            ThreadStatusInner::Running => ThreadStatus::Running,
445            ThreadStatusInner::Normal => ThreadStatus::Normal,
446            ThreadStatusInner::Finished => ThreadStatus::Finished,
447            ThreadStatusInner::Error => ThreadStatus::Error,
448        }
449    }
450
451    /// Gets the status of the thread (internal implementation).
452    fn status_inner(&self, lua: &RawLua) -> ThreadStatusInner {
453        let thread_state = self.state();
454        if thread_state == lua.state() {
455            // The thread is currently running
456            return ThreadStatusInner::Running;
457        }
458        let status = unsafe { ffi::lua_status(thread_state) };
459        let top = unsafe { ffi::lua_gettop(thread_state) };
460        match status {
461            ffi::LUA_YIELD => ThreadStatusInner::Yielded(top),
462            ffi::LUA_OK => {
463                // Active call frames mean this thread has resumed another (still-running) thread.
464                // Without frames it's new or finished.
465                let mut ar = const { unsafe { std::mem::zeroed::<ffi::lua_Debug>() } };
466                #[cfg(not(feature = "luau"))]
467                let has_frames = unsafe { ffi::lua_getstack(thread_state, 0, &mut ar) != 0 };
468                #[cfg(feature = "luau")]
469                let has_frames = unsafe { ffi::lua_getinfo(thread_state, 0, cstr!(""), &mut ar) != 0 };
470                if has_frames {
471                    ThreadStatusInner::Normal
472                } else if top > 0 {
473                    ThreadStatusInner::New(top - 1)
474                } else {
475                    ThreadStatusInner::Finished
476                }
477            }
478            _ => ThreadStatusInner::Error,
479        }
480    }
481
482    /// Returns the pending argument count and whether the thread was interrupted by a hook.
483    #[inline]
484    fn resumable_state(&self, lua: &RawLua) -> Result<(c_int, bool)> {
485        match self.status_inner(lua) {
486            ThreadStatusInner::New(nargs) => Ok((nargs, false)),
487            ThreadStatusInner::Yielded(nargs) => {
488                let hook_yielded = self.is_hook_yielded(lua);
489                Ok((if hook_yielded { 0 } else { nargs }, hook_yielded))
490            }
491            _ => Err(Error::CoroutineUnresumable),
492        }
493    }
494
495    /// Distinguishes a hook interruption from a normal yield.
496    fn is_hook_yielded(&self, lua: &RawLua) -> bool {
497        unsafe { lua.is_hook_yielded(self.state()) }
498    }
499
500    /// Returns `true` if this thread is resumable (meaning it can be resumed by calling
501    /// [`Thread::resume`]).
502    #[inline(always)]
503    pub fn is_resumable(&self) -> bool {
504        self.status() == ThreadStatus::Resumable
505    }
506
507    /// Returns `true` if this thread is currently running.
508    #[inline(always)]
509    pub fn is_running(&self) -> bool {
510        self.status() == ThreadStatus::Running
511    }
512
513    /// Returns `true` if this thread is active but not running.
514    ///
515    /// This is the case when the thread has resumed another thread that has not yet returned
516    /// or yielded.
517    #[inline(always)]
518    pub fn is_normal(&self) -> bool {
519        self.status() == ThreadStatus::Normal
520    }
521
522    /// Returns `true` if this thread has finished executing.
523    #[inline(always)]
524    pub fn is_finished(&self) -> bool {
525        self.status() == ThreadStatus::Finished
526    }
527
528    /// Returns `true` if this thread has raised a Lua error during execution.
529    #[inline(always)]
530    pub fn is_error(&self) -> bool {
531        self.status() == ThreadStatus::Error
532    }
533
534    /// Sets a hook function that will periodically be called as Lua code executes.
535    ///
536    /// This function is similar to [`Lua::set_hook`] except that it sets the hook for the thread.
537    /// You can have multiple hooks for different threads.
538    ///
539    /// To remove a hook call [`Thread::remove_hook`].
540    ///
541    /// [`Lua::set_hook`]: crate::Lua::set_hook
542    #[cfg(not(feature = "luau"))]
543    #[cfg_attr(docsrs, doc(cfg(not(feature = "luau"))))]
544    pub fn set_hook<F>(&self, triggers: HookTriggers, callback: F) -> Result<()>
545    where
546        F: Fn(&crate::Lua, &Debug) -> Result<crate::VmState> + crate::MaybeSend + 'static,
547    {
548        let lua = self.0.lua.lock();
549        unsafe {
550            lua.set_thread_hook(
551                self.state(),
552                HookKind::Thread(triggers, crate::types::XRc::new(callback)),
553            )
554        }
555    }
556
557    /// Removes any hook function from this thread.
558    #[cfg(not(feature = "luau"))]
559    #[cfg_attr(docsrs, doc(cfg(not(feature = "luau"))))]
560    pub fn remove_hook(&self) {
561        let lua = self.0.lua.lock();
562        unsafe {
563            lua.remove_thread_hook(self.state());
564        }
565    }
566
567    /// Resets a thread
568    ///
569    /// In [Lua 5.4]: cleans its call stack and closes all pending to-be-closed variables.
570    /// Returns an error in case of either the original error that stopped the thread or errors
571    /// in closing methods.
572    ///
573    /// In Luau: resets to the initial state of a newly created Lua thread.
574    /// Lua threads in arbitrary states (like yielded or errored) can be reset properly.
575    ///
576    /// Other Lua versions can reset only new or finished threads.
577    ///
578    /// Sets a Lua function for the thread afterwards.
579    ///
580    /// [Lua 5.4]: https://www.lua.org/manual/5.4/manual.html#lua_closethread
581    pub fn reset(&self, func: Function) -> Result<()> {
582        let lua = self.0.lua.lock();
583        assert!(
584            lua.weak() == &func.0.lua,
585            "Lua instance passed Value created from a different main Lua state"
586        );
587        check_thread_reentrancy(self.state(), &lua)?;
588        let thread_state = self.state();
589        unsafe {
590            let status = self.status_inner(&lua);
591            self.reset_inner(status)?;
592
593            // Push function to the top of the thread stack
594            ffi::lua_xpush(lua.ref_thread(), thread_state, func.0.index);
595
596            #[cfg(feature = "luau")]
597            {
598                // Inherit `LUA_GLOBALSINDEX` from the main thread
599                ffi::lua_xpush(lua.main_state(), thread_state, ffi::LUA_GLOBALSINDEX);
600                ffi::lua_replace(thread_state, ffi::LUA_GLOBALSINDEX);
601            }
602
603            Ok(())
604        }
605    }
606
607    unsafe fn reset_inner(&self, status: ThreadStatusInner) -> Result<()> {
608        match status {
609            ThreadStatusInner::New(_) => {
610                // The thread is new, so we can just set the top to 0
611                ffi::lua_settop(self.state(), 0);
612                Ok(())
613            }
614            ThreadStatusInner::Running => Err(Error::runtime("cannot reset a running thread")),
615            ThreadStatusInner::Normal => Err(Error::runtime("cannot reset a normal thread")),
616            ThreadStatusInner::Finished => Ok(()),
617            #[cfg(not(any(feature = "lua55", feature = "lua54", feature = "luau")))]
618            ThreadStatusInner::Yielded(_) | ThreadStatusInner::Error => {
619                Err(Error::runtime("cannot reset non-finished thread"))
620            }
621            #[cfg(any(feature = "lua55", feature = "lua54", feature = "luau"))]
622            ThreadStatusInner::Yielded(_) | ThreadStatusInner::Error => {
623                let thread_state = self.state();
624
625                #[cfg(all(feature = "lua54", not(feature = "vendored")))]
626                let status = ffi::lua_resetthread(thread_state);
627                #[cfg(any(feature = "lua55", all(feature = "lua54", feature = "vendored")))]
628                let status = {
629                    let lua = self.0.lua.lock();
630                    ffi::lua_closethread(thread_state, lua.state())
631                };
632                #[cfg(any(feature = "lua55", feature = "lua54"))]
633                if status != ffi::LUA_OK {
634                    return Err(pop_error(thread_state, status));
635                }
636                #[cfg(feature = "luau")]
637                ffi::lua_resetthread(thread_state);
638
639                Ok(())
640            }
641        }
642    }
643
644    /// Converts [`Thread`] to an [`AsyncThread`] which implements [`Future`] and [`Stream`] traits.
645    ///
646    /// Only resumable threads can be converted to [`AsyncThread`].
647    ///
648    /// `args` are pushed to the thread stack and will be used when the thread is resumed.
649    /// The object calls [`resume`] while polling and also allow to run Rust futures
650    /// to completion using an executor.
651    ///
652    /// Using [`AsyncThread`] as a [`Stream`] allow to iterate through [`coroutine.yield`]
653    /// values whereas [`Future`] version discards that values and poll until the final
654    /// one (returned from the thread function).
655    ///
656    /// [`Future`]: std::future::Future
657    /// [`Stream`]: futures_util::stream::Stream
658    /// [`resume`]: https://www.lua.org/manual/5.4/manual.html#lua_resume
659    /// [`coroutine.yield`]: https://www.lua.org/manual/5.4/manual.html#pdf-coroutine.yield
660    ///
661    /// # Examples
662    ///
663    /// ```
664    /// # use mlua::{Lua, Result, Thread};
665    /// use futures_util::stream::TryStreamExt;
666    /// # #[tokio::main]
667    /// # async fn main() -> Result<()> {
668    /// # let lua = Lua::new();
669    /// let thread: Thread = lua.load(r#"
670    ///     coroutine.create(function (sum)
671    ///         for i = 1,10 do
672    ///             sum = sum + i
673    ///             coroutine.yield(sum)
674    ///         end
675    ///         return sum
676    ///     end)
677    /// "#).eval()?;
678    ///
679    /// let mut stream = thread.into_async::<i64>(1)?;
680    /// let mut sum = 0;
681    /// while let Some(n) = stream.try_next().await? {
682    ///     sum += n;
683    /// }
684    ///
685    /// assert_eq!(sum, 286);
686    ///
687    /// # Ok(())
688    /// # }
689    /// ```
690    #[cfg(feature = "async")]
691    #[cfg_attr(docsrs, doc(cfg(feature = "async")))]
692    pub fn into_async<R>(self, args: impl IntoLuaMulti) -> Result<AsyncThread<R>>
693    where
694        R: FromLuaMulti,
695    {
696        let lua = self.0.lua.lock();
697        check_thread_reentrancy(self.state(), &lua)?;
698        let (_, hook_yielded) = self.resumable_state(&lua)?;
699
700        let state = lua.state();
701        let thread_state = self.state();
702        unsafe {
703            let _sg = StackGuard::new(state);
704
705            if !hook_yielded {
706                let nargs = args.push_into_stack_multi(&lua)?;
707                if nargs > 0 {
708                    check_stack(thread_state, nargs)?;
709                    ffi::lua_xmove(state, thread_state, nargs);
710                }
711            }
712
713            Ok(AsyncThread {
714                thread: self,
715                ret: PhantomData,
716                recycle: false,
717            })
718        }
719    }
720
721    /// Enables sandbox mode on this thread.
722    ///
723    /// Under the hood replaces the global environment table with a new table,
724    /// that performs writes locally and proxies reads to caller's global environment.
725    ///
726    /// This mode ideally should be used together with the global sandbox mode [`Lua::sandbox`].
727    ///
728    /// Please note that Luau links environment table with chunk when loading it into Lua state.
729    /// Therefore you need to load chunks into a thread to link with the thread environment.
730    ///
731    /// [`Lua::sandbox`]: crate::Lua::sandbox
732    ///
733    /// # Examples
734    ///
735    /// ```
736    /// # use mlua::{Lua, Result};
737    /// # #[cfg(feature = "luau")]
738    /// # fn main() -> Result<()> {
739    /// let lua = Lua::new();
740    /// let thread = lua.create_thread(lua.create_function(|lua2, ()| {
741    ///     lua2.load("var = 123").exec()?;
742    ///     assert_eq!(lua2.globals().get::<u32>("var")?, 123);
743    ///     Ok(())
744    /// })?)?;
745    /// thread.sandbox()?;
746    /// thread.resume::<()>(())?;
747    ///
748    /// // The global environment should be unchanged
749    /// assert_eq!(lua.globals().get::<Option<u32>>("var")?, None);
750    /// # Ok(())
751    /// # }
752    ///
753    /// # #[cfg(not(feature = "luau"))]
754    /// # fn main() { }
755    /// ```
756    #[cfg(any(feature = "luau", doc))]
757    #[cfg_attr(docsrs, doc(cfg(feature = "luau")))]
758    pub fn sandbox(&self) -> Result<()> {
759        let lua = self.0.lua.lock();
760        let state = lua.state();
761        let thread_state = self.state();
762        unsafe {
763            check_stack(thread_state, 3)?;
764            check_stack(state, 3)?;
765            protect_lua!(state, 0, 0, |_| ffi::luaL_sandboxthread(thread_state))
766        }
767    }
768
769    /// Converts this thread to a generic C pointer.
770    ///
771    /// There is no way to convert the pointer back to its original value.
772    ///
773    /// Typically this function is used only for hashing and debug information.
774    #[inline]
775    pub fn to_pointer(&self) -> *const c_void {
776        self.0.to_pointer()
777    }
778}
779
780impl fmt::Debug for Thread {
781    fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result {
782        fmt.debug_tuple("Thread").field(&self.0).finish()
783    }
784}
785
786impl LuaType for Thread {
787    const TYPE_ID: c_int = ffi::LUA_TTHREAD;
788}
789
790#[cfg(feature = "async")]
791impl<R> AsyncThread<R> {
792    #[inline(always)]
793    pub(crate) fn set_recyclable(&mut self, recyclable: bool) {
794        self.recycle = recyclable;
795    }
796
797    #[inline(always)]
798    pub(crate) fn thread(&self) -> &Thread {
799        &self.thread
800    }
801}
802
803#[cfg(feature = "async")]
804impl<R> Drop for AsyncThread<R> {
805    fn drop(&mut self) {
806        if self.recycle
807            && let Some(lua) = self.thread.0.lua.try_lock()
808        {
809            unsafe {
810                #[cfg(feature = "luau")]
811                if lua.is_running_gc() {
812                    lua.update_thread_ownership(&self.thread, None);
813                    return;
814                }
815
816                let mut status = self.thread.status_inner(&lua);
817                if matches!(status, ThreadStatusInner::Yielded(0)) && !self.thread.is_hook_yielded(&lua) {
818                    // The thread is dropped while yielded, resume it with the "terminate" signal
819                    ffi::lua_pushlightuserdata(self.thread.1, crate::Lua::poll_terminate().0);
820                    if let Ok((new_status, _)) = self.thread.resume_inner(&lua, 1) {
821                        // `new_status` should always be `ThreadStatusInner::Yielded(0)`
822                        status = new_status;
823                    }
824                }
825
826                // For Lua 5.4 this also closes all pending to-be-closed variables
827                if self.thread.reset_inner(status).is_ok() {
828                    lua.recycle_thread(&mut self.thread);
829                }
830                lua.update_thread_ownership(&self.thread, None);
831            }
832        }
833    }
834}
835
836#[cfg(feature = "async")]
837impl<R: FromLuaMulti> Stream for AsyncThread<R> {
838    type Item = Result<R>;
839
840    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
841        let lua = self.thread.0.lua.lock();
842        check_thread_reentrancy(self.thread.state(), &lua)?;
843        let mut nargs = match self.thread.resumable_state(&lua) {
844            Ok((nargs, _)) => nargs,
845            Err(_) => return Poll::Ready(None),
846        };
847
848        let state = lua.state();
849        let thread_state = self.thread.state();
850        unsafe {
851            let _sg = StackGuard::new(state);
852            let _wg = WakerGuard::new(&lua, cx.waker());
853
854            // If the resume callback runs, it may touch this thread, so re-read the argument count
855            let on_resume = lua.thread_event_triggers().on_resume;
856            if exec_thread_event(&lua, on_resume, thread_state, || {
857                ThreadEvent::Resume(self.thread.clone())
858            })? {
859                nargs = match self.thread.resumable_state(&lua) {
860                    Ok((nargs, _)) => nargs,
861                    Err(_) => return Poll::Ready(None),
862                };
863            }
864
865            let mut thread_sg = StackGuard::with_top(thread_state, 0);
866            let (status, nresults) = (self.thread).resume_inner(&lua, nargs)?;
867            let hook_yielded = status.is_yielded() && self.thread.is_hook_yielded(&lua);
868            if hook_yielded {
869                debug_assert_eq!(nresults, 0);
870                thread_sg.keep(ffi::lua_gettop(thread_state));
871            }
872
873            if status.is_yielded() && !hook_yielded && nresults == 1 && is_poll_pending(thread_state) {
874                // Exec thread yield callback
875                let on_yield = lua.thread_event_triggers().on_yield;
876                exec_thread_event(&lua, on_yield, thread_state, || {
877                    ThreadEvent::Yield(self.thread.clone())
878                })?;
879                return Poll::Pending;
880            }
881
882            check_stack(state, nresults + 1)?;
883            ffi::lua_xmove(thread_state, state, nresults);
884
885            if status.is_yielded() {
886                let on_yield = lua.thread_event_triggers().on_yield;
887                exec_thread_event(&lua, on_yield, thread_state, || {
888                    ThreadEvent::Yield(self.thread.clone())
889                })?;
890                // Continue polling
891                cx.waker().wake_by_ref();
892            }
893
894            Poll::Ready(Some(R::from_stack_multi(nresults, &lua)))
895        }
896    }
897}
898
899#[cfg(feature = "async")]
900impl<R: FromLuaMulti> Future for AsyncThread<R> {
901    type Output = Result<R>;
902
903    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
904        let lua = self.thread.0.lua.lock();
905        check_thread_reentrancy(self.thread.state(), &lua)?;
906        let (mut nargs, _) = self.thread.resumable_state(&lua)?;
907
908        let state = lua.state();
909        let thread_state = self.thread.state();
910        unsafe {
911            let _sg = StackGuard::new(state);
912            let _wg = WakerGuard::new(&lua, cx.waker());
913
914            // If the resume callback runs, it may touch this thread, so re-read the argument count
915            let on_resume = lua.thread_event_triggers().on_resume;
916            if exec_thread_event(&lua, on_resume, thread_state, || {
917                ThreadEvent::Resume(self.thread.clone())
918            })? {
919                (nargs, _) = self.thread.resumable_state(&lua)?;
920            }
921
922            let mut thread_sg = StackGuard::with_top(thread_state, 0);
923            let (status, nresults) = self.thread.resume_inner(&lua, nargs)?;
924            let hook_yielded = status.is_yielded() && self.thread.is_hook_yielded(&lua);
925            if hook_yielded {
926                debug_assert_eq!(nresults, 0);
927                thread_sg.keep(ffi::lua_gettop(thread_state));
928            }
929
930            if status.is_yielded() {
931                let pending = !hook_yielded && nresults == 1 && is_poll_pending(thread_state);
932
933                // Exec thread yield callback
934                let on_yield = lua.thread_event_triggers().on_yield;
935                exec_thread_event(&lua, on_yield, thread_state, || {
936                    ThreadEvent::Yield(self.thread.clone())
937                })?;
938
939                if !pending {
940                    // Ignore values returned via yield()
941                    cx.waker().wake_by_ref();
942                }
943                return Poll::Pending;
944            }
945
946            check_stack(state, nresults + 1)?;
947            ffi::lua_xmove(thread_state, state, nresults);
948
949            Poll::Ready(R::from_stack_multi(nresults, &lua))
950        }
951    }
952}
953
954#[cfg(feature = "async")]
955#[inline(always)]
956unsafe fn is_poll_pending(state: *mut ffi::lua_State) -> bool {
957    ffi::lua_tolightuserdata(state, -1) == crate::Lua::poll_pending().0
958}
959
960#[cfg(feature = "async")]
961struct WakerGuard<'lua, 'a> {
962    lua: &'lua RawLua,
963    prev: NonNull<Waker>,
964    _phantom: PhantomData<&'a ()>,
965}
966
967#[cfg(feature = "async")]
968impl<'lua, 'a> WakerGuard<'lua, 'a> {
969    #[inline]
970    pub fn new(lua: &'lua RawLua, waker: &'a Waker) -> Result<WakerGuard<'lua, 'a>> {
971        let prev = lua.set_waker(NonNull::from(waker));
972        Ok(WakerGuard {
973            lua,
974            prev,
975            _phantom: PhantomData,
976        })
977    }
978}
979
980#[cfg(feature = "async")]
981impl Drop for WakerGuard<'_, '_> {
982    fn drop(&mut self) {
983        self.lua.set_waker(self.prev);
984    }
985}
986
987#[cfg(test)]
988mod assertions {
989    use super::*;
990
991    #[cfg(not(feature = "send"))]
992    static_assertions::assert_not_impl_any!(Thread: Send);
993    #[cfg(feature = "send")]
994    static_assertions::assert_impl_all!(Thread: Send, Sync);
995    #[cfg(all(feature = "async", not(feature = "send")))]
996    static_assertions::assert_not_impl_any!(AsyncThread<()>: Send);
997    #[cfg(all(feature = "async", feature = "send"))]
998    static_assertions::assert_impl_all!(AsyncThread<()>: Send, Sync);
999}
mlua/thread.rs

mlua/
thread.rs
