#!/usr/bin/env python # # Tests for image streaming. # # Copyright (C) 2012 IBM Corp. # # This program is free software; you can redistribute it and/or modify # it under the terms of the GNU General Public License as published by # the Free Software Foundation; either version 2 of the License, or # (at your option) any later version. # # This program is distributed in the hope that it will be useful, # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU General Public License for more details. # # You should have received a copy of the GNU General Public License # along with this program. If not, see <http://www.gnu.org/licenses/>. # import time import os import iotests from iotests import qemu_img, qemu_io import struct backing_img = os.path.join(iotests.test_dir, 'backing.img') mid_img = os.path.join(iotests.test_dir, 'mid.img') test_img = os.path.join(iotests.test_dir, 'test.img') class ImageStreamingTestCase(iotests.QMPTestCase): '''Abstract base class for image streaming test cases''' def assert_no_active_streams(self): result = self.vm.qmp('query-block-jobs') self.assert_qmp(result, 'return', []) def cancel_and_wait(self, drive='drive0'): '''Cancel a block job and wait for it to finish''' result = self.vm.qmp('block-job-cancel', device=drive) self.assert_qmp(result, 'return', {}) cancelled = False while not cancelled: for event in self.vm.get_qmp_events(wait=True): if event['event'] == 'BLOCK_JOB_CANCELLED': self.assert_qmp(event, 'data/type', 'stream') self.assert_qmp(event, 'data/device', drive) cancelled = True self.assert_no_active_streams() def create_image(self, name, size): file = open(name, 'w') i = 0 while i < size: sector = struct.pack('>l504xl', i / 512, i / 512) file.write(sector) i = i + 512 file.close() class TestSingleDrive(ImageStreamingTestCase): image_len = 1 * 1024 * 1024 # MB def setUp(self): self.create_image(backing_img, TestSingleDrive.image_len) qemu_img('create', '-f', iotests.imgfmt, '-o', 'backing_file=%s' % backing_img, mid_img) qemu_img('create', '-f', iotests.imgfmt, '-o', 'backing_file=%s' % mid_img, test_img) self.vm = iotests.VM().add_drive(test_img) self.vm.launch() def tearDown(self): self.vm.shutdown() os.remove(test_img) os.remove(mid_img) os.remove(backing_img) def test_stream(self): self.assert_no_active_streams() result = self.vm.qmp('block-stream', device='drive0') self.assert_qmp(result, 'return', {}) completed = False while not completed: for event in self.vm.get_qmp_events(wait=True): if event['event'] == 'BLOCK_JOB_COMPLETED': self.assert_qmp(event, 'data/type', 'stream') self.assert_qmp(event, 'data/device', 'drive0') self.assert_qmp(event, 'data/offset', self.image_len) self.assert_qmp(event, 'data/len', self.image_len) completed = True self.assert_no_active_streams() self.vm.shutdown() self.assertEqual(qemu_io('-c', 'map', backing_img), qemu_io('-c', 'map', test_img), 'image file map does not match backing file after streaming') def test_stream_pause(self): self.assert_no_active_streams() result = self.vm.qmp('block-stream', device='drive0') self.assert_qmp(result, 'return', {}) result = self.vm.qmp('block-job-pause', device='drive0') self.assert_qmp(result, 'return', {}) time.sleep(1) result = self.vm.qmp('query-block-jobs') offset = self.dictpath(result, 'return[0]/offset') time.sleep(1) result = self.vm.qmp('query-block-jobs') self.assert_qmp(result, 'return[0]/offset', offset) result = self.vm.qmp('block-job-resume', device='drive0') self.assert_qmp(result, 'return', {}) completed = False while not completed: for event in self.vm.get_qmp_events(wait=True): if event['event'] == 'BLOCK_JOB_COMPLETED': self.assert_qmp(event, 'data/type', 'stream') self.assert_qmp(event, 'data/device', 'drive0') self.assert_qmp(event, 'data/offset', self.image_len) self.assert_qmp(event, 'data/len', self.image_len) completed = True self.assert_no_active_streams() self.vm.shutdown() self.assertEqual(qemu_io('-c', 'map', backing_img), qemu_io('-c', 'map', test_img), 'image file map does not match backing file after streaming') def test_stream_partial(self): self.assert_no_active_streams() result = self.vm.qmp('block-stream', device='drive0', base=mid_img) self.assert_qmp(result, 'return', {}) completed = False while not completed: for event in self.vm.get_qmp_events(wait=True): if event['event'] == 'BLOCK_JOB_COMPLETED': self.assert_qmp(event, 'data/type', 'stream') self.assert_qmp(event, 'data/device', 'drive0') self.assert_qmp(event, 'data/offset', self.image_len) self.assert_qmp(event, 'data/len', self.image_len) completed = True self.assert_no_active_streams() self.vm.shutdown() self.assertEqual(qemu_io('-c', 'map', mid_img), qemu_io('-c', 'map', test_img), 'image file map does not match backing file after streaming') def test_device_not_found(self): result = self.vm.qmp('block-stream', device='nonexistent') self.assert_qmp(result, 'error/class', 'DeviceNotFound') class TestSmallerBackingFile(ImageStreamingTestCase): backing_len = 1 * 1024 * 1024 # MB image_len = 2 * backing_len def setUp(self): self.create_image(backing_img, self.backing_len) qemu_img('create', '-f', iotests.imgfmt, '-o', 'backing_file=%s' % backing_img, test_img, str(self.image_len)) self.vm = iotests.VM().add_drive(test_img) self.vm.launch() # If this hangs, then you are missing a fix to complete streaming when the # end of the backing file is reached. def test_stream(self): self.assert_no_active_streams() result = self.vm.qmp('block-stream', device='drive0') self.assert_qmp(result, 'return', {}) completed = False while not completed: for event in self.vm.get_qmp_events(wait=True): if event['event'] == 'BLOCK_JOB_COMPLETED': self.assert_qmp(event, 'data/type', 'stream') self.assert_qmp(event, 'data/device', 'drive0') self.assert_qmp(event, 'data/offset', self.image_len) self.assert_qmp(event, 'data/len', self.image_len) completed = True self.assert_no_active_streams() self.vm.shutdown() class TestErrors(ImageStreamingTestCase): image_len = 2 * 1024 * 1024 # MB # this should match STREAM_BUFFER_SIZE/512 in block/stream.c STREAM_BUFFER_SIZE = 512 * 1024 def create_blkdebug_file(self, name, event, errno): file = open(name, 'w') file.write(''' [inject-error] state = "1" event = "%s" errno = "%d" immediately = "off" once = "on" sector = "%d" [set-state] state = "1" event = "%s" new_state = "2" [set-state] state = "2" event = "%s" new_state = "1" ''' % (event, errno, self.STREAM_BUFFER_SIZE / 512, event, event)) file.close() class TestEIO(TestErrors): def setUp(self): self.blkdebug_file = backing_img + ".blkdebug" self.create_image(backing_img, TestErrors.image_len) self.create_blkdebug_file(self.blkdebug_file, "read_aio", 5) qemu_img('create', '-f', iotests.imgfmt, '-o', 'backing_file=blkdebug:%s:%s,backing_fmt=raw' % (self.blkdebug_file, backing_img), test_img) self.vm = iotests.VM().add_drive(test_img) self.vm.launch() def tearDown(self): self.vm.shutdown() os.remove(test_img) os.remove(backing_img) os.remove(self.blkdebug_file) def test_report(self): self.assert_no_active_streams() result = self.vm.qmp('block-stream', device='drive0') self.assert_qmp(result, 'return', {}) completed = False error = False while not completed: for event in self.vm.get_qmp_events(wait=True): if event['event'] == 'BLOCK_JOB_ERROR': self.assert_qmp(event, 'data/device', 'drive0') self.assert_qmp(event, 'data/operation', 'read') error = True elif event['event'] == 'BLOCK_JOB_COMPLETED': self.assertTrue(error, 'job completed unexpectedly') self.assert_qmp(event, 'data/type', 'stream') self.assert_qmp(event, 'data/device', 'drive0') self.assert_qmp(event, 'data/error', 'Input/output error') self.assert_qmp(event, 'data/offset', self.STREAM_BUFFER_SIZE) self.assert_qmp(event, 'data/len', self.image_len) completed = True self.assert_no_active_streams() self.vm.shutdown() def test_ignore(self): self.assert_no_active_streams() result = self.vm.qmp('block-stream', device='drive0', on_error='ignore') self.assert_qmp(result, 'return', {}) error = False completed = False while not completed: for event in self.vm.get_qmp_events(wait=True): if event['event'] == 'BLOCK_JOB_ERROR': self.assert_qmp(event, 'data/device', 'drive0') self.assert_qmp(event, 'data/operation', 'read') result = self.vm.qmp('query-block-jobs') self.assert_qmp(result, 'return[0]/paused', False) error = True elif event['event'] == 'BLOCK_JOB_COMPLETED': self.assertTrue(error, 'job completed unexpectedly') self.assert_qmp(event, 'data/type', 'stream') self.assert_qmp(event, 'data/device', 'drive0') self.assert_qmp(event, 'data/error', 'Input/output error') self.assert_qmp(event, 'data/offset', self.image_len) self.assert_qmp(event, 'data/len', self.image_len) completed = True self.assert_no_active_streams() self.vm.shutdown() def test_stop(self): self.assert_no_active_streams() result = self.vm.qmp('block-stream', device='drive0', on_error='stop') self.assert_qmp(result, 'return', {}) error = False completed = False while not completed: for event in self.vm.get_qmp_events(wait=True): if event['event'] == 'BLOCK_JOB_ERROR': self.assert_qmp(event, 'data/device', 'drive0') self.assert_qmp(event, 'data/operation', 'read') result = self.vm.qmp('query-block-jobs') self.assert_qmp(result, 'return[0]/paused', True) self.assert_qmp(result, 'return[0]/offset', self.STREAM_BUFFER_SIZE) self.assert_qmp(result, 'return[0]/io-status', 'failed') result = self.vm.qmp('block-job-resume', device='drive0') self.assert_qmp(result, 'return', {}) result = self.vm.qmp('query-block-jobs') self.assert_qmp(result, 'return[0]/paused', False) self.assert_qmp(result, 'return[0]/io-status', 'ok') error = True elif event['event'] == 'BLOCK_JOB_COMPLETED': self.assertTrue(error, 'job completed unexpectedly') self.assert_qmp(event, 'data/type', 'stream') self.assert_qmp(event, 'data/device', 'drive0') self.assert_qmp_absent(event, 'data/error') self.assert_qmp(event, 'data/offset', self.image_len) self.assert_qmp(event, 'data/len', self.image_len) completed = True self.assert_no_active_streams() self.vm.shutdown() def test_enospc(self): self.assert_no_active_streams() result = self.vm.qmp('block-stream', device='drive0', on_error='enospc') self.assert_qmp(result, 'return', {}) completed = False error = False while not completed: for event in self.vm.get_qmp_events(wait=True): if event['event'] == 'BLOCK_JOB_ERROR': self.assert_qmp(event, 'data/device', 'drive0') self.assert_qmp(event, 'data/operation', 'read') error = True elif event['event'] == 'BLOCK_JOB_COMPLETED': self.assertTrue(error, 'job completed unexpectedly') self.assert_qmp(event, 'data/type', 'stream') self.assert_qmp(event, 'data/device', 'drive0') self.assert_qmp(event, 'data/error', 'Input/output error') self.assert_qmp(event, 'data/offset', self.STREAM_BUFFER_SIZE) self.assert_qmp(event, 'data/len', self.image_len) completed = True self.assert_no_active_streams() self.vm.shutdown() class TestENOSPC(TestErrors): def setUp(self): self.blkdebug_file = backing_img + ".blkdebug" self.create_image(backing_img, TestErrors.image_len) self.create_blkdebug_file(self.blkdebug_file, "read_aio", 28) qemu_img('create', '-f', iotests.imgfmt, '-o', 'backing_file=blkdebug:%s:%s,backing_fmt=raw' % (self.blkdebug_file, backing_img), test_img) self.vm = iotests.VM().add_drive(test_img) self.vm.launch() def tearDown(self): self.vm.shutdown() os.remove(test_img) os.remove(backing_img) os.remove(self.blkdebug_file) def test_enospc(self): self.assert_no_active_streams() result = self.vm.qmp('block-stream', device='drive0', on_error='enospc') self.assert_qmp(result, 'return', {}) error = False completed = False while not completed: for event in self.vm.get_qmp_events(wait=True): if event['event'] == 'BLOCK_JOB_ERROR': self.assert_qmp(event, 'data/device', 'drive0') self.assert_qmp(event, 'data/operation', 'read') result = self.vm.qmp('query-block-jobs') self.assert_qmp(result, 'return[0]/paused', True) self.assert_qmp(result, 'return[0]/offset', self.STREAM_BUFFER_SIZE) self.assert_qmp(result, 'return[0]/io-status', 'nospace') result = self.vm.qmp('block-job-resume', device='drive0') self.assert_qmp(result, 'return', {}) result = self.vm.qmp('query-block-jobs') self.assert_qmp(result, 'return[0]/paused', False) self.assert_qmp(result, 'return[0]/io-status', 'ok') error = True elif event['event'] == 'BLOCK_JOB_COMPLETED': self.assertTrue(error, 'job completed unexpectedly') self.assert_qmp(event, 'data/type', 'stream') self.assert_qmp(event, 'data/device', 'drive0') self.assert_qmp_absent(event, 'data/error') self.assert_qmp(event, 'data/offset', self.image_len) self.assert_qmp(event, 'data/len', self.image_len) completed = True self.assert_no_active_streams() self.vm.shutdown() class TestStreamStop(ImageStreamingTestCase): image_len = 8 * 1024 * 1024 * 1024 # GB def setUp(self): qemu_img('create', backing_img, str(TestStreamStop.image_len)) qemu_img('create', '-f', iotests.imgfmt, '-o', 'backing_file=%s' % backing_img, test_img) self.vm = iotests.VM().add_drive(test_img) self.vm.launch() def tearDown(self): self.vm.shutdown() os.remove(test_img) os.remove(backing_img) def test_stream_stop(self): self.assert_no_active_streams() result = self.vm.qmp('block-stream', device='drive0') self.assert_qmp(result, 'return', {}) time.sleep(0.1) events = self.vm.get_qmp_events(wait=False) self.assertEqual(events, [], 'unexpected QMP event: %s' % events) self.cancel_and_wait() class TestSetSpeed(ImageStreamingTestCase): image_len = 80 * 1024 * 1024 # MB def setUp(self): qemu_img('create', backing_img, str(TestSetSpeed.image_len)) qemu_img('create', '-f', iotests.imgfmt, '-o', 'backing_file=%s' % backing_img, test_img) self.vm = iotests.VM().add_drive(test_img) self.vm.launch() def tearDown(self): self.vm.shutdown() os.remove(test_img) os.remove(backing_img) # This is a short performance test which is not run by default. # Invoke "IMGFMT=qed ./030 TestSetSpeed.perf_test_throughput" def perf_test_throughput(self): self.assert_no_active_streams() result = self.vm.qmp('block-stream', device='drive0') self.assert_qmp(result, 'return', {}) result = self.vm.qmp('block-job-set-speed', device='drive0', speed=8 * 1024 * 1024) self.assert_qmp(result, 'return', {}) completed = False while not completed: for event in self.vm.get_qmp_events(wait=True): if event['event'] == 'BLOCK_JOB_COMPLETED': self.assert_qmp(event, 'data/type', 'stream') self.assert_qmp(event, 'data/device', 'drive0') self.assert_qmp(event, 'data/offset', self.image_len) self.assert_qmp(event, 'data/len', self.image_len) completed = True self.assert_no_active_streams() def test_set_speed(self): self.assert_no_active_streams() result = self.vm.qmp('block-stream', device='drive0') self.assert_qmp(result, 'return', {}) # Default speed is 0 result = self.vm.qmp('query-block-jobs') self.assert_qmp(result, 'return[0]/device', 'drive0') self.assert_qmp(result, 'return[0]/speed', 0) result = self.vm.qmp('block-job-set-speed', device='drive0', speed=8 * 1024 * 1024) self.assert_qmp(result, 'return', {}) # Ensure the speed we set was accepted result = self.vm.qmp('query-block-jobs') self.assert_qmp(result, 'return[0]/device', 'drive0') self.assert_qmp(result, 'return[0]/speed', 8 * 1024 * 1024) self.cancel_and_wait() # Check setting speed in block-stream works result = self.vm.qmp('block-stream', device='drive0', speed=4 * 1024 * 1024) self.assert_qmp(result, 'return', {}) result = self.vm.qmp('query-block-jobs') self.assert_qmp(result, 'return[0]/device', 'drive0') self.assert_qmp(result, 'return[0]/speed', 4 * 1024 * 1024) self.cancel_and_wait() def test_set_speed_invalid(self): self.assert_no_active_streams() result = self.vm.qmp('block-stream', device='drive0', speed=-1) self.assert_qmp(result, 'error/class', 'GenericError') self.assert_no_active_streams() result = self.vm.qmp('block-stream', device='drive0') self.assert_qmp(result, 'return', {}) result = self.vm.qmp('block-job-set-speed', device='drive0', speed=-1) self.assert_qmp(result, 'error/class', 'GenericError') self.cancel_and_wait() if __name__ == '__main__': iotests.main(supported_fmts=['qcow2', 'qed'])