From 99faf039621dda0ffaceddd94f2d7f71a81fc2f3 Mon Sep 17 00:00:00 2001 From: zTz01 <1773266173@qq.com> Date: Tue, 27 Oct 2026 22:41:25 +0800 Subject: [PATCH] perf: restore asynchronous route publication for Owner writes --- .../src/client_kv_api/external_api.rs | 58 +++++---------- .../external_client_test.rs | 73 ++++++++++++------- fluxon_rs/fluxon_kv/src/kv_test.rs | 41 +++++++++++ 3 files changed, 107 insertions(+), 65 deletions(-) diff --git a/fluxon_rs/fluxon_kv/src/client_kv_api/external_api.rs b/fluxon_rs/fluxon_kv/src/client_kv_api/external_api.rs index 398400f..7760b64 100755 --- a/fluxon_rs/fluxon_kv/src/client_kv_api/external_api.rs +++ b/fluxon_rs/fluxon_kv/src/client_kv_api/external_api.rs @@ -468,11 +468,11 @@ async fn release_external_local_first_pending_slot( } } -async fn publish_external_local_first( +fn spawn_external_local_first_publish( inner: &client_kv_api::ClientKvApiInner, items: Vec, pending_contexts: Vec, -) -> KvResult>> { +) { assert_eq!( items.len(), pending_contexts.len(), @@ -480,8 +480,7 @@ async fn publish_external_local_first( ); let view = inner.view.clone_view(); let spawn_view = view.clone(); - let (completion_tx, completion_rx) = ::tokio::sync::oneshot::channel(); - // The registered owner task retains the slots and fences if the RPC is cancelled. + // ACK covers local commit. The owner task retains slots and fences through publication. spawn_view.spawn("external_local_first_route_publish", async move { let results = client_kv_api::put::publish_external_owner_local_job( view, @@ -492,13 +491,12 @@ async fn publish_external_local_first( }, ) .await; - let _ = completion_tx.send(results); + for result in results { + if let Err(error) = result { + tracing::warn!(%error, "external local-first background publication failed"); + } + } }); - completion_rx.await.map_err(|_| { - KvError::Api(ApiError::SystemShutdown { - detail: "owner Put publication task stopped before completion".to_string(), - }) - }) } pub(crate) fn normalize_external_get_start_group_lens( @@ -4854,7 +4852,6 @@ impl HandlerForExternalClient for ClientKvApi { let mut revoke_pending = Vec::new(); let mut local_publish_items = Vec::new(); let mut local_publish_contexts = Vec::new(); - let mut local_publish_indices = Vec::new(); for (idx, item) in req.items.into_iter().enumerate() { let Some(put_id) = item.put_id else { @@ -4939,7 +4936,10 @@ impl HandlerForExternalClient for ClientKvApi { radix: pending_ctx.radix.clone(), }); local_publish_contexts.push(pending_ctx); - local_publish_indices.push(idx); + results[idx] = Some(ExternalBatchPutTransferEndItemResp { + error_code: OK, + error_json: String::new(), + }); } Err(err) => { release_external_local_first_pending_slot( @@ -5012,19 +5012,7 @@ impl HandlerForExternalClient for ClientKvApi { } if !local_publish_items.is_empty() { - let published = - publish_external_local_first(inner, local_publish_items, local_publish_contexts) - .await?; - assert_eq!(published.len(), local_publish_indices.len()); - for (idx, result) in local_publish_indices.into_iter().zip(published) { - results[idx] = Some(match result { - Ok(()) => ExternalBatchPutTransferEndItemResp { - error_code: OK, - error_json: String::new(), - }, - Err(error) => ExternalBatchPutTransferEndItemResp::from_error(&error), - }); - } + spawn_external_local_first_publish(inner, local_publish_items, local_publish_contexts); } if !revoke_pending.is_empty() { @@ -5269,7 +5257,6 @@ impl HandlerForExternalClient for ClientKvApi { let mut revoke_pending = Vec::new(); let mut local_publish_items = Vec::new(); let mut local_publish_contexts = Vec::new(); - let mut local_publish_indices = Vec::new(); for (idx, item) in req.items.into_iter().enumerate() { let Some(put_id) = item.put_id else { @@ -5353,7 +5340,10 @@ impl HandlerForExternalClient for ClientKvApi { radix: pending_ctx.radix.clone(), }); local_publish_contexts.push(pending_ctx); - local_publish_indices.push(idx); + results[idx] = Some(ExternalBatchPutCommitItemResp { + error_code: OK, + error_json: String::new(), + }); } Err(err) => { release_external_local_first_pending_slot( @@ -5420,19 +5410,7 @@ impl HandlerForExternalClient for ClientKvApi { } if !local_publish_items.is_empty() { - let published = - publish_external_local_first(inner, local_publish_items, local_publish_contexts) - .await?; - assert_eq!(published.len(), local_publish_indices.len()); - for (idx, result) in local_publish_indices.into_iter().zip(published) { - results[idx] = Some(match result { - Ok(()) => ExternalBatchPutCommitItemResp { - error_code: OK, - error_json: String::new(), - }, - Err(error) => ExternalBatchPutCommitItemResp::from_error(&error), - }); - } + spawn_external_local_first_publish(inner, local_publish_items, local_publish_contexts); } if !revoke_pending.is_empty() { diff --git a/fluxon_rs/fluxon_kv/src/external_client_api/external_client_test.rs b/fluxon_rs/fluxon_kv/src/external_client_api/external_client_test.rs index 47408c2..4dfac8c 100644 --- a/fluxon_rs/fluxon_kv/src/external_client_api/external_client_test.rs +++ b/fluxon_rs/fluxon_kv/src/external_client_api/external_client_test.rs @@ -645,7 +645,7 @@ async fn external_client_basic_crud( } #[tokio::test(flavor = "multi_thread", worker_threads = 8)] -async fn test_external_put_completion_waits_for_route_publication() { +async fn test_external_put_acknowledges_local_commit_before_route_publication() { use crate::client_kv_api::external_api::HandlerForExternalClient; use crate::client_kv_api::msg_pack::{ ExternalBatchPutCommitItemReq, ExternalBatchPutCommitReq, ExternalBatchPutStartItemReq, @@ -653,7 +653,7 @@ async fn test_external_put_completion_waits_for_route_publication() { ExternalBatchPutTransferEndReq, }; use crate::client_kv_api::{ExternalLocalFirstPutKeyReservation, ExternalPutKeyOutcome}; - use crate::rpcresp_kvresult_convert::{msg_and_error::KvError, try_from_code}; + use crate::rpcresp_kvresult_convert::try_from_code; #[derive(Clone, Copy, Debug)] enum Entry { @@ -663,13 +663,12 @@ async fn test_external_put_completion_waits_for_route_publication() { #[derive(Clone, Copy, Debug, PartialEq, Eq)] enum Scenario { Success, - Cancel, RejectIndependent, } let _test_guard = crate::kvcore_test_lib::integration_test_lock().await; tokio::time::timeout(Duration::from_secs(120), async { - fluxon_util::test_util::start_test_etcd().expect("start etcd for publication barrier"); + fluxon_util::test_util::start_test_etcd().expect("start etcd for local commit acknowledgement"); let etcd = fluxon_util::dev_config::read_etcd_endpoint_from_build_config() .expect("read test etcd endpoint"); let suffix = uuid::Uuid::new_v4(); @@ -696,7 +695,7 @@ async fn test_external_put_completion_waits_for_route_publication() { ).await; for entry in [Entry::TransferEnd, Entry::Commit] { - for scenario in [Scenario::Success, Scenario::Cancel, Scenario::RejectIndependent] { + for scenario in [Scenario::Success, Scenario::RejectIndependent] { let keys = (0..3).map(|index| format!("{entry:?}-{scenario:?}-{index}")) .collect::>(); let started = owner_api.external_batch_put_start(ExternalBatchPutStartReq { @@ -762,12 +761,18 @@ async fn test_external_put_completion_waits_for_route_publication() { } } }); - tokio::select! { - biased; - result = &mut completion => panic!("{entry:?} acknowledged before route publication: {result:?}"), - reached = entered => reached.expect("publication worker reached gate"), - } - assert!(futures::poll!(&mut completion).is_pending()); + let results = tokio::time::timeout(Duration::from_secs(5), &mut completion) + .await + .expect("local commit must acknowledge while Master publication is paused"); + assert_eq!(results.len(), keys.len()); + assert!(results.iter().all(Result::is_ok), "local commit results: {results:?}"); + // The caller can leave after the local ACK; the owner retains publication. + drop(completion); + drop(results); + tokio::time::timeout(Duration::from_secs(5), entered) + .await + .expect("publication worker must reach the gate") + .expect("publication worker reached gate"); for key in &keys { assert!(master_inner.kv_routes.get(key).is_none(), "gate must precede the master route"); } @@ -780,14 +785,15 @@ async fn test_external_put_completion_waits_for_route_publication() { _ => panic!("pending publication must retain its per-key fence"), } }).collect::>(); - let results = if scenario == Scenario::Cancel { - drop(completion); - release.send(()).expect("release cancelled RPC's owner worker"); - None - } else { - release.send(()).expect("release publication worker"); - Some(completion.await) - }; + for (index, reader) in readers.iter().enumerate() { + assert_eq!(reader.bytes(), vec![0x30 + index as u8; 4096]); + } + for follower in &followers { + let wait = follower.wait_released(); + futures::pin_mut!(wait); + assert!(futures::poll!(&mut wait).is_pending(), "local ACK must retain the publication fence"); + } + release.send(()).expect("release the departed caller's owner publication"); for (index, follower) in followers.into_iter().enumerate() { let rejected = scenario == Scenario::RejectIndependent && index == 0; assert_eq!(follower.wait().await, if rejected { @@ -798,14 +804,11 @@ async fn test_external_put_completion_waits_for_route_publication() { follower.wait_released().await; assert_eq!(readers[index].bytes(), vec![0x30 + index as u8; 4096]); if rejected { - assert!(matches!(results.as_ref().unwrap()[index], Err(KvError::LeaseMgr(_)))); + // Publication rejection does not retract the local ACK or invalidate holders. assert!(master_inner.kv_routes.get(&keys[index]).is_none()); assert!(inner.local_visible_mem_holder(&keys[index]).is_none()); } else { - if let Some(results) = &results { - assert!(results[index].is_ok(), "{:?}", results[index]); - } - let route = master_inner.kv_routes.get(&keys[index]).expect("successful Put must publish its route"); + let route = master_inner.kv_routes.get(&keys[index]).expect("background publication must eventually expose the route"); assert_eq!(route.atomic_group.as_ref().map(|group| group.members.len()), (index > 0).then_some(2)); drop(route); let (holder, _) = inner.get(&keys[index]).await.expect("get completed value").expect("published value exists"); @@ -815,12 +818,31 @@ async fn test_external_put_completion_waits_for_route_publication() { let after = inner.runtime_observe_snapshot(); assert!(after.grouped_put_done_batches > before.grouped_put_done_batches); assert_eq!(after.legacy_put_done_batches, before.legacy_put_done_batches); + assert_eq!( + after.owner_segment_allocated_bytes, + before.owner_segment_allocated_bytes, + "a rejected publication must retain its physical slot while a reader holds it" + ); + let rejected_capacity = if scenario == Scenario::RejectIndependent { + readers[0].local_reserve_resident_slot_ref() + .expect("rejected local value retains its exact slot").2 + } else { + 0 + }; + let released = inner.owner_local_reserve_rebalance_notify(); drop(readers); + tokio::time::timeout( + Duration::from_secs(5), + fluxon_util::notify_state::wait_until(&released, || { + inner.owner_local_reserve_control_snapshot().allocated_bytes + == after.owner_segment_allocated_bytes - rejected_capacity + }), + ).await.expect("the last rejected reader must release its slot without freeing published values"); } } owner.shutdown().await.expect("stop publication owner"); master.shutdown().await.expect("stop publication master"); - }).await.expect("external publication barriers and fence cleanup must complete"); + }).await.expect("local ACK, background publication, and fence cleanup must complete"); } #[tokio::test(flavor = "multi_thread", worker_threads = 8)] @@ -1247,6 +1269,7 @@ async fn test_external_remote_get_uses_bound_owner_slot_without_master_allocator ) .await .expect("put only on source owner"); + wait_master_route(master_view.clone(), key, Duration::from_secs(5)).await; { let route = master_view .master_kv_router() diff --git a/fluxon_rs/fluxon_kv/src/kv_test.rs b/fluxon_rs/fluxon_kv/src/kv_test.rs index 44c67ad..f816e57 100644 --- a/fluxon_rs/fluxon_kv/src/kv_test.rs +++ b/fluxon_rs/fluxon_kv/src/kv_test.rs @@ -302,6 +302,25 @@ async fn get_owner_probe_value( } } +async fn wait_for_external_route_publication( + view: &crate::external_client_api::ExternalClientApiView, + key: &str, + deadline: Instant, +) -> anyhow::Result<()> { + let api = view.external_client_api(); + timeout(deadline.saturating_duration_since(Instant::now()), async { + loop { + // Default existence checks query Master instead of the local snapshot. + if api.inner().is_exist(key).await? { + return Ok::<(), anyhow::Error>(()); + } + sleep(Duration::from_millis(10)).await; + } + }) + .await + .map_err(|_| anyhow::anyhow!("external route publication timed out: key={}", key))? +} + async fn fetch_transfer_link_te_value( cluster_name: &str, from_instance_key: &str, @@ -381,6 +400,20 @@ async fn verify_rdma_transfer_data_link( continue; } + // The probe reads from another Owner, so it needs the Master route. + wait_for_external_route_publication( + &probe_writer_framework.external_client_api_view(), + attempt_key.as_str(), + deadline, + ) + .await + .unwrap_or_else(|err| { + panic!( + "closed transfer data probe publication failed: cluster={} from={} to={} key={} attempt={} err={}", + cluster_name, from_instance_key, to_instance_key, attempt_key, attempt, err + ) + }); + let payload = match get_owner_probe_value(requester_framework, attempt_key.as_str()).await { Ok(payload) => payload, Err(err) => { @@ -1302,6 +1335,14 @@ async fn perform_external_client_crud_operations( } } + // Same-owner GET above checks local completion; global existence needs publication. + wait_for_external_route_publication( + &view, + key, + Instant::now() + Duration::from_secs(KV_TEST_TRANSFER_PROBE_IO_TIMEOUT_SECS), + ) + .await?; + // EXIST操作 match api.inner().is_exist(key).await { Ok(true) => info!(