perf(socket): replace per-connection maintenance tick with exact deadlines
ZhenYi committed
Modified lib/infra/socket/src/session/run.rs lib/infra/socket/src/session/run.rs +15 −5 View lib/infra/socket/src/session/run.rs
@@ -57,19 +57,22 @@
5757 let mut heartbeat = interval(options.heartbeat_interval);
5858 heartbeat.set_missed_tick_behavior(MissedTickBehavior::Delay);
5959 heartbeat.tick().await;
60 let mut maintenance = interval(std::time::Duration::from_secs(1));
6160
6261 loop {
6362 if *shutdown.borrow() {
6463 break;
6564 }
65 let auth_deadline = session.auth_deadline;
66 let ack_deadline = session
67 .pending_acks
68 .values()
69 .map(|(_, deadline, _)| *deadline)
70 .min();
6671 tokio::select! {
6772 _ = shutdown.changed() => break,
68 _ = maintenance.tick() => {
73 _ = wait_until(auth_deadline) => break,
74 _ = wait_until(ack_deadline) => {
6975 let now = tokio::time::Instant::now();
70 if session.auth_deadline.is_some_and(|deadline| now >= deadline) {
71 break;
72 }
7376 session.pending_acks.retain(|_, (_, deadline, sender)| *deadline > now && !sender.is_closed());
7477 }
7578 incoming = receiver.recv() => {
@@ -98,3 +101,10 @@
98101
99102 session.cleanup().await;
100103}
104
105async fn wait_until(deadline: Option<tokio::time::Instant>) {
106 match deadline {
107 Some(deadline) => tokio::time::sleep_until(deadline).await,
108 None => std::future::pending().await,
109 }
110}