1 # Copyright (c) 2012 The Chromium OS Authors. All rights reserved. 2 # Use of this source code is governed by a BSD-style license that can be 3 # found in the LICENSE file. 4 # Author: Cosimo Alfarano <cosimo.alfarano (at] collabora.co.uk> 5 6 import datetime 7 import logging 8 import os 9 10 from autotest_lib.client.cros import storage as storage_mod 11 from autotest_lib.client.common_lib import autotemp, error 12 from autotest_lib.client.bin import utils 13 14 USECS_IN_SEC = 1000000.0 15 16 class hardware_Usb30Throughput(storage_mod.StorageTester): 17 version = 1 18 preserve_srcdir = True 19 _autosrc = None 20 _autodst = None 21 results = {} 22 23 24 def cleanup(self): 25 if self._autosrc: 26 self._autosrc.clean() 27 if self._autodst: 28 self._autodst.clean() 29 30 self.scanner.unmount_all() 31 32 super(hardware_Usb30Throughput, self).cleanup() 33 34 35 def run_once(self, measurements=5, size=1, min_speed=300.0): 36 """ 37 @param measurements: (int) the number of measurements to do. 38 For the test to fail at least one measurement needs to be 39 below |min_speed| 40 @param size: (int) size of the file to be copied for testing the 41 transfer rate, it represent the size in megabytes. 42 Generally speaking, the bigger is the file used for 43 |measurements| the slower the test will run and the more 44 accurate it will be. 45 e.g.: 10 is 10MB, 101 is 101MB 46 @param min_speed: (float) in Mbit/sec. It's the min throughput a USB 3.0 47 device should perform to be accepted. Conceptually it's the max 48 USB 3.0 throughput minus a tollerance. 49 Defaults to 300Mbit/sec (ie 350Mbits/sec minus ~15% tollerance) 50 """ 51 volume_filter = {'bus': 'usb'} 52 storage = self.wait_for_device(volume_filter, cycles=1, 53 mount_volume=True)[0] 54 55 # in Megabytes (power of 10, to be consistent with the throughput unit) 56 size *= 1000*1000 57 58 self._autosrc = autotemp.tempfile(unique_id='autotest.src', 59 dir=storage['mountpoint']) 60 self._autodst = autotemp.tempfile(unique_id='autotest.dst', 61 dir=self.tmpdir) 62 63 # Create random file 64 storage_mod.create_file(self._autosrc.name, size) 65 66 num_failures = 0 67 for measurement in range(measurements): 68 xfer_rate = get_xfer_rate(self._autosrc.name, self._autodst.name) 69 key = 'Mbit_per_sec_measurement_%d' % measurement 70 self.results[key] = xfer_rate 71 logging.debug('xfer rate (measurement %d) %.2f (min=%.2f)', 72 measurement, xfer_rate, min_speed) 73 74 if xfer_rate < min_speed: 75 num_failures += 1 76 77 # Apparently self.postprocess_iteration is not called on TestFail 78 # so we need to process data here in order to have some performance log 79 # even on TestFail 80 self.results['Mbit_per_sec_average'] = (sum(self.results.values()) / 81 len(self.results)) 82 self.write_perf_keyval(self.results) 83 84 if num_failures > 0: 85 msg = ('%d/%d measured transfer rates under performed ' 86 '(min_speed=%.2fMbit/sec)' % (num_failures, measurements, 87 min_speed)) 88 raise error.TestFail(msg) 89 90 91 def get_xfer_rate(src, dst): 92 """Compute transfer rate from src to dst as Mbit/sec 93 94 Execute a copy from |src| to |dst| and returns the file copy transfer rate 95 in Mbit/sec 96 97 @param src, dst: paths for source and destination 98 99 @return trasfer rate (float) in Mbit/sec 100 """ 101 assert os.path.isfile(src) 102 assert os.path.isfile(dst) 103 104 utils.drop_caches() 105 start = datetime.datetime.now() 106 utils.force_copy(src, dst) 107 end = datetime.datetime.now() 108 delta = end - start 109 110 # compute seconds (as float) from microsecs 111 delta_secs = delta.seconds + (delta.microseconds/USECS_IN_SEC) 112 # compute Mbit from bytes 113 size_Mbit = (os.path.getsize(src)*8.0)/(1000*1000) 114 115 logging.info('file trasferred: size (Mbits): %f, start: %f, end: %d,' 116 ' delta (secs): %f', 117 size_Mbit, 118 start.second+start.microsecond/USECS_IN_SEC, 119 end.second+end.microsecond/USECS_IN_SEC, 120 delta_secs) 121 122 # return the xfer rate in Mbits/secs having bytes/microsec 123 return size_Mbit / delta_secs 124
