swadhin / justapi public

⚡ p3: native_async + sse_specs dispatch in handler pipeline (ADR-088/089)

swadhinbiswas committed 2 months ago
d8c16dd
1 parent 8d3c5dc
commit d8c16dd
1 file changed
+265 -123
Showing 1 changed file+265-123
crates/justapi-py/src/native/handlers.rs
▼crates/justapi-py/src/native/handlers.rs
+265-123
@@ -103,6 +103,79 @@ fn handle_ok_result(
103
103
serialize_response(py, res, helper)
104
104
}
105
105
106
+/// Build the Python `Request` + multipart form dict and invoke the handler.
107
+/// Shared by the sync (GIL-pool) and async (parallel pool) dispatch paths
108
+/// (ADR-087).
109
+#[allow(clippy::too_many_arguments)]
110
+fn dispatch_python_handler(
111
+ py: Python<'_>,
112
+ handlers: &Arc<Vec<Py<PyAny>>>,
113
+ schemas: &Arc<Vec<Option<Py<PyAny>>>>,
114
+ handler_id: usize,
115
+ path_params: &[(String, String)],
116
+ query_string: &[u8],
117
+ headers: &[(Vec<u8>, Vec<u8>)],
118
+ body: &[u8],
119
+ db_url: Option<&str>,
120
+ trace_ctx: Option<(String, String)>,
121
+ multipart_form_res: Option<justapi_core::multipart::MultipartForm>,
122
+ method: &str,
123
+ path: &str,
124
+ scheme: String,
125
+ client: Option<(String, u16)>,
126
+ app: Option<Py<PyAny>>,
127
+ needs_request: bool,
128
+ http_version: String,
129
+ auth_claims: Option<String>,
130
+) -> NativeResponse {
131
+ let mut form_dict_py: Option<pyo3::Py<pyo3::types::PyDict>> = None;
132
+ if let Some(form) = multipart_form_res {
133
+ let d = pyo3::types::PyDict::new(py);
134
+ for (k, v) in form.fields.iter() {
135
+ d.set_item(k, v).ok();
136
+ }
137
+ for f in form.files.iter() {
138
+ let headers_dict = pyo3::types::PyDict::new(py);
139
+ for (k, v) in f.headers.iter() {
140
+ headers_dict.set_item(k, v).ok();
141
+ }
142
+ let upload_file = crate::multipart::UploadFile::new(
143
+ f.filename.clone().unwrap_or_default(),
144
+ f.content_type.clone().unwrap_or_default(),
145
+ f.size,
146
+ headers_dict.into(),
147
+ f.temp_path.clone(),
148
+ );
149
+ let Ok(p) = pyo3::Bound::new(py, upload_file) else {
150
+ continue;
151
+ };
152
+ d.set_item(&f.field_name, p).ok();
153
+ }
154
+ form_dict_py = Some(d.into());
155
+ }
156
+
157
+ call_python_handler(
158
+ py,
159
+ &handlers[handler_id],
160
+ schemas[handler_id].as_ref(),
161
+ path_params,
162
+ query_string,
163
+ headers,
164
+ body,
165
+ db_url,
166
+ trace_ctx,
167
+ form_dict_py,
168
+ method,
169
+ path,
170
+ scheme,
171
+ client,
172
+ app,
173
+ needs_request,
174
+ http_version,
175
+ auth_claims,
176
+ )
177
+}
178
+
106
179
#[allow(clippy::too_many_arguments)]
107
180
pub(crate) fn call_python_handler(
108
181
py: Python<'_>,
@@ -521,23 +594,52 @@ pub(crate) fn serialize_response(
521
594
/// serialize the resolved value with the same fast serializer the sync path
522
595
/// uses. Runs on a `spawn_blocking` thread so no worker thread is tied up.
523
596
async fn resolve_async_response(py_future: Py<PyAny>) -> Result<NativeResponse, anyhow::Error> {
524
- // Block on the future's .result(). If the coroutine raised (HTTPException,
525
- // RequestValidationError, ...), the exception surfaces as a PyErr from
526
- // .result() — convert it with the same logic as the sync Err path.
527
- let outcome: Result<Py<PyAny>, pyo3::PyErr> = tokio::task::spawn_blocking(move || {
528
- Python::attach(|py| py_future.bind(py).call_method0("result").map(|v| v.unbind()))
597
+ // Callback-driven completion: register a `_DoneNotifier` on the
598
+ // concurrent.futures.Future so we can await completion on the tokio
599
+ // runtime without blocking a thread (no spawn_blocking) or polling.
600
+ // The callback fires on the asyncio loop thread the instant the coroutine
601
+ // finishes. This removes one thread hop vs the old spawn_blocking path.
602
+ let (done_tx, done_rx) = tokio::sync::oneshot::channel::<()>();
603
+ // Register the callback on a spawn_blocking thread so no GIL attach ever
604
+ // happens on the server threads (which would contend with sync handlers).
605
+ // Returns the future so the result-read closure can use it.
606
+ let register = tokio::task::spawn_blocking(move || {
607
+ Python::attach(|py| -> Result<Py<PyAny>, anyhow::Error> {
608
+ let notifier = pyo3::Bound::new(
609
+ py,
610
+ crate::native::types::DoneNotifier { tx: std::sync::Mutex::new(Some(done_tx)) },
611
+ )
612
+ .map_err(|e| anyhow::anyhow!("failed to create done notifier: {}", e))?;
613
+ let fut = py_future.bind(py);
614
+ fut.call_method1("add_done_callback", (notifier,))
615
+ .map_err(|e| anyhow::anyhow!("async handler callback error: {}", e))?;
616
+ Ok(py_future)
617
+ })
618
+ });
619
+ let py_future =
620
+ register.await.map_err(|e| anyhow::anyhow!("async callback join error: {}", e))??;
621
+
622
+ // Wait for the coroutine to finish (pure tokio await, no thread blocked).
623
+ done_rx.await.map_err(|_| anyhow::anyhow!("async handler dropped"))?;
624
+
625
+ // Read + serialize the result on spawn_blocking so the GIL attaches never
626
+ // contend with sync handlers on the server threads (the coroutine is
627
+ // already done, so .result() returns instantly).
628
+ tokio::task::spawn_blocking(move || {
629
+ Python::attach(|py| {
630
+ let outcome: Result<Py<PyAny>, pyo3::PyErr> =
631
+ py_future.bind(py).call_method0("result").map(|v| v.unbind());
632
+ match outcome {
633
+ Ok(res) => {
634
+ let helper = get_helper(py);
635
+ handle_ok_result(py, res.bind(py).into(), helper)
636
+ }
637
+ Err(e) => handle_py_error(py, e),
638
+ }
639
+ })
529
640
})
530
641
.await
531
- .map_err(|e| anyhow::anyhow!("async handler join error: {}", e))?;
532
-
533
- let nr = Python::attach(|py| match outcome {
534
- Ok(res) => {
535
- let helper = get_helper(py);
536
- handle_ok_result(py, res.bind(py).into(), helper)
537
- }
538
- Err(e) => handle_py_error(py, e),
539
- });
540
- Ok(nr)
642
+ .map_err(|e| anyhow::anyhow!("async handler join error: {}", e))
541
643
}
542
644
543
645
pub(crate) fn nr_to_response(nr: NativeResponse) -> hyper::Response<ResponseBody> {
@@ -748,7 +850,9 @@ pub(crate) fn make_native_handler<B>(
748
850
app: Option<Py<PyAny>>,
749
851
needs_request: Arc<Vec<bool>>,
750
852
native: Arc<Vec<bool>>,
853
+ _native_async: Arc<Vec<bool>>,
751
854
crud_specs: CrudConfig,
855
+ sse_specs: Arc<Vec<Option<(u64, u64)>>>,
752
856
schema_validators: Arc<Vec<Option<justapi_core::validate::CompiledValidator>>>,
753
857
max_body_size: usize,
754
858
) -> justapi_core::middleware::HandlerFn<B>
@@ -782,7 +886,10 @@ where
782
886
.to_string();
783
887
let needs_request_clone = needs_request.clone();
784
888
let native_clone = native.clone();
889
+ #[cfg(Py_GIL_DISABLED)]
890
+ let native_async_clone = _native_async.clone();
785
891
let crud_specs_clone = crud_specs.clone();
892
+ let sse_specs_clone = sse_specs.clone();
786
893
let schema_validators_clone = schema_validators.clone();
787
894
Box::pin(async move {
788
895
let method = req.method().clone();
@@ -824,7 +931,7 @@ where
824
931
let (body_bytes, multipart_form_res) = if is_multipart {
825
932
let ct = content_type.unwrap();
826
933
match justapi_core::multipart::parse_multipart(req_body, &ct).await {
827
- Ok(form) => (Bytes::new(), Some(Ok::<_, anyhow::Error>(form))),
934
+ Ok(form) => (Bytes::new(), Some(form)),
828
935
Err(e) => {
829
936
let msg = e.to_string();
830
937
if msg.contains("exceeds maximum size") {
@@ -906,6 +1013,12 @@ where
906
1013
return Ok(nr_to_response(res));
907
1014
}
908
1015
1016
+ // Rust-native SSE stream (ADR-088): serve events generated
1017
+ // entirely in Rust — no Python handler, no GIL, no pump.
1018
+ if let Some(Some((count, interval_ms))) = sse_specs_clone.get(handler_id) {
1019
+ return Ok(justapi_core::server::sse_stream_response(*count, *interval_ms));
1020
+ }
1021
+
909
1022
// Rust-native CRUD (ADR-056 Step C): validate body + run INSERT /
910
1023
// SELECT / UPDATE / DELETE entirely in Rust, no GIL/Python hop. The
911
1024
// operation is inferred from the request method; the body was already
@@ -998,60 +1111,72 @@ where
998
1111
let db_url_str = db_url.clone();
999
1112
let trace_ctx = justapi_core::trace_context::get_current_trace_context();
1000
1113
1001
- let nr = crate::gil_pool::run_python(move |py| {
1002
- let mut form_dict_py: Option<pyo3::Py<pyo3::types::PyDict>> = None;
1003
- if let Some(Ok(form)) = multipart_form_res {
1004
- let d = pyo3::types::PyDict::new(py);
1005
- for (k, v) in form.fields.iter() {
1006
- d.set_item(k, v).ok();
1007
- }
1008
- for f in form.files.iter() {
1009
- let headers_dict = pyo3::types::PyDict::new(py);
1010
- for (k, v) in f.headers.iter() {
1011
- headers_dict.set_item(k, v).ok();
1012
- }
1013
- let upload_file = crate::multipart::UploadFile::new(
1014
- f.filename.clone().unwrap_or_default(),
1015
- f.content_type.clone().unwrap_or_default(),
1016
- f.size,
1017
- headers_dict.into(),
1018
- f.temp_path.clone(),
1019
- );
1020
- let Ok(p) = pyo3::Bound::new(py, upload_file) else {
1021
- continue;
1022
- };
1023
- d.set_item(&f.field_name, p).ok();
1024
- }
1025
- form_dict_py = Some(d.into());
1026
- }
1027
-
1028
- call_python_handler(
1029
- py,
1030
- &handlers_clone[handler_id],
1031
- schemas_clone[handler_id].as_ref(),
1032
- &path_params,
1033
- &query_string,
1034
- &headers,
1035
- &body_bytes,
1036
- db_url_str.as_deref(),
1037
- trace_ctx,
1038
- form_dict_py,
1039
- method.as_str(),
1040
- &path,
1041
- req_scheme.clone(),
1042
- req_client.clone(),
1043
- app.as_ref().as_ref().map(|a| a.clone_ref(py)),
1044
- needs_request_clone[handler_id],
1045
- http_version.clone(),
1046
- auth_claims,
1047
- )
1048
- })
1049
- .await
1050
- .map_err(|e| anyhow::anyhow!("GIL pool error: {}", e))?;
1114
+ // `@native_async` + free-threaded CPython: dispatch in TRUE
1115
+ // parallel (no GIL to thrash) — the GIL pool scales to the
1116
+ // hardware. On GIL-locked builds, native_async uses the standard
1117
+ // pool + callback resolution (the optimal path there).
1118
+ #[cfg(Py_GIL_DISABLED)]
1119
+ let is_parallel = native_async_clone.get(handler_id).copied().unwrap_or(false);
1120
+ #[cfg(not(Py_GIL_DISABLED))]
1121
+ let is_parallel = false;
1122
+
1123
+ let nr = if is_parallel {
1124
+ crate::gil_pool::run_python_parallel(move |py| {
1125
+ dispatch_python_handler(
1126
+ py,
1127
+ &handlers_clone,
1128
+ &schemas_clone,
1129
+ handler_id,
1130
+ &path_params,
1131
+ &query_string,
1132
+ &headers,
1133
+ &body_bytes,
1134
+ db_url_str.as_deref(),
1135
+ trace_ctx,
1136
+ multipart_form_res.clone(),
1137
+ method.as_str(),
1138
+ &path,
1139
+ req_scheme.clone(),
1140
+ req_client.clone(),
1141
+ app.as_ref().as_ref().map(|a| a.clone_ref(py)),
1142
+ needs_request_clone[handler_id],
1143
+ http_version.clone(),
1144
+ auth_claims,
1145
+ )
1146
+ })
1147
+ .await
1148
+ .map_err(|e| anyhow::anyhow!("GIL pool error: {}", e))?
1149
+ } else {
1150
+ crate::gil_pool::run_python(move |py| {
1151
+ dispatch_python_handler(
1152
+ py,
1153
+ &handlers_clone,
1154
+ &schemas_clone,
1155
+ handler_id,
1156
+ &path_params,
1157
+ &query_string,
1158
+ &headers,
1159
+ &body_bytes,
1160
+ db_url_str.as_deref(),
1161
+ trace_ctx,
1162
+ multipart_form_res.clone(),
1163
+ method.as_str(),
1164
+ &path,
1165
+ req_scheme.clone(),
1166
+ req_client.clone(),
1167
+ app.as_ref().as_ref().map(|a| a.clone_ref(py)),
1168
+ needs_request_clone[handler_id],
1169
+ http_version.clone(),
1170
+ auth_claims,
1171
+ )
1172
+ })
1173
+ .await
1174
+ .map_err(|e| anyhow::anyhow!("GIL pool error: {}", e))?
1175
+ };
1051
1176
1052
- // Async handlers: the GIL worker returned the future without
1053
- // blocking. Await it here (spawn_blocking, GIL released while
1054
- // waiting) so concurrent async requests interleave.
1177
+ // Async handlers: await the future here via the callback-driven
1178
+ // resolution (GIL released while waiting), so concurrent async
1179
+ // requests interleave.
1055
1180
let nr = match nr.body {
1056
1181
NativeBody::Async(fut) => resolve_async_response(fut).await?,
1057
1182
_ => nr,
@@ -1083,6 +1208,8 @@ pub fn make_test_handler<B>(
1083
1208
app: Option<Py<PyAny>>,
1084
1209
needs_request: Arc<Vec<bool>>,
1085
1210
native: Arc<Vec<bool>>,
1211
+ _native_async: Arc<Vec<bool>>,
1212
+ sse_specs: Arc<Vec<Option<(u64, u64)>>>,
1086
1213
schema_validators: Arc<Vec<Option<justapi_core::validate::CompiledValidator>>>,
1087
1214
max_body_size: usize,
1088
1215
) -> justapi_core::middleware::HandlerFn<B>
@@ -1108,6 +1235,9 @@ where
1108
1235
let http_version = http_version.clone();
1109
1236
let needs_request_clone = needs_request.clone();
1110
1237
let native_clone = native.clone();
1238
+ #[cfg(Py_GIL_DISABLED)]
1239
+ let native_async_clone = _native_async.clone();
1240
+ let sse_specs_clone = sse_specs.clone();
1111
1241
let schema_validators_clone = schema_validators.clone();
1112
1242
Box::pin(async move {
1113
1243
let method = req.method().clone();
@@ -1153,9 +1283,8 @@ where
1153
1283
Err(e) => return Err(anyhow::anyhow!("Body error: {}", e)),
1154
1284
};
1155
1285
1156
- let multipart_form_res: Option<
1157
- Result<justapi_core::multipart::MultipartForm, anyhow::Error>,
1158
- > = if is_multipart {
1286
+ let multipart_form_res: Option<justapi_core::multipart::MultipartForm> = if is_multipart
1287
+ {
1159
1288
let ct = content_type.unwrap();
1160
1289
match justapi_core::multipart::parse_multipart(
1161
1290
http_body_util::Full::new(body_bytes.clone()),
@@ -1163,7 +1292,7 @@ where
1163
1292
)
1164
1293
.await
1165
1294
{
1166
- Ok(form) => Some(Ok::<_, anyhow::Error>(form)),
1295
+ Ok(form) => Some(form),
1167
1296
Err(e) => {
1168
1297
let msg = e.to_string();
1169
1298
if msg.contains("exceeds maximum size") {
@@ -1200,6 +1329,11 @@ where
1200
1329
1201
1330
let handler_id = *matched.handler;
1202
1331
1332
+ // Rust-native SSE stream (ADR-088): serve events entirely in Rust.
1333
+ if let Some(Some((count, interval_ms))) = sse_specs_clone.get(handler_id) {
1334
+ return Ok(justapi_core::server::sse_stream_response(*count, *interval_ms));
1335
+ }
1336
+
1203
1337
// Native fast path: validate the body in Rust and echo it back as the
1204
1338
// response, with no blocking-pool thread hop, no GIL acquisition, and
1205
1339
// no Python handler call.
@@ -1250,56 +1384,64 @@ where
1250
1384
1251
1385
let method_clone = method.to_string();
1252
1386
let path_clone = path.clone();
1253
- let nr = crate::gil_pool::run_python(move |py| {
1254
- let mut form_dict_py: Option<pyo3::Py<pyo3::types::PyDict>> = None;
1255
- if let Some(Ok(form)) = &multipart_form_res {
1256
- let d = pyo3::types::PyDict::new(py);
1257
- for (k, v) in form.fields.iter() {
1258
- d.set_item(k, v).ok();
1259
- }
1260
- for f in form.files.iter() {
1261
- let headers_dict = pyo3::types::PyDict::new(py);
1262
- for (k, v) in f.headers.iter() {
1263
- headers_dict.set_item(k, v).ok();
1264
- }
1265
- let upload_file = crate::multipart::UploadFile::new(
1266
- f.filename.clone().unwrap_or_default(),
1267
- f.content_type.clone().unwrap_or_default(),
1268
- f.size,
1269
- headers_dict.into(),
1270
- f.temp_path.clone(),
1271
- );
1272
- let Ok(p) = pyo3::Bound::new(py, upload_file) else {
1273
- continue;
1274
- };
1275
- d.set_item(&f.field_name, p).ok();
1276
- }
1277
- form_dict_py = Some(d.into());
1278
- }
1279
-
1280
- call_python_handler(
1281
- py,
1282
- &handlers_clone[handler_id],
1283
- schemas_clone[handler_id].as_ref(),
1284
- &path_params,
1285
- &query_string,
1286
- &headers,
1287
- &body_bytes,
1288
- db_url_str.as_deref(),
1289
- trace_ctx,
1290
- form_dict_py,
1291
- method_clone.as_str(),
1292
- &path_clone,
1293
- req_scheme.clone(),
1294
- req_client.clone(),
1295
- app.as_ref().as_ref().map(|a| a.clone_ref(py)),
1296
- needs_request_clone[handler_id],
1297
- http_version.clone(),
1298
- None, // auth_claims: test client has no middleware
1299
- )
1300
- })
1301
- .await
1302
- .map_err(|e| anyhow::anyhow!("GIL pool error: {}", e))?;
1387
+ #[cfg(Py_GIL_DISABLED)]
1388
+ let is_parallel = native_async_clone.get(handler_id).copied().unwrap_or(false);
1389
+ #[cfg(not(Py_GIL_DISABLED))]
1390
+ let is_parallel = false;
1391
+
1392
+ let nr = if is_parallel {
1393
+ crate::gil_pool::run_python_parallel(move |py| {
1394
+ dispatch_python_handler(
1395
+ py,
1396
+ &handlers_clone,
1397
+ &schemas_clone,
1398
+ handler_id,
1399
+ &path_params,
1400
+ &query_string,
1401
+ &headers,
1402
+ &body_bytes,
1403
+ db_url_str.as_deref(),
1404
+ trace_ctx,
1405
+ multipart_form_res.clone(),
1406
+ method_clone.as_str(),
1407
+ &path_clone,
1408
+ req_scheme.clone(),
1409
+ req_client.clone(),
1410
+ app.as_ref().as_ref().map(|a| a.clone_ref(py)),
1411
+ needs_request_clone[handler_id],
1412
+ http_version.clone(),
1413
+ None,
1414
+ )
1415
+ })
1416
+ .await
1417
+ .map_err(|e| anyhow::anyhow!("GIL pool error: {}", e))?
1418
+ } else {
1419
+ crate::gil_pool::run_python(move |py| {
1420
+ dispatch_python_handler(
1421
+ py,
1422
+ &handlers_clone,
1423
+ &schemas_clone,
1424
+ handler_id,
1425
+ &path_params,
1426
+ &query_string,
1427
+ &headers,
1428
+ &body_bytes,
1429
+ db_url_str.as_deref(),
1430
+ trace_ctx,
1431
+ multipart_form_res.clone(),
1432
+ method_clone.as_str(),
1433
+ &path_clone,
1434
+ req_scheme.clone(),
1435
+ req_client.clone(),
1436
+ app.as_ref().as_ref().map(|a| a.clone_ref(py)),
1437
+ needs_request_clone[handler_id],
1438
+ http_version.clone(),
1439
+ None,
1440
+ )
1441
+ })
1442
+ .await
1443
+ .map_err(|e| anyhow::anyhow!("GIL pool error: {}", e))?
1444
+ };
1303
1445
1304
1446
// Async handlers: await the future here (GIL released while
1305
1447
// waiting), mirroring make_native_handler.
1306
1448