//! _asyncio module - provides native asyncio support

//!

//! This module provides native implementations of Future and Task classes,

pub(crate) use _asyncio::module_def;

#[pymodule]

pub(crate) mod _asyncio {

use crate::common::wtf8::{Wtf8Buf, wtf8_concat};

use crate::{

common::lock::PyRwLock,

vm::{

AsObject, Py, PyObject, PyObjectRef, PyPayload, PyRef, PyResult, VirtualMachine,

builtins::{

PyBaseException, PyBaseExceptionRef, PyDict, PyDictRef, PyGenericAlias, PyList,

PyListRef, PyModule, PySet, PyTuple, PyType, PyTypeRef,

},

extend_module,

function::{FuncArgs, KwArgs, OptionalArg, OptionalOption, PySetterValue},

protocol::PyIterReturn,

recursion::ReprGuard,

types::{

Callable, Constructor, Destructor, Initializer, IterNext, Iterable, Representable,

SelfIter,

},

warn,

},

};

use core::sync::atomic::{AtomicBool, AtomicI32, AtomicU64, Ordering};

use crossbeam_utils::atomic::AtomicCell;

pub(crate) fn module_exec(vm: &VirtualMachine, module: &Py<PyModule>) -> PyResult<()> {

__module_exec(vm, module);

// Initialize module-level state

let weakref_module = vm.import("weakref", 0)?;

let weak_set_class = vm

.get_attribute_opt(weakref_module, vm.ctx.intern_str("WeakSet"))?

.ok_or_else(|| vm.new_attribute_error("WeakSet not found"))?;

let scheduled_tasks = weak_set_class.call((), vm)?;

let eager_tasks = PySet::default().into_ref(&vm.ctx);

let current_tasks = PyDict::default().into_ref(&vm.ctx);

extend_module!(vm, module, {

"_scheduled_tasks" => scheduled_tasks,

"_eager_tasks" => eager_tasks,

"_current_tasks" => current_tasks,

});

// Register fork handler to clear task state in child process

#[cfg(unix)]

{

let on_fork = vm

.get_attribute_opt(module.to_owned().into(), vm.ctx.intern_str("_on_fork"))?

.expect("_on_fork not found in _asyncio module");

vm.state.after_forkers_child.lock().push(on_fork);

}

Ok(())

}

#[derive(FromArgs)]

struct AddDoneCallbackArgs {

#[pyarg(positional)]

func: PyObjectRef,

#[pyarg(named, optional)]

context: OptionalOption<PyObjectRef>,

}

#[derive(FromArgs)]

struct CancelArgs {

#[pyarg(any, optional)]

msg: OptionalOption<PyObjectRef>,

}

#[derive(FromArgs)]

struct LoopArg {

#[pyarg(any, name = "loop", optional)]

loop_: OptionalOption<PyObjectRef>,

}

#[derive(FromArgs)]

struct GetStackArgs {

#[pyarg(named, optional)]

limit: OptionalOption<PyObjectRef>,

}

#[derive(FromArgs)]

struct PrintStackArgs {

#[pyarg(named, optional)]

limit: OptionalOption<PyObjectRef>,

#[pyarg(named, optional)]

file: OptionalOption<PyObjectRef>,

}

#[derive(Clone, Copy, PartialEq, Eq, Debug)]

enum FutureState {

Pending,

Cancelled,

Finished,

}

impl FutureState {

fn as_str(&self) -> &'static str {

match self {

FutureState::Pending => "PENDING",

FutureState::Cancelled => "CANCELLED",

FutureState::Finished => "FINISHED",

}

}

}

/// asyncio.Future implementation

#[pyattr]

#[pyclass(name = "Future", module = "_asyncio", traverse)]

#[derive(Debug, PyPayload)]

#[repr(C)] // Required for inheritance - ensures base field is at offset 0 in subclasses

struct PyFuture {

fut_loop: PyRwLock<Option<PyObjectRef>>,

fut_callback0: PyRwLock<Option<PyObjectRef>>,

fut_context0: PyRwLock<Option<PyObjectRef>>,

fut_callbacks: PyRwLock<Option<PyObjectRef>>,

fut_exception: PyRwLock<Option<PyObjectRef>>,

fut_exception_tb: PyRwLock<Option<PyObjectRef>>,

fut_result: PyRwLock<Option<PyObjectRef>>,

fut_source_tb: PyRwLock<Option<PyObjectRef>>,

fut_cancel_msg: PyRwLock<Option<PyObjectRef>>,

fut_cancelled_exc: PyRwLock<Option<PyObjectRef>>,

fut_awaited_by: PyRwLock<Option<PyObjectRef>>,

#[pytraverse(skip)]

fut_state: AtomicCell<FutureState>,

#[pytraverse(skip)]

fut_awaited_by_is_set: AtomicBool,

#[pytraverse(skip)]

fut_log_tb: AtomicBool,

#[pytraverse(skip)]

fut_blocking: AtomicBool,

}

impl Constructor for PyFuture {

type Args = FuncArgs;

fn py_new(_cls: &Py<PyType>, _args: Self::Args, _vm: &VirtualMachine) -> PyResult<Self> {

Ok(PyFuture::new_empty())

}

}

impl Initializer for PyFuture {

type Args = FuncArgs;

fn init(zelf: PyRef<Self>, args: Self::Args, vm: &VirtualMachine) -> PyResult<()> {

// Future does not accept positional arguments

if !args.args.is_empty() {

return Err(vm.new_type_error("Future() takes no positional arguments"));

}

// Extract only 'loop' keyword argument

let loop_ = args.kwargs.get("loop").cloned();

PyFuture::py_init(&zelf, loop_, vm)

}

}

#[pyclass(

flags(BASETYPE, HAS_DICT, HAS_WEAKREF),

with(Constructor, Initializer, Destructor, Representable, Iterable)

)]

impl PyFuture {

fn new_empty() -> Self {

Self {

fut_loop: PyRwLock::new(None),

fut_callback0: PyRwLock::new(None),

fut_context0: PyRwLock::new(None),

fut_callbacks: PyRwLock::new(None),

fut_exception: PyRwLock::new(None),

fut_exception_tb: PyRwLock::new(None),

fut_result: PyRwLock::new(None),

fut_source_tb: PyRwLock::new(None),

fut_cancel_msg: PyRwLock::new(None),

fut_cancelled_exc: PyRwLock::new(None),

fut_awaited_by: PyRwLock::new(None),

fut_state: AtomicCell::new(FutureState::Pending),

fut_awaited_by_is_set: AtomicBool::new(false),

fut_log_tb: AtomicBool::new(false),

fut_blocking: AtomicBool::new(false),

}

}

fn py_init(

zelf: &PyRef<Self>,

loop_: Option<PyObjectRef>,

vm: &VirtualMachine,

) -> PyResult<()> {

// Get the event loop

let loop_obj = match loop_ {

Some(l) if !vm.is_none(&l) => l,

_ => get_event_loop(vm)?,

};

*zelf.fut_loop.write() = Some(loop_obj.clone());

// Check if loop has get_debug method and call it

if let Ok(Some(get_debug)) =

vm.get_attribute_opt(loop_obj.clone(), vm.ctx.intern_str("get_debug"))

&& let Ok(debug) = get_debug.call((), vm)

&& debug.try_to_bool(vm).unwrap_or(false)

{

// Get source traceback

if let Ok(tb_module) = vm.import("traceback", 0)

&& let Ok(Some(extract_stack)) =

vm.get_attribute_opt(tb_module, vm.ctx.intern_str("extract_stack"))

&& let Ok(tb) = extract_stack.call((), vm)

{

*zelf.fut_source_tb.write() = Some(tb);

}

}

Ok(())

}

#[pymethod]

fn result(&self, vm: &VirtualMachine) -> PyResult {

match self.fut_state.load() {

FutureState::Pending => Err(new_invalid_state_error(vm, "Result is not ready.")),

FutureState::Cancelled => {

let exc = self.make_cancelled_error_impl(vm);

Err(exc)

}

FutureState::Finished => {

self.fut_log_tb.store(false, Ordering::Relaxed);

if let Some(exc) = self.fut_exception.read().clone() {

let exc: PyBaseExceptionRef = exc.downcast().unwrap();

// Restore the original traceback to prevent traceback accumulation

if let Some(tb) = self.fut_exception_tb.read().clone() {

let _ = exc.set___traceback__(tb, vm);

}

Err(exc)

} else {

Ok(self

.fut_result

.read()

.clone()

.unwrap_or_else(|| vm.ctx.none()))

}

}

}

}

#[pymethod]

fn exception(&self, vm: &VirtualMachine) -> PyResult {

match self.fut_state.load() {

FutureState::Pending => Err(new_invalid_state_error(vm, "Exception is not set.")),

FutureState::Cancelled => {

let exc = self.make_cancelled_error_impl(vm);

Err(exc)

}

FutureState::Finished => {

self.fut_log_tb.store(false, Ordering::Relaxed);

Ok(self

.fut_exception

.read()

.clone()

.unwrap_or_else(|| vm.ctx.none()))

}

}

}

#[pymethod]

fn set_result(zelf: PyRef<Self>, result: PyObjectRef, vm: &VirtualMachine) -> PyResult<()> {

if zelf.fut_loop.read().is_none() {

return Err(vm.new_runtime_error("Future object is not initialized."));

}

if zelf.fut_state.load() != FutureState::Pending {

return Err(new_invalid_state_error(vm, "invalid state"));

}

*zelf.fut_result.write() = Some(result);

zelf.fut_state.store(FutureState::Finished);

Self::schedule_callbacks(&zelf, vm)?;

Ok(())

}

#[pymethod]

fn set_exception(

zelf: PyRef<Self>,

exception: PyObjectRef,

vm: &VirtualMachine,

) -> PyResult<()> {

if zelf.fut_loop.read().is_none() {

return Err(vm.new_runtime_error("Future object is not initialized."));

}

if zelf.fut_state.load() != FutureState::Pending {

return Err(new_invalid_state_error(vm, "invalid state"));

}

// Normalize the exception

let exc = if exception.fast_isinstance(vm.ctx.types.type_type) {

exception.call((), vm)?

} else {

exception

};

if !exc.fast_isinstance(vm.ctx.exceptions.base_exception_type) {

return Err(vm.new_type_error(format!(

"exception must be a BaseException, not {}",

exc.class().name()

)));

}

// Wrap StopIteration in RuntimeError

let exc = if exc.fast_isinstance(vm.ctx.exceptions.stop_iteration) {

let msg = "StopIteration interacts badly with generators and cannot be raised into a Future";

let runtime_err = vm.new_runtime_error(msg.to_string());

// Set cause and context to the original StopIteration

let stop_iter: PyRef<PyBaseException> = exc.downcast().unwrap();

runtime_err.set___cause__(Some(stop_iter.clone()));

runtime_err.set___context__(Some(stop_iter));

runtime_err.into()

} else {

exc

};

// Save the original traceback for later restoration

if let Ok(exc_ref) = exc.clone().downcast::<PyBaseException>() {

let tb = exc_ref.__traceback__().map(|tb| tb.into());

*zelf.fut_exception_tb.write() = tb;

}

*zelf.fut_exception.write() = Some(exc);

zelf.fut_state.store(FutureState::Finished);

zelf.fut_log_tb.store(true, Ordering::Relaxed);

Self::schedule_callbacks(&zelf, vm)?;

Ok(())

}

#[pymethod]

fn add_done_callback(

zelf: PyRef<Self>,

args: AddDoneCallbackArgs,

vm: &VirtualMachine,

) -> PyResult<()> {

if zelf.fut_loop.read().is_none() {

return Err(vm.new_runtime_error("Future object is not initialized."));

}

let ctx = match args.context.flatten() {

Some(c) => c,

None => get_copy_context(vm)?,

};

if zelf.fut_state.load() != FutureState::Pending {

Self::call_soon_with_context(&zelf, args.func, Some(ctx), vm)?;

} else if zelf.fut_callback0.read().is_none() {

*zelf.fut_callback0.write() = Some(args.func);

*zelf.fut_context0.write() = Some(ctx);

} else {

let tuple = vm.ctx.new_tuple(vec![args.func, ctx]);

let mut callbacks = zelf.fut_callbacks.write();

if callbacks.is_none() {

*callbacks = Some(vm.ctx.new_list(vec![tuple.into()]).into());

} else {

let list = callbacks.as_ref().unwrap();

vm.call_method(list, "append", (tuple,))?;

}

}

Ok(())

}

#[pymethod]

fn remove_done_callback(&self, func: PyObjectRef, vm: &VirtualMachine) -> PyResult<usize> {

if self.fut_loop.read().is_none() {

return Err(vm.new_runtime_error("Future object is not initialized."));

}

let mut cleared_callback0 = 0usize;

// Check fut_callback0 first

// Clone to release lock before comparison (which may run Python code)

let cb0 = self.fut_callback0.read().clone();

if let Some(cb0) = cb0 {

let cmp = vm.identical_or_equal(&cb0, &func)?;

if cmp {

*self.fut_callback0.write() = None;

*self.fut_context0.write() = None;

cleared_callback0 = 1;

}

}

// Check if fut_callbacks exists

let callbacks = self.fut_callbacks.read().clone();

let callbacks = match callbacks {

Some(c) => c,

None => return Ok(cleared_callback0),

};

let list: PyListRef = callbacks.downcast().unwrap();

let len = list.borrow_vec().len();

if len == 0 {

*self.fut_callbacks.write() = None;

return Ok(cleared_callback0);

}

// Special case for single callback

if len == 1 {

let item = list.borrow_vec().first().cloned();

if let Some(item) = item {

let tuple: &PyTuple = item.downcast_ref().unwrap();

let cb = tuple.first().unwrap().clone();

let cmp = vm.identical_or_equal(&cb, &func)?;

if cmp {

*self.fut_callbacks.write() = None;

return Ok(1 + cleared_callback0);

}

}

return Ok(cleared_callback0);

}

// Multiple callbacks - iterate with index, checking validity each time

// to handle evil comparisons

let mut new_callbacks = Vec::with_capacity(len);

let mut i = 0usize;

let mut removed = 0usize;

loop {

// Re-check fut_callbacks on each iteration (evil code may have cleared it)

let callbacks = self.fut_callbacks.read().clone();

let callbacks = match callbacks {

Some(c) => c,

None => break,

};

let list: PyListRef = callbacks.downcast().unwrap();

let current_len = list.borrow_vec().len();

if i >= current_len {

break;

}

// Get item and release lock before comparison

let item = list.borrow_vec().get(i).cloned();

let item = match item {

Some(item) => item,

None => break,

};

let tuple: &PyTuple = item.downcast_ref().unwrap();

let cb = tuple.first().unwrap().clone();

let cmp = vm.identical_or_equal(&cb, &func)?;

if !cmp {

new_callbacks.push(item);

} else {

removed += 1;

}

i += 1;

}

// Update fut_callbacks with filtered list

if new_callbacks.is_empty() {

*self.fut_callbacks.write() = None;

} else {

*self.fut_callbacks.write() = Some(vm.ctx.new_list(new_callbacks).into());

}

Ok(removed + cleared_callback0)

}

#[pymethod]

fn cancel(zelf: PyRef<Self>, args: CancelArgs, vm: &VirtualMachine) -> PyResult<bool> {

if zelf.fut_loop.read().is_none() {

return Err(vm.new_runtime_error("Future object is not initialized."));

}

if zelf.fut_state.load() != FutureState::Pending {

// Clear log_tb even when cancel fails

zelf.fut_log_tb.store(false, Ordering::Relaxed);

return Ok(false);

}

*zelf.fut_cancel_msg.write() = args.msg.flatten();

zelf.fut_state.store(FutureState::Cancelled);

Self::schedule_callbacks(&zelf, vm)?;

Ok(true)

}

#[pymethod]

fn cancelled(&self) -> bool {

self.fut_state.load() == FutureState::Cancelled

}

#[pymethod]

fn done(&self) -> bool {

self.fut_state.load() != FutureState::Pending

}

#[pymethod]

fn get_loop(&self, vm: &VirtualMachine) -> PyResult {

self.fut_loop

.read()

.clone()

.ok_or_else(|| vm.new_runtime_error("Future object is not initialized."))

}

#[pymethod]

fn _make_cancelled_error(&self, vm: &VirtualMachine) -> PyBaseExceptionRef {

self.make_cancelled_error_impl(vm)

}

fn make_cancelled_error_impl(&self, vm: &VirtualMachine) -> PyBaseExceptionRef {

// If a saved CancelledError exists, take it (clearing the stored reference)

if let Some(exc) = self.fut_cancelled_exc.write().take()

&& let Ok(exc) = exc.downcast::<PyBaseException>()

{

return exc;

}

let msg = self.fut_cancel_msg.read().clone();

let args = if let Some(m) = msg { vec![m] } else { vec![] };

match get_cancelled_error_type(vm) {

Ok(cancelled_error) => vm.new_exception(cancelled_error, args),

Err(_) => vm.new_runtime_error("cancelled"),

}

}

fn schedule_callbacks(zelf: &PyRef<Self>, vm: &VirtualMachine) -> PyResult<()> {

// Collect all callbacks first to avoid holding locks during callback execution

// This prevents deadlock when callbacks access the future's properties

let mut callbacks_to_call: Vec<(PyObjectRef, Option<PyObjectRef>)> = Vec::new();

// Take callback0 - release lock before collecting from list

let cb0 = zelf.fut_callback0.write().take();

let ctx0 = zelf.fut_context0.write().take();

if let Some(cb) = cb0 {

callbacks_to_call.push((cb, ctx0));

}

// Take callbacks list and collect items

let callbacks_list = zelf.fut_callbacks.write().take();

if let Some(callbacks) = callbacks_list

&& let Ok(list) = callbacks.downcast::<PyList>()

{

// Clone the items while holding the list lock, then release

let items: Vec<_> = list.borrow_vec().iter().cloned().collect();

for item in items {

if let Some(tuple) = item.downcast_ref::<PyTuple>()

&& let (Some(cb), Some(ctx)) = (tuple.first(), tuple.get(1))

{

callbacks_to_call.push((cb.clone(), Some(ctx.clone())));

}

}

}

// Now call all callbacks without holding any locks

for (cb, ctx) in callbacks_to_call {

Self::call_soon_with_context(zelf, cb, ctx, vm)?;

}

Ok(())

}

fn call_soon_with_context(

zelf: &PyRef<Self>,

callback: PyObjectRef,

context: Option<PyObjectRef>,

vm: &VirtualMachine,

) -> PyResult<()> {

let loop_obj = zelf.fut_loop.read().clone();

if let Some(loop_obj) = loop_obj {

// call_soon(callback, *args, context=context)

// callback receives the future as its argument

let future_arg: PyObjectRef = zelf.clone().into();

let args = if let Some(ctx) = context {

FuncArgs::new(

vec![callback, future_arg],

KwArgs::new([("context".to_owned(), ctx)].into_iter().collect()),

)

} else {

FuncArgs::new(vec![callback, future_arg], KwArgs::default())

};

vm.call_method(&loop_obj, "call_soon", args)?;

}

Ok(())

}

// Properties

#[pygetset]

fn _state(&self) -> &'static str {

self.fut_state.load().as_str()

}

#[pygetset]

fn _asyncio_future_blocking(&self) -> bool {

self.fut_blocking.load(Ordering::Relaxed)

}

#[pygetset(setter)]

fn set__asyncio_future_blocking(

&self,

value: PySetterValue<bool>,

vm: &VirtualMachine,

) -> PyResult<()> {

match value {

PySetterValue::Assign(v) => {

self.fut_blocking.store(v, Ordering::Relaxed);

Ok(())

}

PySetterValue::Delete => Err(vm.new_attribute_error("cannot delete attribute")),

}

}

#[pygetset]

fn _loop(&self, vm: &VirtualMachine) -> PyObjectRef {

self.fut_loop

.read()

.clone()

.unwrap_or_else(|| vm.ctx.none())

}

#[pygetset]

fn _callbacks(&self, vm: &VirtualMachine) -> PyResult<PyObjectRef> {

let mut result = Vec::new();

if let Some(cb0) = self.fut_callback0.read().clone() {

let ctx0 = self

.fut_context0

.read()

.clone()

.unwrap_or_else(|| vm.ctx.none());

result.push(vm.ctx.new_tuple(vec![cb0, ctx0]).into());

}

if let Some(callbacks) = self.fut_callbacks.read().clone() {

let list: PyListRef = callbacks.downcast().unwrap();

for item in list.borrow_vec().iter() {

result.push(item.clone());

}

}

// Return None if no callbacks

if result.is_empty() {

Ok(vm.ctx.none())

} else {

Ok(vm.ctx.new_list(result).into())

}

}

#[pygetset]

fn _result(&self, vm: &VirtualMachine) -> PyObjectRef {

self.fut_result

.read()

.clone()

.unwrap_or_else(|| vm.ctx.none())

}

#[pygetset]

fn _exception(&self, vm: &VirtualMachine) -> PyObjectRef {

self.fut_exception

.read()

.clone()

.unwrap_or_else(|| vm.ctx.none())

}

#[pygetset]

fn _log_traceback(&self) -> bool {

self.fut_log_tb.load(Ordering::Relaxed)

}

#[pygetset(setter)]

fn set__log_traceback(

&self,

value: PySetterValue<bool>,

vm: &VirtualMachine,

) -> PyResult<()> {

match value {

PySetterValue::Assign(v) => {

if v {

return Err(vm.new_value_error("_log_traceback can only be set to False"));

}

self.fut_log_tb.store(false, Ordering::Relaxed);

Ok(())

}

PySetterValue::Delete => Err(vm.new_attribute_error("cannot delete attribute")),

}

}

#[pygetset]

fn _source_traceback(&self, vm: &VirtualMachine) -> PyObjectRef {

self.fut_source_tb

.read()

.clone()

.unwrap_or_else(|| vm.ctx.none())

}

#[pygetset]

fn _cancel_message(&self, vm: &VirtualMachine) -> PyObjectRef {

self.fut_cancel_msg

.read()

.clone()

.unwrap_or_else(|| vm.ctx.none())

}

#[pygetset(setter)]

fn set__cancel_message(&self, value: PySetterValue) {

match value {

PySetterValue::Assign(v) => *self.fut_cancel_msg.write() = Some(v),

PySetterValue::Delete => *self.fut_cancel_msg.write() = None,

}

}

#[pygetset]

fn _asyncio_awaited_by(&self, vm: &VirtualMachine) -> PyResult<PyObjectRef> {

let awaited_by = self.fut_awaited_by.read().clone();

match awaited_by {

None => Ok(vm.ctx.none()),

Some(obj) => {

if self.fut_awaited_by_is_set.load(Ordering::Relaxed) {

// Already a Set

Ok(obj)

} else {

// Single object - create a Set for the return value

let new_set = PySet::default().into_ref(&vm.ctx);

new_set.add(obj, vm)?;

Ok(new_set.into())

}

}

}

}

/// Add waiter to fut_awaited_by with single-object optimization

fn awaited_by_add(&self, waiter: PyObjectRef, vm: &VirtualMachine) -> PyResult<()> {

let mut awaited_by = self.fut_awaited_by.write();

if awaited_by.is_none() {

// First waiter - store directly

*awaited_by = Some(waiter);

return Ok(());

}

if self.fut_awaited_by_is_set.load(Ordering::Relaxed) {

// Already a Set - add to it

let set = awaited_by.as_ref().unwrap();

vm.call_method(set, "add", (waiter,))?;

} else {

// Single object - convert to Set

let existing = awaited_by.take().unwrap();

let new_set = PySet::default().into_ref(&vm.ctx);

new_set.add(existing, vm)?;

new_set.add(waiter, vm)?;

*awaited_by = Some(new_set.into());

self.fut_awaited_by_is_set.store(true, Ordering::Relaxed);

}

Ok(())

}

/// Discard waiter from fut_awaited_by with single-object optimization

fn awaited_by_discard(&self, waiter: &PyObject, vm: &VirtualMachine) -> PyResult<()> {

let mut awaited_by = self.fut_awaited_by.write();

if awaited_by.is_none() {

return Ok(());

}

let obj = awaited_by.as_ref().unwrap();

if !self.fut_awaited_by_is_set.load(Ordering::Relaxed) {

// Single object - check if it matches

if obj.is(waiter) {

*awaited_by = None;

}

} else {

// It's a Set - use discard

vm.call_method(obj, "discard", (waiter.to_owned(),))?;

}

Ok(())

}

#[pymethod]

fn __await__(zelf: PyRef<Self>, _vm: &VirtualMachine) -> PyResult<PyFutureIter> {

Ok(PyFutureIter {

future: PyRwLock::new(Some(zelf.into())),

})

}

#[pyclassmethod]

fn __class_getitem__(

cls: PyTypeRef,

args: PyObjectRef,

vm: &VirtualMachine,

) -> PyGenericAlias {

PyGenericAlias::from_args(cls, args, vm)

}

}

impl Destructor for PyFuture {

fn del(zelf: &Py<Self>, vm: &VirtualMachine) -> PyResult<()> {

// Check if we should log the traceback

// Don't log if log_tb is false or if the future was cancelled

if !zelf.fut_log_tb.load(Ordering::Relaxed) {

return Ok(());

}

if zelf.fut_state.load() == FutureState::Cancelled {

return Ok(());

}

let exc = zelf.fut_exception.read().clone();

let exc = match exc {

Some(e) => e,

None => return Ok(()),

};

let loop_obj = zelf.fut_loop.read().clone();

let loop_obj = match loop_obj {

Some(l) => l,

None => return Ok(()),

};

// Create context dict for call_exception_handler

let context = PyDict::default().into_ref(&vm.ctx);

let class_name = zelf.class().name().to_string();

let message = format!("{} exception was never retrieved", class_name);

context.set_item(

vm.ctx.intern_str("message"),

vm.ctx.new_str(message).into(),

vm,

)?;

context.set_item(vm.ctx.intern_str("exception"), exc, vm)?;

context.set_item(vm.ctx.intern_str("future"), zelf.to_owned().into(), vm)?;

if let Some(tb) = zelf.fut_source_tb.read().clone() {

context.set_item(vm.ctx.intern_str("source_traceback"), tb, vm)?;

}

// Call loop.call_exception_handler(context)

let _ = vm.call_method(&loop_obj, "call_exception_handler", (context,));

Ok(())

}

}

impl Representable for PyFuture {

fn repr_str(zelf: &Py<Self>, vm: &VirtualMachine) -> PyResult<String> {

let class_name = zelf.class().name().to_string();

if let Some(_guard) = ReprGuard::enter(vm, zelf.as_object()) {

let info = get_future_repr_info(zelf.as_object(), vm)?;

Ok(format!("<{} {}>", class_name, info))

} else {

Ok(format!("<{} ...>", class_name))

}

}

}

impl Iterable for PyFuture {

fn iter(zelf: PyRef<Self>, _vm: &VirtualMachine) -> PyResult {

Ok(PyFutureIter {

future: PyRwLock::new(Some(zelf.into())),

}

.into_pyobject(_vm))

}

}

fn get_future_repr_info(future: &PyObject, vm: &VirtualMachine) -> PyResult<Wtf8Buf> {

// Try to use asyncio.base_futures._future_repr_info

// Import from sys.modules if available, otherwise try regular import

let sys_modules = vm.sys_module.get_attr("modules", vm)?;

let module =

if let Ok(m) = sys_modules.get_item(&*vm.ctx.new_str("asyncio.base_futures"), vm) {

m

} else {

// vm.import returns the top-level module, get base_futures submodule

match vm

.import("asyncio.base_futures", 0)

.and_then(|asyncio| asyncio.get_attr(vm.ctx.intern_str("base_futures"), vm))

{

Ok(m) => m,

Err(_) => return get_future_repr_info_fallback(future, vm),

}

};

let func = match vm.get_attribute_opt(module, vm.ctx.intern_str("_future_repr_info")) {

Ok(Some(f)) => f,

_ => return get_future_repr_info_fallback(future, vm),

};

let info = match func.call((future.to_owned(),), vm) {

Ok(i) => i,

Err(_) => return get_future_repr_info_fallback(future, vm),

};

let list: PyListRef = match info.downcast() {

Ok(l) => l,

Err(_) => return get_future_repr_info_fallback(future, vm),

};

let mut result = Wtf8Buf::new();

let parts = list.borrow_vec();

for (i, x) in parts.iter().enumerate() {

if i > 0 {

result.push_str(" ");

}

if let Ok(s) = x.str(vm) {

result.push_wtf8(s.as_wtf8());

}

}

Ok(result)

}

fn get_future_repr_info_fallback(future: &PyObject, vm: &VirtualMachine) -> PyResult<Wtf8Buf> {

// Fallback: build repr from properties directly

if let Ok(Some(state)) =

vm.get_attribute_opt(future.to_owned(), vm.ctx.intern_str("_state"))

{

let s = state

.str(vm)

.map(|s| s.as_wtf8().to_lowercase())

.unwrap_or_else(|_| Wtf8Buf::from("unknown"));

return Ok(s);

}

Ok(Wtf8Buf::from("state=unknown"))

}

fn get_task_repr_info(task: &PyObject, vm: &VirtualMachine) -> PyResult<Wtf8Buf> {

// vm.import returns the top-level module, get base_tasks submodule

match vm

.import("asyncio.base_tasks", 0)

.and_then(|asyncio| asyncio.get_attr(vm.ctx.intern_str("base_tasks"), vm))

{

Ok(base_tasks) => {

match vm.get_attribute_opt(base_tasks, vm.ctx.intern_str("_task_repr_info")) {

Ok(Some(func)) => {

let info: PyObjectRef = func.call((task.to_owned(),), vm)?;

let list: PyListRef = info.downcast().map_err(|_| {

vm.new_type_error("_task_repr_info should return a list")

})?;

let mut result = Wtf8Buf::new();

let parts = list.borrow_vec();

for (i, x) in parts.iter().enumerate() {

if i > 0 {

result.push_str(" ");

}

result.push_wtf8(x.str(vm)?.as_wtf8());

}

Ok(result)

}

_ => get_future_repr_info(task, vm),

}

}

Err(_) => get_future_repr_info(task, vm),

}

}

#[pyattr]

#[pyclass(name = "FutureIter", module = "_asyncio", traverse)]

#[derive(Debug, PyPayload)]

struct PyFutureIter {

future: PyRwLock<Option<PyObjectRef>>,

}

#[pyclass(with(IterNext, Iterable))]

impl PyFutureIter {

#[pymethod]

fn send(&self, _value: PyObjectRef, vm: &VirtualMachine) -> PyResult {

let future = self.future.read().clone();

let future = match future {

Some(f) => f,

None => return Err(vm.new_stop_iteration(None)),

};

// Try to get blocking flag (check Task first since it inherits from Future)

let blocking = if let Some(task) = future.downcast_ref::<PyTask>() {

task.base.fut_blocking.load(Ordering::Relaxed)

} else if let Some(fut) = future.downcast_ref::<PyFuture>() {

fut.fut_blocking.load(Ordering::Relaxed)

} else {

// For non-native futures, check the attribute

vm.get_attribute_opt(

future.clone(),

vm.ctx.intern_str("_asyncio_future_blocking"),

)?

.map(|v| v.try_to_bool(vm))

.transpose()?

.unwrap_or(false)

};

// Check if future is done

let done = vm.call_method(&future, "done", ())?;

if done.try_to_bool(vm)? {

*self.future.write() = None;

let result = vm.call_method(&future, "result", ())?;

return Err(vm.new_stop_iteration(Some(result)));

}

// If still pending and blocking is already set, raise RuntimeError

// This means await wasn't used with future

if blocking {

return Err(vm.new_runtime_error("await wasn't used with future"));

}

// First call: set blocking flag and yield the future (check Task first)

if let Some(task) = future.downcast_ref::<PyTask>() {

task.base.fut_blocking.store(true, Ordering::Relaxed);

} else if let Some(fut) = future.downcast_ref::<PyFuture>() {

fut.fut_blocking.store(true, Ordering::Relaxed);

} else {

future.set_attr(

vm.ctx.intern_str("_asyncio_future_blocking"),

vm.ctx.true_value.clone(),

vm,

)?;

}

Ok(future)