diff --git a/crates/irtt-client/src/async_client.rs b/crates/irtt-client/src/async_client.rs index 3d3e759..51a02fb 100644 --- a/crates/irtt-client/src/async_client.rs +++ b/crates/irtt-client/src/async_client.rs @@ -93,8 +93,6 @@ pub(crate) struct AsyncOpenState { deadline: Option, deadline_timer: Option>>, request_submitted: bool, - stop_after_current_attempt: bool, - stopped_after_current_attempt: bool, buffer: [u8; MAX_OPEN_PACKET_SIZE], cleanup: Option, } @@ -106,8 +104,6 @@ impl AsyncOpenState { deadline: None, deadline_timer: None, request_submitted: false, - stop_after_current_attempt: false, - stopped_after_current_attempt: false, buffer: [0_u8; MAX_OPEN_PACKET_SIZE], cleanup: None, } @@ -128,21 +124,12 @@ impl AsyncOpenState { self.deadline = None; self.deadline_timer = None; self.request_submitted = false; - self.stopped_after_current_attempt = self.stop_after_current_attempt; } pub(crate) fn has_in_flight_work(&self) -> bool { self.request_submitted || self.cleanup.is_some() } - pub(crate) fn request_stop_after_current_attempt(&mut self) { - self.stop_after_current_attempt = true; - } - - pub(crate) fn stopped_after_current_attempt(&self) -> bool { - self.stopped_after_current_attempt - } - fn deadline(&self) -> Instant { self.deadline .expect("an open attempt has an active deadline") @@ -324,10 +311,6 @@ impl AsyncClient { if state.attempt >= attempt_count { return Poll::Ready(Err(ClientError::OpenTimeout)); } - if state.stop_after_current_attempt && state.deadline.is_none() { - state.stopped_after_current_attempt = true; - return Poll::Ready(Err(ClientError::OpenTimeout)); - } if state.deadline.is_none() { let timeout = self.machine.config().open.timeouts[state.attempt]; if let Err(error) = state.start_attempt(timeout) { @@ -368,9 +351,6 @@ impl AsyncClient { Poll::Ready(Err(error)) => return Poll::Ready(Err(error)), Poll::Ready(Ok(false)) => { state.finish_attempt(); - if state.stopped_after_current_attempt() { - return Poll::Ready(Err(ClientError::OpenTimeout)); - } continue; } Poll::Ready(Ok(true)) => {} @@ -396,9 +376,6 @@ impl AsyncClient { }; if datagram.received_at.mono > state.deadline() { state.finish_attempt(); - if state.stopped_after_current_attempt() { - return Poll::Ready(Err(ClientError::OpenTimeout)); - } continue; } @@ -778,6 +755,16 @@ impl AsyncClient { .map_err(ClientError::Socket) } + /// Cancel opening without waiting for a reply or socket writability. + /// Retained post-token cleanup gets one best-effort nonblocking send. + pub(crate) fn cancel_open(&self, state: &mut AsyncOpenState) -> Option { + let cleanup = state.cleanup.take()?; + if let Some(packet) = cleanup.packet() { + let _ = self.socket.try_send(packet); + } + Some(cleanup.into_primary()) + } + fn poll_open_cleanup( &self, state: &mut AsyncOpenState, diff --git a/crates/irtt-client/src/managed/task/target.rs b/crates/irtt-client/src/managed/task/target.rs index 0a71ef4..e405558 100644 --- a/crates/irtt-client/src/managed/task/target.rs +++ b/crates/irtt-client/src/managed/task/target.rs @@ -409,12 +409,12 @@ impl ManagedClientTask { let retirement = retirement .clone() .unwrap_or(ManagedTargetEndReason::Stopped); - if !open.has_in_flight_work() { - self.targets[index].sync_live_packets_sent(); - self.finish_target(index, retirement, None); - return false; - } - open.request_stop_after_current_attempt(); + let cleanup_failure = client.cancel_open(open).map(|error| { + classify_client_error(ManagedTargetFailurePhase::Opening, &error) + }); + self.targets[index].sync_live_packets_sent(); + self.finish_target(index, retirement, cleanup_failure); + return false; } match client.poll_open(open, cx) { Poll::Pending => false, @@ -438,64 +438,25 @@ impl ManagedClientTask { index, vec![ClientEvent::SessionStarted(started)], ); - if stopping { - self.targets[index].sync_live_packets_sent(); - self.begin_drain( - index, - retirement - .clone() - .unwrap_or(ManagedTargetEndReason::Stopped), - now, - ) - } else { - let previous = self.targets[index].take_phase(); - self.targets[index].restore_phase(TargetPhase::Active { - client: previous.into_client(), - schedule, - send_waiting: false, - }); - self.publish_target_phase(index, ManagedTargetLifecycle::Opening); - true - } + let previous = self.targets[index].take_phase(); + self.targets[index].restore_phase(TargetPhase::Active { + client: previous.into_client(), + schedule, + send_waiting: false, + }); + self.publish_target_phase(index, ManagedTargetLifecycle::Opening); + true } Poll::Ready(Ok(OpenOutcome::NoTestCompleted(completed))) => { self.publish_client_events( index, vec![ClientEvent::NoTestCompleted(completed)], ); - let end_reason = if stopping { - retirement - .clone() - .unwrap_or(ManagedTargetEndReason::Stopped) - } else { - ManagedTargetEndReason::NoTestComplete - }; - self.finish_target(index, end_reason, None); + self.finish_target(index, ManagedTargetEndReason::NoTestComplete, None); false } Poll::Ready(Err(error)) => { - if stopping { - let TargetPhase::Opening { open, .. } = self.targets[index].phase() - else { - unreachable!() - }; - let cleanup_failure = - (!open.stopped_after_current_attempt()).then(|| { - classify_client_error( - ManagedTargetFailurePhase::Opening, - &error, - ) - }); - self.finish_target( - index, - retirement - .clone() - .unwrap_or(ManagedTargetEndReason::Stopped), - cleanup_failure, - ); - } else { - self.fail_target(index, ManagedTargetFailurePhase::Opening, error); - } + self.fail_target(index, ManagedTargetFailurePhase::Opening, error); false } }