--- src/lib.rs.before
+++ src/lib.rs.experiment
@@ -1,4 +1,5 @@
//! Linux HTTP/1 ASGI prototype. One event-loop thread owns all native state.
+mod channel;
mod http;
mod net;
use pyo3::exceptions::{PyOSError, PyRuntimeError, PyStopIteration, PyValueError};
@@ -189,6 +190,8 @@
trailers_started: bool,
}
struct State {
+ channel_key: Cell<u64>,
+ channel_type: Py<pyo3::types::PyType>,
listener: TcpListener,
poller: net::Poller,
app: Py<PyAny>,
@@ -597,20 +600,9 @@
trailers_started: false,
}));
c.borrow_mut().active = Some(r.clone());
- let send = Py::new(
- py,
- Sender {
- state: Rc::downgrade(self),
- request: r.clone(),
- },
- )?;
- let receive = Py::new(
- py,
- Receiver {
- state: Rc::downgrade(self),
- request: r.clone(),
- },
- )?;
+ let token = c.borrow().token;
+ let send = channel::new(py, self, token, id, true)?;
+ let receive = channel::new(py, self, token, id, false)?;
let result = (|| -> PyResult<()> {
let coro = self.app.bind(py).call1((scope, receive, send))?;
// Task's eager mode runs until the first real wait. Preserve embedding
@@ -735,14 +727,12 @@
Ok(())
}
-#[pyclass(unsendable)]
struct Sender {
state: Weak<State>,
request: Request,
}
-#[pymethods]
impl Sender {
- fn __call__(&self, py: Python<'_>, msg: &Bound<'_, PyDict>) -> PyResult<Py<PyAny>> {
+ fn call(&self, py: Python<'_>, msg: &Bound<'_, PyDict>) -> PyResult<Py<PyAny>> {
let s = self
.state
.upgrade()
@@ -1037,14 +1027,12 @@
}
}
-#[pyclass(unsendable)]
struct Receiver {
state: Weak<State>,
request: Request,
}
-#[pymethods]
impl Receiver {
- fn __call__(&self, py: Python<'_>) -> PyResult<Py<PyAny>> {
+ fn call(&self, py: Python<'_>) -> PyResult<Py<PyAny>> {
let Some(s) = self.state.upgrade() else {
return ready(py, message(py, "http.disconnect", None)?, None);
};
@@ -1183,6 +1171,11 @@
#[pyclass(unsendable)]
struct Worker {
state: Rc<State>,
+}
+impl Drop for Worker {
+ fn drop(&mut self) {
+ channel::unregister(self.state.channel_key.get());
+ }
}
#[pymethods]
impl Worker {
@@ -1216,28 +1209,30 @@
let task_options = PyDict::new(py);
task_options.set_item("loop", loop_.bind(py))?;
task_options.set_item("eager_start", true)?;
- Ok(Self {
- state: Rc::new(State {
- listener,
- poller,
- app,
- loop_,
- lifespan,
- prototype: prototype.unbind(),
- done: done.unbind(),
- task_type,
- task_options: task_options.unbind(),
- connections: RefCell::new(HashMap::new()),
- tasks: RefCell::new(HashMap::new()),
- next: Cell::new(0),
- sends: Cell::new(0),
- stopping: Cell::new(false),
- date: RefCell::new(Vec::new()),
- date_at: Cell::new(0),
- read_buffer: RefCell::new(vec![0; 65536]),
- header_names: RefCell::new(HashMap::new()),
- }),
- })
+ let state = Rc::new(State {
+ channel_key: Cell::new(0),
+ channel_type: channel::kind(py)?,
+ listener,
+ poller,
+ app,
+ loop_,
+ lifespan,
+ prototype: prototype.unbind(),
+ done: done.unbind(),
+ task_type,
+ task_options: task_options.unbind(),
+ connections: RefCell::new(HashMap::new()),
+ tasks: RefCell::new(HashMap::new()),
+ next: Cell::new(0),
+ sends: Cell::new(0),
+ stopping: Cell::new(false),
+ date: RefCell::new(Vec::new()),
+ date_at: Cell::new(0),
+ read_buffer: RefCell::new(vec![0; 65536]),
+ header_names: RefCell::new(HashMap::new()),
+ });
+ state.channel_key.set(channel::register(&state)?);
+ Ok(Self { state })
}
fn fileno(&self) -> i32 {
self.state.poller.fd()
--- /dev/null
+++ src/channel.rs
@@ -0,0 +1,193 @@
+//! CPython vectorcall channels contain only identities, never thread-bound Rust
+//! pointers. Lookup is confined to the owning thread; stale channels cannot
+//! access a subsequent request or keep a connection alive.
+use super::*;
+use pyo3::{exceptions::PyTypeError, ffi, types::PyType};
+use std::ffi::c_void;
+use std::ptr;
+use std::sync::atomic::{AtomicU64, Ordering};
+
+thread_local! {
+ static WORKERS: RefCell<Vec<(u64, Weak<State>)>> = const { RefCell::new(Vec::new()) };
+}
+static NEXT: AtomicU64 = AtomicU64::new(1);
+pub fn register(state: &Rc<State>) -> PyResult<u64> {
+ let id = NEXT
+ .fetch_update(Ordering::Relaxed, Ordering::Relaxed, |v| v.checked_add(1))
+ .map_err(|_| PyRuntimeError::new_err("Worker identity exhausted"))?;
+ WORKERS.with(|w| w.borrow_mut().push((id, Rc::downgrade(state))));
+ Ok(id)
+}
+pub fn unregister(id: u64) {
+ WORKERS.with(|w| w.borrow_mut().retain(|(key, _)| *key != id));
+}
+#[repr(C)]
+struct Channel {
+ head: ffi::PyObject,
+ vectorcall: Option<ffi::vectorcallfunc>,
+ worker: u64,
+ connection: u64,
+ request: u64,
+}
+// SAFETY: CPython calls this only for our non-GC, non-subclassable heap type.
+// The payload has no Python references or Rust destructors, so destruction is
+// also safe on a different Python thread. GenericAlloc owns one type reference.
+unsafe extern "C" fn dealloc(obj: *mut ffi::PyObject) {
+ unsafe {
+ let kind = ffi::Py_TYPE(obj);
+ ffi::PyObject_Free(obj.cast());
+ ffi::Py_DECREF(kind.cast());
+ }
+}
+fn invoke(
+ py: Python<'_>,
+ channel: &Channel,
+ msg: Option<&Bound<'_, PyAny>>,
+) -> PyResult<Py<PyAny>> {
+ let state = WORKERS.with(|w| {
+ w.borrow()
+ .iter()
+ .find(|(id, _)| *id == channel.worker)
+ .and_then(|(_, s)| s.upgrade())
+ });
+ let request = state.as_ref().and_then(|s| {
+ let c = s.connections.borrow().get(&channel.connection).cloned()?;
+ let r = c.borrow().active.clone()?;
+ let matches = r.borrow().id == channel.request;
+ matches.then_some(r)
+ });
+ if let (Some(s), Some(request)) = (&state, request) {
+ let state = Rc::downgrade(s);
+ if let Some(msg) = msg {
+ Sender { state, request }.call(py, msg.cast::<PyDict>()?)
+ } else {
+ Receiver { state, request }.call(py)
+ }
+ } else if msg.is_some()
+ && state
+ .as_ref()
+ .is_some_and(|s| s.connections.borrow().contains_key(&channel.connection))
+ {
+ Err(PyRuntimeError::new_err("Response already completed"))
+ } else if msg.is_some() {
+ Err(PyOSError::new_err(
+ "ASGI channel is closed or belongs to another thread",
+ ))
+ } else {
+ ready(py, message(py, "http.disconnect", None)?, state.as_deref())
+ }
+}
+// SAFETY: vectorcall provides a live Channel, an array of borrowed arguments,
+// and the attached Python thread. No Rust unwind may cross the C ABI boundary.
+unsafe fn call(
+ obj: *mut ffi::PyObject,
+ args: *const *mut ffi::PyObject,
+ nargs: usize,
+ names: *mut ffi::PyObject,
+ send: bool,
+) -> *mut ffi::PyObject {
+ let py = unsafe { Python::assume_attached() };
+ let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
+ if unsafe { ffi::PyVectorcall_NARGS(nargs) } != isize::from(send)
+ || (!names.is_null() && unsafe { ffi::PyTuple_Size(names) } != 0)
+ {
+ return Err(PyTypeError::new_err(
+ "ASGI channel expects positional arguments",
+ ));
+ }
+ let msg = send.then(|| unsafe { Bound::from_borrowed_ptr(py, *args) });
+ invoke(py, unsafe { &*obj.cast::<Channel>() }, msg.as_ref())
+ }))
+ .unwrap_or_else(|_| Err(PyRuntimeError::new_err("Native ASGI channel panicked")));
+ match result {
+ Ok(value) => value.into_ptr(),
+ Err(error) => {
+ error.restore(py);
+ ptr::null_mut()
+ }
+ }
+}
+unsafe extern "C" fn send(
+ obj: *mut ffi::PyObject,
+ args: *const *mut ffi::PyObject,
+ nargs: usize,
+ names: *mut ffi::PyObject,
+) -> *mut ffi::PyObject {
+ unsafe { call(obj, args, nargs, names, true) }
+}
+unsafe extern "C" fn receive(
+ obj: *mut ffi::PyObject,
+ args: *const *mut ffi::PyObject,
+ nargs: usize,
+ names: *mut ffi::PyObject,
+) -> *mut ffi::PyObject {
+ unsafe { call(obj, args, nargs, names, false) }
+}
+pub fn kind(py: Python<'_>) -> PyResult<Py<PyType>> {
+ let mut members = [
+ ffi::PyMemberDef {
+ name: c"__vectorcalloffset__".as_ptr(),
+ type_code: ffi::Py_T_PYSSIZET,
+ offset: std::mem::offset_of!(Channel, vectorcall) as isize,
+ flags: ffi::Py_READONLY,
+ doc: ptr::null(),
+ },
+ unsafe { std::mem::zeroed() },
+ ];
+ let mut slots = [
+ ffi::PyType_Slot {
+ slot: ffi::Py_tp_dealloc,
+ pfunc: dealloc as *mut c_void,
+ },
+ ffi::PyType_Slot {
+ slot: ffi::Py_tp_call,
+ pfunc: ffi::PyVectorcall_Call as *mut c_void,
+ },
+ ffi::PyType_Slot {
+ slot: ffi::Py_tp_members,
+ pfunc: members.as_mut_ptr().cast(),
+ },
+ ffi::PyType_Slot {
+ slot: 0,
+ pfunc: ptr::null_mut(),
+ },
+ ];
+ let mut spec = ffi::PyType_Spec {
+ name: c"skirr.Channel".as_ptr(),
+ basicsize: std::mem::size_of::<Channel>() as i32,
+ itemsize: 0,
+ flags: (ffi::Py_TPFLAGS_DEFAULT
+ | ffi::Py_TPFLAGS_HAVE_VECTORCALL
+ | ffi::Py_TPFLAGS_IMMUTABLETYPE
+ | ffi::Py_TPFLAGS_DISALLOW_INSTANTIATION) as u32,
+ slots: slots.as_mut_ptr(),
+ };
+ // SAFETY: FromSpec copies slots/member definitions into the heap type;
+ // their names have static storage. Errors transfer to PyO3 as usual.
+ unsafe {
+ Bound::from_owned_ptr_or_err(py, ffi::PyType_FromSpec(&mut spec))?
+ .cast_into::<PyType>()
+ .map(Bound::unbind)
+ .map_err(Into::into)
+ }
+}
+pub fn new(
+ py: Python<'_>,
+ s: &State,
+ connection: u64,
+ request: u64,
+ sending: bool,
+) -> PyResult<Py<PyAny>> {
+ // SAFETY: allocate the exact immutable type above; initialize all POD fields
+ // before exposing it. GenericAlloc zeroes memory and retains the heap type.
+ unsafe {
+ let obj = ffi::PyType_GenericAlloc(s.channel_type.as_ptr().cast(), 0);
+ let value = Bound::from_owned_ptr_or_err(py, obj)?;
+ let c = obj.cast::<Channel>();
+ (*c).vectorcall = Some(if sending { send } else { receive });
+ (*c).worker = s.channel_key.get();
+ (*c).connection = connection;
+ (*c).request = request;
+ Ok(value.unbind())
+ }
+}