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

mlua/
thread.rs
