iscsi_client_rs/state_machine/
nop_states.rs

1//! This module defines the state machine for the iSCSI NOP-Out/NOP-In exchange.
2//! It includes the states, context, and transitions for handling ping-like
3//! messages.
4
5// SPDX-License-Identifier: AGPL-3.0-or-later
6// Copyright (C) 2012-2025 Andrei Maltsev
7
8use std::{
9    marker::PhantomData,
10    pin::Pin,
11    sync::{
12        Arc,
13        atomic::{AtomicU32, Ordering},
14    },
15};
16
17use anyhow::{Context, Result, anyhow, bail};
18use tokio_util::sync::CancellationToken;
19use tracing::debug;
20
21use crate::{
22    client::client::ClientConnection,
23    models::{
24        common::HEADER_LEN,
25        data_fromat::{PduRequest, PduResponse},
26        nop::{
27            request::{NopOutRequest, NopOutRequestBuilder},
28            response::NopInResponse,
29        },
30    },
31    state_machine::common::{StateMachine, StateMachineCtx, Transition},
32};
33
34/// This structure represents the context for a NOP-Out/NOP-In exchange.
35///
36/// It holds all the necessary information to manage the state of a
37/// NOP-Out/NOP-In operation, including connection details and command
38/// parameters.
39#[derive(Debug)]
40pub struct NopCtx<'a> {
41    _lt: PhantomData<&'a ()>,
42
43    pub conn: Arc<ClientConnection>,
44    pub lun: u64,
45    pub itt: u32,
46    pub cmd_sn: u32,
47    pub exp_stat_sn: Arc<AtomicU32>,
48    pub ttt: u32,
49    pub buf: [u8; HEADER_LEN],
50
51    last_response: Option<PduResponse<NopInResponse>>,
52    state: Option<NopStates>,
53}
54
55impl<'a> NopCtx<'a> {
56    /// Creates a new `NopCtx` for initiating a NOP-Out exchange.
57    pub fn new(
58        conn: Arc<ClientConnection>,
59        lun: u64,
60        itt: Arc<AtomicU32>,
61        cmd_sn: Arc<AtomicU32>,
62        exp_stat_sn: Arc<AtomicU32>,
63        ttt: u32,
64    ) -> Self {
65        Self {
66            conn,
67            lun,
68            itt: itt.fetch_add(1, Ordering::SeqCst),
69            cmd_sn: cmd_sn.load(Ordering::SeqCst),
70            exp_stat_sn,
71            ttt,
72            buf: [0u8; HEADER_LEN],
73            state: Some(NopStates::Start(Start)),
74            last_response: None,
75            _lt: PhantomData,
76        }
77    }
78
79    /// Resets the state of the context to the initial state.
80    pub fn set_default_state(&mut self) {
81        self.state = Some(NopStates::Start(Start));
82    }
83
84    /// Validates the header of the last received NOP-In response.
85    pub fn validate_last_response_header(&mut self) -> Result<&NopInResponse> {
86        match &self.last_response {
87            Some(l) => match l.header_view() {
88                Ok(last) => {
89                    self.exp_stat_sn
90                        .store(last.stat_sn.get().wrapping_add(1), Ordering::SeqCst);
91                    Ok(last)
92                },
93                Err(e) => Err(e),
94            },
95            None => Err(anyhow!("no last response in ctx")),
96        }
97    }
98
99    /// Creates a new `NopCtx` for replying to a NOP-In.
100    pub fn for_reply(
101        conn: Arc<ClientConnection>,
102        _itt: Arc<AtomicU32>,
103        cmd_sn: Arc<AtomicU32>,
104        exp_stat_sn: Arc<AtomicU32>,
105        response: PduResponse<NopInResponse>,
106    ) -> Result<Self> {
107        let header = response.header_view()?;
108        Ok(Self {
109            conn,
110            lun: 0,
111            itt: 0,
112            cmd_sn: cmd_sn.load(Ordering::SeqCst),
113            exp_stat_sn,
114            ttt: header.target_task_tag.get(),
115            buf: [0u8; HEADER_LEN],
116            last_response: Some(response),
117            state: Some(NopStates::Reply(Reply)),
118            _lt: PhantomData,
119        })
120    }
121
122    async fn send_nop_out(&mut self) -> Result<()> {
123        let exp_stat_sn = self.exp_stat_sn.load(Ordering::SeqCst);
124
125        let header = NopOutRequestBuilder::new()
126            .cmd_sn(self.cmd_sn)
127            .lun(self.lun)
128            .initiator_task_tag(self.itt)
129            .target_task_tag(self.ttt)
130            .exp_stat_sn(exp_stat_sn)
131            .immediate();
132
133        header.header.to_bhs_bytes(self.buf.as_mut_slice())?;
134
135        let builder = PduRequest::<NopOutRequest>::new_request(self.buf, &self.conn.cfg);
136        self.conn.send_request(self.itt, builder).await?;
137        Ok(())
138    }
139
140    async fn recieve_nop_in(&mut self) -> Result<()> {
141        match self.conn.read_response::<NopInResponse>(self.itt).await {
142            Ok(rsp) => {
143                self.last_response = Some(rsp);
144                Ok(())
145            },
146            Err(other) => bail!("got unexpected PDU: {:?}", other.to_string()),
147        }
148    }
149}
150
151/// Represents the initial state of a NOP-Out operation.
152#[derive(Debug)]
153pub struct Start;
154
155/// Represents the state of waiting for a NOP-In response.
156#[derive(Debug)]
157pub struct Wait;
158
159/// Represents the state of sending a NOP-Out in reply to a NOP-In.
160#[derive(Debug)]
161pub struct Reply;
162
163/// Defines the possible states for a NOP-Out/NOP-In operation state machine.
164#[derive(Debug)]
165pub enum NopStates {
166    Start(Start),
167    Wait(Wait),
168    Reply(Reply),
169}
170
171type NopStepOut = Transition<NopStates, Result<()>>;
172
173impl<'ctx> StateMachine<NopCtx<'ctx>, NopStepOut> for Start {
174    type StepResult<'a>
175        = Pin<Box<dyn Future<Output = NopStepOut> + Send + 'a>>
176    where
177        Self: 'a,
178        NopCtx<'ctx>: 'a;
179
180    fn step<'a>(&'a self, ctx: &'a mut NopCtx<'ctx>) -> Self::StepResult<'a> {
181        Box::pin(async move {
182            match ctx.send_nop_out().await {
183                Ok(st) => Transition::Next(NopStates::Wait(Wait), Ok(st)),
184                Err(e) => Transition::Done(Err(e)),
185            }
186        })
187    }
188}
189
190impl<'ctx> StateMachine<NopCtx<'ctx>, NopStepOut> for Wait {
191    type StepResult<'a>
192        = Pin<Box<dyn Future<Output = NopStepOut> + Send + 'a>>
193    where
194        Self: 'a,
195        NopCtx<'ctx>: 'a;
196
197    fn step<'a>(&'a self, ctx: &'a mut NopCtx<'ctx>) -> Self::StepResult<'a> {
198        Box::pin(async move {
199            match ctx.recieve_nop_in().await {
200                Ok(()) => Transition::Done(Ok(())),
201                Err(e) => Transition::Done(Err(e)),
202            }
203        })
204    }
205}
206
207impl<'ctx> StateMachine<NopCtx<'ctx>, NopStepOut> for Reply {
208    type StepResult<'a>
209        = Pin<Box<dyn Future<Output = NopStepOut> + Send + 'a>>
210    where
211        Self: 'a,
212        NopCtx<'ctx>: 'a;
213
214    fn step<'a>(&'a self, ctx: &'a mut NopCtx<'ctx>) -> Self::StepResult<'a> {
215        Box::pin(async move {
216            if let Err(e) = ctx.validate_last_response_header() {
217                return Transition::Done(Err(e));
218            }
219
220            // ITT for response NOP-In = 0xFFFF_FFFF
221            // TTT — copy from NOP-In (ctx.ttt)
222            let hdr = NopOutRequestBuilder::new()
223                .immediate()
224                .lun(0)
225                .initiator_task_tag(u32::MAX)
226                .target_task_tag(ctx.ttt)
227                .cmd_sn(ctx.cmd_sn)
228                .exp_stat_sn(ctx.exp_stat_sn.load(Ordering::SeqCst))
229                .header;
230
231            if let Err(e) = hdr.to_bhs_bytes(ctx.buf.as_mut_slice()) {
232                return Transition::Done(Err(e));
233            }
234            let pdu = PduRequest::<NopOutRequest>::new_request(ctx.buf, &ctx.conn.cfg);
235
236            // Response — fire-and-forget
237            if let Err(e) = ctx.conn.send_request(u32::MAX, pdu).await {
238                return Transition::Done(Err(e));
239            }
240
241            Transition::Done(Ok(()))
242        })
243    }
244}
245
246impl<'s> StateMachineCtx<NopCtx<'s>, PduResponse<NopInResponse>> for NopCtx<'s> {
247    async fn execute(
248        &mut self,
249        _cancel: &CancellationToken,
250    ) -> Result<PduResponse<NopInResponse>> {
251        debug!("Loop Nop");
252        loop {
253            let state = self.state.take().context("state must be set NopCtx")?;
254            let trans = match state {
255                NopStates::Start(s) => s.step(self).await,
256                NopStates::Wait(s) => s.step(self).await,
257                NopStates::Reply(s) => s.step(self).await,
258            };
259
260            match trans {
261                Transition::Next(next, r) => {
262                    r?;
263                    self.state = Some(next);
264                },
265                Transition::Stay(Ok(_)) => {},
266                Transition::Stay(Err(e)) => return Err(e),
267                Transition::Done(r) => {
268                    r?;
269                    return self
270                        .last_response
271                        .take()
272                        .ok_or_else(|| anyhow!("no last response in ctx"));
273                },
274            }
275        }
276    }
277}
iscsi_client_rs/state_machine/nop_states.rs

iscsi_client_rs/state_machine/
nop_states.rs
