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

mlua/
thread.rs
