iscsi_client_rs/state_machine/
read_states.rs

1//! This module defines the state machine for the iSCSI SCSI Read command.
2//! It includes the states, context, and transitions for handling the read
3//! operation.
4
5// SPDX-License-Identifier: AGPL-3.0-or-later
6// Copyright (C) 2012-2025 Andrei Maltsev
7
8use std::{
9    future::Future,
10    marker::PhantomData,
11    pin::Pin,
12    sync::{
13        Arc,
14        atomic::{AtomicU32, Ordering},
15    },
16};
17
18use anyhow::{Context, Result, anyhow};
19use bytes::Bytes;
20use tokio_util::sync::CancellationToken;
21use tracing::debug;
22
23use crate::{
24    client::client::ClientConnection,
25    models::{
26        command::{
27            common::{ScsiStatus, TaskAttribute},
28            request::{ScsiCommandRequest, ScsiCommandRequestBuilder},
29            response::ScsiCommandResponse,
30        },
31        common::HEADER_LEN,
32        data::{response::ScsiDataIn, sense_data::SenseData},
33        data_fromat::{PduRequest, PduResponse},
34        opcode::{BhsOpcode, Opcode},
35        parse::Pdu,
36    },
37    state_machine::common::{StateMachine, StateMachineCtx, Transition},
38};
39
40/// Represents the types of PDUs that can be received during a SCSI Read
41/// operation.
42#[derive(Debug)]
43pub enum ReadPdu {
44    /// A SCSI Data-In PDU.
45    DataIn(PduResponse<ScsiDataIn>),
46    /// A SCSI Command Response PDU.
47    CmdResp(PduResponse<ScsiCommandResponse>),
48}
49
50/// Holds the runtime state for a SCSI Read operation.
51#[derive(Debug)]
52pub struct ReadRuntime {
53    /// The accumulated data from Data-In PDUs.
54    pub acc: Vec<u8>,
55    /// The command sequence number of the current command.
56    pub cur_cmd_sn: Option<u32>,
57    /// The SCSI status received in a Data-In PDU.
58    pub status_in_datain: Option<ScsiStatus>,
59    /// The residual count received in a Data-In PDU.
60    pub residual_in_datain: Option<u32>,
61}
62
63/// This structure represents the context for a SCSI Read operation.
64#[derive(Debug)]
65pub struct ReadCtx<'a> {
66    _lt: PhantomData<&'a ()>,
67
68    /// The client connection.
69    pub conn: Arc<ClientConnection>,
70    /// The Logical Unit Number.
71    pub lun: u64,
72    /// The Initiator Task Tag.
73    pub itt: u32,
74    /// The Command Sequence Number.
75    pub cmd_sn: Arc<AtomicU32>,
76    /// The Expected Status Sequence Number.
77    pub exp_stat_sn: Arc<AtomicU32>,
78    /// The number of bytes to read.
79    pub read_len: u32,
80    /// The SCSI Command Descriptor Block.
81    pub cdb: [u8; 16],
82    /// A buffer for the BHS.
83    pub buf: [u8; HEADER_LEN],
84
85    /// The last received command response.
86    pub last_response: Option<PduResponse<ScsiCommandResponse>>,
87    /// The runtime state of the read operation.
88    pub rt: ReadRuntime,
89    state: Option<ReadStates>,
90}
91
92impl<'a> ReadCtx<'a> {
93    /// Creates a new `ReadCtx` for a SCSI Read operation.
94    pub fn new(
95        conn: Arc<ClientConnection>,
96        lun: u64,
97        itt: Arc<AtomicU32>,
98        cmd_sn: Arc<AtomicU32>,
99        exp_stat_sn: Arc<AtomicU32>,
100        read_len: u32,
101        cdb: [u8; 16],
102    ) -> Self {
103        Self {
104            conn,
105            lun,
106            itt: itt.fetch_add(1, Ordering::SeqCst),
107            cmd_sn,
108            exp_stat_sn,
109            read_len,
110            cdb,
111            buf: [0u8; HEADER_LEN],
112            last_response: None,
113            rt: ReadRuntime {
114                acc: Vec::with_capacity(read_len as usize),
115                cur_cmd_sn: None,
116                status_in_datain: None,
117                residual_in_datain: None,
118            },
119            state: Some(ReadStates::Start(Start)),
120            _lt: PhantomData,
121        }
122    }
123
124    /// Receives any PDU related to the read operation.
125    pub async fn recv_any(&self, itt: u32) -> anyhow::Result<ReadPdu> {
126        let (p_any, data): (PduResponse<Pdu>, Bytes) =
127            self.conn.read_response_raw(itt).await?;
128        let op = BhsOpcode::try_from(p_any.header_buf[0])?.opcode;
129
130        let pdu_local = match op {
131            Opcode::ScsiDataIn => Ok(ReadPdu::DataIn({
132                let mut pdu = p_any.rebind_pdu::<ScsiDataIn>()?;
133                pdu.parse_with_buff(&data)?;
134                pdu
135            })),
136            Opcode::ScsiCommandResp => Ok(ReadPdu::CmdResp({
137                let mut pdu = p_any.rebind_pdu::<ScsiCommandResponse>()?;
138                pdu.parse_with_buff(&data)?;
139                pdu
140            })),
141            other => anyhow::bail!("unexpected PDU opcode for read path: {other:?}"),
142        };
143        debug!("READ {pdu_local:?}");
144        pdu_local
145    }
146
147    async fn send_read_request(&mut self) -> Result<u32> {
148        let sn = self.cmd_sn.fetch_add(1, Ordering::SeqCst);
149        let esn = self.exp_stat_sn.load(Ordering::SeqCst);
150
151        let header = ScsiCommandRequestBuilder::new()
152            .lun(self.lun)
153            .initiator_task_tag(self.itt)
154            .cmd_sn(sn)
155            .exp_stat_sn(esn)
156            .expected_data_transfer_length(self.read_len)
157            .scsi_descriptor_block(&self.cdb)
158            .read()
159            .task_attribute(TaskAttribute::Simple);
160
161        header.header.to_bhs_bytes(self.buf.as_mut_slice())?;
162        let builder =
163            PduRequest::<ScsiCommandRequest>::new_request(self.buf, &self.conn.cfg);
164        self.conn.send_request(self.itt, builder).await?;
165
166        self.rt.cur_cmd_sn = Some(sn);
167        Ok(esn)
168    }
169
170    /// Receives a Data-In PDU.
171    pub async fn recv_datain(&self, itt: u32) -> Result<PduResponse<ScsiDataIn>> {
172        self.conn.read_response(itt).await
173    }
174
175    /// Appends the data from a Data-In PDU to the accumulator.
176    pub fn apply_datain_append(&mut self, pdu: &PduResponse<ScsiDataIn>) -> Result<bool> {
177        let h = pdu.header_view()?;
178
179        let off = h.buffer_offset.get() as usize;
180        if off != self.rt.acc.len() {
181            return Err(anyhow!(
182                "unexpected buffer_offset: got {}, expected {}",
183                off,
184                self.rt.acc.len()
185            ));
186        }
187
188        let data = pdu.data()?;
189
190        if !data.is_empty() {
191            self.rt.acc.extend_from_slice(data);
192        }
193
194        if h.stat_sn_or_rsvd.get() != 0 {
195            self.exp_stat_sn
196                .store(h.stat_sn_or_rsvd.get().wrapping_add(1), Ordering::SeqCst);
197        }
198        if h.get_status_bit() {
199            self.rt.status_in_datain = h.scsi_status();
200            self.rt.residual_in_datain = Some(h.residual_effective());
201        }
202
203        Ok(h.get_real_final_bit())
204    }
205
206    /// Finalizes the status of the read operation after all data has been
207    /// received.
208    pub async fn finalize_status_after_datain(
209        &mut self,
210        itt: u32,
211    ) -> Result<(ScsiStatus, u32, Option<Vec<u8>>)> {
212        if let Some(ScsiStatus::Good) = self.rt.status_in_datain {
213            return Ok((
214                ScsiStatus::Good,
215                self.rt.residual_in_datain.unwrap_or_default(),
216                None,
217            ));
218        }
219
220        let rsp: PduResponse<ScsiCommandResponse> = match self.last_response.take() {
221            Some(r) => r,
222            None => self.conn.read_response(itt).await?,
223        };
224        self.last_response = Some(rsp);
225
226        let lr = self.last_response.as_ref().expect("saved above");
227        let h = lr.header_view()?;
228
229        let status = h
230            .status
231            .decode()
232            .map_err(|e| anyhow!("SCSI status decode: {e}"))?;
233        self.exp_stat_sn
234            .store(h.stat_sn.get().wrapping_add(1), Ordering::SeqCst);
235
236        let data = lr.data()?;
237
238        let sense = if data.is_empty() {
239            None
240        } else {
241            Some(data.to_vec())
242        };
243        Ok((status, h.residual_effective(), sense))
244    }
245}
246
247/// Represents the initial state of a read operation.
248#[derive(Debug)]
249pub struct Start;
250
251/// Represents the state of waiting for data from the target.
252#[derive(Debug)]
253pub struct ReadWait;
254
255/// Represents the final state of a read operation.
256#[derive(Debug)]
257pub struct Finish;
258
259/// Defines the possible states for a SCSI Read operation state machine.
260#[derive(Debug)]
261pub enum ReadStates {
262    /// The initial state.
263    Start(Start),
264    /// Waiting for data.
265    Wait(ReadWait),
266    /// The final state.
267    Finish(Finish),
268}
269
270type ReadStepOut = Transition<ReadStates, Result<()>>;
271
272impl<'ctx> StateMachine<ReadCtx<'ctx>, ReadStepOut> for Start {
273    type StepResult<'a>
274        = Pin<Box<dyn Future<Output = ReadStepOut> + Send + 'a>>
275    where
276        Self: 'a,
277        ReadCtx<'ctx>: 'a;
278
279    fn step<'a>(&'a self, ctx: &'a mut ReadCtx<'ctx>) -> Self::StepResult<'a> {
280        Box::pin(async move {
281            if let Err(e) = ctx.send_read_request().await {
282                return Transition::Done(Err(e));
283            }
284            Transition::Next(ReadStates::Wait(ReadWait), Ok(()))
285        })
286    }
287}
288
289impl<'ctx> StateMachine<ReadCtx<'ctx>, ReadStepOut> for ReadWait {
290    type StepResult<'a>
291        = Pin<Box<dyn Future<Output = ReadStepOut> + Send + 'a>>
292    where
293        Self: 'a,
294        ReadCtx<'ctx>: 'a;
295
296    fn step<'a>(&'a self, ctx: &'a mut ReadCtx<'ctx>) -> Self::StepResult<'a> {
297        Box::pin(async move {
298            loop {
299                match ctx.recv_any(ctx.itt).await {
300                    Ok(ReadPdu::DataIn(pdu)) => {
301                        let is_final = match ctx.apply_datain_append(&pdu) {
302                            Ok(f) => f,
303                            Err(e) => return Transition::Done(Err(e)),
304                        };
305                        if is_final {
306                            break;
307                        }
308                    },
309                    Ok(ReadPdu::CmdResp(rsp)) => {
310                        ctx.last_response = Some(rsp);
311                        break;
312                    },
313                    Err(e) => {
314                        return Transition::Done(Err(anyhow!(
315                            "unexpected PDU while read: {e}"
316                        )));
317                    },
318                }
319            }
320            Transition::Next(ReadStates::Finish(Finish), Ok(()))
321        })
322    }
323}
324
325impl<'ctx> StateMachine<ReadCtx<'ctx>, ReadStepOut> for Finish {
326    type StepResult<'a>
327        = Pin<Box<dyn Future<Output = ReadStepOut> + Send + 'a>>
328    where
329        Self: 'a,
330        ReadCtx<'ctx>: 'a;
331
332    fn step<'a>(&'a self, ctx: &'a mut ReadCtx<'ctx>) -> Self::StepResult<'a> {
333        Box::pin(async move {
334            let (status, residual, sense_opt) =
335                match ctx.finalize_status_after_datain(ctx.itt).await {
336                    Ok(v) => v,
337                    Err(e) => return Transition::Done(Err(e)),
338                };
339
340            if status != ScsiStatus::Good {
341                if let Some(sb) = sense_opt {
342                    if let Ok(sense) = SenseData::parse(&sb) {
343                        return Transition::Done(Err(anyhow!(
344                            "SCSI CheckCondition: {:?}",
345                            sense
346                        )));
347                    }
348                    return Transition::Done(Err(anyhow!(
349                        "SCSI CheckCondition (sense {} bytes): {:02X?}",
350                        sb.len(),
351                        sb
352                    )));
353                }
354                return Transition::Done(Err(anyhow!(
355                    "SCSI status != GOOD ({:?}) and no sense provided",
356                    status
357                )));
358            }
359
360            let requested = ctx.read_len as usize;
361            let expected_after_residual = requested.saturating_sub(residual as usize);
362            let got = ctx.rt.acc.len();
363
364            if got != expected_after_residual {
365                return Transition::Done(Err(anyhow!(
366                    "read length mismatch: requested={}, residual={}, \
367                     expected_after_residual={}, got={}",
368                    requested,
369                    residual,
370                    expected_after_residual,
371                    got
372                )));
373            }
374
375            Transition::Done(Ok(()))
376        })
377    }
378}
379
380/// Represents the outcome of a completed SCSI Read operation.
381#[derive(Debug)]
382pub struct ReadOutcome {
383    /// The data received from the target.
384    pub data: Vec<u8>,
385    /// The final SCSI Command Response, if one was sent.
386    pub last_response: Option<PduResponse<ScsiCommandResponse>>,
387}
388
389impl<'ctx> StateMachineCtx<ReadCtx<'ctx>, ReadOutcome> for ReadCtx<'ctx> {
390    async fn execute(&mut self, _cancel: &CancellationToken) -> Result<ReadOutcome> {
391        debug!("Loop Read");
392
393        loop {
394            let state = self.state.take().context("state must be set ReadCtx")?;
395            let tr = match state {
396                ReadStates::Start(s) => s.step(self).await,
397                ReadStates::Wait(s) => s.step(self).await,
398                ReadStates::Finish(s) => s.step(self).await,
399            };
400
401            match tr {
402                Transition::Next(next, r) => {
403                    r?;
404                    self.state = Some(next);
405                },
406                Transition::Stay(Ok(_)) => {},
407                Transition::Stay(Err(e)) => return Err(e),
408                Transition::Done(r) => {
409                    r?;
410                    return Ok(ReadOutcome {
411                        data: std::mem::take(&mut self.rt.acc),
412                        last_response: self.last_response.take(),
413                    });
414                },
415            }
416        }
417    }
418}
iscsi_client_rs/state_machine/read_states.rs

iscsi_client_rs/state_machine/
read_states.rs
