Skip to content

FIO

fio (Flexible I/O Tester) — the standard Linux I/O benchmarking and stress-testing tool. Wraps job file generation, execution, and result parsing.

sts.fio.fio

FIO test execution.

FIO pydantic-model

Bases: StsBaseModel

FIO test execution.

Example 
fio = FIO()  # Uses default parameters
fio = FIO('/dev/sda')  # Uses device with default parameters
fio = FIO(parameters=DefaultParameters(name='test'))  # Custom parameters
Show JSON schema: 
{
  "$defs": {
    "FIOParameters": {
      "additionalProperties": false,
      "description": "Base FIO parameters.",
      "properties": {
        "name": {
          "default": "sts-fio",
          "title": "Name",
          "type": "string"
        },
        "ioengine": {
          "default": "libaio",
          "enum": [
            "libaio",
            "sync",
            "posixaio",
            "mmap",
            "splice"
          ],
          "title": "Ioengine",
          "type": "string"
        },
        "direct": {
          "default": true,
          "title": "Direct",
          "type": "boolean"
        },
        "rw": {
          "default": "randrw",
          "enum": [
            "read",
            "write",
            "randread",
            "randwrite",
            "randrw",
            "trim"
          ],
          "title": "Rw",
          "type": "string"
        },
        "bs": {
          "default": "4k",
          "minLength": 1,
          "title": "Bs",
          "type": "string"
        },
        "numjobs": {
          "default": 1,
          "title": "Numjobs",
          "type": "integer"
        },
        "group_reporting": {
          "default": false,
          "title": "Group Reporting",
          "type": "boolean"
        },
        "verify_fatal": {
          "default": false,
          "title": "Verify Fatal",
          "type": "boolean"
        },
        "end_fsync": {
          "default": false,
          "title": "End Fsync",
          "type": "boolean"
        },
        "time_based": {
          "default": false,
          "title": "Time Based",
          "type": "boolean"
        },
        "filename": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Filename"
        },
        "size": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Size"
        },
        "runtime": {
          "anyOf": [
            {
              "exclusiveMinimum": 0,
              "type": "integer"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Runtime"
        },
        "iodepth": {
          "anyOf": [
            {
              "type": "integer"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Iodepth"
        },
        "iodepth_batch_submit": {
          "anyOf": [
            {
              "exclusiveMinimum": 0,
              "type": "integer"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Iodepth Batch Submit"
        },
        "iodepth_batch_complete_min": {
          "anyOf": [
            {
              "exclusiveMinimum": 0,
              "type": "integer"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Iodepth Batch Complete Min"
        },
        "verify": {
          "anyOf": [
            {
              "enum": [
                "crc32",
                "crc32c",
                "crc32c-intel",
                "md5",
                "sha1",
                "sha256",
                "sha512",
                "xxhash",
                "meta"
              ],
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Verify"
        },
        "verify_backlog": {
          "anyOf": [
            {
              "type": "integer"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Verify Backlog"
        }
      },
      "title": "FIOParameters",
      "type": "object"
    }
  },
  "additionalProperties": false,
  "description": "FIO test execution.\n\nExample:\n    ```python\n    fio = FIO()  # Uses default parameters\n    fio = FIO('/dev/sda')  # Uses device with default parameters\n    fio = FIO(parameters=DefaultParameters(name='test'))  # Custom parameters\n    ```",
  "properties": {
    "filename": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "format": "path",
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Filename"
    },
    "parameters": {
      "anyOf": [
        {
          "$ref": "#/$defs/FIOParameters"
        },
        {
          "type": "null"
        }
      ],
      "default": null
    },
    "options": {
      "items": {
        "type": "string"
      },
      "title": "Options",
      "type": "array"
    },
    "config_file": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "format": "path",
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Config File"
    }
  },
  "title": "FIO",
  "type": "object"
}

Fields:

  • filename (PathOrStr | None)
  • parameters (FIOParameters | None)
  • options (list[str])
  • config_file (PathOrStr | None)
  • _UNSET (int)

Validators:

  • _init_parameters
  • _normalize_options → options
Source code in sts_libs/src/sts/fio/fio.py 
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
class FIO(StsBaseModel):
    """FIO test execution.

    Example:
        ```python
        fio = FIO()  # Uses default parameters
        fio = FIO('/dev/sda')  # Uses device with default parameters
        fio = FIO(parameters=DefaultParameters(name='test'))  # Custom parameters
        ```
    """

    # Optional parameters
    filename: PathOrStr | None = None
    parameters: FIOParameters | None = None
    options: list[str] = Field(default_factory=list)
    config_file: PathOrStr | None = None

    @model_validator(mode='after')
    def _init_parameters(self) -> Self:
        """Initialize FIO parameters and config."""
        if self.config_file:
            self.load_config_file(self.config_file)
            if not self.filename and self.parameters:
                self.filename = self.parameters.filename
        elif not self.parameters:
            self.parameters = DefaultParameters(name='sts-fio-default')
            if not self.filename:
                self.filename = self.parameters.filename
        return self

    @field_validator('options')
    @classmethod
    def _normalize_options(cls, options: list[str]) -> list[str]:
        """Normalize long-option spelling; last value wins for each key."""
        seen: dict[str, int] = {}
        normalized: list[str] = []
        for option in options:
            normalized_option = f'--{option.removeprefix("--")}'
            key = normalized_option.split('=', maxsplit=1)[0]
            if key in seen:
                normalized[seen[key]] = normalized_option
            else:
                seen[key] = len(normalized)
                normalized.append(normalized_option)
        return normalized

    def load_config_file(self, config_file: str | Path) -> None:
        """Load parameters from FIO config file.

        Raises:
            FIOConfigError: If config file is invalid or cannot be read
        """
        config_path = Path(config_file)
        if not config_path.exists():
            raise FIOConfigError(f'Config file not found: {config_path}')

        try:
            config = configparser.ConfigParser()
            config.read(config_path)

            # Get the first job section
            job_section = next(s for s in config.sections() if s != 'global')
            params = dict(config[job_section])

            # Convert parameters
            self.parameters = FIOParameters(
                name=job_section,
                **{k: v for k, v in params.items() if not k.startswith('_')},  # type: ignore[arg-type]
            )
        except (configparser.Error, StopIteration) as e:
            raise FIOConfigError(f'Invalid config file: {e}') from e

    def save_config_file(self, config_file: str | Path) -> None:
        """Save parameters to FIO config file.

        Raises:
            FIOConfigError: If parameters are not set or file cannot be written
        """
        if not self.parameters:
            raise FIOConfigError('No parameters to save')

        config = configparser.ConfigParser()
        config[self.parameters.name] = self.parameters.to_dict()

        try:
            with Path(config_file).open('w') as f:
                config.write(f)
        except OSError as e:
            raise FIOConfigError(f'Failed to write config file: {e}') from e

    def update_parameters(self, params: dict[str, Any]) -> None:
        """Update parameters."""
        if not self.parameters:
            self.parameters = DefaultParameters(name='sts-fio-default')
        for key, value in params.items():
            setattr(self.parameters, key, value)

    def update_options(self, options: list[str]) -> None:
        """Add options, deduplicating against existing ones."""
        self.options = self._normalize_options(self.options + options)

    def load_default_params(self) -> None:
        """Load default parameters."""
        self.parameters = DefaultParameters(name='sts-fio-default')

    def load_fs_params(self) -> None:
        """Load filesystem parameters."""
        self.parameters = FileSystemParameters(name='sts-fio-fs')

    def load_block_params(self) -> None:
        """Load block device parameters."""
        self.parameters = BlockDeviceParameters(name='sts-fio-block')

    def load_stress_params(self) -> None:
        """Load stress test parameters."""
        self.parameters = StressParameters(name='sts-fio-stress')

    def _create_argv(self) -> list[str]:
        """Create an argument vector for fio execution.

        Raises:
            FIOConfigError: If parameters are not set or filename is missing
        """
        if not self.parameters:
            raise FIOConfigError('No parameters set')

        argv = ['fio']
        if self.config_file:
            return [*argv, str(self.config_file), *self.options]

        parameter_filename = self.parameters.filename
        if self.filename and parameter_filename and self.filename != Path(parameter_filename):
            raise FIOConfigError(f'Conflicting filename values: {self.filename} and {parameter_filename}')
        filename = self.filename or parameter_filename
        if not filename:
            raise FIOConfigError('No filename set')

        argv.append(f'--filename={filename}')
        argv.extend(f'--{key}={value}' for key, value in self.parameters.to_dict().items() if key != 'filename')
        return [*argv, *self.options]

    def _create_command(self) -> str:
        """Create a displayable fio command string.

        Returns:
            Shell-escaped representation of the execution argument vector.
        """
        return shlex.join(self._create_argv())

    _UNSET: int = -1

    def run(self, *, timeout: int | None = _UNSET) -> CommandResult:
        """Run FIO test.

        Args:
            timeout: Subprocess timeout in seconds.  When omitted the timeout
                is derived from the configured FIO runtime (+ 120 s buffer),
                falling back to 600 s when no runtime is set.  Pass an explicit
                value to override.

        Raises:
            FIOExecutionError: If the fio package cannot be installed or the command cannot be built.
        """
        if not ensure_installed('fio'):
            raise FIOExecutionError('Failed to install fio package')

        try:
            argv = self._create_argv()
        except FIOConfigError as e:
            raise FIOExecutionError(f'Failed to create command: {e}') from e

        if timeout == self._UNSET:
            timeout = self.parameters.runtime + 120 if self.parameters and self.parameters.runtime else 600

        result = run_argv(argv, timeout=timeout)
        if result.failed:
            logger.error(f'FIO run failed:\n{result.stderr}')
        else:
            logger.debug('FIO executed successfully')

        return result

load_block_params()

Load block device parameters.

Source code in sts_libs/src/sts/fio/fio.py 
143
144
145
def load_block_params(self) -> None:
    """Load block device parameters."""
    self.parameters = BlockDeviceParameters(name='sts-fio-block')

load_config_file(config_file)

Load parameters from FIO config file.

Raises:

Type Description
FIOConfigError

If config file is invalid or cannot be read
Source code in sts_libs/src/sts/fio/fio.py 
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
def load_config_file(self, config_file: str | Path) -> None:
    """Load parameters from FIO config file.

    Raises:
        FIOConfigError: If config file is invalid or cannot be read
    """
    config_path = Path(config_file)
    if not config_path.exists():
        raise FIOConfigError(f'Config file not found: {config_path}')

    try:
        config = configparser.ConfigParser()
        config.read(config_path)

        # Get the first job section
        job_section = next(s for s in config.sections() if s != 'global')
        params = dict(config[job_section])

        # Convert parameters
        self.parameters = FIOParameters(
            name=job_section,
            **{k: v for k, v in params.items() if not k.startswith('_')},  # type: ignore[arg-type]
        )
    except (configparser.Error, StopIteration) as e:
        raise FIOConfigError(f'Invalid config file: {e}') from e

load_default_params()

Load default parameters.

Source code in sts_libs/src/sts/fio/fio.py 
135
136
137
def load_default_params(self) -> None:
    """Load default parameters."""
    self.parameters = DefaultParameters(name='sts-fio-default')

load_fs_params()

Load filesystem parameters.

Source code in sts_libs/src/sts/fio/fio.py 
139
140
141
def load_fs_params(self) -> None:
    """Load filesystem parameters."""
    self.parameters = FileSystemParameters(name='sts-fio-fs')

load_stress_params()

Load stress test parameters.

Source code in sts_libs/src/sts/fio/fio.py 
147
148
149
def load_stress_params(self) -> None:
    """Load stress test parameters."""
    self.parameters = StressParameters(name='sts-fio-stress')

run(*, timeout=_UNSET)

Run FIO test.

Parameters:

Name Type Description Default
timeout int | None

Subprocess timeout in seconds. When omitted the timeout is derived from the configured FIO runtime (+ 120 s buffer), falling back to 600 s when no runtime is set. Pass an explicit value to override.

 _UNSET

Raises:

Type Description
FIOExecutionError

If the fio package cannot be installed or the command cannot be built.
Source code in sts_libs/src/sts/fio/fio.py 
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
def run(self, *, timeout: int | None = _UNSET) -> CommandResult:
    """Run FIO test.

    Args:
        timeout: Subprocess timeout in seconds.  When omitted the timeout
            is derived from the configured FIO runtime (+ 120 s buffer),
            falling back to 600 s when no runtime is set.  Pass an explicit
            value to override.

    Raises:
        FIOExecutionError: If the fio package cannot be installed or the command cannot be built.
    """
    if not ensure_installed('fio'):
        raise FIOExecutionError('Failed to install fio package')

    try:
        argv = self._create_argv()
    except FIOConfigError as e:
        raise FIOExecutionError(f'Failed to create command: {e}') from e

    if timeout == self._UNSET:
        timeout = self.parameters.runtime + 120 if self.parameters and self.parameters.runtime else 600

    result = run_argv(argv, timeout=timeout)
    if result.failed:
        logger.error(f'FIO run failed:\n{result.stderr}')
    else:
        logger.debug('FIO executed successfully')

    return result

save_config_file(config_file)

Save parameters to FIO config file.

Raises:

Type Description
FIOConfigError

If parameters are not set or file cannot be written
Source code in sts_libs/src/sts/fio/fio.py 
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
def save_config_file(self, config_file: str | Path) -> None:
    """Save parameters to FIO config file.

    Raises:
        FIOConfigError: If parameters are not set or file cannot be written
    """
    if not self.parameters:
        raise FIOConfigError('No parameters to save')

    config = configparser.ConfigParser()
    config[self.parameters.name] = self.parameters.to_dict()

    try:
        with Path(config_file).open('w') as f:
            config.write(f)
    except OSError as e:
        raise FIOConfigError(f'Failed to write config file: {e}') from e

update_options(options)

Add options, deduplicating against existing ones.

Source code in sts_libs/src/sts/fio/fio.py 
131
132
133
def update_options(self, options: list[str]) -> None:
    """Add options, deduplicating against existing ones."""
    self.options = self._normalize_options(self.options + options)

update_parameters(params)

Update parameters.

Source code in sts_libs/src/sts/fio/fio.py 
124
125
126
127
128
129
def update_parameters(self, params: dict[str, Any]) -> None:
    """Update parameters."""
    if not self.parameters:
        self.parameters = DefaultParameters(name='sts-fio-default')
    for key, value in params.items():
        setattr(self.parameters, key, value)

sts.fio.parameters

FIO parameter configurations.

BlockDeviceParameters pydantic-model

Bases: FIOParameters

Parameters optimized for block device IOPS testing: random reads with high concurrency.

Show JSON schema: 
{
  "additionalProperties": false,
  "description": "Parameters optimized for block device IOPS testing: random reads with high concurrency.",
  "properties": {
    "name": {
      "default": "sts-fio",
      "title": "Name",
      "type": "string"
    },
    "ioengine": {
      "default": "libaio",
      "enum": [
        "libaio",
        "sync",
        "posixaio",
        "mmap",
        "splice"
      ],
      "title": "Ioengine",
      "type": "string"
    },
    "direct": {
      "default": true,
      "title": "Direct",
      "type": "boolean"
    },
    "rw": {
      "default": "randread",
      "enum": [
        "read",
        "write",
        "randread",
        "randwrite",
        "randrw",
        "trim"
      ],
      "title": "Rw",
      "type": "string"
    },
    "bs": {
      "default": "512",
      "title": "Bs",
      "type": "string"
    },
    "numjobs": {
      "default": 4,
      "title": "Numjobs",
      "type": "integer"
    },
    "group_reporting": {
      "default": true,
      "title": "Group Reporting",
      "type": "boolean"
    },
    "verify_fatal": {
      "default": false,
      "title": "Verify Fatal",
      "type": "boolean"
    },
    "end_fsync": {
      "default": false,
      "title": "End Fsync",
      "type": "boolean"
    },
    "time_based": {
      "default": false,
      "title": "Time Based",
      "type": "boolean"
    },
    "filename": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Filename"
    },
    "size": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Size"
    },
    "runtime": {
      "anyOf": [
        {
          "exclusiveMinimum": 0,
          "type": "integer"
        },
        {
          "type": "null"
        }
      ],
      "default": 1800,
      "title": "Runtime"
    },
    "iodepth": {
      "anyOf": [
        {
          "type": "integer"
        },
        {
          "type": "null"
        }
      ],
      "default": 32,
      "title": "Iodepth"
    },
    "iodepth_batch_submit": {
      "anyOf": [
        {
          "exclusiveMinimum": 0,
          "type": "integer"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Iodepth Batch Submit"
    },
    "iodepth_batch_complete_min": {
      "anyOf": [
        {
          "exclusiveMinimum": 0,
          "type": "integer"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Iodepth Batch Complete Min"
    },
    "verify": {
      "anyOf": [
        {
          "enum": [
            "crc32",
            "crc32c",
            "crc32c-intel",
            "md5",
            "sha1",
            "sha256",
            "sha512",
            "xxhash",
            "meta"
          ],
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Verify"
    },
    "verify_backlog": {
      "anyOf": [
        {
          "type": "integer"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Verify Backlog"
    }
  },
  "title": "BlockDeviceParameters",
  "type": "object"
}

Fields:

  • name (str)
  • ioengine (IOEngine)
  • direct (bool)
  • verify_fatal (bool)
  • end_fsync (bool)
  • time_based (bool)
  • filename (str | None)
  • size (str | None)
  • iodepth_batch_submit (int | None)
  • iodepth_batch_complete_min (int | None)
  • verify (VerifyType | None)
  • verify_backlog (int | None)
  • rw (RWType)
  • bs (str)
  • numjobs (int)
  • group_reporting (bool)
  • iodepth (int | None)
  • runtime (int | None)
Source code in sts_libs/src/sts/fio/parameters.py 
109
110
111
112
113
114
115
116
117
class BlockDeviceParameters(FIOParameters):
    """Parameters optimized for block device IOPS testing: random reads with high concurrency."""

    rw: RWType = 'randread'
    bs: str = '512'
    numjobs: int = 4
    group_reporting: bool = True
    iodepth: int | None = 32
    runtime: int | None = Field(default=1800, gt=0)

DefaultParameters pydantic-model

Bases: FIOParameters

Default parameters for general FIO testing: random read/write with verification.

Show JSON schema: 
{
  "additionalProperties": false,
  "description": "Default parameters for general FIO testing: random read/write with verification.",
  "properties": {
    "name": {
      "default": "sts-fio",
      "title": "Name",
      "type": "string"
    },
    "ioengine": {
      "default": "libaio",
      "enum": [
        "libaio",
        "sync",
        "posixaio",
        "mmap",
        "splice"
      ],
      "title": "Ioengine",
      "type": "string"
    },
    "direct": {
      "default": true,
      "title": "Direct",
      "type": "boolean"
    },
    "rw": {
      "default": "randrw",
      "enum": [
        "read",
        "write",
        "randread",
        "randwrite",
        "randrw",
        "trim"
      ],
      "title": "Rw",
      "type": "string"
    },
    "bs": {
      "default": "4k",
      "minLength": 1,
      "title": "Bs",
      "type": "string"
    },
    "numjobs": {
      "default": 1,
      "title": "Numjobs",
      "type": "integer"
    },
    "group_reporting": {
      "default": true,
      "title": "Group Reporting",
      "type": "boolean"
    },
    "verify_fatal": {
      "default": true,
      "title": "Verify Fatal",
      "type": "boolean"
    },
    "end_fsync": {
      "default": false,
      "title": "End Fsync",
      "type": "boolean"
    },
    "time_based": {
      "default": false,
      "title": "Time Based",
      "type": "boolean"
    },
    "filename": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Filename"
    },
    "size": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Size"
    },
    "runtime": {
      "anyOf": [
        {
          "exclusiveMinimum": 0,
          "type": "integer"
        },
        {
          "type": "null"
        }
      ],
      "default": 60,
      "title": "Runtime"
    },
    "iodepth": {
      "anyOf": [
        {
          "type": "integer"
        },
        {
          "type": "null"
        }
      ],
      "default": 32,
      "title": "Iodepth"
    },
    "iodepth_batch_submit": {
      "anyOf": [
        {
          "exclusiveMinimum": 0,
          "type": "integer"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Iodepth Batch Submit"
    },
    "iodepth_batch_complete_min": {
      "anyOf": [
        {
          "exclusiveMinimum": 0,
          "type": "integer"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Iodepth Batch Complete Min"
    },
    "verify": {
      "anyOf": [
        {
          "enum": [
            "crc32",
            "crc32c",
            "crc32c-intel",
            "md5",
            "sha1",
            "sha256",
            "sha512",
            "xxhash",
            "meta"
          ],
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": "crc32",
      "title": "Verify"
    },
    "verify_backlog": {
      "anyOf": [
        {
          "type": "integer"
        },
        {
          "type": "null"
        }
      ],
      "default": 1024,
      "title": "Verify Backlog"
    }
  },
  "title": "DefaultParameters",
  "type": "object"
}

Fields:

  • name (str)
  • ioengine (IOEngine)
  • direct (bool)
  • rw (RWType)
  • bs (str)
  • numjobs (int)
  • end_fsync (bool)
  • time_based (bool)
  • filename (str | None)
  • size (str | None)
  • iodepth_batch_submit (int | None)
  • iodepth_batch_complete_min (int | None)
  • group_reporting (bool)
  • verify_fatal (bool)
  • iodepth (int | None)
  • runtime (int | None)
  • verify (VerifyType | None)
  • verify_backlog (int | None)
Source code in sts_libs/src/sts/fio/parameters.py 
88
89
90
91
92
93
94
95
96
class DefaultParameters(FIOParameters):
    """Default parameters for general FIO testing: random read/write with verification."""

    group_reporting: bool = True
    verify_fatal: bool = True
    iodepth: int | None = 32
    runtime: int | None = Field(default=60, gt=0)
    verify: VerifyType | None = 'crc32'
    verify_backlog: int | None = 1024

FIOParameters pydantic-model

Bases: StsBaseModel

Base FIO parameters.

Show JSON schema: 
{
  "additionalProperties": false,
  "description": "Base FIO parameters.",
  "properties": {
    "name": {
      "default": "sts-fio",
      "title": "Name",
      "type": "string"
    },
    "ioengine": {
      "default": "libaio",
      "enum": [
        "libaio",
        "sync",
        "posixaio",
        "mmap",
        "splice"
      ],
      "title": "Ioengine",
      "type": "string"
    },
    "direct": {
      "default": true,
      "title": "Direct",
      "type": "boolean"
    },
    "rw": {
      "default": "randrw",
      "enum": [
        "read",
        "write",
        "randread",
        "randwrite",
        "randrw",
        "trim"
      ],
      "title": "Rw",
      "type": "string"
    },
    "bs": {
      "default": "4k",
      "minLength": 1,
      "title": "Bs",
      "type": "string"
    },
    "numjobs": {
      "default": 1,
      "title": "Numjobs",
      "type": "integer"
    },
    "group_reporting": {
      "default": false,
      "title": "Group Reporting",
      "type": "boolean"
    },
    "verify_fatal": {
      "default": false,
      "title": "Verify Fatal",
      "type": "boolean"
    },
    "end_fsync": {
      "default": false,
      "title": "End Fsync",
      "type": "boolean"
    },
    "time_based": {
      "default": false,
      "title": "Time Based",
      "type": "boolean"
    },
    "filename": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Filename"
    },
    "size": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Size"
    },
    "runtime": {
      "anyOf": [
        {
          "exclusiveMinimum": 0,
          "type": "integer"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Runtime"
    },
    "iodepth": {
      "anyOf": [
        {
          "type": "integer"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Iodepth"
    },
    "iodepth_batch_submit": {
      "anyOf": [
        {
          "exclusiveMinimum": 0,
          "type": "integer"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Iodepth Batch Submit"
    },
    "iodepth_batch_complete_min": {
      "anyOf": [
        {
          "exclusiveMinimum": 0,
          "type": "integer"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Iodepth Batch Complete Min"
    },
    "verify": {
      "anyOf": [
        {
          "enum": [
            "crc32",
            "crc32c",
            "crc32c-intel",
            "md5",
            "sha1",
            "sha256",
            "sha512",
            "xxhash",
            "meta"
          ],
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Verify"
    },
    "verify_backlog": {
      "anyOf": [
        {
          "type": "integer"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Verify Backlog"
    }
  },
  "title": "FIOParameters",
  "type": "object"
}

Config:

  • validate_assignment: True

Fields:

  • name (str)
  • ioengine (IOEngine)
  • direct (bool)
  • rw (RWType)
  • bs (str)
  • numjobs (int)
  • group_reporting (bool)
  • verify_fatal (bool)
  • end_fsync (bool)
  • time_based (bool)
  • filename (str | None)
  • size (str | None)
  • runtime (int | None)
  • iodepth (int | None)
  • iodepth_batch_submit (int | None)
  • iodepth_batch_complete_min (int | None)
  • verify (VerifyType | None)
  • verify_backlog (int | None)
Source code in sts_libs/src/sts/fio/parameters.py 
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
class FIOParameters(StsBaseModel):
    """Base FIO parameters."""

    model_config = ConfigDict(validate_assignment=True)

    # Optional parameters with defaults
    name: str = 'sts-fio'
    ioengine: IOEngine = 'libaio'
    direct: bool = True
    rw: RWType = 'randrw'
    bs: str = Field(default='4k', min_length=1)
    numjobs: int = 1
    group_reporting: bool = False
    verify_fatal: bool = False
    end_fsync: bool = False
    time_based: bool = False

    # Optional parameters without defaults
    filename: str | None = None
    size: str | None = None
    runtime: int | None = Field(default=None, gt=0)
    iodepth: int | None = None
    iodepth_batch_submit: int | None = Field(default=None, gt=0)
    iodepth_batch_complete_min: int | None = Field(default=None, gt=0)
    verify: VerifyType | None = None
    verify_backlog: int | None = None

    def to_dict(self) -> dict[str, str]:
        """Convert parameters to FIO command format (bools become '1', None fields omitted)."""
        result: dict[str, str] = {}
        for key, value in self.model_dump().items():
            if value is None:
                continue
            if isinstance(value, bool):
                if value:
                    result[key] = '1'
            else:
                result[key] = str(value)
        return result

    @classmethod
    def from_file(cls, path: str | Path) -> FIOParameters | None:
        """Create parameters from FIO config file, or None on parse failure."""
        try:
            config = configparser.ConfigParser()
            config.read(path)

            # Get the first job section
            job_section = next(s for s in config.sections() if s != 'global')
            params = dict(config[job_section])

            # Convert parameters
            return cls(
                name=job_section,
                **{k: v for k, v in params.items() if not k.startswith('_')},  # type: ignore[arg-type]
            )
        except (configparser.Error, StopIteration) as e:
            logger.warning(f'Invalid config file: {e}')
            return None

from_file(path) classmethod

Create parameters from FIO config file, or None on parse failure.

Source code in sts_libs/src/sts/fio/parameters.py 
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
@classmethod
def from_file(cls, path: str | Path) -> FIOParameters | None:
    """Create parameters from FIO config file, or None on parse failure."""
    try:
        config = configparser.ConfigParser()
        config.read(path)

        # Get the first job section
        job_section = next(s for s in config.sections() if s != 'global')
        params = dict(config[job_section])

        # Convert parameters
        return cls(
            name=job_section,
            **{k: v for k, v in params.items() if not k.startswith('_')},  # type: ignore[arg-type]
        )
    except (configparser.Error, StopIteration) as e:
        logger.warning(f'Invalid config file: {e}')
        return None

to_dict()

Convert parameters to FIO command format (bools become '1', None fields omitted).

Source code in sts_libs/src/sts/fio/parameters.py 
54
55
56
57
58
59
60
61
62
63
64
65
def to_dict(self) -> dict[str, str]:
    """Convert parameters to FIO command format (bools become '1', None fields omitted)."""
    result: dict[str, str] = {}
    for key, value in self.model_dump().items():
        if value is None:
            continue
        if isinstance(value, bool):
            if value:
                result[key] = '1'
        else:
            result[key] = str(value)
    return result

FileSystemParameters pydantic-model

Bases: FIOParameters

Parameters optimized for filesystem testing: sequential writes with large block size.

Show JSON schema: 
{
  "additionalProperties": false,
  "description": "Parameters optimized for filesystem testing: sequential writes with large block size.",
  "properties": {
    "name": {
      "default": "sts-fio",
      "title": "Name",
      "type": "string"
    },
    "ioengine": {
      "default": "sync",
      "enum": [
        "libaio",
        "sync",
        "posixaio",
        "mmap",
        "splice"
      ],
      "title": "Ioengine",
      "type": "string"
    },
    "direct": {
      "default": true,
      "title": "Direct",
      "type": "boolean"
    },
    "rw": {
      "default": "write",
      "enum": [
        "read",
        "write",
        "randread",
        "randwrite",
        "randrw",
        "trim"
      ],
      "title": "Rw",
      "type": "string"
    },
    "bs": {
      "default": "1M",
      "title": "Bs",
      "type": "string"
    },
    "numjobs": {
      "default": 1,
      "title": "Numjobs",
      "type": "integer"
    },
    "group_reporting": {
      "default": false,
      "title": "Group Reporting",
      "type": "boolean"
    },
    "verify_fatal": {
      "default": false,
      "title": "Verify Fatal",
      "type": "boolean"
    },
    "end_fsync": {
      "default": true,
      "title": "End Fsync",
      "type": "boolean"
    },
    "time_based": {
      "default": false,
      "title": "Time Based",
      "type": "boolean"
    },
    "filename": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Filename"
    },
    "size": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": "10G",
      "title": "Size"
    },
    "runtime": {
      "anyOf": [
        {
          "exclusiveMinimum": 0,
          "type": "integer"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Runtime"
    },
    "iodepth": {
      "anyOf": [
        {
          "type": "integer"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Iodepth"
    },
    "iodepth_batch_submit": {
      "anyOf": [
        {
          "exclusiveMinimum": 0,
          "type": "integer"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Iodepth Batch Submit"
    },
    "iodepth_batch_complete_min": {
      "anyOf": [
        {
          "exclusiveMinimum": 0,
          "type": "integer"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Iodepth Batch Complete Min"
    },
    "verify": {
      "anyOf": [
        {
          "enum": [
            "crc32",
            "crc32c",
            "crc32c-intel",
            "md5",
            "sha1",
            "sha256",
            "sha512",
            "xxhash",
            "meta"
          ],
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Verify"
    },
    "verify_backlog": {
      "anyOf": [
        {
          "type": "integer"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Verify Backlog"
    }
  },
  "title": "FileSystemParameters",
  "type": "object"
}

Fields:

  • name (str)
  • direct (bool)
  • numjobs (int)
  • group_reporting (bool)
  • verify_fatal (bool)
  • time_based (bool)
  • filename (str | None)
  • runtime (int | None)
  • iodepth (int | None)
  • iodepth_batch_submit (int | None)
  • iodepth_batch_complete_min (int | None)
  • verify (VerifyType | None)
  • verify_backlog (int | None)
  • ioengine (IOEngine)
  • rw (RWType)
  • bs (str)
  • end_fsync (bool)
  • size (str | None)
Source code in sts_libs/src/sts/fio/parameters.py 
 99
100
101
102
103
104
105
106
class FileSystemParameters(FIOParameters):
    """Parameters optimized for filesystem testing: sequential writes with large block size."""

    ioengine: IOEngine = 'sync'
    rw: RWType = 'write'
    bs: str = '1M'
    end_fsync: bool = True
    size: str | None = '10G'

StressParameters pydantic-model

Bases: FIOParameters

Parameters for stress/endurance testing: high concurrency random read/write.

Show JSON schema: 
{
  "additionalProperties": false,
  "description": "Parameters for stress/endurance testing: high concurrency random read/write.",
  "properties": {
    "name": {
      "default": "sts-fio",
      "title": "Name",
      "type": "string"
    },
    "ioengine": {
      "default": "libaio",
      "enum": [
        "libaio",
        "sync",
        "posixaio",
        "mmap",
        "splice"
      ],
      "title": "Ioengine",
      "type": "string"
    },
    "direct": {
      "default": true,
      "title": "Direct",
      "type": "boolean"
    },
    "rw": {
      "default": "randrw",
      "enum": [
        "read",
        "write",
        "randread",
        "randwrite",
        "randrw",
        "trim"
      ],
      "title": "Rw",
      "type": "string"
    },
    "bs": {
      "default": "4k",
      "minLength": 1,
      "title": "Bs",
      "type": "string"
    },
    "numjobs": {
      "default": 64,
      "title": "Numjobs",
      "type": "integer"
    },
    "group_reporting": {
      "default": true,
      "title": "Group Reporting",
      "type": "boolean"
    },
    "verify_fatal": {
      "default": false,
      "title": "Verify Fatal",
      "type": "boolean"
    },
    "end_fsync": {
      "default": false,
      "title": "End Fsync",
      "type": "boolean"
    },
    "time_based": {
      "default": false,
      "title": "Time Based",
      "type": "boolean"
    },
    "filename": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Filename"
    },
    "size": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Size"
    },
    "runtime": {
      "anyOf": [
        {
          "exclusiveMinimum": 0,
          "type": "integer"
        },
        {
          "type": "null"
        }
      ],
      "default": 3600,
      "title": "Runtime"
    },
    "iodepth": {
      "anyOf": [
        {
          "type": "integer"
        },
        {
          "type": "null"
        }
      ],
      "default": 64,
      "title": "Iodepth"
    },
    "iodepth_batch_submit": {
      "anyOf": [
        {
          "exclusiveMinimum": 0,
          "type": "integer"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Iodepth Batch Submit"
    },
    "iodepth_batch_complete_min": {
      "anyOf": [
        {
          "exclusiveMinimum": 0,
          "type": "integer"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Iodepth Batch Complete Min"
    },
    "verify": {
      "anyOf": [
        {
          "enum": [
            "crc32",
            "crc32c",
            "crc32c-intel",
            "md5",
            "sha1",
            "sha256",
            "sha512",
            "xxhash",
            "meta"
          ],
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Verify"
    },
    "verify_backlog": {
      "anyOf": [
        {
          "type": "integer"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Verify Backlog"
    }
  },
  "title": "StressParameters",
  "type": "object"
}

Fields:

  • name (str)
  • ioengine (IOEngine)
  • direct (bool)
  • rw (RWType)
  • bs (str)
  • verify_fatal (bool)
  • end_fsync (bool)
  • time_based (bool)
  • filename (str | None)
  • size (str | None)
  • iodepth_batch_submit (int | None)
  • iodepth_batch_complete_min (int | None)
  • verify (VerifyType | None)
  • verify_backlog (int | None)
  • numjobs (int)
  • group_reporting (bool)
  • iodepth (int | None)
  • runtime (int | None)
Source code in sts_libs/src/sts/fio/parameters.py 
120
121
122
123
124
125
126
class StressParameters(FIOParameters):
    """Parameters for stress/endurance testing: high concurrency random read/write."""

    numjobs: int = 64
    group_reporting: bool = True
    iodepth: int | None = 64
    runtime: int | None = Field(default=3600, gt=0)