From 6c1003fd03e8347bcc50ceb7937230c34db982c3 Mon Sep 17 00:00:00 2001 From: "Jeong, YunWon" <69878+youknowone@users.noreply.github.com> Date: Tue, 15 Sep 2026 22:27:23 +0900 Subject: [PATCH] Honor asyncgen ag_closed and wasm BIO EOF (#8718) * address review: drain plaintext before BIO EOF Skip feed_tls([]) while rustls still has application data, matching the native SSL read EOF order. Assisted-by: Grok:grok-4.6 * Honor asyncgen ag_closed and asend reuse Keep ag_closed apart from a finished frame. aclose() sets it before GeneratorExit. asend() already-running closes the awaitable so it does not warn unawaited. Assisted-by: Grok:grok-4.6 (cherry picked from commit 823f6c6e0e932f1549da31ceca5a4f8538da5eda) Assisted-by: Codex:gpt-6 --- crates/vm/src/builtins/asyncgenerator.rs | 133 +++++++++++++---------- crates/wasm/src/ssl.rs | 2 +- 2 files changed, 75 insertions(+), 60 deletions(-) diff --git a/crates/vm/src/builtins/asyncgenerator.rs b/crates/vm/src/builtins/asyncgenerator.rs index 79652be39..cd971aecd 100644 --- a/crates/vm/src/builtins/asyncgenerator.rs +++ b/crates/vm/src/builtins/asyncgenerator.rs @@ -29,6 +29,10 @@ pub struct PyAsyncGen { running_async: AtomicCell, // whether hooks have been initialized ag_hooks_inited: AtomicCell, + // Distinct from the frame being finished: aclose() sets this + // before throwing GeneratorExit, and unwrap sets it on + // StopAsyncIteration / GeneratorExit. + ag_closed: AtomicCell, // ag_origin_or_finalizer - stores the finalizer callback ag_finalizer: PyMutex>, } @@ -63,6 +67,7 @@ impl PyAsyncGen { inner: Coro::new(frame, name, qualname), running_async: AtomicCell::new(false), ag_hooks_inited: AtomicCell::new(false), + ag_closed: AtomicCell::new(false), ag_finalizer: PyMutex::new(None), } } @@ -97,7 +102,7 @@ impl PyAsyncGen { fn call_finalizer(zelf: &Py, vm: &VirtualMachine) { let finalizer = zelf.ag_finalizer.lock().clone(); if let Some(finalizer) = finalizer - && !zelf.inner.closed.load() + && !zelf.ag_closed.load() { // Create a strong reference for the finalizer call. // This keeps the object alive during the finalizer execution. @@ -263,15 +268,21 @@ impl PyAsyncGenWrappedValue {} impl PyAsyncGenWrappedValue { fn unbox(ag: &PyAsyncGen, val: PyResult, vm: &VirtualMachine) -> PyResult { - let (closed, async_done) = match &val { - Ok(PyIterReturn::StopIteration(_)) => (true, true), - Err(e) if e.fast_isinstance(vm.ctx.exceptions.generator_exit) => (true, true), - Err(_) => (false, true), - _ => (false, false), + let (frame_done, mark_ag_closed, async_done) = match &val { + Ok(PyIterReturn::StopIteration(_)) => (true, true, true), + Err(e) if e.fast_isinstance(vm.ctx.exceptions.generator_exit) => (true, true, true), + Err(e) if e.fast_isinstance(vm.ctx.exceptions.stop_async_iteration) => { + (false, true, true) + } + Err(_) => (false, false, true), + _ => (false, false, false), }; - if closed { + if frame_done { ag.inner.closed.store(true); } + if mark_ag_closed { + ag.ag_closed.store(true); + } if async_done { ag.running_async.store(false); } @@ -333,6 +344,7 @@ impl PyAsyncGenASend { AwaitableState::Iter => val, // already running, all good AwaitableState::Init => { if self.ag.running_async.load() { + self.state.store(AwaitableState::Closed); return Err( vm.new_runtime_error("anext(): asynchronous generator is already running") ); @@ -473,61 +485,65 @@ impl PyAsyncGenAThrow { #[pymethod] fn send(&self, val: PyObjectRef, vm: &VirtualMachine) -> PyResult { - match self.state.load() { - AwaitableState::Closed => { - Err(vm.new_runtime_error("cannot reuse already awaited aclose()/athrow()")) - } - AwaitableState::Init => { - if self.ag.running_async.load() { - self.state.store(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.store(AwaitableState::Closed); - return Err(vm.new_stop_iteration(None)); - } - if !vm.is_none(&val) { - return Err(vm.new_runtime_error( - "can't send non-None value to a just-started async generator", - )); - } - self.state.store(AwaitableState::Iter); - self.ag.running_async.store(true); - - let (ty, val, tb) = self.value.clone(); - let ret = self.ag.inner.throw(self.ag.as_object(), ty, val, tb, vm); - if self.aclose && self.ignored_close(&ret) { - return Err(self.yield_close(vm)); - } - let ret = if self.aclose { - ret.and_then(|o| o.into_async_pyresult(vm)) + if matches!(self.state.load(), AwaitableState::Closed) { + return Err(vm.new_runtime_error("cannot reuse already awaited aclose()/athrow()")); + } + if self.ag.inner.closed() { + self.state.store(AwaitableState::Closed); + return Err(vm.new_stop_iteration(None)); + } + if matches!(self.state.load(), AwaitableState::Init) { + if self.ag.running_async.load() { + self.state.store(AwaitableState::Closed); + let msg = if self.aclose { + "aclose(): asynchronous generator is already running" } else { - PyAsyncGenWrappedValue::unbox(&self.ag, ret, vm) + "athrow(): asynchronous generator is already running" }; - ret.map_err(|e| self.check_error(e, vm)) + return Err(vm.new_runtime_error(msg.to_owned())); } - AwaitableState::Iter => { - let ret = self.ag.inner.send(self.ag.as_object(), val, vm); - if self.aclose { - match ret { - Ok(PyIterReturn::Return(v)) - if v.downcastable::() => - { - Err(self.yield_close(vm)) - } - other => other - .and_then(|o| o.into_async_pyresult(vm)) - .map_err(|e| self.check_error(e, vm)), - } - } else { - PyAsyncGenWrappedValue::unbox(&self.ag, ret, vm) + if self.ag.ag_closed.load() { + self.state.store(AwaitableState::Closed); + return Err( + vm.new_exception_empty(vm.ctx.exceptions.stop_async_iteration.to_owned()) + ); + } + if !vm.is_none(&val) { + return Err( + vm.new_runtime_error("can't send non-None value to a just-started coroutine") + ); + } + self.state.store(AwaitableState::Iter); + self.ag.running_async.store(true); + if self.aclose { + self.ag.ag_closed.store(true); + } + + let (ty, val, tb) = self.value.clone(); + let ret = self.ag.inner.throw(self.ag.as_object(), ty, val, tb, vm); + if self.aclose && self.ignored_close(&ret) { + return Err(self.yield_close(vm)); + } + let ret = if self.aclose { + ret.and_then(|o| o.into_async_pyresult(vm)) + } else { + PyAsyncGenWrappedValue::unbox(&self.ag, ret, vm) + }; + return ret.map_err(|e| self.check_error(e, vm)); + } + + let ret = self.ag.inner.send(self.ag.as_object(), val, vm); + if self.aclose { + match ret { + Ok(PyIterReturn::Return(v)) if v.downcastable::() => { + Err(self.yield_close(vm)) } + other => other + .and_then(|o| o.into_async_pyresult(vm)) + .map_err(|e| self.check_error(e, vm)), } + } else { + PyAsyncGenWrappedValue::unbox(&self.ag, ret, vm) } } @@ -619,7 +635,6 @@ impl PyAsyncGenAThrow { } fn check_error(&self, exc: PyBaseExceptionRef, vm: &VirtualMachine) -> PyBaseExceptionRef { self.ag.running_async.store(false); - self.ag.inner.closed.store(true); self.state.store(AwaitableState::Closed); if self.aclose && (exc.fast_isinstance(vm.ctx.exceptions.stop_async_iteration) @@ -829,7 +844,7 @@ impl Destructor for PyAsyncGen { if zelf.inner.closed() { return Ok(()); } - if zelf.ag_finalizer.lock().clone().is_some() { + if zelf.ag_finalizer.lock().clone().is_some() && !zelf.ag_closed.load() { Self::call_finalizer(zelf, vm); return Ok(()); } diff --git a/crates/wasm/src/ssl.rs b/crates/wasm/src/ssl.rs index 26e687c9d..443aadf16 100644 --- a/crates/wasm/src/ssl.rs +++ b/crates/wasm/src/ssl.rs @@ -781,7 +781,7 @@ mod _ssl { }; if !incoming.is_empty() { conn.feed_tls(&incoming).map_err(|err| map_tls(vm, err))?; - } else if eof { + } else if eof && conn.pending_plaintext() == 0 { conn.feed_tls(&[]).map_err(|err| map_tls(vm, err))?; } conn.process_packets().map_err(|err| map_tls(vm, err))?;