// Copyright 2023 The ChromiumOS Authors

// Use of this source code is governed by a BSD-style license that can be

// found in the LICENSE file.

#![deny(missing_docs)]

//! A SCSI controller has SCSI target(s), a SCSI target has logical unit(s).

//! crosvm currently supports only one logical unit in a target (LUN0), therefore a SCSI target is

//! tied to a logical unit and a disk image belongs to a logical unit in crosvm.

use std::cell::RefCell;

use std::collections::BTreeMap;

use std::collections::BTreeSet;

use std::io;

use std::io::Read;

use std::io::Write;

use std::rc::Rc;

use anyhow::Context;

use base::error;

use base::warn;

use base::Event;

use base::WorkerThread;

use cros_async::EventAsync;

use cros_async::Executor;

use cros_async::ExecutorKind;

use disk::AsyncDisk;

use disk::DiskFile;

use futures::pin_mut;

use futures::stream::FuturesUnordered;

use futures::FutureExt;

use futures::StreamExt;

use remain::sorted;

use thiserror::Error as ThisError;

use virtio_sys::virtio_scsi::virtio_scsi_config;

use virtio_sys::virtio_scsi::virtio_scsi_ctrl_an_resp;

use virtio_sys::virtio_scsi::virtio_scsi_ctrl_tmf_req;

use virtio_sys::virtio_scsi::virtio_scsi_ctrl_tmf_resp;

use virtio_sys::virtio_scsi::virtio_scsi_event;

use virtio_sys::virtio_scsi::VIRTIO_SCSI_CDB_DEFAULT_SIZE;

use virtio_sys::virtio_scsi::VIRTIO_SCSI_SENSE_DEFAULT_SIZE;

use virtio_sys::virtio_scsi::VIRTIO_SCSI_S_BAD_TARGET;

use virtio_sys::virtio_scsi::VIRTIO_SCSI_S_FUNCTION_REJECTED;

use virtio_sys::virtio_scsi::VIRTIO_SCSI_S_FUNCTION_SUCCEEDED;

use virtio_sys::virtio_scsi::VIRTIO_SCSI_S_INCORRECT_LUN;

use virtio_sys::virtio_scsi::VIRTIO_SCSI_S_OK;

use virtio_sys::virtio_scsi::VIRTIO_SCSI_T_AN_QUERY;

use virtio_sys::virtio_scsi::VIRTIO_SCSI_T_AN_SUBSCRIBE;

use virtio_sys::virtio_scsi::VIRTIO_SCSI_T_TMF;

use virtio_sys::virtio_scsi::VIRTIO_SCSI_T_TMF_I_T_NEXUS_RESET;

use virtio_sys::virtio_scsi::VIRTIO_SCSI_T_TMF_LOGICAL_UNIT_RESET;

use vm_memory::GuestMemory;

use zerocopy::FromBytes;

use zerocopy::Immutable;

use zerocopy::IntoBytes;

use zerocopy::KnownLayout;

use crate::virtio::async_utils;

use crate::virtio::block::sys::get_seg_max;

use crate::virtio::copy_config;

use crate::virtio::scsi::commands::execute_cdb;

use crate::virtio::scsi::constants::CHECK_CONDITION;

use crate::virtio::scsi::constants::GOOD;

use crate::virtio::scsi::constants::ILLEGAL_REQUEST;

use crate::virtio::scsi::constants::MEDIUM_ERROR;

use crate::virtio::DescriptorChain;

use crate::virtio::DeviceType as VirtioDeviceType;

use crate::virtio::Interrupt;

use crate::virtio::Queue;

use crate::virtio::Reader;

use crate::virtio::VirtioDevice;

use crate::virtio::Writer;

// The following values reflects the virtio v1.2 spec:

// <https://docs.oasis-open.org/virtio/virtio/v1.2/csd01/virtio-v1.2-csd01.html#x1-3470004>

// Should have one controlq, one eventq, and at least one request queue.

const MIN_NUM_QUEUES: usize = 3;

// The number of queues exposed by the device.

// First crosvm pass this value through `VirtioDevice::read_config`, and then the driver determines

// the number of queues which does not exceed the passed value. The determined value eventually

// shows as the length of `queues` in `VirtioDevice::activate`.

const MAX_NUM_QUEUES: usize = 16;

// Max channel should be 0.

const DEFAULT_MAX_CHANNEL: u16 = 0;

// Max target should be less than or equal to 255.

const DEFAULT_MAX_TARGET: u16 = 255;

// Max lun should be less than or equal to 16383

const DEFAULT_MAX_LUN: u32 = 16383;

const DEFAULT_QUEUE_SIZE: u16 = 1024;

// The maximum number of linked commands.

const MAX_CMD_PER_LUN: u32 = 1024;

// We do not set a limit on the transfer size.

const MAX_SECTORS: u32 = u32::MAX;

// The length of sense data in fixed format. Details are in SPC-3 t10 revision 23:

// <https://www.t10.org/cgi-bin/ac.pl?t=f&f=spc3r23.pdf>

const FIXED_FORMAT_SENSE_SIZE: u32 = 18;

#[repr(C, packed)]

#[derive(Debug, Default, Copy, Clone, FromBytes, Immutable, IntoBytes, KnownLayout)]

struct VirtioScsiCmdReqHeader {

    lun: [u8; 8usize],

    tag: u64,

    task_attr: u8,

    prio: u8,

    crn: u8,

}

#[repr(C, packed)]

#[derive(Debug, Default, Copy, Clone, FromBytes, Immutable, IntoBytes, KnownLayout)]

struct VirtioScsiCmdRespHeader {

    sense_len: u32,

    resid: u32,

    status_qualifier: u16,

    status: u8,

    response: u8,

}

impl VirtioScsiCmdRespHeader {

    fn ok() -> Self {

        VirtioScsiCmdRespHeader {

            sense_len: 0,

            resid: 0,

            status_qualifier: 0,

            status: GOOD,

            response: VIRTIO_SCSI_S_OK as u8,

        }

    }

}

/// Errors that happen while handling scsi commands.

#[sorted]

#[derive(ThisError, Debug)]

pub enum ExecuteError {

    #[error("invalid cdb field")]

    InvalidField,

    #[error("invalid parameter length")]

    InvalidParamLen,

    #[error("{xfer_blocks} blocks from LBA {lba} exceeds end of this device {last_lba}")]

    LbaOutOfRange {

        lba: u64,

        xfer_blocks: usize,

        last_lba: u64,

    },

    #[error("failed to read message: {0}")]

    Read(io::Error),

    #[error("failed to read command from cdb")]

    ReadCommand,

    #[error("io error {resid} bytes remained to be read: {desc_error}")]

    ReadIo {

        resid: usize,

        desc_error: disk::Error,

    },

    #[error("writing to a read only device")]

    ReadOnly,

    #[error("saving parameters not supported")]

    SavingParamNotSupported,

    #[error("synchronization error")]

    SynchronizationError,

    #[error("unsupported scsi command: {0}")]

    Unsupported(u8),

    #[error("failed to write message: {0}")]

    Write(io::Error),

    #[error("io error {resid} bytes remained to be written: {desc_error}")]

    WriteIo {

        resid: usize,

        desc_error: disk::Error,

    },

}

impl ExecuteError {

    // converts ExecuteError to (VirtioScsiCmdReqHeader, Sense)

    fn as_resp(&self) -> (VirtioScsiCmdRespHeader, Sense) {

        let resp = VirtioScsiCmdRespHeader::ok();

        // The asc and ascq assignments are taken from the t10 SPC spec.

        // cf) Table 28 of <https://www.t10.org/cgi-bin/ac.pl?t=f&f=spc3r23.pdf>

        let sense = match self {

            Self::Read(_) | Self::ReadCommand => {

                // UNRECOVERED READ ERROR

                Sense {

                    key: MEDIUM_ERROR,

                    asc: 0x11,

                    ascq: 0x00,

                }

            }

            Self::Write(_) => {

                // WRITE ERROR

                Sense {

                    key: MEDIUM_ERROR,

                    asc: 0x0c,

                    ascq: 0x00,

                }

            }

            Self::InvalidField => {

                // INVALID FIELD IN CDB

                Sense {

                    key: ILLEGAL_REQUEST,

                    asc: 0x24,

                    ascq: 0x00,

                }

            }

            Self::InvalidParamLen => {

                // INVALID PARAMETER LENGTH

                Sense {

                    key: ILLEGAL_REQUEST,

                    asc: 0x1a,

                    ascq: 0x00,

                }

            }

            Self::Unsupported(_) => {

                // INVALID COMMAND OPERATION CODE

                Sense {

                    key: ILLEGAL_REQUEST,

                    asc: 0x20,

                    ascq: 0x00,

                }

            }

            Self::ReadOnly | Self::LbaOutOfRange { .. } => {

                // LOGICAL BLOCK ADDRESS OUT OF RANGE

                Sense {

                    key: ILLEGAL_REQUEST,

                    asc: 0x21,

                    ascq: 0x00,

                }

            }

            Self::SavingParamNotSupported => Sense {

                // SAVING PARAMETERS NOT SUPPORTED

                key: ILLEGAL_REQUEST,

                asc: 0x39,

                ascq: 0x00,

            },

            Self::SynchronizationError => Sense {

                // SYNCHRONIZATION ERROR

                key: MEDIUM_ERROR,

                asc: 0x16,

                ascq: 0x00,

            },

            // Ignore these errors.

            Self::ReadIo { resid, desc_error } | Self::WriteIo { resid, desc_error } => {

                warn!("error while performing I/O {}", desc_error);

                let hdr = VirtioScsiCmdRespHeader {

                    resid: (*resid).try_into().unwrap_or(u32::MAX).to_be(),

                    ..resp

                };

                return (hdr, Sense::default());

            }

        };

        (

            VirtioScsiCmdRespHeader {

                sense_len: FIXED_FORMAT_SENSE_SIZE,

                status: CHECK_CONDITION,

                ..resp

            },

            sense,

        )

    }

}

/// Sense code representation

#[derive(Copy, Clone, Debug, Default, PartialEq, Eq)]

pub struct Sense {

    /// Provides generic information describing an error or exception condition.

    pub key: u8,

    /// Additional Sense Code.

    /// Indicates further information related to the error or exception reported in the key field.

    pub asc: u8,

    /// Additional Sense Code Qualifier.

    /// Indicates further detailed information related to the additional sense code.

    pub ascq: u8,

}

impl Sense {

    fn write_to(&self, writer: &mut Writer, sense_size: u32) -> Result<(), ExecuteError> {

        let mut sense_data = [0u8; FIXED_FORMAT_SENSE_SIZE as usize];

        // Fixed format sense data has response code:

        // 1) 0x70 for current errors

        // 2) 0x71 for deferred errors

        sense_data[0] = 0x70;

        // sense_data[1]: Obsolete

        // Sense key

        sense_data[2] = self.key;

        // sense_data[3..7]: Information field, which we do not support.

        // Additional length. The data is 18 bytes, and this byte is 8th.

        sense_data[7] = 10;

        // sense_data[8..12]: Command specific information, which we do not support.

        // Additional sense code

        sense_data[12] = self.asc;

        // Additional sense code qualifier

        sense_data[13] = self.ascq;

        // sense_data[14]: Field replaceable unit code, which we do not support.

        // sense_data[15..18]: Field replaceable unit code, which we do not support.

        writer.write_all(&sense_data).map_err(ExecuteError::Write)?;

        writer.consume_bytes(sense_size as usize - sense_data.len());

        Ok(())

    }

}

/// Describes each SCSI logical unit.

struct LogicalUnit {

    /// The logical block address of the last logical block on the target device.

    last_lba: u64,

    /// Block size of the target device.

    block_size: u32,

    read_only: bool,

    // Represents the image on disk.

    disk_image: Box<dyn DiskFile>,

}

impl LogicalUnit {

    fn make_async(self, ex: &Executor) -> anyhow::Result<AsyncLogicalUnit> {

        let disk_image = self

            .disk_image

            .to_async_disk(ex)

            .context("Failed to create async disk")?;

        Ok(AsyncLogicalUnit {

            last_lba: self.last_lba,

            block_size: self.block_size,

            read_only: self.read_only,

            disk_image,

        })

    }

}

/// A logical unit with an AsyncDisk as the disk.

pub struct AsyncLogicalUnit {

    pub last_lba: u64,

    pub block_size: u32,

    pub read_only: bool,

    // Represents the async image on disk.

    pub disk_image: Box<dyn AsyncDisk>,

}

type TargetId = u8;

struct Targets(BTreeMap<TargetId, LogicalUnit>);

impl Targets {

    fn try_clone(&self) -> io::Result<Self> {

        let logical_units = self

            .0

            .iter()

            .map(|(id, logical_unit)| {

                let disk_image = logical_unit.disk_image.try_clone()?;

                Ok((

                    *id,

                    LogicalUnit {

                        disk_image,

                        last_lba: logical_unit.last_lba,

                        block_size: logical_unit.block_size,

                        read_only: logical_unit.read_only,

                    },

                ))

            })

            .collect::<io::Result<_>>()?;

        Ok(Self(logical_units))

    }

    fn target_ids(&self) -> BTreeSet<TargetId> {

        self.0.keys().cloned().collect()

    }

}

/// Configuration of each SCSI device.

pub struct DiskConfig {

    /// The disk file of the device.

    pub file: Box<dyn DiskFile>,

    /// The block size of the SCSI disk.

    pub block_size: u32,

    /// Indicates whether the SCSI disk is read only.

    pub read_only: bool,

}

/// Vitio device for exposing SCSI command operations on a host file.

pub struct Controller {

    // Bitmap of virtio-scsi feature bits.

    avail_features: u64,

    // Sizes for the virtqueue.

    queue_sizes: Vec<u16>,

    // The maximum number of segments that can be in a command.

    seg_max: u32,

    // The size of the sense data.

    sense_size: u32,

    // The byte size of the CDB that the driver will write.

    cdb_size: u32,

    executor_kind: ExecutorKind,

    worker_threads: Vec<WorkerThread<()>>,

    // Stores target devices by its target id. Currently we only support bus id 0.

    targets: Option<Targets>,

    // Whether the devices handles requests in multiple request queues.

    // If true, each virtqueue will be handled in a separate worker thread.

    multi_queue: bool,

}

impl Controller {

    /// Creates a virtio-scsi device.

    pub fn new(base_features: u64, disks: Vec<DiskConfig>) -> anyhow::Result<Self> {

        let multi_queue = disks.iter().all(|disk| disk.file.try_clone().is_ok());

        let num_queues = if multi_queue {

            MAX_NUM_QUEUES

        } else {

            MIN_NUM_QUEUES

        };

        let logical_units = disks

            .into_iter()

            .enumerate()

            .map(|(i, disk)| {

                let num_blocks = disk

                    .file

                    .get_len()

                    .context("Failed to get the length of the disk image")?

                    / disk.block_size as u64;

                let last_lba = num_blocks

                    .checked_sub(1)

                    .context("Invalid zero-length SCSI LUN")?;

                let target = LogicalUnit {

                    last_lba,

                    block_size: disk.block_size,

                    read_only: disk.read_only,

                    disk_image: disk.file,

                };

                Ok((i as TargetId, target))

            })

            .collect::<anyhow::Result<_>>()?;

        // b/300560198: Support feature bits in virtio-scsi.

        Ok(Self {

            avail_features: base_features,

            queue_sizes: vec![DEFAULT_QUEUE_SIZE; num_queues],

            seg_max: get_seg_max(DEFAULT_QUEUE_SIZE),

            sense_size: VIRTIO_SCSI_SENSE_DEFAULT_SIZE,

            cdb_size: VIRTIO_SCSI_CDB_DEFAULT_SIZE,

            executor_kind: ExecutorKind::default(),

            worker_threads: vec![],

            targets: Some(Targets(logical_units)),

            multi_queue,

        })

    }

    fn build_config_space(&self) -> virtio_scsi_config {

        virtio_scsi_config {

            // num_queues is the number of request queues only so we subtract 2 for the control

            // queue and the event queue.

            num_queues: self.queue_sizes.len() as u32 - 2,

            seg_max: self.seg_max,

            max_sectors: MAX_SECTORS,

            cmd_per_lun: MAX_CMD_PER_LUN,

            event_info_size: std::mem::size_of::<virtio_scsi_event>() as u32,

            sense_size: self.sense_size,

            cdb_size: self.cdb_size,

            max_channel: DEFAULT_MAX_CHANNEL,

            max_target: DEFAULT_MAX_TARGET,

            max_lun: DEFAULT_MAX_LUN,

        }

    }

    // Executes a request in the controlq.

    fn execute_control(

        reader: &mut Reader,

        writer: &mut Writer,

        target_ids: &BTreeSet<TargetId>,

    ) -> Result<(), ExecuteError> {

        let typ = reader.peek_obj::<u32>().map_err(ExecuteError::Read)?;

        match typ {

            VIRTIO_SCSI_T_TMF => {

                let tmf = reader

                    .read_obj::<virtio_scsi_ctrl_tmf_req>()

                    .map_err(ExecuteError::Read)?;

                let resp = Self::execute_tmf(tmf, target_ids);

                writer.write_obj(resp).map_err(ExecuteError::Write)?;

                Ok(())

            }

            VIRTIO_SCSI_T_AN_QUERY | VIRTIO_SCSI_T_AN_SUBSCRIBE => {

                // We do not support any asynchronous notification queries hence `event_actual`

                // will be 0.

                let resp = virtio_scsi_ctrl_an_resp {

                    event_actual: 0,

                    response: VIRTIO_SCSI_S_OK as u8,

                };

                writer.write_obj(resp).map_err(ExecuteError::Write)?;

                Ok(())

            }

            _ => {

                error!("invalid type of a control request: {typ}");

                Err(ExecuteError::InvalidField)

            }

        }

    }

    // Executes a TMF (task management function) request.

    fn execute_tmf(

        tmf: virtio_scsi_ctrl_tmf_req,

        target_ids: &BTreeSet<TargetId>,

    ) -> virtio_scsi_ctrl_tmf_resp {

        match tmf.subtype {

            VIRTIO_SCSI_T_TMF_LOGICAL_UNIT_RESET | VIRTIO_SCSI_T_TMF_I_T_NEXUS_RESET => {

                // We only have LUN0.

                let lun = tmf.lun;

                let target_id = lun[1];

                let response = if target_ids.contains(&target_id) {

                    let is_lun0 = u16::from_be_bytes([lun[2], lun[3]]) & 0x3fff == 0;

                    if is_lun0 {

                        VIRTIO_SCSI_S_FUNCTION_SUCCEEDED as u8

                    } else {

                        VIRTIO_SCSI_S_INCORRECT_LUN as u8

                    }

                } else {

                    VIRTIO_SCSI_S_BAD_TARGET as u8

                };

                virtio_scsi_ctrl_tmf_resp { response }

            }

            subtype => {

                error!("TMF request {subtype} is not supported");

                virtio_scsi_ctrl_tmf_resp {

                    response: VIRTIO_SCSI_S_FUNCTION_REJECTED as u8,

                }

            }

        }

    }

    async fn execute_request(

        reader: &mut Reader,

        resp_writer: &mut Writer,

        data_writer: &mut Writer,

        targets: &BTreeMap<TargetId, AsyncLogicalUnit>,

        sense_size: u32,

        cdb_size: u32,

    ) -> Result<(), ExecuteError> {

        let req_header = reader

            .read_obj::<VirtioScsiCmdReqHeader>()

            .map_err(ExecuteError::Read)?;

        match Self::get_logical_unit(req_header.lun, targets) {

            Some(target) => {

                let mut cdb = vec![0; cdb_size as usize];

                reader.read_exact(&mut cdb).map_err(ExecuteError::Read)?;

                match execute_cdb(&cdb, reader, data_writer, target).await {

                    Ok(()) => {

                        let hdr = VirtioScsiCmdRespHeader {

                            sense_len: 0,

                            resid: 0,

                            status_qualifier: 0,

                            status: GOOD,

                            response: VIRTIO_SCSI_S_OK as u8,

                        };

                        resp_writer.write_obj(hdr).map_err(ExecuteError::Write)?;

                        resp_writer.consume_bytes(sense_size as usize);

                        Ok(())

                    }

                    Err(err) => {

                        error!("error while executing a scsi request: {err}");

                        let (hdr, sense) = err.as_resp();

                        resp_writer.write_obj(hdr).map_err(ExecuteError::Write)?;

                        sense.write_to(resp_writer, sense_size)

                    }

                }

            }

            None => {

                let hdr = VirtioScsiCmdRespHeader {

                    response: VIRTIO_SCSI_S_BAD_TARGET as u8,

                    ..Default::default()

                };

                resp_writer.write_obj(hdr).map_err(ExecuteError::Write)?;

                resp_writer.consume_bytes(sense_size as usize);

                Ok(())

            }

        }

    }

    fn get_logical_unit(

        lun: [u8; 8],

        targets: &BTreeMap<TargetId, AsyncLogicalUnit>,

    ) -> Option<&AsyncLogicalUnit> {

        // First byte should be 1.

        if lun[0] != 1 {

            return None;

        }

        // General search strategy for scsi devices is as follows:

        // 1) Look for a device which has the same bus id and lun indicated by the given lun. If

        //    there is one, that is the target device.

        // 2) If we cannot find such device, then we return the first device that has the same bus

        //    id.

        // Since we only support one LUN per target, we only need to use the target id.

        let target_id = lun[1];

        targets.get(&target_id)

    }

}

impl VirtioDevice for Controller {

    fn keep_rds(&self) -> Vec<base::RawDescriptor> {

        match &self.targets {

            Some(targets) => targets

                .0

                .values()

                .flat_map(|t| t.disk_image.as_raw_descriptors())

                .collect(),

            None => vec![],

        }

    }

    fn features(&self) -> u64 {

        self.avail_features

    }

    fn device_type(&self) -> VirtioDeviceType {

        VirtioDeviceType::Scsi

    }

    fn queue_max_sizes(&self) -> &[u16] {

        &self.queue_sizes

    }

    fn read_config(&self, offset: u64, data: &mut [u8]) {

        let config_space = self.build_config_space();

        copy_config(data, 0, config_space.as_bytes(), offset);

    }

    fn write_config(&mut self, offset: u64, data: &[u8]) {

        let mut config = self.build_config_space();

        copy_config(config.as_mut_bytes(), offset, data, 0);

        // Only `sense_size` and `cdb_size` are modifiable by the driver.

        self.sense_size = config.sense_size;

        self.cdb_size = config.cdb_size;

    }

    fn activate(

        &mut self,

        _mem: GuestMemory,

        _interrupt: Interrupt,

        mut queues: BTreeMap<usize, Queue>,

    ) -> anyhow::Result<()> {

        let executor_kind = self.executor_kind;

        // 0th virtqueue is the controlq.

        let controlq = queues.remove(&0).context("controlq should be present")?;

        // 1st virtqueue is the eventq.

        // We do not send any events through eventq.

        let _eventq = queues.remove(&1).context("eventq should be present")?;

        let targets = self.targets.take().context("failed to take SCSI targets")?;

        let target_ids = targets.target_ids();

        let sense_size = self.sense_size;

        let cdb_size = self.cdb_size;

        // The rest of the queues are request queues.

        let request_queues = if self.multi_queue {

            queues

                .into_values()

                .map(|queue| {

                    let targets = targets

                        .try_clone()

                        .context("Failed to clone a disk image")?;

                    Ok((queue, targets))

                })

                .collect::<anyhow::Result<_>>()?

        } else {

            // Handle all virtio requests with one thread.

            vec![(

                queues

                    .remove(&2)

                    .context("request queue should be present")?,

                targets,

            )]

        };

        let worker_thread = WorkerThread::start("v_scsi_ctrlq", move |kill_evt| {

            let ex =

                Executor::with_executor_kind(executor_kind).expect("Failed to create an executor");

            if let Err(err) = ex

                .run_until(run_worker(

                    &ex,

                    controlq,

                    kill_evt,

                    QueueType::Control { target_ids },

                    sense_size,

                    cdb_size,

                ))

                .expect("run_until failed")

            {

                error!("run_worker failed: {err}");

            }

        });

        self.worker_threads.push(worker_thread);

        for (i, (queue, targets)) in request_queues.into_iter().enumerate() {

            let worker_thread =

                WorkerThread::start(format!("v_scsi_req_{}", i + 2), move |kill_evt| {

                    let ex = Executor::with_executor_kind(executor_kind)

                        .expect("Failed to create an executor");

                    let async_logical_unit = targets

                        .0

                        .into_iter()

                        .map(|(idx, unit)| match unit.make_async(&ex) {

                            Ok(async_unit) => (idx, async_unit),

                            Err(err) => panic!("{err}"),

                        })

                        .collect();

                    if let Err(err) = ex

                        .run_until(run_worker(

                            &ex,

                            queue,

                            kill_evt,

                            QueueType::Request(async_logical_unit),

                            sense_size,

                            cdb_size,

                        ))

                        .expect("run_until failed")

                    {

                        error!("run_worker failed: {err}");

                    }

                });

            self.worker_threads.push(worker_thread);

        }

        Ok(())

    }

}

enum QueueType {

    Control { target_ids: BTreeSet<TargetId> },

    Request(BTreeMap<TargetId, AsyncLogicalUnit>),

}

async fn run_worker(

    ex: &Executor,

    queue: Queue,

    kill_evt: Event,

    queue_type: QueueType,

    sense_size: u32,

    cdb_size: u32,

) -> anyhow::Result<()> {

    let kill = async_utils::await_and_exit(ex, kill_evt).fuse();

    pin_mut!(kill);

    let kick_evt = queue

        .event()

        .try_clone()

        .expect("Failed to clone queue event");

    let queue_handler = handle_queue(

        Rc::new(RefCell::new(queue)),

        EventAsync::new(kick_evt, ex).expect("Failed to create async event for queue"),

        queue_type,

        sense_size,

        cdb_size,

    )

    .fuse();

    pin_mut!(queue_handler);

    futures::select! {

        _ = queue_handler => anyhow::bail!("queue handler exited unexpectedly"),

        r = kill => r.context("failed to wait on the kill event"),

    }

}

async fn handle_queue(

    queue: Rc<RefCell<Queue>>,

    evt: EventAsync,

    queue_type: QueueType,

    sense_size: u32,

    cdb_size: u32,

) {

    let mut background_tasks = FuturesUnordered::new();

    let evt_future = evt.next_val().fuse();

    pin_mut!(evt_future);

    loop {

        futures::select! {

            _ = background_tasks.next() => continue,

            res = evt_future => {

                evt_future.set(evt.next_val().fuse());

                if let Err(e) = res {

                    error!("Failed to read the next queue event: {e}");

                    continue;

                }

            }

        }

        while let Some(chain) = queue.borrow_mut().pop() {

            background_tasks.push(process_one_chain(

                &queue,

                chain,

                &queue_type,

                sense_size,

                cdb_size,

            ));

        }

    }

}

async fn process_one_chain(

    queue: &RefCell<Queue>,

    mut avail_desc: DescriptorChain,

    queue_type: &QueueType,

    sense_size: u32,

    cdb_size: u32,

) {

    let _trace = cros_tracing::trace_event!(VirtioScsi, "process_one_chain");

    let len = process_one_request(&mut avail_desc, queue_type, sense_size, cdb_size).await;

    let mut queue = queue.borrow_mut();

    queue.add_used_with_bytes_written(avail_desc, len as u32);

    queue.trigger_interrupt();

}

async fn process_one_request(

    avail_desc: &mut DescriptorChain,

    queue_type: &QueueType,

    sense_size: u32,

    cdb_size: u32,

) -> usize {

    let reader = &mut avail_desc.reader;

    let resp_writer = &mut avail_desc.writer;

    match queue_type {

        QueueType::Control { target_ids } => {

            if let Err(err) = Controller::execute_control(reader, resp_writer, target_ids) {

                error!("failed to execute control request: {err}");

            }

            resp_writer.bytes_written()

        }

        QueueType::Request(async_targets) => {

            let mut data_writer = resp_writer

                .split_at(std::mem::size_of::<VirtioScsiCmdRespHeader>() + sense_size as usize);

            if let Err(err) = Controller::execute_request(

                reader,

                resp_writer,

                &mut data_writer,

                async_targets,

                sense_size,

                cdb_size,

            )

            .await

            {

                // If the write of the virtio_scsi_cmd_resp fails, there is nothing we can do to

                // inform the error to the guest driver (we usually propagate errors with sense

                // field, which is in the struct virtio_scsi_cmd_resp). The guest driver should

                // have at least sizeof(virtio_scsi_cmd_resp) bytes of device-writable part

                // regions. For now we simply emit an error message.

                let (hdr, sense) = err.as_resp();

                if let Err(e) = resp_writer.write_obj(hdr) {

                    error!("failed to write VirtioScsiCmdRespHeader: {e}");

                }

                if let Err(e) = sense.write_to(resp_writer, sense_size) {

                    error!("failed to write sense data: {e}");

                }

            }

            resp_writer.bytes_written() + data_writer.bytes_written()

        }

    }

}

#[cfg(test)]

mod tests {

    use std::fs::File;

    use std::mem::size_of;

    use std::mem::size_of_val;

    use std::rc::Rc;

    use cros_async::Executor;

    use disk::SingleFileDisk;

    use tempfile::tempfile;

    use virtio_sys::virtio_scsi::virtio_scsi_cmd_req;

    use virtio_sys::virtio_scsi::virtio_scsi_cmd_resp;

    use virtio_sys::virtio_scsi::VIRTIO_SCSI_S_OK;

    use vm_memory::GuestAddress;

    use vm_memory::GuestMemory;

    use super::*;

    use crate::virtio::create_descriptor_chain;

    use crate::virtio::scsi::constants::READ_10;

    use crate::virtio::DescriptorType;

    fn setup_disk(disk_size: u64) -> (File, Vec<u8>) {

        let mut file_content = vec![0; disk_size as usize];

        for i in 0..disk_size {

            file_content[i as usize] = (i % 10) as u8;

        }

        let mut f = tempfile().unwrap();

        f.set_len(disk_size).unwrap();

        f.write_all(file_content.as_slice()).unwrap();

        (f, file_content)

    }

    fn build_read_req_header(target_id: u8, start_lba: u8, xfer_blocks: u8) -> virtio_scsi_cmd_req {

        let mut cdb = [0; 32];

        cdb[0] = READ_10;

        cdb[5] = start_lba;

        cdb[8] = xfer_blocks;

        virtio_scsi_cmd_req {

            lun: [1, 0, 0, target_id, 0, 0, 0, 0],

            cdb,

            ..Default::default()

        }

    }

    fn setup_desciptor_chain(

        target_id: TargetId,

        start_lba: u8,

        xfer_blocks: u8,

        block_size: u32,

        mem: &Rc<GuestMemory>,

    ) -> DescriptorChain {

        let req_hdr = build_read_req_header(target_id, start_lba, xfer_blocks);

        let xfer_bytes = xfer_blocks as u32 * block_size;

        create_descriptor_chain(

            mem,

            GuestAddress(0x100),  // Place descriptor chain at 0x100.

            GuestAddress(0x1000), // Describe buffer at 0x1000.

            vec![

                // Request header

                (DescriptorType::Readable, size_of_val(&req_hdr) as u32),

                // Response header

                (

                    DescriptorType::Writable,

                    size_of::<virtio_scsi_cmd_resp>() as u32,

                ),

                (DescriptorType::Writable, xfer_bytes),

            ],

            0,

        )

        .expect("create_descriptor_chain failed")

    }

    fn read_blocks(

        ex: &Executor,

        file_disks: &[File],

        target_id: u8,

        start_lba: u8,

        xfer_blocks: u8,

        block_size: u32,

    ) -> (virtio_scsi_cmd_resp, Vec<u8>) {

        let xfer_bytes = xfer_blocks as u32 * block_size;

        let mem = Rc::new(

            GuestMemory::new(&[(GuestAddress(0u64), 4 * 1024 * 1024)])

                .expect("Creating guest memory failed."),

        );

        let req_hdr = build_read_req_header(target_id, start_lba, xfer_blocks);

        mem.write_obj_at_addr(req_hdr, GuestAddress(0x1000))

            .expect("writing req failed");

        let mut avail_desc = setup_desciptor_chain(target_id, 0, xfer_blocks, block_size, &mem);

        let targets = file_disks

            .iter()

            .enumerate()

            .map(|(i, file)| {

                let file = file.try_clone().unwrap();

                let disk_image = Box::new(SingleFileDisk::new(file, ex).unwrap());

                let logical_unit = AsyncLogicalUnit {

                    last_lba: 0xFFF,

                    block_size,

                    read_only: false,

                    disk_image,

                };

                (i as TargetId, logical_unit)

            })

            .collect();

        ex.run_until(process_one_request(

            &mut avail_desc,

            &QueueType::Request(targets),

            VIRTIO_SCSI_SENSE_DEFAULT_SIZE,

            VIRTIO_SCSI_CDB_DEFAULT_SIZE,

        ))

        .expect("running executor failed");

        let resp_offset = GuestAddress((0x1000 + size_of::<virtio_scsi_cmd_resp>()) as u64);

        let resp = mem

            .read_obj_from_addr::<virtio_scsi_cmd_resp>(resp_offset)

            .unwrap();

        let dataout_offset = GuestAddress(

            (0x1000 + size_of::<virtio_scsi_cmd_req>() + size_of::<virtio_scsi_cmd_resp>()) as u64,