Merge pull request #1801 from RustPython/coolreader18/async-gen

Implement async generators

Lib/asyncio/events.py

@@ -6,7 +6,8 @@ __all__ = ['AbstractEventLoopPolicy',
            'get_event_loop_policy', 'set_event_loop_policy',
            'get_event_loop', 'set_event_loop', 'new_event_loop',
            'get_child_watcher', 'set_child_watcher',
-           '_set_running_loop', '_get_running_loop',
+           '_set_running_loop', 'get_running_loop',
+           '_get_running_loop',
            ]
 
 import functools
@@ -614,6 +615,18 @@ class _RunningLoop(threading.local):
 _running_loop = _RunningLoop()
 
 
+def get_running_loop():
+    """Return the running event loop.  Raise a RuntimeError if there is none.
+
+    This function is thread-specific.
+    """
+    # NOTE: this function is implemented in C (see _asynciomodule.c)
+    loop = _get_running_loop()
+    if loop is None:
+        raise RuntimeError('no running event loop')
+    return loop
+
+
 def _get_running_loop():
     """Return the running event loop or None.
 

Lib/asyncio/tasks.py

@@ -1,6 +1,6 @@
 """Support for tasks, coroutines and the scheduler."""
 
-__all__ = ['Task',
+__all__ = ['Task', 'create_task',
            'FIRST_COMPLETED', 'FIRST_EXCEPTION', 'ALL_COMPLETED',
            'wait', 'wait_for', 'as_completed', 'sleep', 'async',
            'gather', 'shield', 'ensure_future', 'run_coroutine_threadsafe',
@@ -44,6 +44,16 @@ def _all_tasks_compat(loop=None):
     return {t for t in _all_tasks if futures._get_loop(t) is loop}
 
 
+def _set_task_name(task, name):
+    if name is not None:
+        try:
+            set_name = task.set_name
+        except AttributeError:
+            pass
+        else:
+            set_name(name)
+
+
 class Task(futures.Future):
     """A coroutine wrapped in a Future."""
 
@@ -292,6 +302,17 @@ else:
     Task = _CTask = _asyncio.Task
 
 
+def create_task(coro, *, name=None):
+    """Schedule the execution of a coroutine object in a spawn task.
+
+    Return a Task object.
+    """
+    loop = events.get_running_loop()
+    task = loop.create_task(coro)
+    _set_task_name(task, name)
+    return task
+
+
 # wait() and as_completed() similar to those in PEP 3148.
 
 FIRST_COMPLETED = concurrent.futures.FIRST_COMPLETED

compiler/src/compile.rs

@@ -39,7 +39,7 @@ struct CompileContext {
     func: FunctionContext,
 }
 
-#[derive(Clone, Copy)]
+#[derive(Clone, Copy, PartialEq)]
 enum FunctionContext {
     NoFunction,
     Function,
@@ -562,6 +562,14 @@ impl<O: OutputStream> Compiler<O> {
                 }
                 match value {
                     Some(v) => {
+                        if self.ctx.func == FunctionContext::AsyncFunction
+                            && self.current_output().is_generator()
+                        {
+                            return Err(self.error_loc(
+                                CompileErrorType::AsyncReturnValue,
+                                statement.location.clone(),
+                            ));
+                        }
                         self.compile_expression(v)?;
                     }
                     None => {
@@ -1644,6 +1652,9 @@ impl<O: OutputStream> Compiler<O> {
                 self.emit(Instruction::YieldValue);
             }
             Await { value } => {
+                if self.ctx.func != FunctionContext::AsyncFunction {
+                    return Err(self.error(CompileErrorType::InvalidAwait));
+                }
                 self.compile_expression(value)?;
                 self.emit(Instruction::GetAwaitable);
                 self.emit(Instruction::LoadConst {
@@ -1652,6 +1663,15 @@ impl<O: OutputStream> Compiler<O> {
                 self.emit(Instruction::YieldFrom);
             }
             YieldFrom { value } => {
+                match self.ctx.func {
+                    FunctionContext::NoFunction => {
+                        return Err(self.error(CompileErrorType::InvalidYieldFrom))
+                    }
+                    FunctionContext::AsyncFunction => {
+                        return Err(self.error(CompileErrorType::AsyncYieldFrom))
+                    }
+                    FunctionContext::Function => {}
+                }
                 self.mark_generator();
                 self.compile_expression(value)?;
                 self.emit(Instruction::GetIter);
@@ -1670,7 +1690,7 @@ impl<O: OutputStream> Compiler<O> {
                     value: bytecode::Constant::Boolean { value: false },
                 });
             }
-            None => {
+            ast::ExpressionType::None => {
                 self.emit(Instruction::LoadConst {
                     value: bytecode::Constant::None,
                 });

compiler/src/error.rs

@@ -54,6 +54,10 @@ pub enum CompileErrorType {
     InvalidContinue,
     InvalidReturn,
     InvalidYield,
+    InvalidYieldFrom,
+    InvalidAwait,
+    AsyncYieldFrom,
+    AsyncReturnValue,
 }
 
 impl CompileError {
@@ -96,6 +100,12 @@ impl fmt::Display for CompileError {
             CompileErrorType::InvalidContinue => "'continue' outside loop".to_owned(),
             CompileErrorType::InvalidReturn => "'return' outside function".to_owned(),
             CompileErrorType::InvalidYield => "'yield' outside function".to_owned(),
+            CompileErrorType::InvalidYieldFrom => "'yield from' outside function".to_owned(),
+            CompileErrorType::InvalidAwait => "'await' outside async function".to_owned(),
+            CompileErrorType::AsyncYieldFrom => "'yield from' inside async function".to_owned(),
+            CompileErrorType::AsyncReturnValue => {
+                "'return' with value inside async generator".to_owned()
+            }
         };
 
         if let Some(statement) = &self.statement {

compiler/src/output_stream.rs

@@ -7,6 +7,8 @@ pub trait OutputStream: From<CodeObject> + Into<CodeObject> {
     fn set_label(&mut self, label: Label);
     /// Mark the inner CodeObject as a generator
     fn mark_generator(&mut self);
+    /// Check to see if the inner CodeObject is a generator
+    fn is_generator(&self) -> bool;
 }
 
 pub struct CodeObjectStream {
@@ -36,4 +38,7 @@ impl OutputStream for CodeObjectStream {
     fn mark_generator(&mut self) {
         self.code.flags |= CodeFlags::IS_GENERATOR;
     }
+    fn is_generator(&self) -> bool {
+        self.code.flags.contains(CodeFlags::IS_GENERATOR)
+    }
 }

compiler/src/peephole.rs

@@ -112,6 +112,9 @@ where
     fn mark_generator(&mut self) {
         self.inner.mark_generator()
     }
+    fn is_generator(&self) -> bool {
+        self.inner.is_generator()
+    }
 }
 
 impl<O: OutputStream> OptimizationBuffer for PeepholeOptimizer<O> {

parser/src/lexer.rs

@@ -686,6 +686,8 @@ where
                     tabs = 0;
                 }
                 None => {
+                    spaces = 0;
+                    tabs = 0;
                     break;
                 }
                 _ => {

vm/src/frame.rs

@@ -7,6 +7,7 @@ use itertools::Itertools;
 use crate::bytecode;
 use crate::exceptions::{self, ExceptionCtor, PyBaseExceptionRef};
 use crate::function::{single_or_tuple_any, PyFuncArgs};
+use crate::obj::objasyncgenerator::PyAsyncGenWrappedValue;
 use crate::obj::objbool;
 use crate::obj::objcode::PyCodeRef;
 use crate::obj::objcoroinner::Coro;
@@ -118,11 +119,15 @@ impl ExecutionResult {
     }
 
     /// Turn an ExecutionResult into a PyResult that would be returned from a generator or coroutine
-    pub fn into_result(self, vm: &VirtualMachine) -> PyResult {
+    pub fn into_result(self, async_stopiter: bool, vm: &VirtualMachine) -> PyResult {
         match self {
             ExecutionResult::Yield(value) => Ok(value),
             ExecutionResult::Return(value) => {
-                let stop_iteration = vm.ctx.exceptions.stop_iteration.clone();
+                let stop_iteration = if async_stopiter {
+                    vm.ctx.exceptions.stop_async_iteration.clone()
+                } else {
+                    vm.ctx.exceptions.stop_iteration.clone()
+                };
                 let args = if vm.is_none(&value) {
                     vec![]
                 } else {
@@ -379,6 +384,11 @@ impl Frame {
             }
             bytecode::Instruction::YieldValue => {
                 let value = self.pop_value();
+                let value = if self.code.flags.contains(bytecode::CodeFlags::IS_COROUTINE) {
+                    PyAsyncGenWrappedValue(value).into_ref(vm).into_object()
+                } else {
+                    value
+                };
                 Ok(Some(ExecutionResult::Yield(value)))
             }
             bytecode::Instruction::YieldFrom => self.execute_yield_from(vm),

vm/src/obj/mod.rs

@@ -1,5 +1,6 @@
 //! This package contains the python basic/builtin types
 
+pub mod objasyncgenerator;
 pub mod objbool;
 pub mod objbuiltinfunc;
 pub mod objbytearray;

vm/src/obj/objasyncgenerator.rs

@@ -0,0 +1,401 @@
+use super::objcode::PyCodeRef;
+use super::objcoroinner::{Coro, Variant};
+use super::objtype::{self, PyClassRef};
+use crate::exceptions::PyBaseExceptionRef;
+use crate::frame::FrameRef;
+use crate::function::OptionalArg;
+use crate::pyobject::{PyClassImpl, PyContext, PyObjectRef, PyRef, PyResult, PyValue};
+use crate::vm::VirtualMachine;
+
+use std::cell::Cell;
+
+#[pyclass(name = "async_generator")]
+#[derive(Debug)]
+pub struct PyAsyncGen {
+    inner: Coro,
+    running_async: Cell<bool>,
+}
+pub type PyAsyncGenRef = PyRef<PyAsyncGen>;
+
+impl PyValue for PyAsyncGen {
+    fn class(vm: &VirtualMachine) -> PyClassRef {
+        vm.ctx.types.async_generator.clone()
+    }
+}
+
+#[pyimpl]
+impl PyAsyncGen {
+    pub fn as_coro(&self) -> &Coro {
+        &self.inner
+    }
+
+    pub fn new(frame: FrameRef, vm: &VirtualMachine) -> PyAsyncGenRef {
+        PyAsyncGen {
+            inner: Coro::new(frame, Variant::AsyncGen),
+            running_async: Cell::new(false),
+        }
+        .into_ref(vm)
+    }
+
+    // TODO: fix function names situation
+    #[pyproperty(magic)]
+    fn name(&self, vm: &VirtualMachine) -> PyObjectRef {
+        vm.get_none()
+    }
+
+    #[pymethod(name = "__aiter__")]
+    fn aiter(zelf: PyRef<Self>, _vm: &VirtualMachine) -> PyRef<Self> {
+        zelf
+    }
+
+    #[pymethod(name = "__anext__")]
+    fn anext(zelf: PyRef<Self>, vm: &VirtualMachine) -> PyAsyncGenASend {
+        Self::asend(zelf, vm.get_none(), vm)
+    }
+
+    #[pymethod]
+    fn asend(zelf: PyRef<Self>, value: PyObjectRef, _vm: &VirtualMachine) -> PyAsyncGenASend {
+        PyAsyncGenASend {
+            ag: zelf,
+            state: Cell::new(AwaitableState::Init),
+            value,
+        }
+    }
+
+    #[pymethod]
+    fn athrow(
+        zelf: PyRef<Self>,
+        exc_type: PyObjectRef,
+        exc_val: OptionalArg,
+        exc_tb: OptionalArg,
+        vm: &VirtualMachine,
+    ) -> PyAsyncGenAThrow {
+        PyAsyncGenAThrow {
+            ag: zelf,
+            aclose: false,
+            state: Cell::new(AwaitableState::Init),
+            value: (
+                exc_type,
+                exc_val.unwrap_or_else(|| vm.get_none()),
+                exc_tb.unwrap_or_else(|| vm.get_none()),
+            ),
+        }
+    }
+
+    #[pymethod]
+    fn aclose(zelf: PyRef<Self>, vm: &VirtualMachine) -> PyAsyncGenAThrow {
+        PyAsyncGenAThrow {
+            ag: zelf,
+            aclose: true,
+            state: Cell::new(AwaitableState::Init),
+            value: (
+                vm.ctx.exceptions.generator_exit.clone().into_object(),
+                vm.get_none(),
+                vm.get_none(),
+            ),
+        }
+    }
+
+    #[pyproperty]
+    fn ag_await(&self, _vm: &VirtualMachine) -> Option<PyObjectRef> {
+        self.inner.frame().yield_from_target()
+    }
+    #[pyproperty]
+    fn ag_frame(&self, _vm: &VirtualMachine) -> FrameRef {
+        self.inner.frame()
+    }
+    #[pyproperty]
+    fn ag_running(&self, _vm: &VirtualMachine) -> bool {
+        self.inner.running()
+    }
+    #[pyproperty]
+    fn ag_code(&self, _vm: &VirtualMachine) -> PyCodeRef {
+        self.inner.frame().code.clone()
+    }
+}
+
+#[pyclass(name = "async_generator_wrapped_value")]
+#[derive(Debug)]
+pub(crate) struct PyAsyncGenWrappedValue(pub PyObjectRef);
+impl PyValue for PyAsyncGenWrappedValue {
+    fn class(vm: &VirtualMachine) -> PyClassRef {
+        vm.ctx.types.async_generator_wrapped_value.clone()
+    }
+}
+
+impl PyAsyncGenWrappedValue {
+    fn unbox(ag: &PyAsyncGen, val: PyResult, vm: &VirtualMachine) -> PyResult {
+        if let Err(ref e) = val {
+            if objtype::isinstance(&e, &vm.ctx.exceptions.stop_async_iteration)
+                || objtype::isinstance(&e, &vm.ctx.exceptions.generator_exit)
+            {
+                ag.inner.closed.set(true);
+            }
+            ag.running_async.set(false);
+        }
+        let val = val?;
+
+        match_class!(match val {
+            val @ Self => {
+                ag.running_async.set(false);
+                Err(vm.new_exception(
+                    vm.ctx.exceptions.stop_iteration.clone(),
+                    vec![val.0.clone()],
+                ))
+            }
+            val => Ok(val),
+        })
+    }
+}
+
+#[derive(Debug, Clone, Copy)]
+enum AwaitableState {
+    Init,
+    Iter,
+    Closed,
+}
+
+#[pyclass(name = "async_generator_asend")]
+#[derive(Debug)]
+struct PyAsyncGenASend {
+    ag: PyAsyncGenRef,
+    state: Cell<AwaitableState>,
+    value: PyObjectRef,
+}
+
+impl PyValue for PyAsyncGenASend {
+    fn class(vm: &VirtualMachine) -> PyClassRef {
+        vm.ctx.types.async_generator_asend.clone()
+    }
+}
+
+#[pyimpl]
+impl PyAsyncGenASend {
+    #[pymethod(name = "__iter__")]
+    fn iter(zelf: PyRef<Self>, _vm: &VirtualMachine) -> PyRef<Self> {
+        zelf
+    }
+    #[pymethod(name = "__await__")]
+    fn r#await(zelf: PyRef<Self>, _vm: &VirtualMachine) -> PyRef<Self> {
+        zelf
+    }
+
+    #[pymethod(name = "__next__")]
+    fn next(&self, vm: &VirtualMachine) -> PyResult {
+        self.send(vm.get_none(), vm)
+    }
+
+    #[pymethod]
+    fn send(&self, val: PyObjectRef, vm: &VirtualMachine) -> PyResult {
+        let val = match self.state.get() {
+            AwaitableState::Closed => {
+                return Err(vm.new_runtime_error(
+                    "cannot reuse already awaited __anext__()/asend()".to_owned(),
+                ))
+            }
+            AwaitableState::Iter => val, // already running, all good
+            AwaitableState::Init => {
+                if self.ag.running_async.get() {
+                    return Err(vm.new_runtime_error(
+                        "anext(): asynchronous generator is already running".to_owned(),
+                    ));
+                }
+                self.ag.running_async.set(true);
+                self.state.set(AwaitableState::Iter);
+                if vm.is_none(&val) {
+                    self.value.clone()
+                } else {
+                    val
+                }
+            }
+        };
+        let res = self.ag.inner.send(val, vm);
+        let res = PyAsyncGenWrappedValue::unbox(&self.ag, res, vm);
+        if res.is_err() {
+            self.close();
+        }
+        res
+    }
+
+    #[pymethod]
+    fn throw(
+        &self,
+        exc_type: PyObjectRef,
+        exc_val: OptionalArg,
+        exc_tb: OptionalArg,
+        vm: &VirtualMachine,
+    ) -> PyResult {
+        if let AwaitableState::Closed = self.state.get() {
+            return Err(
+                vm.new_runtime_error("cannot reuse already awaited __anext__()/asend()".to_owned())
+            );
+        }
+
+        let res = self.ag.inner.throw(
+            exc_type,
+            exc_val.unwrap_or_else(|| vm.get_none()),
+            exc_tb.unwrap_or_else(|| vm.get_none()),
+            vm,
+        );
+        let res = PyAsyncGenWrappedValue::unbox(&self.ag, res, vm);
+        if res.is_err() {
+            self.close();
+        }
+        res
+    }
+
+    #[pymethod]
+    fn close(&self) {
+        self.state.set(AwaitableState::Closed);
+    }
+}
+
+#[pyclass(name = "async_generator_athrow")]
+#[derive(Debug)]
+struct PyAsyncGenAThrow {
+    ag: PyAsyncGenRef,
+    aclose: bool,
+    state: Cell<AwaitableState>,
+    value: (PyObjectRef, PyObjectRef, PyObjectRef),
+}
+
+impl PyValue for PyAsyncGenAThrow {
+    fn class(vm: &VirtualMachine) -> PyClassRef {
+        vm.ctx.types.async_generator_athrow.clone()
+    }
+}
+
+#[pyimpl]
+impl PyAsyncGenAThrow {
+    #[pymethod(name = "__iter__")]
+    fn iter(zelf: PyRef<Self>, _vm: &VirtualMachine) -> PyRef<Self> {
+        zelf
+    }
+    #[pymethod(name = "__await__")]
+    fn r#await(zelf: PyRef<Self>, _vm: &VirtualMachine) -> PyRef<Self> {
+        zelf
+    }
+
+    #[pymethod(name = "__next__")]
+    fn next(&self, vm: &VirtualMachine) -> PyResult {
+        self.send(vm.get_none(), vm)
+    }
+
+    #[pymethod]
+    fn send(&self, val: PyObjectRef, vm: &VirtualMachine) -> PyResult {
+        match self.state.get() {
+            AwaitableState::Closed => {
+                Err(vm
+                    .new_runtime_error("cannot reuse already awaited aclose()/athrow()".to_owned()))
+            }
+            AwaitableState::Init => {
+                if self.ag.running_async.get() {
+                    self.state.set(AwaitableState::Closed);
+                    let msg = if self.aclose {
+                        "aclose(): asynchronous generator is already running"
+                    } else {
+                        "athrow(): asynchronous generator is already running"
+                    };
+                    return Err(vm.new_runtime_error(msg.to_owned()));
+                }
+                if self.ag.inner.closed() {
+                    self.state.set(AwaitableState::Closed);
+                    return Err(vm.new_exception_empty(vm.ctx.exceptions.stop_iteration.clone()));
+                }
+                if !vm.is_none(&val) {
+                    return Err(vm.new_runtime_error(
+                        "can't send non-None value to a just-started async generator".to_owned(),
+                    ));
+                }
+                self.state.set(AwaitableState::Iter);
+                self.ag.running_async.set(true);
+
+                let (ty, val, tb) = self.value.clone();
+                let ret = self.ag.inner.throw(ty, val, tb, vm);
+                let ret = if self.aclose {
+                    if self.ignored_close(&ret) {
+                        Err(self.yield_close(vm))
+                    } else {
+                        ret
+                    }
+                } else {
+                    PyAsyncGenWrappedValue::unbox(&self.ag, ret, vm)
+                };
+                ret.map_err(|e| self.check_error(e, vm))
+            }
+            AwaitableState::Iter => {
+                let ret = self.ag.inner.send(val, vm);
+                if self.aclose {
+                    match ret {
+                        Ok(v) if v.payload_is::<PyAsyncGenWrappedValue>() => {
+                            Err(self.yield_close(vm))
+                        }
+                        Ok(v) => Ok(v),
+                        Err(e) => Err(self.check_error(e, vm)),
+                    }
+                } else {
+                    PyAsyncGenWrappedValue::unbox(&self.ag, ret, vm)
+                }
+            }
+        }
+    }
+
+    #[pymethod]
+    fn throw(
+        &self,
+        exc_type: PyObjectRef,
+        exc_val: OptionalArg,
+        exc_tb: OptionalArg,
+        vm: &VirtualMachine,
+    ) -> PyResult {
+        let ret = self.ag.inner.throw(
+            exc_type,
+            exc_val.unwrap_or_else(|| vm.get_none()),
+            exc_tb.unwrap_or_else(|| vm.get_none()),
+            vm,
+        );
+        let res = if self.aclose {
+            if self.ignored_close(&ret) {
+                Err(self.yield_close(vm))
+            } else {
+                ret
+            }
+        } else {
+            PyAsyncGenWrappedValue::unbox(&self.ag, ret, vm)
+        };
+        res.map_err(|e| self.check_error(e, vm))
+    }
+
+    #[pymethod]
+    fn close(&self) {
+        self.state.set(AwaitableState::Closed);
+    }
+
+    fn ignored_close(&self, res: &PyResult) -> bool {
+        res.as_ref()
+            .map_or(false, |v| v.payload_is::<PyAsyncGenWrappedValue>())
+    }
+    fn yield_close(&self, vm: &VirtualMachine) -> PyBaseExceptionRef {
+        self.ag.running_async.set(false);
+        self.state.set(AwaitableState::Closed);
+        vm.new_runtime_error("async generator ignored GeneratorExit".to_owned())
+    }
+    fn check_error(&self, exc: PyBaseExceptionRef, vm: &VirtualMachine) -> PyBaseExceptionRef {
+        self.ag.running_async.set(false);
+        self.state.set(AwaitableState::Closed);
+        if self.aclose
+            && (objtype::isinstance(&exc, &vm.ctx.exceptions.stop_async_iteration)
+                || objtype::isinstance(&exc, &vm.ctx.exceptions.generator_exit))
+        {
+            vm.new_exception_empty(vm.ctx.exceptions.stop_iteration.clone())
+        } else {
+            exc
+        }
+    }
+}
+
+pub fn init(ctx: &PyContext) {
+    PyAsyncGen::extend_class(ctx, &ctx.types.async_generator);
+    PyAsyncGenASend::extend_class(ctx, &ctx.types.async_generator_asend);
+    PyAsyncGenAThrow::extend_class(ctx, &ctx.types.async_generator_athrow);
+}

vm/src/obj/objcoroinner.rs

@@ -1,4 +1,4 @@
-use super::{objiter, objtype};
+use super::objtype::{self, PyClassRef};
 use crate::exceptions::{self, PyBaseExceptionRef};
 use crate::frame::{ExecutionResult, FrameRef};
 use crate::pyobject::{PyObjectRef, PyResult};
@@ -6,21 +6,50 @@ use crate::vm::VirtualMachine;
 
 use std::cell::{Cell, RefCell};
 
+#[derive(Debug, PartialEq, Clone, Copy)]
+pub enum Variant {
+    Gen,
+    Coroutine,
+    AsyncGen,
+}
+impl Variant {
+    fn exec_result(self, res: ExecutionResult, vm: &VirtualMachine) -> PyResult {
+        res.into_result(self == Self::AsyncGen, vm)
+    }
+    fn name(self) -> &'static str {
+        match self {
+            Self::Gen => "generator",
+            Self::Coroutine => "coroutine",
+            Self::AsyncGen => "async generator",
+        }
+    }
+    fn stop_iteration(self, vm: &VirtualMachine) -> PyClassRef {
+        match self {
+            Self::AsyncGen => vm.ctx.exceptions.stop_async_iteration.clone(),
+            _ => vm.ctx.exceptions.stop_iteration.clone(),
+        }
+    }
+}
+
 #[derive(Debug)]
 pub struct Coro {
     frame: FrameRef,
-    closed: Cell<bool>,
+    pub closed: Cell<bool>,
     running: Cell<bool>,
     exceptions: RefCell<Vec<PyBaseExceptionRef>>,
+    started: Cell<bool>,
+    variant: Variant,
 }
 
 impl Coro {
-    pub fn new(frame: FrameRef) -> Self {
+    pub fn new(frame: FrameRef, variant: Variant) -> Self {
         Coro {
             frame,
             closed: Cell::new(false),
             running: Cell::new(false),
             exceptions: RefCell::new(vec![]),
+            started: Cell::new(false),
+            variant,
         }
     }
 
@@ -47,20 +76,44 @@ impl Coro {
                 .split_off(curr_exception_stack_len),
         );
         self.running.set(false);
+        self.started.set(true);
         result
     }
 
     pub fn send(&self, value: PyObjectRef, vm: &VirtualMachine) -> PyResult {
         if self.closed.get() {
-            return Err(objiter::new_stop_iteration(vm));
+            return Err(vm.new_exception_empty(self.variant.stop_iteration(vm)));
+        }
+        if !self.started.get() && !vm.is_none(&value) {
+            return Err(vm.new_type_error(format!(
+                "can't send non-None value to a just-started {}",
+                self.variant.name()
+            )));
         }
-
         self.frame.push_value(value.clone());
         let result = self.run_with_context(|| vm.run_frame(self.frame.clone()), vm);
         self.maybe_close(&result);
-        result?.into_result(vm)
+        match result {
+            Ok(exec_res) => self.variant.exec_result(exec_res, vm),
+            Err(e) => {
+                if objtype::isinstance(&e, &vm.ctx.exceptions.stop_iteration) {
+                    let err = vm
+                        .new_runtime_error(format!("{} raised StopIteration", self.variant.name()));
+                    err.set_cause(Some(e));
+                    Err(err)
+                } else if self.variant == Variant::AsyncGen
+                    && objtype::isinstance(&e, &vm.ctx.exceptions.stop_async_iteration)
+                {
+                    let err = vm
+                        .new_runtime_error("async generator raised StopAsyncIteration".to_owned());
+                    err.set_cause(Some(e));
+                    Err(err)
+                } else {
+                    Err(e)
+                }
+            }
+        }
     }
-
     pub fn throw(
         &self,
         exc_type: PyObjectRef,
@@ -76,7 +129,7 @@ impl Coro {
             self.run_with_context(|| self.frame.gen_throw(vm, exc_type, exc_val, exc_tb), vm);
         self.maybe_close(&result);
         vm.frames.borrow_mut().pop();
-        result?.into_result(vm)
+        self.variant.exec_result(result?, vm)
     }
 
     pub fn close(&self, vm: &VirtualMachine) -> PyResult<()> {
@@ -99,22 +152,28 @@ impl Coro {
         self.closed.set(true);
         match result {
             Ok(ExecutionResult::Yield(_)) => {
-                Err(vm.new_runtime_error("generator ignored GeneratorExit".to_owned()))
+                Err(vm.new_runtime_error(format!("{} ignored GeneratorExit", self.variant.name())))
             }
             Err(e) if !is_gen_exit(&e, vm) => Err(e),
             _ => Ok(()),
         }
     }
 
-    pub fn closed(&self) -> bool {
-        self.closed.get()
+    pub fn started(&self) -> bool {
+        self.started.get()
     }
     pub fn running(&self) -> bool {
         self.running.get()
     }
+    pub fn closed(&self) -> bool {
+        self.closed.get()
+    }
     pub fn frame(&self) -> FrameRef {
         self.frame.clone()
     }
+    pub fn name(&self) -> String {
+        self.frame.code.obj_name.clone()
+    }
 }
 
 pub fn is_gen_exit(exc: &PyBaseExceptionRef, vm: &VirtualMachine) -> bool {

vm/src/obj/objcoroutine.rs

@@ -1,5 +1,5 @@
 use super::objcode::PyCodeRef;
-use super::objcoroinner::Coro;
+use super::objcoroinner::{Coro, Variant};
 use super::objstr::PyStringRef;
 use super::objtype::PyClassRef;
 use crate::frame::FrameRef;
@@ -29,11 +29,17 @@ impl PyCoroutine {
 
     pub fn new(frame: FrameRef, vm: &VirtualMachine) -> PyCoroutineRef {
         PyCoroutine {
-            inner: Coro::new(frame),
+            inner: Coro::new(frame, Variant::Coroutine),
         }
         .into_ref(vm)
     }
 
+    // TODO: fix function names situation
+    #[pyproperty(magic)]
+    fn name(&self, vm: &VirtualMachine) -> PyObjectRef {
+        vm.get_none()
+    }
+
     #[pymethod]
     fn send(&self, value: PyObjectRef, vm: &VirtualMachine) -> PyResult {
         self.inner.send(value, vm)

vm/src/obj/objfunction.rs

@@ -6,6 +6,7 @@ use super::objtype::PyClassRef;
 use crate::bytecode;
 use crate::frame::Frame;
 use crate::function::{OptionalArg, PyFuncArgs};
+use crate::obj::objasyncgenerator::PyAsyncGen;
 use crate::obj::objcoroutine::PyCoroutine;
 use crate::obj::objgenerator::PyGenerator;
 use crate::pyobject::{
@@ -221,12 +222,13 @@ impl PyFunction {
         let frame = Frame::new(code.clone(), scope).into_ref(vm);
 
         // If we have a generator, create a new generator
-        if code.flags.contains(bytecode::CodeFlags::IS_GENERATOR) {
-            Ok(PyGenerator::new(frame, vm).into_object())
-        } else if code.flags.contains(bytecode::CodeFlags::IS_COROUTINE) {
-            Ok(PyCoroutine::new(frame, vm).into_object())
-        } else {
-            vm.run_frame_full(frame)
+        let is_gen = code.flags.contains(bytecode::CodeFlags::IS_GENERATOR);
+        let is_coro = code.flags.contains(bytecode::CodeFlags::IS_COROUTINE);
+        match (is_gen, is_coro) {
+            (true, false) => Ok(PyGenerator::new(frame, vm).into_object()),
+            (false, true) => Ok(PyCoroutine::new(frame, vm).into_object()),
+            (true, true) => Ok(PyAsyncGen::new(frame, vm).into_object()),
+            (false, false) => vm.run_frame_full(frame),
         }
     }
 

vm/src/obj/objgenerator.rs

@@ -3,7 +3,7 @@
  */
 
 use super::objcode::PyCodeRef;
-use super::objcoroinner::Coro;
+use super::objcoroinner::{Coro, Variant};
 use super::objtype::PyClassRef;
 use crate::frame::FrameRef;
 use crate::function::OptionalArg;
@@ -32,11 +32,17 @@ impl PyGenerator {
 
     pub fn new(frame: FrameRef, vm: &VirtualMachine) -> PyGeneratorRef {
         PyGenerator {
-            inner: Coro::new(frame),
+            inner: Coro::new(frame, Variant::Gen),
         }
         .into_ref(vm)
     }
 
+    // TODO: fix function names situation
+    #[pyproperty(magic)]
+    fn name(&self, vm: &VirtualMachine) -> PyObjectRef {
+        vm.get_none()
+    }
+
     #[pymethod(name = "__iter__")]
     fn iter(zelf: PyGeneratorRef) -> PyGeneratorRef {
         zelf

vm/src/types.rs

@@ -1,3 +1,4 @@
+use crate::obj::objasyncgenerator;
 use crate::obj::objbool;
 use crate::obj::objbuiltinfunc;
 use crate::obj::objbytearray;
@@ -47,6 +48,10 @@ use std::rc::Rc;
 /// Holder of references to builtin types.
 #[derive(Debug)]
 pub struct TypeZoo {
+    pub async_generator: PyClassRef,
+    pub async_generator_asend: PyClassRef,
+    pub async_generator_athrow: PyClassRef,
+    pub async_generator_wrapped_value: PyClassRef,
     pub bytes_type: PyClassRef,
     pub bytesiterator_type: PyClassRef,
     pub bytearray_type: PyClassRef,
@@ -134,6 +139,12 @@ impl TypeZoo {
         let generator_type = create_type("generator", &type_type, &object_type);
         let coroutine_type = create_type("coroutine", &type_type, &object_type);
         let coroutine_wrapper_type = create_type("coroutine_wrapper", &type_type, &object_type);
+        let async_generator = create_type("async_generator", &type_type, &object_type);
+        let async_generator_asend = create_type("async_generator_asend", &type_type, &object_type);
+        let async_generator_athrow =
+            create_type("async_generator_athrow", &type_type, &object_type);
+        let async_generator_wrapped_value =
+            create_type("async_generator_wrapped_value", &type_type, &object_type);
         let bound_method_type = create_type("method", &type_type, &object_type);
         let str_type = create_type("str", &type_type, &object_type);
         let list_type = create_type("list", &type_type, &object_type);
@@ -175,6 +186,10 @@ impl TypeZoo {
         let traceback_type = create_type("traceback", &type_type, &object_type);
 
         Self {
+            async_generator,
+            async_generator_asend,
+            async_generator_athrow,
+            async_generator_wrapped_value,
             bool_type,
             memoryview_type,
             bytearray_type,
@@ -379,6 +394,7 @@ pub fn initialize_types(context: &PyContext) {
     objclassmethod::init(&context);
     objgenerator::init(&context);
     objcoroutine::init(&context);
+    objasyncgenerator::init(&context);
     objint::init(&context);
     objfloat::init(&context);
     objcomplex::init(&context);