Skip to content

Simplify polling logic in ConnectionActor #164

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
leynos wants to merge 2 commits into from
Closed

Simplify polling logic in ConnectionActor #164

Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
bfc5223
Simplify polling logic in ConnectionActor
leynos Jul 4, 2025
8f238a8
Merge branch 'main' into codex/refactor-polling-logic-in-connectionactor
leynos Jul 4, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
63 changes: 36 additions & 27 deletions src/connection.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -7,7 +7,7 @@

use std::future::Future;

use futures::StreamExt;

Check failure on line 10 in src/connection.rs

View workflow job for this annotation

GitHub Actions / build-test 

unused import: `futures::StreamExt`
use tokio::{
sync::mpsc,
time::{Duration, Instant},
Expand Down Expand Up @@ -163,33 +163,11 @@
state: &mut ActorState,
out: &mut Vec<F>,
) -> Result<(), WireframeError<E>> {
let high_available = self.high_rx.is_some();
let low_available = self.low_rx.is_some();
let resp_available = self.response.is_some();

tokio::select! {
biased;

() = Self::wait_shutdown(self.shutdown.clone()), if state.is_active() => {
self.process_shutdown(state);
}

res = Self::poll_optional(self.high_rx.as_mut(), Self::recv_push), if high_available => {
self.process_high(res, state, out);
}

res = Self::poll_optional(self.low_rx.as_mut(), Self::recv_push), if low_available => {
self.process_low(res, state, out);
}

// `tokio::select!` is biased so the shutdown branch runs before
// this one. `process_shutdown` removes the response stream, making
// `resp_available` false on the next loop iteration. The explicit
// `!state.is_shutting_down()` check avoids polling the stream after
// shutdown has begun.
res = Self::poll_optional(self.response.as_mut(), |s| s.next()), if resp_available && !state.is_shutting_down() => {
self.process_response(res, state, out)?;
}
match self.next_event(state).await {
PollEvent::Shutdown => self.process_shutdown(state),
PollEvent::High(res) => self.process_high(res, state, out),
PollEvent::Low(res) => self.process_low(res, state, out),
PollEvent::Response(res) => self.process_response(res, state, out)?,
}

Ok(())
Expand Down Expand Up @@ -364,6 +342,37 @@
}
}
}

/// Determine which event should be processed next.
async fn next_event(&mut self, state: &ActorState) -> PollEvent<F, E> {
let high_available = self.high_rx.is_some();
let low_available = self.low_rx.is_some();
Comment on lines +346 to +349

@sourcery-ai sourcery-ai Bot Jul 4, 2025

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

suggestion: Consider documenting or clarifying the bias order in next_event's select.

Since the order of branches in 'tokio::select! { biased; ... }' affects event prioritization, please clarify if prioritizing the shutdown signal is intentional. Documenting this will help maintainers understand the impact on shutdown responsiveness and event throughput.

Suggested change
/// Determine which event should be processed next.
async fn next_event(&mut self, state: &ActorState) -> PollEvent<F, E> {
let high_available = self.high_rx.is_some();
let low_available = self.low_rx.is_some();
/// Determine which event should be processed next.
///
/// # Event Selection Bias
///
/// This function uses `tokio::select! { biased; ... }` to prioritize the shutdown signal branch.
/// The order of branches in the select block is intentional: the shutdown signal is checked first,
/// ensuring that shutdown requests are handled promptly, even if other events are available.
/// This improves shutdown responsiveness, but may slightly reduce event throughput if shutdown
/// signals and other events arrive simultaneously.
async fn next_event(&mut self, state: &ActorState) -> PollEvent<F, E> {
let high_available = self.high_rx.is_some();
let low_available = self.low_rx.is_some();

let resp_available = self.response.is_some() && !state.is_shutting_down();

tokio::select! {
biased;

() = Self::wait_shutdown(self.shutdown.clone()), if state.is_active() => PollEvent::Shutdown,

res = Self::poll_receiver(self.high_rx.as_mut()), if high_available => PollEvent::High(res),

Check failure on line 357 in src/connection.rs

View workflow job for this annotation

GitHub Actions / build-test 

no function or associated item named `poll_receiver` found for struct `connection::ConnectionActor` in the current scope

res = Self::poll_receiver(self.low_rx.as_mut()), if low_available => PollEvent::Low(res),

Check failure on line 359 in src/connection.rs

View workflow job for this annotation

GitHub Actions / build-test 

no function or associated item named `poll_receiver` found for struct `connection::ConnectionActor` in the current scope

res = Self::poll_response(self.response.as_mut()), if resp_available => PollEvent::Response(res),

Check failure on line 361 in src/connection.rs

View workflow job for this annotation

GitHub Actions / build-test 

no function or associated item named `poll_response` found for struct `connection::ConnectionActor` in the current scope
}
}
}

/// Outcome of polling the various sources.
enum PollEvent<F, E> {

@sourcery-ai sourcery-ai Bot Jul 4, 2025

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

suggestion: Consider deriving standard traits for PollEvent for easier debugging and usage.

Adding derives like Debug, PartialEq, or Clone (if applicable) can improve logging, testing, and code manipulation. Consider including them unless there are specific constraints.

Suggested implementation:

#[derive(Debug, PartialEq, Clone)]
/// Outcome of polling the various sources.
enum PollEvent<F, E> {
enum PollEvent<F, E>
where
    F: Debug + PartialEq + Clone,
    E: Debug + PartialEq + Clone,
{

/// Shutdown signal triggered.
Shutdown,
/// Result of polling the high-priority queue.
High(Option<F>),
/// Result of polling the low-priority queue.
Low(Option<F>),
/// Result of polling the response stream.
Response(Option<Result<F, WireframeError<E>>>),
}

/// Internal run state for the connection actor.
Expand Down
Loading