Skip to content

Add UNMAP command #148

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
rosjat merged 3 commits into from
May 6, 2026
Merged

Add UNMAP command #148

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
b845e01
Add UNMAP command
bmeagherix Apr 24, 2026
fef21e6
Bump black and reuse pre-commit hooks for Python 3.12/3.14 compatibility
bmeagherix Apr 29, 2026
ba84ec2
Reformat all files with black 24.4.2
bmeagherix Apr 29, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions .pre-commit-config.yaml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,11 +16,11 @@ repos:
additional_dependencies:
- toml
- repo: https://github.com/fsfe/reuse-tool
rev: v1.1.2
rev: v4.0.3
hooks:
- id: reuse
- repo: https://github.com/psf/black
rev: 23.3.0
rev: 24.4.2
hooks:
- id: black
language_version: python3
17 changes: 17 additions & 0 deletions pyscsi/pyscsi/scsi.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -41,6 +41,7 @@
from pyscsi.pyscsi.scsi_cdb_synchronize_cache10 import SynchronizeCache10
from pyscsi.pyscsi.scsi_cdb_synchronize_cache16 import SynchronizeCache16
from pyscsi.pyscsi.scsi_cdb_testunitready import TestUnitReady
from pyscsi.pyscsi.scsi_cdb_unmap import Unmap
from pyscsi.pyscsi.scsi_cdb_write10 import Write10
from pyscsi.pyscsi.scsi_cdb_write12 import Write12
from pyscsi.pyscsi.scsi_cdb_write16 import Write16
Expand Down Expand Up @@ -526,6 +527,22 @@ def testunitready(self):
self.execute(cmd)
return cmd

def unmap(self, lbas, **kwargs):
"""
Returns an Unmap Instance

:param lbas: a list of dicts, each with 'lba' and 'num_blocks' keys,
specifying the LBA ranges to unmap
:param kwargs: a dict with key/value pairs
anchor = 0, Anchor flag
group = 0, Group Number
:return: an Unmap instance
"""
opcode = self.device.opcodes.UNMAP
cmd = Unmap(opcode, lbas, **kwargs)
self.execute(cmd)
return cmd

def write10(self, lba, tl, data, **kwargs):
"""
Returns a Write10 Instance
Expand Down
6 changes: 3 additions & 3 deletions pyscsi/pyscsi/scsi_cdb_report_target_port_groups.py
View file
Open in desktop
Comment thread
bmeagherix marked this conversation as resolved.
Original file line number Diff line number Diff line change
Expand Up @@ -97,9 +97,9 @@ def unmarshall_datain(cls, data):
result["implicit_transition_time"] = _r["implicit_transition_time"]
_data = _data[4:]
else:
result[
"format_type"
] = DATA_FORMAT_TYPE.LENGTH_ONLY_HEADER_PARAMETER_DATA_FORMAT
result["format_type"] = (
DATA_FORMAT_TYPE.LENGTH_ONLY_HEADER_PARAMETER_DATA_FORMAT
)

_tpg_descriptors = [] # Target Port Group Descriptors
while len(_data):
Expand Down
74 changes: 74 additions & 0 deletions pyscsi/pyscsi/scsi_cdb_unmap.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,74 @@
# coding: utf-8

# Copyright (C) 2026 by Brian Meagher<[email protected]>
# SPDX-FileCopyrightText: 2014 The python-scsi Authors
#
# SPDX-License-Identifier: LGPL-2.1-or-later

from pyscsi.pyscsi.scsi_command import SCSICommand
from pyscsi.utils.converter import scsi_int_to_ba

#
# SCSI UNMAP command and definitions
#
# See SBC-4 5.35 UNMAP command
#


class Unmap(SCSICommand):
"""
A class to send an UNMAP command to a scsi device
"""

_cdb_bits = {
"opcode": [0xFF, 0],
"anchor": [0x01, 1],
"group": [0x3F, 6],
"parameter_list_length": [0xFFFF, 7],
}

@classmethod
def marshall_dataout(cls, lbas):
"""
Build the UNMAP parameter list (SBC-4 5.35.1).

:param lbas: a list of dicts, each with 'lba' and 'num_blocks' keys
:return: a bytearray
"""
descriptors = bytearray()
for entry in lbas:
d = bytearray(16)
d[0:8] = scsi_int_to_ba(entry["lba"], 8)
d[8:12] = scsi_int_to_ba(entry["num_blocks"], 4)
# bytes 12-15: reserved
descriptors += d

desc_len = len(descriptors)
header = bytearray(8)
# UNMAP DATA LENGTH: length of remaining bytes (total - 2)
header[0:2] = scsi_int_to_ba(6 + desc_len, 2)
# UNMAP BLOCK DESCRIPTOR DATA LENGTH
header[2:4] = scsi_int_to_ba(desc_len, 2)
# bytes 4-7: reserved

return header + descriptors

def __init__(self, opcode, lbas, anchor=0, group=0):
"""
initialize a new instance

:param opcode: an OpCode instance
:param lbas: a list of dicts, each with 'lba' and 'num_blocks' keys,
specifying the LBA ranges to unmap
:param anchor: Anchor flag, 0 or 1
:param group: Group Number
"""
_data = Unmap.marshall_dataout(lbas)
SCSICommand.__init__(self, opcode, 0, 0)
self.dataout = _data
self.cdb = self.build_cdb(
opcode=self.opcode.value,
anchor=anchor,
parameter_list_length=len(_data),
group=group,
)
103 changes: 103 additions & 0 deletions tests/test_cdb_unmap.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,103 @@
# coding: utf-8

# Copyright (C) 2026 by Brian Meagher<[email protected]>
# SPDX-FileCopyrightText: 2014 The python-scsi Authors
#
# SPDX-License-Identifier: LGPL-2.1-or-later

import unittest

from pyscsi.pyscsi.scsi_cdb_unmap import Unmap
from pyscsi.pyscsi.scsi_enum_command import sbc
from pyscsi.utils.converter import scsi_ba_to_int
from tests.mock_device import MockDevice, MockSCSI


class CdbUnmapTest(unittest.TestCase):
def test_main(self):
with MockSCSI(MockDevice(sbc)) as s:

# Single descriptor, default flags
u = s.unmap([{"lba": 0, "num_blocks": 0}])
cdb = u.cdb
self.assertEqual(cdb[0], s.device.opcodes.UNMAP.value)
self.assertEqual(cdb[1], 0) # anchor=0
self.assertEqual(scsi_ba_to_int(cdb[2:6]), 0) # reserved
self.assertEqual(cdb[6], 0) # group=0
self.assertEqual(scsi_ba_to_int(cdb[7:9]), 24) # 8 header + 16 descriptor
self.assertEqual(cdb[9], 0)
cdb = u.unmarshall_cdb(cdb)
self.assertEqual(cdb["opcode"], s.device.opcodes.UNMAP.value)
self.assertEqual(cdb["anchor"], 0)
self.assertEqual(cdb["group"], 0)
self.assertEqual(cdb["parameter_list_length"], 24)

d = Unmap.unmarshall_cdb(Unmap.marshall_cdb(cdb))
self.assertEqual(d, cdb)

# Single descriptor, anchor=1, group=0x3F, non-zero LBA and num_blocks
u = s.unmap(
[{"lba": 0x0102030405060708, "num_blocks": 0x090A0B0C}],
anchor=1,
group=0x3F,
)
cdb = u.cdb
self.assertEqual(cdb[0], s.device.opcodes.UNMAP.value)
self.assertEqual(cdb[1], 0x01) # anchor=1
self.assertEqual(scsi_ba_to_int(cdb[2:6]), 0) # reserved
self.assertEqual(cdb[6], 0x3F) # group=63
self.assertEqual(scsi_ba_to_int(cdb[7:9]), 24)
self.assertEqual(cdb[9], 0)
cdb = u.unmarshall_cdb(cdb)
self.assertEqual(cdb["anchor"], 1)
self.assertEqual(cdb["group"], 0x3F)
self.assertEqual(cdb["parameter_list_length"], 24)

d = Unmap.unmarshall_cdb(Unmap.marshall_cdb(cdb))
self.assertEqual(d, cdb)

# Two descriptors — parameter_list_length = 8 + 2*16 = 40
u = s.unmap(
[{"lba": 0x100, "num_blocks": 0x10}, {"lba": 0x200, "num_blocks": 0x20}]
)
cdb = u.cdb
self.assertEqual(scsi_ba_to_int(cdb[7:9]), 40)
cdb = u.unmarshall_cdb(cdb)
self.assertEqual(cdb["parameter_list_length"], 40)

def test_dataout(self):
with MockSCSI(MockDevice(sbc)) as s:

# Single descriptor: verify parameter list bytes
u = s.unmap([{"lba": 0x0102030405060708, "num_blocks": 0x090A0B0C}])
data = u.dataout
self.assertEqual(len(data), 24)
# UNMAP DATA LENGTH = 22
self.assertEqual(scsi_ba_to_int(data[0:2]), 22)
# UNMAP BLOCK DESCRIPTOR DATA LENGTH = 16
self.assertEqual(scsi_ba_to_int(data[2:4]), 16)
# reserved bytes 4-7
self.assertEqual(scsi_ba_to_int(data[4:8]), 0)
# descriptor: LBA
self.assertEqual(scsi_ba_to_int(data[8:16]), 0x0102030405060708)
# descriptor: NUMBER OF LOGICAL BLOCKS
self.assertEqual(scsi_ba_to_int(data[16:20]), 0x090A0B0C)
# descriptor: reserved
self.assertEqual(scsi_ba_to_int(data[20:24]), 0)

# Two descriptors: verify lengths and both descriptors
u = s.unmap(
[{"lba": 0x100, "num_blocks": 0x10}, {"lba": 0x200, "num_blocks": 0x20}]
)
data = u.dataout
self.assertEqual(len(data), 40)
# UNMAP DATA LENGTH = 38
self.assertEqual(scsi_ba_to_int(data[0:2]), 38)
# UNMAP BLOCK DESCRIPTOR DATA LENGTH = 32
self.assertEqual(scsi_ba_to_int(data[2:4]), 32)
# descriptor 0
self.assertEqual(scsi_ba_to_int(data[8:16]), 0x100)
self.assertEqual(scsi_ba_to_int(data[16:20]), 0x10)
# descriptor 1
self.assertEqual(scsi_ba_to_int(data[24:32]), 0x200)
self.assertEqual(scsi_ba_to_int(data[32:36]), 0x20)
8 changes: 5 additions & 3 deletions tools/inquiry.py
View file
Open in desktop
Comment thread
bmeagherix marked this conversation as resolved.
Original file line number Diff line number Diff line change
Expand Up @@ -118,9 +118,11 @@ def inquiry_logical_block_prov(s):
" Threshold=%d blocks [%s]"
% (
1 << i["threshold_exponent"],
"NO LOGICAL BLOCK PROVISIONING SUPPORT"
if not i["threshold_exponent"]
else "exponent=%d" % (i["threshold_exponent"]),
(
"NO LOGICAL BLOCK PROVISIONING SUPPORT"
if not i["threshold_exponent"]
else "exponent=%d" % (i["threshold_exponent"])
),
)
)
print(
Expand Down
Loading