--- src/lib.rs.before
+++ src/lib.rs
@@ -19,13 +19,24 @@
type Link = Rc<RefCell<Connection>>;
type Request = Rc<RefCell<RequestState>>;
-fn ready(py: Python<'_>, value: Py<PyAny>, state: Option<&State>) -> PyResult<Py<PyAny>> {
- let yield_first = state.is_some_and(|s| {
- let n = s.sends.get() + 1;
- s.sends.set(if n >= 128 { 0 } else { n });
- n >= 128
- });
- completed::value(py, Some(value), yield_first, state)
+// Fairness belongs to the producing request. Unrelated short requests must
+// not spend each other's quota and delay cleanup after an already-written body.
+fn cooperate(request: &Request) -> bool {
+ let mut r = request.borrow_mut();
+ r.operations += 1;
+ if r.operations < 128 {
+ return false;
+ }
+ r.operations = 0;
+ true
+}
+fn ready(
+ py: Python<'_>,
+ value: Py<PyAny>,
+ request: &Request,
+ state: Option<&State>,
+) -> PyResult<Py<PyAny>> {
+ completed::value(py, Some(value), cooperate(request), state)
}
fn message(py: Python<'_>, kind: &str, body: Option<(Py<PyBytes>, bool)>) -> PyResult<Py<PyAny>> {
let d = PyDict::new(py);
@@ -156,6 +167,7 @@
head: bool,
minor: u8,
task_done: bool,
+ operations: usize,
trailers: bool,
body_finished: bool,
trailers_started: bool,
@@ -173,7 +185,7 @@
connections: RefCell<HashMap<u64, Link>>,
tasks: RefCell<HashMap<u64, Py<PyAny>>>,
next: Cell<u64>,
- sends: Cell<usize>,
+ starts: RefCell<HashMap<u64, Py<Start>>>,
stopping: Cell<bool>,
date: RefCell<Vec<u8>>,
date_at: Cell<u64>,
@@ -276,6 +288,10 @@
"set_exception",
(PyOSError::new_err("Client disconnected").value(py),),
)?;
+ }
+ let start = self.starts.borrow_mut().remove(&id);
+ if let Some(start) = start {
+ start.borrow(py).discard(py)?;
}
let task = self.tasks.borrow().get(&id).map(|t| t.clone_ref(py));
if let Some(t) = task {
@@ -570,6 +586,7 @@
head: h.method == "HEAD",
minor: h.minor,
task_done: false,
+ operations: 0,
trailers: false,
body_finished: false,
trailers_started: false,
@@ -590,7 +607,35 @@
},
)?;
let result = (|| -> PyResult<()> {
- let coro = self.app.bind(py).call1((scope, receive, send))?;
+ let coro = self.app.bind(py).call1((scope, receive, send))?.unbind();
+ let start = Py::new(
+ py,
+ Start {
+ state: Rc::downgrade(self),
+ request: r,
+ coroutine: RefCell::new(Some(coro)),
+ },
+ )?;
+ self.starts.borrow_mut().insert(id, start.clone_ref(py));
+ self.loop_
+ .bind(py)
+ .call_method1(pyo3::intern!(py, "call_soon"), (start,))?;
+ Ok(())
+ })();
+ if let Err(e) = result {
+ e.print(py);
+ self.error(py, c, 500)?;
+ }
+ Ok(())
+ }
+ fn launch(
+ self: &Rc<Self>,
+ py: Python<'_>,
+ c: &Link,
+ request: &Request,
+ coro: Py<PyAny>,
+ ) -> PyResult<()> {
+ let result = (|| -> PyResult<()> {
// Task's eager mode runs until the first real wait. Preserve embedding
// applications' custom task factories by using their normal loop path.
let task = if self
@@ -609,19 +654,21 @@
};
let done = Done {
state: Rc::downgrade(self),
- request: r.clone(),
+ request: request.clone(),
};
if task.call_method0(pyo3::intern!(py, "done"))?.is_truthy()? {
// An eager task can be finished without allocating another Python
- // callback or scheduling another event-loop turn. The outer pump
- // owns pipeline progress, so this path must not recurse into it.
- done.finish(py, &task, false)?;
+ // callback or scheduling another event-loop turn. Pipeline progress
+ // queues the next Start; it cannot recurse into application code.
+ done.finish(py, &task, true)?;
} else {
task.call_method1(
pyo3::intern!(py, "add_done_callback"),
(Py::new(py, done)?,),
)?;
- self.tasks.borrow_mut().insert(id, task.unbind());
+ self.tasks
+ .borrow_mut()
+ .insert(request.borrow().id, task.unbind());
}
Ok(())
})();
@@ -1000,10 +1047,7 @@
} else {
return Err(PyValueError::new_err("Unsupported ASGI message"));
}
- let n = s.sends.get() + 1;
- s.sends.set(n);
- if n >= 128 {
- s.sends.set(0);
+ if cooperate(&self.request) {
return completed::value(py, None, true, Some(&s));
}
Ok(s.done.clone_ref(py))
@@ -1019,13 +1063,28 @@
impl Receiver {
fn __call__(&self, py: Python<'_>) -> PyResult<Py<PyAny>> {
let Some(s) = self.state.upgrade() else {
- return ready(py, message(py, "http.disconnect", None)?, None);
+ return ready(
+ py,
+ message(py, "http.disconnect", None)?,
+ &self.request,
+ None,
+ );
};
let Some(c) = self.request.borrow().connection.upgrade() else {
- return ready(py, message(py, "http.disconnect", None)?, Some(&s));
+ return ready(
+ py,
+ message(py, "http.disconnect", None)?,
+ &self.request,
+ Some(&s),
+ );
};
if c.borrow().dead || self.request.borrow().complete || self.request.borrow().task_done {
- return ready(py, message(py, "http.disconnect", None)?, Some(&s));
+ return ready(
+ py,
+ message(py, "http.disconnect", None)?,
+ &self.request,
+ Some(&s),
+ );
}
s.pump(py, &c)?;
if !self.request.borrow().body.is_empty()
@@ -1033,7 +1092,7 @@
{
let msg = take_message(py, &self.request)?;
s.pump(py, &c)?;
- return ready(py, msg, Some(&s));
+ return ready(py, msg, &self.request, Some(&s));
}
let old = self.request.borrow_mut().receive.take();
if let Some(f) = old
@@ -1061,6 +1120,44 @@
.unbind();
self.request.borrow_mut().receive = Some(f.clone_ref(py));
Ok(f)
+ }
+}
+
+#[pyclass(unsendable, frozen)]
+struct Start {
+ state: Weak<State>,
+ request: Request,
+ coroutine: RefCell<Option<Py<PyAny>>>,
+}
+impl Start {
+ fn discard(&self, py: Python<'_>) -> PyResult<()> {
+ let coro = self.coroutine.borrow_mut().take();
+ if let Some(coro) = coro {
+ coro.bind(py).call_method0(pyo3::intern!(py, "close"))?;
+ }
+ Ok(())
+ }
+}
+#[pymethods]
+impl Start {
+ fn __call__(&self, py: Python<'_>) -> PyResult<()> {
+ let Some(s) = self.state.upgrade() else {
+ return self.discard(py);
+ };
+ s.starts.borrow_mut().remove(&self.request.borrow().id);
+ let Some(coro) = self.coroutine.borrow_mut().take() else {
+ return Ok(());
+ };
+ let c = self.request.borrow().connection.upgrade();
+ let alive = c.as_ref().is_some_and(|c| !c.borrow().dead);
+ if let Some(c) = c
+ && alive
+ {
+ s.launch(py, &c, &self.request, coro)
+ } else {
+ coro.bind(py).call_method0(pyo3::intern!(py, "close"))?;
+ Ok(())
+ }
}
}
@@ -1218,7 +1315,7 @@
connections: RefCell::new(HashMap::new()),
tasks: RefCell::new(HashMap::new()),
next: Cell::new(0),
- sends: Cell::new(0),
+ starts: RefCell::new(HashMap::new()),
stopping: Cell::new(false),
date: RefCell::new(Vec::new()),
date_at: Cell::new(0),
@@ -1236,7 +1333,6 @@
}
fn poll(&self, py: Python<'_>) -> PyResult<()> {
let s = &self.state;
- s.sends.set(0);
for (token, mask) in s.poller.events()? {
if token == 0 {
if s.stopping.get() {
@@ -1338,7 +1434,8 @@
Ok(())
}
fn pending(&self) -> usize {
- self.state.tasks.borrow().len()
+ self.state.starts.borrow().len()
+ + self.state.tasks.borrow().len()
+ self
.state
.connections