初探pynvme - pynvme/pynvme GitHub Wiki
有病早治，有问题早解决，大家都好。 ——《深入浅出SSD》7.4 回归测试
SSD的市场越来越大，国内外进入SSD研发和生产领域的厂商也越来越多。在技术开发方面，大家八仙过海，各显神通。各自通过自己的技术特色，满足各自客户的差异化需求。但是在测试方面，并不是所有团队都有足够的资源和技术积累来做充分的测试，大部分的测试借用了现成的测试软件，由测试工程师手动执行。
这种测试方式会导致一些问题。首先，SSD客户对SSD的理解不断深入，对SSD的需求也在不断调整，导致SSD固件的迭代节奏非常快。每次更新都需要做完整的回归测试，消耗大量的人力、物力和时间资源，甚至有时需要固件开发人员来执行测试，影响了测试结果的客观性。其次，大部分的手动测试都不具备可重复性，更不用说由于测试人员操作习惯的不同而引入的差异了。在这样的情况下，测试不能帮助开发人员直接定位问题，也很难验证一个bugfix是不是真正解决了问题。
如上图所示，在一个项目开发过程中，大部分的bug会在设计和coding阶段引入。越早开始测试，就能越早发现bug，并且解决bug的成本也最低。具体到SSD固件开发的例子，可能在项目初期只能支持顺序读写，那就要开始针对这部分功能开展各种测试，而不是等待随机读写和其他功能的开发。而要实现这个目的，就不能依赖现成的测试软件，一定要开发自己的有针对性的测试程序或者脚本。反过来，当系统测试出现fail,我们尽量在功能测试中通过构造脚本来重现，既能帮助分析问题，也能方便日后回归测试。这样，我们就可以在功能测试阶段尽可能提升固件的质量，减少系统测试阶段的资源开销。Fail early! Fail fast! 从软件工程学的角度讲，我们要从瀑布模型走向敏捷开发。
反观互联网等行业的软件开发，已经接受持续集成和持续发布的实践，有敏捷开发、看板、测试驱动等各种成熟的方法，以及大量自动化的工具。这些都极大地提高软件的开发效率和节奏。当一个bug被引入后，可能几分钟或者几个小时就会被发现，这个时候开发者趁热打铁，解决问题的成本是最低的，同时能阻止bug向主干和其他分支扩散。SSD，特别是SSD的固件，为什么不能像软件那样测试呢？
SSD到测试脚本（或者测试软件）之间，隔着一个驱动程序。所以，一个专门为测试定制的驱动程序可以把固件的测试转化为软件的测试，从而把软件开发的工具和方法引入固件开发的实践中。这个测试驱动需要：
- 尽可能轻量级，能够把SSD的功能、特性和缺陷完全暴露给上层测试软件。不要引入文件系统、cache等复杂逻辑。这样有助于控制测试驱动的复杂度，提高测试驱动的可靠性，以及测试的全面性。
- 驱动的接口要简洁，容易操作，降低测试人员的开发成本。最好能用Python等脚本语言来开发脚本。另外，应用层实现的驱动更容易让固件开发和测试工程师上手，降低测试环境部署的难度。
- IO性能要高，否则无法压出极限使用条件下可能存在的bug。
- 要能和现代软件开发测试的工具集成，将软件的测试方法和流程引入固件开发，提高测试开发的效率。在最大程度上，用脚本和工具来替代测试的人工执行。
基于这些原因，很多厂家会选择ULink的DriveMaster。但其专有的脚本语法和封闭的工具，以及加密的测试脚本，使得其采购成本、开发成本和执行成本都非常高，甚至是痛苦的。IoL所使用的内核态的dNVMe驱动在性能方面也无法满足要求。
2015年Intel开源了SPDK，为数据中心存储提供一系列软件支持，其中包括了NVMe驱动。SPDK的整体架构非常清晰，NVMe驱动模块的接口完整，实现了大部分NVMe设备的功能。SPDK也经过数据中心的实践检验，有稳定性的保障。但另一方面，SPDK毕竟是以数据中心存储系统为设计目标，会缺少一些SSD测试所需要的功能，譬如：
- SPDK采用轮询，不支持中断；
- SPDK只提供C语言的接口，直接在SPDK上面开发测试程序的成本高，并且容易引入bug；
- 没有提供cmdlog等专门用于测试和调试的功能，测试过程中遇到问题不容易分析；
- NVMe的SQ和CQ被绑定为1:1的Qpair；
- 只支持NVMe协议。
但是作为一个Intel的开源项目，SPDK的文档和代码非常规范，模块之间高度解耦。我们以SPDK的NVMe驱动为起点，开发了NVMe设备的测试驱动。这就是pynvme的缘起。
在SPDK的基础上，我们在pynvme中实现了这些功能：
- 支持MSIx中断：通过软件实现的中断控制器，提供中断相关的测试接口；
- 通过Cython的封装，为SPDK的NVMe驱动提供了Python接口，抽象出controller, namespace, qpair, buffer, ioworker等对象和方法；
- 实现了cmdlog：将NVMe测试驱动最近发送的若干条NVMe command及其completion结构记录到cmdlog中；
- 提供通用化的命令接口，可以发送任何NVMe command，包括VS command，甚至是不符合NVMe协议要求的命令；
- 结合pytest及其fixture，进一步简化测试脚本的开发和执行；
- 通过vscode及其插件，在IDE中开发和调试测试脚本，并能查看测试过程中寄存器、Qpair以及cmdlog的数据；
- 通过S3/standby模式实现SSD设备的电源控制，无需任何额外设备就可以在普通PC上实现部分上下电测试；
- 实现ioworker，以多进程的方式对同一个SSD设备进行并发读写操作，加大测试压力；
- 自动计算、记录、校验每个LBA的CRC，实现数据完整性的测试；
- 自动在每个LBA数据中插入LBA地址等信息，以发现其他数据一致性问题；
- 可以读写PCI配置空间，以及BAR空间。
pynvme实现了众多NVMe SSD测试所需的功能，并通过Python提供一致的脚本接口。就像美味的煎饼果子，用煎饼包裹各种不同的食材，呈现统一而诱人的风味。pynvme就是专为NVMe SSD测试而生的煎饼大侠！:)
在pynvme中，我们保留了SQ/CQ的Qpair模型，因为这符合实际的使用方式。另外，我们也没有考虑对SATA协议的支持。在整个pynvme开发过程中，我们同步实现对NVMe测试驱动的测试，并在GitLab的CI环境中做到持续测试，以保证pynvme测试驱动的质量。pynvme是一个独立的开源项目，不同的团队可以使用这个测试驱动和框架，甚至共享一部分测试脚本，以提高测试的自动化程度，加快开发的迭代节奏。
如上图所示，pynvme在SPDK NVMe驱动的基础上，扩充了一些测试相关的功能，并对外提供Python API接口。
pynvme的安装可以参考文档：https://pynvme.readthedocs.io/install.html 下面通过几个例子来展示如何基于pynvme开发NVMe设备的测试脚本。
import pytest
import nvme as d
def test_hello_world(nvme0, nvme0n1:d.Namespace):
read_buf = d.Buffer(512)
data_buf = d.Buffer(512)
data_buf[10:21] = b'hello world'
qpair = d.Qpair(nvme0, 16)
def write_cb(cdw0, status1):
nvme0n1.read(qpair, read_buf, 0, 1)
nvme0n1.write(qpair, data_buf, 0, 1, cb=write_cb)
qpair.waitdone(2)
assert read_buf[10:21] == b'hello world'SPDK自带了一个hello world示例程序，用了370行C代码。通过pynvme，只需要14行Python代码。下面是具体解释。
import pytest
import nvme as dpynvme推荐在pytest框架下编写测试脚本，所以我们先引入pytest库。这里也引入pynvme的nvme驱动库，我们习惯使用：import nvme as d。
def test_hello_world(nvme0, nvme0n1:d.Namespace):这里定义一个测试函数，pytest定义以test开头的函数为测试函数。这个函数有两个参数，nvme0和nvme0n1，其实是pytest的fixture。当一个测试函数被执行时，pytest首先调用参数列表中定义的fixture，以完成测试的准备工作。pynvme的常用fixture定义在conftest.py中。这里的两个参数会返回待测NVMe设备的controller对象和namespace对象。可以给nvme0n1指定类型d.Namespace，某些编辑器，譬如VSCode，会给我们更多coding辅助。
read_buf = d.Buffer(512)
data_buf = d.Buffer(512)
data_buf[10:21] = b'hello world'这里申请两个buffer，每个buffer有512字节，并且给其中一个buffer填入一些测试字符。
qpair = d.Qpair(nvme0, 16)为后续读写命令建立一个新的IO Qpair，深度为16。
def write_cb(cdw0, status1):
nvme0n1.read(qpair, read_buf, 0, 1)
nvme0n1.write(qpair, data_buf, 0, 1, cb=write_cb)
qpair.waitdone(2)通过nvme0n1.write发出一条写命令，将一个buffer的数据写入SSD盘的LBA0。这个命令带有一个回调函数，当命令结束时，会调用这个回调函数。这里的回调函数write_cb里面会继续发出一条读命令，将LBA0的数据读入另一个buffer。这个例子我们会发出2条command。NVMe的命令都是异步发送，所以我们要先等这2条命令结束再做后续处理。
assert read_buf[10:21] == b'hello world'命令结束后，检查读数据的buffer内容是不是如预期。
我们可以在VSCode中编辑、调试这段脚本。关于VSCode的配置可以参考文档：https://pynvme.readthedocs.io/vscode.html
或者在命令行环境中用如下命令执行这个测试函数：
make test TESTS=scripts/demo_test.py::test_hello_world测试的log保存在test.log中。
def test_sanitize(nvme0, nvme0n1, buf):
if nvme0.id_data(331, 328) == 0:
warnings.warn("sanitize operation is not supported")
return
logging.info("supported sanitize operation: %d" % nvme0.id_data(331, 328))
nvme0.sanitize().waitdone()
# sanitize status log page
nvme0.getlogpage(0x81, buf, 20).waitdone()
while buf.data(3, 2) & 0x7 != 1: # sanitize is not completed
progress = buf.data(1, 0)*100//0xffff
sg.OneLineProgressMeter('sanitize progress', progress, 100,
'progress', orientation='h')
nvme0.getlogpage(0x81, buf, 20).waitdone()
time.sleep(1)这个函数启动了一次sanitize，并在GUI中显示sanitize的进度。
def test_sanitize(nvme0, nvme0n1, buf):
if nvme0.id_data(331, 328) == 0:
warnings.warn("sanitize operation is not supported")
return检查待测设备是否支持sanitize。
logging.info("supported sanitize operation: %d" % nvme0.id_data(331, 328))
nvme0.sanitize().waitdone()启动sanitize，该命令会立刻结束。
# sanitize status log page
nvme0.getlogpage(0x81, buf, 20).waitdone()
while buf.data(3, 2) & 0x7 != 1: # sanitize is not completed
progress = buf.data(1, 0)*100//0xffff
sg.OneLineProgressMeter('sanitize progress', progress, 100,
'progress', orientation='h')
nvme0.getlogpage(0x81, buf, 20).waitdone()
time.sleep(1)每秒钟检查一次sanitize的进度，并刷新GUI。运行效果如下图。你可以在scripts/utility_test.py里面找个这个工具以及其他一些SSD日常开发中会用到的工具。
def test_trim_basic(nvme0: d.Controller, nvme0n1: d.Namespace, verify):
GB = 1024*1024*1024
all_zero_databuf = d.Buffer(512)
trimbuf = d.Buffer(4096)
q = d.Qpair(nvme0, 32)
# DUT info
logging.info("model number: %s" % nvme0.id_data(63, 24, str))
logging.info("firmware revision: %s" % nvme0.id_data(71, 64, str))
# write
logging.info("write data in 10G ~ 20G")
io_size = 128*1024//512
start_lba = 10*GB//512
lba_count = 10*GB//512
nvme0n1.ioworker(io_size = io_size,
lba_align = io_size,
lba_random = False,
read_percentage = 0,
lba_start = start_lba,
io_count = lba_count//io_size,
qdepth = 128).start().close()
# verify data after write, data should be modified
with pytest.warns(UserWarning, match="ERROR status: 02/85"):
nvme0n1.compare(q, all_zero_databuf, start_lba).waitdone()
# get the empty trim time
trimbuf.set_dsm_range(0, 0, 0)
trim_cmd = nvme0n1.dsm(q, trimbuf, 1).waitdone() # first call is longer, due to cache?
start_time = time.time()
trim_cmd = nvme0n1.dsm(q, trimbuf, 1).waitdone()
empty_trim_time = time.time()-start_time
# the trim time of 10G data
logging.info("trim the 10G data from LBA 0x%lx" % start_lba)
trimbuf.set_dsm_range(0, start_lba, lba_count)
start_time = time.time()
trim_cmd = nvme0n1.dsm(q, trimbuf, 1).waitdone()
trim_time = time.time()-start_time-empty_trim_time
logging.info("trim bandwidth: %0.2fGB/s" % (10/trim_time))
# verify after trim
nvme0n1.compare(q, all_zero_databuf, start_lba).waitdone()@pytest.mark.parametrize("qcount", [1, 2, 4, 8, 16])
def test_ioworker_iops_multiple_queue(nvme0n1, qcount):
l = []
io_total = 0
for i in range(qcount):
a = nvme0n1.ioworker(io_size=8, lba_align=8,
region_start=0, region_end=256*1024*8, # 1GB space
lba_random=False, qdepth=16,
read_percentage=100, time=10).start()
l.append(a)
for a in l:
r = a.close()
io_total += (r.io_count_read+r.io_count_write)
logging.info("Q %d IOPS: %dK" % (qcount, io_total/10000))pytest可以参数化测试函数。这个例子中，我们开启不同数目的IOWorker，在不同的进程中对NVMe设备进行读写。
更多pytest测试脚本编写的指导，请参考pytest文档：https://docs.pytest.org/en/latest/
更多pynvme的脚本示例，可以参考driver_test.py，以及scripts目录下面的脚本文件。也欢迎您将您的脚本贡献到scripts目录下。
pynvme整合了Python、SPDK、VSCode等开源工具，提供一套高效的软件定义的NVMe设备测试方案。
下面我们看一下pynvme的实际应用。