From 32382c2577d67f2c7f95bb189ce5ede0f31a8c84 Mon Sep 17 00:00:00 2001 From: Lawrence Schlosser Date: Wed, 30 May 2018 19:50:29 -0700 Subject: [PATCH 1/3] CT-525 added profile.py for local file system profiling --- tests/profile.py | 600 +++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 600 insertions(+) create mode 100755 tests/profile.py diff --git a/tests/profile.py b/tests/profile.py new file mode 100755 index 00000000..d9188cea --- /dev/null +++ b/tests/profile.py @@ -0,0 +1,600 @@ +#!/usr/bin/env python + +import argparse +import csv +import functools +import hashlib +import inspect +import json +import logging +from multiprocessing.pool import ThreadPool +import os +import time + +from conductor.lib import api_client, loggeria + +try: + import xxhash + XXHASH = True +except ImportError as e: + print "warning: Could not import xxhash. xxhash functionality disabled" + XXHASH = False + + +DEFAULT_READ_SIZE = 65536 + +DESCRIPTION = "DESCRIPTION" +TIME_STAT = "STAT TIME" +TIME_READ = "READ TIME" +TIME_MD5 = "MD5 TIME" +TIME_XXHASH = "XXHASH TIME" +TIME_WARMUP = "WARMUP TIME" +FILEPATH = "FILEPATH" +SIZE = "SIZE" +NAME = "NAME" +VALUE = "VALUE" +FILE_COUNT = "FILE COUNT" + + +# HEADERS for different csv/table sections +DATA_HEADERS = (DESCRIPTION, FILEPATH, SIZE, TIME_STAT, TIME_WARMUP, TIME_READ, TIME_MD5, TIME_XXHASH) +SUMMARY_HEADERS = (DESCRIPTION, FILE_COUNT, SIZE, TIME_STAT, TIME_WARMUP, TIME_READ, TIME_MD5, TIME_XXHASH) +ARGS_HEADERS = [NAME, VALUE] + +logger = logging.getLogger("conductor") + + +def parse_args(): + + # ------------------------------------------------ + # common parser + # ------------------------------------------------ + common_parser = argparse.ArgumentParser(add_help=False) + + common_parser.add_argument( + "--stat", + choices=[False, True], + type=cast_to_bool, + default=True, + help='Perform a stat on each file' + ) + + common_parser.add_argument( + "--read", + choices=[False, True], + type=cast_to_bool, + default=True, + help='Read the entire contents of each file' + ) + + common_parser.add_argument( + "--md5", + choices=[False, True], + type=cast_to_bool, + default=True, + help='Generate an md5 hash for each file' + ) + + if XXHASH: + common_parser.add_argument( + "--xxhash", + choices=[False, True], + type=cast_to_bool, + default=True, + help='Generate an xxhash hash for each file' + ) + + common_parser.add_argument( + "--warmup", + choices=[False, True], + type=cast_to_bool, + default=True, + help=("If True, will \"warm up\" each file before performing any further operations on it." + "This essentially loads the file into any OS/disk cache so that subsequent reads to " + "that file will yield consistent performance between tests.") + ) + + common_parser.add_argument( + "--threads", + type=int, + default=1, + help=("The number of threads to use so that parallel reads can be tested. " + "Note that parallelizing reads my result in a faster overall test, but may also lead " + "to slower per-file reads on average") + ) + + common_parser.add_argument( + "--csv_path", + help="A csv filepath to write the the results to. Results contain metrics per each file", + ) + + common_parser.add_argument( + "--log_level", + choices=loggeria.LEVELS, + default=loggeria.LEVEL_INFO, + help="The logging level to display") + + common_parser.add_argument( + "--read_size", + type=int, + default=65536, + help="The number of bytes to read at a time (when reading a file)", + ) + + common_parser.add_argument( + "--skip_failures", + choices=[False, True], + type=cast_to_bool, + default=False, + help=("If True, will skip any failures upon file reads. Otherwise an exception will halt " + "testing.") + ) + + # ------------------------------------------------ + # Main parser + # ------------------------------------------------ + parser = argparse.ArgumentParser( + description=("Profile the performance of your file system by processing a set of files. " + "A set of files is dictated by either providing a conductor job id, or a " + "list of directories. Use the appropriate sub command to use either methodology.") + ) + subparsers = parser.add_subparsers(title="actions") + + # ------------------------------------------------ + # Job/jid parser + # ------------------------------------------------ + job_parser = subparsers.add_parser( + "job", + parents=[common_parser], + description=("Profile the performance of your file system by processing files from the provided " + "Conductor Job ID (jid)"), + formatter_class=argparse.ArgumentDefaultsHelpFormatter, + ) + + job_parser.add_argument( + "jid", + help="The jid (job id) for a job whose files to target", + ) + + job_parser.set_defaults(func=run_job_profiler) + + # ------------------------------------------------ + # Directory parser + # ------------------------------------------------ + dir_parser = subparsers.add_parser( + "dirs", + parents=[common_parser], + description=("Profile the performance of your file system by processing files found in the" + "the given directories"), + formatter_class=argparse.ArgumentDefaultsHelpFormatter + ) + + dir_parser.add_argument( + "dirs", + nargs="+", + help="One or more directories to read files from (recursively)", + ) + + dir_parser.set_defaults(func=run_dir_profiler) + + return parser.parse_args() + + +def cast_to_bool(string): + ''' + Ensure that the argument provided is either "True" or "False (or "true" or + "false") and convert that argument to an actual bool value (True or False). + ''' + string_lower = string.lower() + if string_lower == "true": + return True + elif string_lower == "false": + return False + raise argparse.ArgumentTypeError('Argument must be True or False') + + +def run_job_profiler(args): + ''' + Profile the files found from the provided Job id (jid) + ''' + + # ------------------------------------------------ + # Fetch Job + # ------------------------------------------------ + logger.info("Fetching Job %s", args.jid) + client = api_client.ApiClient() + r_body, _ = client.make_request('api/v1/jobs', + params="filter=jid_eq_%s" % args.jid, + use_api_key=True + ) + + jobs = json.loads(r_body).get("data") + if not jobs: + raise Exception("No Job %s found" % args.jid) + assert len(jobs) < 2, "Found more than 1 Job with jid %s" % args.jid + job = jobs[0] + upload_id = job["upload"] + + # ------------------------------------------------ + # Fetch Job's Upload + # ------------------------------------------------ + logger.info("Fetching Upload %s", upload_id) + r_body, _ = client.make_request('api/v1/uploads/%s' % upload_id, + use_api_key=True + ) + upload = json.loads(r_body).get("data") + filepaths = upload["upload_files"].keys() + + profile( + filepaths, + stat=args.stat, + read=args.read, + hash_md5=args.md5, + hash_xx=args.xxhash if XXHASH else None, + warmup=args.warmup, + read_size=args.read_size, + thread_count=args.threads, + csv_filepath=args.csv_path, + ) + + +def run_dir_profiler(args): + ''' + Profile the files found from the provided directories + ''' + logger.info("Testing files in the following directories:\n\t%s", "\n\t".join(args.dirs)) + # Validate that provided directories exist + for dirpath in args.dirs: + if not os.path.isdir(dirpath): + raise Exception("Directory does not exist: %s" % dirpath) + + filepaths = [] + for dirpath in args.dirs: + filepaths.extend(get_files(dirpath, recurse=True)) + + profile( + filepaths, + stat=args.stat, + read=args.read, + hash_md5=args.md5, + hash_xx=args.xxhash if XXHASH else None, + warmup=args.warmup, + read_size=args.read_size, + thread_count=args.threads, + csv_filepath=args.csv_path, + skip_failures=args.skip_failures, + ) + + +def profile(filepaths, stat=True, read=True, hash_md5=True, hash_xx=True, warmup=True, + read_size=DEFAULT_READ_SIZE, thread_count=1, skip_failures=False, csv_filepath=None): + ''' + Profile the given list of files. see "profile_file" function for + additional argument details. If a csv_filepath is provided, verbose profiling data + will be written to that destination. Otherwise, only summary data will be logged to stdout, e.g + + ---------- SUMMARY ----------- + DESCRIPTION FILE COUNT SIZE STAT TIME WARMUP TIME READ TIME MD5 TIME XXHASH TIME + Averaged Time 970 files (0 skipped) 206185 000.000005209814642 000.000095234703772 000.000074392495696 000.000403518037698 000.000095770776886 + Summed Time 970 files (0 skipped) 200000000 000.005053520202637 000.092377662658691 000.072160720825195 000.391412496566772 000.092897653579712 + Test Time 0.661609172821 + ------------------------------ + ''' + + # Capture/record the arguments that were provided to the function. We'll log/write them out as + # part of the report. + args_record = inspect.getargvalues(inspect.currentframe())[3] + # Instead of recording each filepaths arg (as there may be millions), simply record the file count + args_record["filepaths"] = len(args_record["filepaths"]) + + # Log out the arguments + for arg_name, arg_value in sorted(args_record.iteritems()): + logger.info("%s: %s", arg_name, arg_value) + + # Record the start time for the entire test + test_time_start = time.time() + logger.info("Profiling %s files using %s threads...", len(filepaths), thread_count) + + # Run the profile + results = profile_files(filepaths, stat=stat, read=read, hash_md5=hash_md5, hash_xx=hash_xx, warmup=warmup, + read_size=read_size, thread_count=thread_count, skip_failures=skip_failures) + + # Record the duration of the entire test. + test_time_duration = time.time() - test_time_start + logger.info('Profile runtime: %.2f seconds', test_time_duration) + + # Generate summary information + summary = generate_summary(results) + # inject the duration of the entire test into the summary. Pretty hacky, could be better! + summary.append({DESCRIPTION: "Test Time", FILE_COUNT: test_time_duration}) + + # Generate a summary string that can be logged out + summary_str = SummaryTableStr(summary, + SUMMARY_HEADERS, + title=" SUMMARY ".center(30, "-"), + footer="-" * 30).make_table_str() + # print out summary + print summary_str + + # If a csv filepath is provided (for output), generate a more verbose output + if csv_filepath: + + # Generate a summary of the arguments that were used to run this test + args_summary = generate_args_summary(args_record) + logger.debug("Writing to: %s", csv_filepath) + + # There are three different sections in the csv (each have different headers): + # - arguments used to run the test + # - the summary of the test + # - the per-file results of the test + with open(csv_filepath, 'w') as f: + for title, headers, data in ( + (" ARGUMENTS ".center(30, "-"), [NAME, VALUE], args_summary), + (" SUMMARY ".center(30, "-"), SUMMARY_HEADERS, summary), + (" FILE RESULTS ".center(30, "-"), DATA_HEADERS, results), + ): + write_csv_section(f, title, headers, data) + + logger.info("Summary written to: %s", csv_filepath) + + +def profile_files(filepaths, stat=True, read=True, hash_md5=True, hash_xx=True, warmup=True, + read_size=DEFAULT_READ_SIZE, thread_count=1, skip_failures=False): + ''' + Profile the given list of files within a threading pool. see "profile_file" function for + additional argument details. + ''' + + # Run Profile using the number of threads specified + pool = ThreadPool(processes=thread_count) + return pool.map(functools.partial(profile_file, + stat=stat, + read=read, + hash_md5=hash_md5, + hash_xx=hash_xx, + warmup=warmup, + read_size=read_size, + skip_failures=skip_failures, + ), + filepaths) + + +def profile_file(filepath, stat=True, read=True, hash_md5=True, hash_xx=True, warmup=True, + read_size=DEFAULT_READ_SIZE, skip_failures=False): + ''' + Perform the indicated actions on the given file. Record and return the time that it takes for + each action. + + args: + stat: bool. If True, stat the file. + read: bool. If True, read the entire file. + hash_md5: bool. If True, generate an md5 hash from the file. + hash_xx: bool. If True, generate an xx hash from the file. + warmup: bool. If True, read the entire file first (i.e. "warm it up") before performing any + operations on it. This essentially loads the file into any OS/disk cache so that + subsequent reads to that file will yield consistent performance between tests. + read_size: int. The number of bytes to read when + skip_failures: bool. If True, skip/suppress any failures (rather than raising an exception). + return: dict of metrics data + ''' + # Create the boilerplate data structure for profile data + profile_data = { + DESCRIPTION: "file", + FILEPATH: filepath, + SIZE: None, + TIME_STAT: None, + TIME_READ: None, + TIME_MD5: None, + TIME_XXHASH: None, + TIME_WARMUP: None, + } + try: + if warmup: + start_time = time.time() + read_file(filepath) + profile_data[TIME_WARMUP] = time.time() - start_time + + if stat: + start_time = time.time() + size = stat_file(filepath).st_size + profile_data[TIME_STAT] = time.time() - start_time + profile_data[SIZE] = size + + if read: + start_time = time.time() + read_file(filepath) + profile_data[TIME_READ] = time.time() - start_time + + if hash_md5: + start_time = time.time() + generate_md5(filepath, read_size=read_size) + profile_data[TIME_MD5] = time.time() - start_time + + if hash_xx and XXHASH: + start_time = time.time() + generate_xxhash(filepath, read_size=read_size) + profile_data[TIME_XXHASH] = time.time() - start_time + + except Exception as e: + if not skip_failures: + raise + logger.warning(e) + profile_data[DESCRIPTION] = "skipped file due to error: %s" % e + + return profile_data + + +def stat_file(filepath): + return os.stat(filepath) + + +def read_file(filepath): + with open(filepath) as f: + f.read() + + +def generate_md5(filepath, read_size=DEFAULT_READ_SIZE): + hasher = hashlib.md5() + return generate_hash(filepath, hasher, read_size=read_size).hexdigest() + + +def generate_xxhash(filepath, read_size=DEFAULT_READ_SIZE): + hasher = xxhash.xxh32() + return generate_hash(filepath, hasher, read_size=read_size).hexdigest() + + +def generate_hash(filepath, hasher, read_size=DEFAULT_READ_SIZE): + ''' + Hash the given filepath using the given hasher object. + + filepath: str. The file path of the file to hash. + + hasher: hashing object. + ''' + file_obj = open(filepath, 'rb') + buffer_count = 1 + file_buffer = file_obj.read(read_size) + while len(file_buffer) > 0: + hasher.update(file_buffer) + file_buffer = file_obj.read(read_size) + buffer_count += 1 + return hasher + + +def write_csv_section(file_obj, title, headers, data): + ''' + Write a section of csv data to an open file object + + ''' + writer = csv.DictWriter(file_obj, fieldnames=headers) + writer.writer.writerow([title]) + writer.writeheader() + for row in data: + writer.writerow(row) + writer.writerow({}) + writer.writerow({}) + + +def generate_args_summary(args): + + args_summary = [] + for arg_name, arg_value in args.iteritems(): + args_summary.append( + { + NAME: arg_name, + VALUE: arg_value, + } + + ) + return args_summary + + +def generate_summary(data): + ''' + Generate the summary data from the profiling results. Summary data consists for two rows: + - summed totals of all columns + - averages of all columns + + Filter out any rows/files that were + skipped/failed. + + args: + data: a list of dicts, where each dict represents a row of file information + return: a list of dicts, where each dict represents a row of summary data. + ''' + logger.info("Creating summary rows...") + + # Filter our skipped files + filtered_rows = [] + skipped_rows = [] + for row in data: + if "skipped" in row[DESCRIPTION]: + skipped_rows.append(row) + else: + filtered_rows.append(row) + + sum_summary = { + DESCRIPTION: "Summed Time", + FILE_COUNT: "%s files (%s skipped)" % (len(filtered_rows), len(skipped_rows)), + } + avg_summary = { + DESCRIPTION: "Averaged Time", + FILE_COUNT: "%s files (%s skipped)" % (len(filtered_rows), len(skipped_rows)), + } + + for header in (SIZE, TIME_STAT, TIME_WARMUP, TIME_READ, TIME_MD5, TIME_XXHASH): + logger.debug('Summarizing "%s" data...', header) + column_entries = [row[header] for row in filtered_rows if row[header] is not None] + sum_summary[header] = sum(column_entries) if column_entries else "" + avg_summary[header] = (sum_summary[header] / float(len(column_entries))) if column_entries else "" + + return [avg_summary, sum_summary] + + +def get_files(dirpath, recurse=True): + ''' + Return all files found in the given directory. + + Optionally recurse the directory to also include files that are located + in subdirectories as well + ''' + files = [] + + if not os.path.isdir(dirpath): + raise Exception("Directory does not exist: '%s'" % dirpath) + + # If operating recursively, use os.walk to grab sub files + if recurse: + for sub_dirpath, _, filenames in os.walk(dirpath): + for filename in filenames: + filepath = os.path.join(sub_dirpath, filename) + if os.path.isfile(filepath): + files.append(filepath) + else: + for filename in os.listdir(dirpath): + if os.path.isfile(os.path.join(dirpath, filename)): + files.append(os.path.join(dirpath, filename)) + + return files + + +def precision_formatter(value, float_precision=15, zfill=4, omit_empty=True): + ''' + args: + float_precision: int. The number of decimal places to limit the value to + zfill: the number of decimal places to fill/pad (with zeros) *to the left* of the + decimal place (if necesary). + ''' + if value is "" and omit_empty: + return "" + + # Pads/prefixes the float with 3 decimal places of zeros + return ('%.*f' % (float_precision, value)).zfill(float_precision + zfill) + + +class SummaryTableStr(loggeria.TableStr): + ''' + Subclass TableStr to provide custom formatters for our data + + ---------- SUMMARY ----------- + DESCRIPTION FILE COUNT SIZE STAT TIME WARMUP TIME READ TIME MD5 TIME XXHASH TIME + Averaged Time 970 files (0 skipped) 206185 000.000005209814642 000.000095234703772 000.000074392495696 000.000403518037698 000.000095770776886 + Summed Time 970 files (0 skipped) 200000000 000.005053520202637 000.092377662658691 000.072160720825195 000.391412496566772 000.092897653579712 + Test Time 0.661609172821 + ------------------------------ + ''' + + cell_modifiers = { + TIME_STAT: precision_formatter, + TIME_READ: precision_formatter, + TIME_MD5: precision_formatter, + TIME_XXHASH: precision_formatter, + TIME_WARMUP: precision_formatter, + SIZE: lambda x: int(x) if x is not "" else x, + } + + +if __name__ == '__main__': + args = parse_args() + loggeria.setup_conductor_logging(logger_level=loggeria.LEVEL_MAP.get(args.log_level)) + args.func(args) From de1b6bd8ad211af4818054c773260c3573014438 Mon Sep 17 00:00:00 2001 From: Lawrence Schlosser Date: Wed, 27 Jun 2018 16:11:53 -0700 Subject: [PATCH 2/3] added cache dropping and test_suite functionality - also added device benchmarking option - also changed all file-reading operations to use io.open --- tests/profile.py | 348 ++++++++++++++++++++++++++++++++------- tests/profile_tests.json | 24 +++ 2 files changed, 312 insertions(+), 60 deletions(-) create mode 100644 tests/profile_tests.json diff --git a/tests/profile.py b/tests/profile.py index d9188cea..0c7fb821 100755 --- a/tests/profile.py +++ b/tests/profile.py @@ -1,14 +1,22 @@ #!/usr/bin/env python +''' +sudo -HE env PATH=$PATH PYTHONPATH=$PYTHONPATH HOME=$HOME ./profile.py dirs /mnt/WD-Passport-WDBYFT0040BBL-1/test_data/random_data/07 --read=False --xxhash=True --md5=False --warm_cache=True --benchmark_device=True +''' import argparse +import collections import csv import functools import hashlib +import io import inspect import json import logging from multiprocessing.pool import ThreadPool import os +import re +import socket +import subprocess import time from conductor.lib import api_client, loggeria @@ -20,6 +28,7 @@ print "warning: Could not import xxhash. xxhash functionality disabled" XXHASH = False +IS_ROOT = os.geteuid() is 0 DEFAULT_READ_SIZE = 65536 @@ -28,18 +37,17 @@ TIME_READ = "READ TIME" TIME_MD5 = "MD5 TIME" TIME_XXHASH = "XXHASH TIME" -TIME_WARMUP = "WARMUP TIME" FILEPATH = "FILEPATH" SIZE = "SIZE" NAME = "NAME" VALUE = "VALUE" FILE_COUNT = "FILE COUNT" - # HEADERS for different csv/table sections -DATA_HEADERS = (DESCRIPTION, FILEPATH, SIZE, TIME_STAT, TIME_WARMUP, TIME_READ, TIME_MD5, TIME_XXHASH) -SUMMARY_HEADERS = (DESCRIPTION, FILE_COUNT, SIZE, TIME_STAT, TIME_WARMUP, TIME_READ, TIME_MD5, TIME_XXHASH) ARGS_HEADERS = [NAME, VALUE] +DEVICE_HEADERS = [DESCRIPTION, VALUE] +DATA_HEADERS = (DESCRIPTION, FILEPATH, SIZE, TIME_STAT, TIME_READ, TIME_MD5, TIME_XXHASH) +SUMMARY_HEADERS = (DESCRIPTION, FILE_COUNT, SIZE, TIME_STAT, TIME_READ, TIME_MD5, TIME_XXHASH) logger = logging.getLogger("conductor") @@ -85,13 +93,12 @@ def parse_args(): ) common_parser.add_argument( - "--warmup", + "--warm_cache", choices=[False, True], type=cast_to_bool, - default=True, - help=("If True, will \"warm up\" each file before performing any further operations on it." - "This essentially loads the file into any OS/disk cache so that subsequent reads to " - "that file will yield consistent performance between tests.") + default=False if IS_ROOT else True, + help=("If True, will \"warm up\"(i.e.read) each file before performing any further operation on it. " + "If False (which requires running as root)will clear OS cache before performing read operations (recommended)") ) common_parser.add_argument( @@ -108,6 +115,12 @@ def parse_args(): help="A csv filepath to write the the results to. Results contain metrics per each file", ) + common_parser.add_argument( + "--limit", + type=int, + help="Limit the number of files tested to the given value", + ) + common_parser.add_argument( "--log_level", choices=loggeria.LEVELS, @@ -130,6 +143,26 @@ def parse_args(): "testing.") ) + common_parser.add_argument( + "--benchmark_device", + choices=[False, True], + type=cast_to_bool, + default=True if IS_ROOT else False, + help=("If True, will test read-throughput of disk before performing file profiling." + "Requires root privileges") + ) + + common_parser.add_argument( + "--suite_file", + action=ValidatePath, + help=("The path to the suite of test configurations to be run"), + ) + + common_parser.add_argument( + "--output_dir", + help=("The directory to write verbose results to"), + ) + # ------------------------------------------------ # Main parser # ------------------------------------------------ @@ -138,7 +171,7 @@ def parse_args(): "A set of files is dictated by either providing a conductor job id, or a " "list of directories. Use the appropriate sub command to use either methodology.") ) - subparsers = parser.add_subparsers(title="actions") + subparsers = parser.add_subparsers(title="Target files from") # ------------------------------------------------ # Job/jid parser @@ -146,6 +179,7 @@ def parse_args(): job_parser = subparsers.add_parser( "job", parents=[common_parser], + help="Target files from an existing Conductor job for profiling", description=("Profile the performance of your file system by processing files from the provided " "Conductor Job ID (jid)"), formatter_class=argparse.ArgumentDefaultsHelpFormatter, @@ -164,8 +198,9 @@ def parse_args(): dir_parser = subparsers.add_parser( "dirs", parents=[common_parser], - description=("Profile the performance of your file system by processing files found in the" - "the given directories"), + help="Target files from a provided directory for profiling", + description=("Profile the performance of your file system by processing files found in the " + "given directories"), formatter_class=argparse.ArgumentDefaultsHelpFormatter ) @@ -193,6 +228,17 @@ def cast_to_bool(string): raise argparse.ArgumentTypeError('Argument must be True or False') +class ValidatePath(argparse.Action): + ''' + Validate that the give path exists + ''' + + def __call__(self, parser, namespace, values, option_string): + if not os.path.exists(values): + raise argparse.ArgumentError(self, "Path does not exist: %s" % values) + setattr(namespace, self.dest, values) + + def run_job_profiler(args): ''' Profile the files found from the provided Job id (jid) @@ -225,17 +271,7 @@ def run_job_profiler(args): upload = json.loads(r_body).get("data") filepaths = upload["upload_files"].keys() - profile( - filepaths, - stat=args.stat, - read=args.read, - hash_md5=args.md5, - hash_xx=args.xxhash if XXHASH else None, - warmup=args.warmup, - read_size=args.read_size, - thread_count=args.threads, - csv_filepath=args.csv_path, - ) + return run_profiler(filepaths, args) def run_dir_profiler(args): @@ -252,25 +288,59 @@ def run_dir_profiler(args): for dirpath in args.dirs: filepaths.extend(get_files(dirpath, recurse=True)) - profile( + return run_profiler(filepaths, args) + + +def run_profiler(filepaths, args): + + if not args.suite_file: + return _run_profiler(filepaths, args) + + with open(args.suite_file) as f: + suite_file = json.load(f) + for test_args in suite_file: + new_args = vars(args) + new_args.update(test_args) + _run_profiler(filepaths, argparse.Namespace(**new_args)) + + +def _run_profiler(filepaths, args): + if args.output_dir: + filename = "%s_threads=%s_stat=%s_read=%s_md5=%s_xxhash=%s_warm_cache=%s.csv" + args.csv_path = os.path.join(args.output_dir, filename % ( + socket.gethostname(), + args.threads, + bool(args.stat), + bool(args.read), + bool(args.md5), + bool(args.xxhash), + bool(args.warm_cache), + + )) + + if args.limit: + filepaths = filepaths[:args.limit] + + return profile( filepaths, stat=args.stat, read=args.read, hash_md5=args.md5, hash_xx=args.xxhash if XXHASH else None, - warmup=args.warmup, + warm_cache=args.warm_cache, read_size=args.read_size, thread_count=args.threads, csv_filepath=args.csv_path, skip_failures=args.skip_failures, + benchmark_device=args.benchmark_device, ) -def profile(filepaths, stat=True, read=True, hash_md5=True, hash_xx=True, warmup=True, +def profile(filepaths, stat=True, read=True, hash_md5=True, hash_xx=True, warm_cache=False, benchmark_device=True, read_size=DEFAULT_READ_SIZE, thread_count=1, skip_failures=False, csv_filepath=None): ''' Profile the given list of files. see "profile_file" function for - additional argument details. If a csv_filepath is provided, verbose profiling data + additional argument details. If a output_dir is provided, verbose profiling data will be written to that destination. Otherwise, only summary data will be logged to stdout, e.g ---------- SUMMARY ----------- @@ -291,12 +361,29 @@ def profile(filepaths, stat=True, read=True, hash_md5=True, hash_xx=True, warmup for arg_name, arg_value in sorted(args_record.iteritems()): logger.info("%s: %s", arg_name, arg_value) + bench_results = [] + if benchmark_device: + logger.info("Using device used by file: %s" % filepaths[0]) + device = get_device(filepaths[0]) + logger.info("Benchmarking device: %s", device) + bench_results.append({DESCRIPTION: "Device", VALUE: device}) + + for bench_type, result in run_hdparm(device, runs=3).iteritems(): + bench_results.append({DESCRIPTION: bench_type, VALUE: "%s bytes/sec" % result}) + + # Generate a summary string that can be logged out + disk_bench_str = SummaryTableStr(bench_results, + DEVICE_HEADERS, + title=" DEVICE BENCHMARK ".center(30, "-"), + footer="-" * 30).make_table_str() + print disk_bench_str + # Record the start time for the entire test test_time_start = time.time() logger.info("Profiling %s files using %s threads...", len(filepaths), thread_count) # Run the profile - results = profile_files(filepaths, stat=stat, read=read, hash_md5=hash_md5, hash_xx=hash_xx, warmup=warmup, + results = profile_files(filepaths, stat=stat, read=read, hash_md5=hash_md5, hash_xx=hash_xx, warm_cache=warm_cache, read_size=read_size, thread_count=thread_count, skip_failures=skip_failures) # Record the duration of the entire test. @@ -322,6 +409,8 @@ def profile(filepaths, stat=True, read=True, hash_md5=True, hash_xx=True, warmup # Generate a summary of the arguments that were used to run this test args_summary = generate_args_summary(args_record) logger.debug("Writing to: %s", csv_filepath) + if not os.path.isdir(os.path.dirname(csv_filepath)): + safe_mkdirs(os.path.dirname(csv_filepath)) # There are three different sections in the csv (each have different headers): # - arguments used to run the test @@ -329,7 +418,8 @@ def profile(filepaths, stat=True, read=True, hash_md5=True, hash_xx=True, warmup # - the per-file results of the test with open(csv_filepath, 'w') as f: for title, headers, data in ( - (" ARGUMENTS ".center(30, "-"), [NAME, VALUE], args_summary), + (" ARGUMENTS ".center(30, "-"), ARGS_HEADERS, args_summary), + (" DEVICE BENCHMARK ".center(30, "-"), DEVICE_HEADERS, bench_results), (" SUMMARY ".center(30, "-"), SUMMARY_HEADERS, summary), (" FILE RESULTS ".center(30, "-"), DATA_HEADERS, results), ): @@ -338,7 +428,7 @@ def profile(filepaths, stat=True, read=True, hash_md5=True, hash_xx=True, warmup logger.info("Summary written to: %s", csv_filepath) -def profile_files(filepaths, stat=True, read=True, hash_md5=True, hash_xx=True, warmup=True, +def profile_files(filepaths, stat=True, read=True, hash_md5=True, hash_xx=True, warm_cache=False, read_size=DEFAULT_READ_SIZE, thread_count=1, skip_failures=False): ''' Profile the given list of files within a threading pool. see "profile_file" function for @@ -352,14 +442,14 @@ def profile_files(filepaths, stat=True, read=True, hash_md5=True, hash_xx=True, read=read, hash_md5=hash_md5, hash_xx=hash_xx, - warmup=warmup, + warm_cache=warm_cache, read_size=read_size, skip_failures=skip_failures, ), filepaths) -def profile_file(filepath, stat=True, read=True, hash_md5=True, hash_xx=True, warmup=True, +def profile_file(filepath, stat=True, read=True, hash_md5=True, hash_xx=True, warm_cache=False, read_size=DEFAULT_READ_SIZE, skip_failures=False): ''' Perform the indicated actions on the given file. Record and return the time that it takes for @@ -386,34 +476,26 @@ def profile_file(filepath, stat=True, read=True, hash_md5=True, hash_xx=True, wa TIME_READ: None, TIME_MD5: None, TIME_XXHASH: None, - TIME_WARMUP: None, } + profiler = Profile(use_cache=warm_cache, read_size=read_size) try: - if warmup: - start_time = time.time() - read_file(filepath) - profile_data[TIME_WARMUP] = time.time() - start_time if stat: - start_time = time.time() - size = stat_file(filepath).st_size - profile_data[TIME_STAT] = time.time() - start_time - profile_data[SIZE] = size + result, duration = profiler(stat_file)(filepath) + profile_data[TIME_STAT] = duration + profile_data[SIZE] = result.st_size if read: - start_time = time.time() - read_file(filepath) - profile_data[TIME_READ] = time.time() - start_time - - if hash_md5: - start_time = time.time() - generate_md5(filepath, read_size=read_size) - profile_data[TIME_MD5] = time.time() - start_time + _, duration = profiler(read_file)(filepath, read_size=read_size) + profile_data[TIME_READ] = duration if hash_xx and XXHASH: - start_time = time.time() - generate_xxhash(filepath, read_size=read_size) - profile_data[TIME_XXHASH] = time.time() - start_time + _, duration = profiler(generate_xxhash)(filepath, read_size=read_size) + profile_data[TIME_XXHASH] = duration + + if hash_md5: + _, duration = profiler(generate_md5)(filepath, read_size=read_size) + profile_data[TIME_MD5] = duration except Exception as e: if not skip_failures: @@ -428,9 +510,11 @@ def stat_file(filepath): return os.stat(filepath) -def read_file(filepath): - with open(filepath) as f: - f.read() +def read_file(filepath, read_size): + with io.open(filepath, 'rb') as f: + file_buffer = f.read(read_size) + while len(file_buffer) > 0: + file_buffer = f.read(read_size) def generate_md5(filepath, read_size=DEFAULT_READ_SIZE): @@ -451,7 +535,7 @@ def generate_hash(filepath, hasher, read_size=DEFAULT_READ_SIZE): hasher: hashing object. ''' - file_obj = open(filepath, 'rb') + file_obj = io.open(filepath, 'rb') buffer_count = 1 file_buffer = file_obj.read(read_size) while len(file_buffer) > 0: @@ -489,6 +573,26 @@ def generate_args_summary(args): return args_summary +def drop_caches(value=3, sync=True): + ''' + Flush Operating System caches (Linux only!) + + Uses /proc/sys/vm/drop_caches + + sync: bool. If True will flush dirty objects to disk. Otherwise dirty objects will continue to + be in use until written out to disk and are not freeable. + + value: int + 1: free pagecache + 2: free dentries and inodes + 3: free pagecache, dentries and inodes + ''' + cmd = "echo %s > /proc/sys/vm/drop_caches" % value + if sync: + cmd = "sync;" + cmd + subprocess.check_call(cmd, shell=True) + + def generate_summary(data): ''' Generate the summary data from the profiling results. Summary data consists for two rows: @@ -514,15 +618,15 @@ def generate_summary(data): filtered_rows.append(row) sum_summary = { - DESCRIPTION: "Summed Time", + DESCRIPTION: "Sum", FILE_COUNT: "%s files (%s skipped)" % (len(filtered_rows), len(skipped_rows)), } avg_summary = { - DESCRIPTION: "Averaged Time", + DESCRIPTION: "Average", FILE_COUNT: "%s files (%s skipped)" % (len(filtered_rows), len(skipped_rows)), } - for header in (SIZE, TIME_STAT, TIME_WARMUP, TIME_READ, TIME_MD5, TIME_XXHASH): + for header in (SIZE, TIME_STAT, TIME_READ, TIME_MD5, TIME_XXHASH): logger.debug('Summarizing "%s" data...', header) column_entries = [row[header] for row in filtered_rows if row[header] is not None] sum_summary[header] = sum(column_entries) if column_entries else "" @@ -558,6 +662,27 @@ def get_files(dirpath, recurse=True): return files +def safe_mkdirs(dirpath): + ''' + Create the given directory. If it already exists, suppress the exception. + This function is useful when handling concurrency issues where it's not + possible to reliably check whether a directory exists before creating it. + ''' + + # Attempt to create the directory path + try: + os.makedirs(dirpath) + + # only catch OSERrror exceptions + except OSError: + # This exception might happen for various reasons. So to be sure that + # it's due to the path already existing, check the path existence. + # If the path doesn't exist, then raise the original exception. + # Otherwise ignore the exception bc the path exists. + if not os.path.isdir(dirpath): + raise + + def precision_formatter(value, float_precision=15, zfill=4, omit_empty=True): ''' args: @@ -572,6 +697,79 @@ def precision_formatter(value, float_precision=15, zfill=4, omit_empty=True): return ('%.*f' % (float_precision, value)).zfill(float_precision + zfill) +def get_device(path): + ''' + Derive the device from the given filepath + ''' + dev = os.stat(path).st_dev + major, minor = os.major(dev), os.minor(dev) + res = {} + for line in file("/proc/partitions"): + fields = line.split() + try: + tmaj = int(fields[0]) + tmin = int(fields[1]) + name = fields[3] + res[(tmaj, tmin)] = name + except Exception: + # just ignore parse errors in header/separator lines + pass + + return "/dev/%s" % res[(major, minor)] + + +def run_hdparm(device, runs=3): + run_results = collections.defaultdict(list) + for _ in range(runs): + print "Executing %s/%s runs:" % (_ + 1, runs), + for bench_type, result in _run_hdparm(device).iteritems(): + run_results[bench_type].append(result) + + result_avg = {} + + for bench_type, results in run_results.iteritems(): + result_avg[bench_type] = int(sum(results) / float(len(results))) + + return result_avg + + +def _run_hdparm(device): + ''' + + regex samples: + Timing cached reads: 18784 MB in 2.00 seconds = 9400.60 MB/sec + Timing buffered disk reads: 3220 MB in 3.00 seconds = 1072.86 MB/sec + Timing buffered disk reads: 80 MB in 3.01 seconds = 26.60 MB/sec + + ''' + rx = r"Timing ((?:\w|\s)+)+:\s+\d+\s+\w+\s+in\s+\d+\.\d+\s+seconds\s+=\s+(\d+\.\d+)\s+(\w+)/sec" + cmd = "hdparm -Tt %s" % device + print cmd + results = {} + for line in subprocess.check_output(cmd, shell=True).split("\n"): + if not line.strip() or line.startswith("/dev/"): + continue + match = re.search(rx, line.strip()) + if not match: + raise Exception('Failed to parse benchmarking output: "%s"' % line.strip()) + + bench_type, data_value, data_unit = match.groups() + # convert to bytes + if data_unit == "GB": + bytes_ = float(data_value) * 1000000000 + elif data_unit == "MB": + bytes_ = float(data_value) * 1000000 + elif data_unit == "KB": + bytes_ = float(data_value) * 1000 + elif data_unit == "B": + bytes_ = float(data_value) + else: + raise Exception("Got unexpected unit of measurement: %s" % data_unit) + results[bench_type] = int(bytes_) + + return results + + class SummaryTableStr(loggeria.TableStr): ''' Subclass TableStr to provide custom formatters for our data @@ -589,11 +787,41 @@ class SummaryTableStr(loggeria.TableStr): TIME_READ: precision_formatter, TIME_MD5: precision_formatter, TIME_XXHASH: precision_formatter, - TIME_WARMUP: precision_formatter, SIZE: lambda x: int(x) if x is not "" else x, } +class Profile(object): + ''' + Wrapped function's signuature must use a filepath as the first argument + ''' + + def __init__(self, use_cache=False, read_size=DEFAULT_READ_SIZE): + ''' + args + use_cache: bool. If True, will read file first before calling wrapped function + ''' + self.use_cache = use_cache + self.read_size = read_size + + def __call__(self, function): + + @functools.wraps(function) + def _decorator(*args, **kwargs): + filepath = args[0] + if self.use_cache: + logger.debug("warming file..") + read_file(filepath, read_size=self.read_size) + else: + logger.debug("flushing os caches") + drop_caches(value=3, sync=True) + start_time = time.time() + result = function(*args, **kwargs) + duration = time.time() - start_time + return result, duration + return _decorator + + if __name__ == '__main__': args = parse_args() loggeria.setup_conductor_logging(logger_level=loggeria.LEVEL_MAP.get(args.log_level)) diff --git a/tests/profile_tests.json b/tests/profile_tests.json new file mode 100644 index 00000000..1ce400a4 --- /dev/null +++ b/tests/profile_tests.json @@ -0,0 +1,24 @@ +[ + { + "threads": 1, + "stat": true, + "xxhash": true, + "md5": true, + "read": true, + "warm_cache": false, + "benchmark_device": true, + "limit": 100 + }, + + { + "threads": 4, + "stat": true, + "xxhash": true, + "md5": true, + "read": true, + "warm_cache": false, + "benchmark_device": true, + "limit": 100 + } + +] From c9a4213b88986ca85d4a24518b9a4d08901beae1 Mon Sep 17 00:00:00 2001 From: Lawrence Schlosser Date: Mon, 1 Jun 2020 00:50:01 -0700 Subject: [PATCH 3/3] add cache flushing for Windows - improve general cross-platform support - now flushes cache once per test run (not per file) --- tests/profile.py | 206 +++++++++++++++++++++----- tests/windows/Clear-StandByMemory.ps1 | 147 ++++++++++++++++++ 2 files changed, 316 insertions(+), 37 deletions(-) create mode 100644 tests/windows/Clear-StandByMemory.ps1 diff --git a/tests/profile.py b/tests/profile.py index 0c7fb821..9cb9c754 100755 --- a/tests/profile.py +++ b/tests/profile.py @@ -1,10 +1,11 @@ #!/usr/bin/env python ''' -sudo -HE env PATH=$PATH PYTHONPATH=$PYTHONPATH HOME=$HOME ./profile.py dirs /mnt/WD-Passport-WDBYFT0040BBL-1/test_data/random_data/07 --read=False --xxhash=True --md5=False --warm_cache=True --benchmark_device=True +sudo -HE env PATH=$PATH PYTHONPATH=$PYTHONPATH HOME=$HOME ./profile.py dirs /mnt/WD-Passport-WDBYFT0040BBL-1/test_data/random_data/07 --read=False --xxhash=True --md5=False --clear_cache=True --benchmark_device=True ''' import argparse import collections +import ctypes import csv import functools import hashlib @@ -14,6 +15,7 @@ import logging from multiprocessing.pool import ThreadPool import os +import platform import re import socket import subprocess @@ -28,8 +30,6 @@ print "warning: Could not import xxhash. xxhash functionality disabled" XXHASH = False -IS_ROOT = os.geteuid() is 0 - DEFAULT_READ_SIZE = 65536 DESCRIPTION = "DESCRIPTION" @@ -49,6 +49,8 @@ DATA_HEADERS = (DESCRIPTION, FILEPATH, SIZE, TIME_STAT, TIME_READ, TIME_MD5, TIME_XXHASH) SUMMARY_HEADERS = (DESCRIPTION, FILE_COUNT, SIZE, TIME_STAT, TIME_READ, TIME_MD5, TIME_XXHASH) +HERE = os.path.realpath(__file__) + logger = logging.getLogger("conductor") @@ -92,13 +94,24 @@ def parse_args(): help='Generate an xxhash hash for each file' ) + common_parser.add_argument( + "--clear_cache", + choices=[False, True], + type=cast_to_bool, + default=True if IS_ADMIN else False, + help=("Flushes the OS' file-system cache to ensure that files are read from disk (rather " + "than benefitting from cached data). This is the default behavior when running as " + "root/admin, and is highly recommended. If False, rather than flushing the cache, " + "each file will be \"warmed up\" by reading it once before running any tests on it") + ) + common_parser.add_argument( "--warm_cache", choices=[False, True], type=cast_to_bool, - default=False if IS_ROOT else True, + default=True if not IS_ADMIN else False, help=("If True, will \"warm up\"(i.e.read) each file before performing any further operation on it. " - "If False (which requires running as root)will clear OS cache before performing read operations (recommended)") + "It's preferable that --clear_cache be used instead. But not both options.") ) common_parser.add_argument( @@ -147,9 +160,9 @@ def parse_args(): "--benchmark_device", choices=[False, True], type=cast_to_bool, - default=True if IS_ROOT else False, + default=True if IS_ADMIN and IS_LINUX else False, help=("If True, will test read-throughput of disk before performing file profiling." - "Requires root privileges") + "Requires root privileges. Only available on Linux.") ) common_parser.add_argument( @@ -306,7 +319,7 @@ def run_profiler(filepaths, args): def _run_profiler(filepaths, args): if args.output_dir: - filename = "%s_threads=%s_stat=%s_read=%s_md5=%s_xxhash=%s_warm_cache=%s.csv" + filename = "%s_threads=%s_stat=%s_read=%s_md5=%s_xxhash=%s_clear_cache=%s_warm_cache=%s.csv" args.csv_path = os.path.join(args.output_dir, filename % ( socket.gethostname(), args.threads, @@ -314,6 +327,7 @@ def _run_profiler(filepaths, args): bool(args.read), bool(args.md5), bool(args.xxhash), + bool(args.clear_cache), bool(args.warm_cache), )) @@ -327,6 +341,7 @@ def _run_profiler(filepaths, args): read=args.read, hash_md5=args.md5, hash_xx=args.xxhash if XXHASH else None, + clear_cache=args.clear_cache, warm_cache=args.warm_cache, read_size=args.read_size, thread_count=args.threads, @@ -336,8 +351,19 @@ def _run_profiler(filepaths, args): ) -def profile(filepaths, stat=True, read=True, hash_md5=True, hash_xx=True, warm_cache=False, benchmark_device=True, - read_size=DEFAULT_READ_SIZE, thread_count=1, skip_failures=False, csv_filepath=None): +def profile(filepaths, + stat=True, + read=True, + hash_md5=True, + hash_xx=True, + clear_cache=True, + warm_cache=False, + benchmark_device=True, + read_size=DEFAULT_READ_SIZE, + thread_count=1, + skip_failures=False, + csv_filepath=None, + ): ''' Profile the given list of files. see "profile_file" function for additional argument details. If a output_dir is provided, verbose profiling data @@ -383,8 +409,18 @@ def profile(filepaths, stat=True, read=True, hash_md5=True, hash_xx=True, warm_c logger.info("Profiling %s files using %s threads...", len(filepaths), thread_count) # Run the profile - results = profile_files(filepaths, stat=stat, read=read, hash_md5=hash_md5, hash_xx=hash_xx, warm_cache=warm_cache, - read_size=read_size, thread_count=thread_count, skip_failures=skip_failures) + results = profile_files( + filepaths, + stat=stat, + read=read, + hash_md5=hash_md5, + hash_xx=hash_xx, + clear_cache=clear_cache, + warm_cache=warm_cache, + read_size=read_size, + thread_count=thread_count, + skip_failures=skip_failures, + ) # Record the duration of the entire test. test_time_duration = time.time() - test_time_start @@ -428,29 +464,63 @@ def profile(filepaths, stat=True, read=True, hash_md5=True, hash_xx=True, warm_c logger.info("Summary written to: %s", csv_filepath) -def profile_files(filepaths, stat=True, read=True, hash_md5=True, hash_xx=True, warm_cache=False, - read_size=DEFAULT_READ_SIZE, thread_count=1, skip_failures=False): +def profile_files(filepaths, + stat=True, + read=True, + hash_md5=True, + hash_xx=True, + clear_cache=True, + warm_cache=False, + read_size=DEFAULT_READ_SIZE, + thread_count=1, + skip_failures=False, + ): ''' Profile the given list of files within a threading pool. see "profile_file" function for additional argument details. ''' + if clear_cache: + logger.debug("flushing OS caches") + flush_cache() # Run Profile using the number of threads specified pool = ThreadPool(processes=thread_count) - return pool.map(functools.partial(profile_file, - stat=stat, - read=read, - hash_md5=hash_md5, - hash_xx=hash_xx, - warm_cache=warm_cache, - read_size=read_size, - skip_failures=skip_failures, - ), - filepaths) + func = functools.partial( + threaded_profile, + stat=stat, + read=read, + hash_md5=hash_md5, + hash_xx=hash_xx, + clear_cache=False, # don't flush cache per file (already flushed OS earlier) + warm_cache=warm_cache, + read_size=read_size, + skip_failures=skip_failures, + ) + return pool.map(func, filepaths) -def profile_file(filepath, stat=True, read=True, hash_md5=True, hash_xx=True, warm_cache=False, - read_size=DEFAULT_READ_SIZE, skip_failures=False): +# define function +def threaded_profile(*args, **kwargs): + ''' + Slim wrapper to log out exception tracebacks that occur in thread. + ''' + try: + return profile_file(*args, **kwargs) + except: + logger.exception("Encountered error:") + raise + + +def profile_file(filepath, + stat=True, + read=True, + hash_md5=True, + hash_xx=True, + clear_cache=True, + warm_cache=False, + read_size=DEFAULT_READ_SIZE, + skip_failures=False, + ): ''' Perform the indicated actions on the given file. Record and return the time that it takes for each action. @@ -460,13 +530,16 @@ def profile_file(filepath, stat=True, read=True, hash_md5=True, hash_xx=True, wa read: bool. If True, read the entire file. hash_md5: bool. If True, generate an md5 hash from the file. hash_xx: bool. If True, generate an xx hash from the file. - warmup: bool. If True, read the entire file first (i.e. "warm it up") before performing any + clear_cache: bool. If True, flushes the OS' file-system cache to ensure that files are read + from disk (rather than cache). + warm_cache: bool. If True, read the entire file first (i.e. "warm it up") before performing any operations on it. This essentially loads the file into any OS/disk cache so that subsequent reads to that file will yield consistent performance between tests. read_size: int. The number of bytes to read when skip_failures: bool. If True, skip/suppress any failures (rather than raising an exception). return: dict of metrics data ''' + # Create the boilerplate data structure for profile data profile_data = { DESCRIPTION: "file", @@ -477,7 +550,7 @@ def profile_file(filepath, stat=True, read=True, hash_md5=True, hash_xx=True, wa TIME_MD5: None, TIME_XXHASH: None, } - profiler = Profile(use_cache=warm_cache, read_size=read_size) + profiler = Profile(clear_cache=clear_cache, warm_cache=warm_cache, read_size=read_size) try: if stat: @@ -573,9 +646,21 @@ def generate_args_summary(args): return args_summary +def flush_cache(): + ''' + Platform ignostic cache flushing + ''' + if IS_LINUX: + return drop_caches(value=3, sync=True) + if IS_WINDOWS: + return clear_filesystem_cache() + logger.warning('cache flush not implemented for OS "%s"', platform.system()) + + def drop_caches(value=3, sync=True): ''' - Flush Operating System caches (Linux only!) + Linux only! Requires root. + Flush Operating System caches. Uses /proc/sys/vm/drop_caches @@ -593,6 +678,17 @@ def drop_caches(value=3, sync=True): subprocess.check_call(cmd, shell=True) +def clear_filesystem_cache(): + ''' + Windows only! Requires admin privs. + Flush Windows' file-system cache + ''' + script_filepath = os.path.join(os.path.dirname(HERE), "windows", "Clear-StandByMemory.ps1") + cmd = 'powershell.exe -NoLogo -NoProfile -ExecutionPolicy RemoteSigned -File "%s"' % script_filepath + logger.debug(cmd) + subprocess.check_call(cmd, shell=True) + + def generate_summary(data): ''' Generate the summary data from the profiling results. Summary data consists for two rows: @@ -791,17 +887,49 @@ class SummaryTableStr(loggeria.TableStr): } +def get_oses(): + ''' + Return a three item tuple where each item is bool, indicating whether the corresponding OS is + active/present. The order of the tuple is: + 1. Linux + 2. Windows + 3. MacOs (Darwin) + + e.g (True, False, False) would indicate that Linux is the active operating system. + + ''' + os = platform.system() + return (os == "Linux", os == "Windows", os == "Darwin") + + +def get_admin(): + ''' + Return True if the user has admin/root permissions + TODO: implement for macOS + ''' + # linux admin test + if hasattr(os, 'geteuid'): + return os.geteuid() == 0 + + # Windows admin test + if hasattr(ctypes, 'windll'): + return ctypes.windll.shell32.IsUserAnAdmin() != 0 + + logger.warning("Unable to determine admin privileges") + + class Profile(object): ''' - Wrapped function's signuature must use a filepath as the first argument + Wrapped function's signature must use a filepath as the first argument ''' - def __init__(self, use_cache=False, read_size=DEFAULT_READ_SIZE): + def __init__(self, clear_cache=True, warm_cache=False, read_size=DEFAULT_READ_SIZE): ''' args - use_cache: bool. If True, will read file first before calling wrapped function + clear_cache: bool. If True, flush OS file-system cache before running tests. ''' - self.use_cache = use_cache + self.clear_cache = clear_cache + self.warm_cache = warm_cache self.read_size = read_size def __call__(self, function): @@ -809,12 +937,12 @@ def __call__(self, function): @functools.wraps(function) def _decorator(*args, **kwargs): filepath = args[0] - if self.use_cache: + if self.clear_cache: + logger.debug("flushing os caches") + flush_cache() + if self.warm_cache: logger.debug("warming file..") read_file(filepath, read_size=self.read_size) - else: - logger.debug("flushing os caches") - drop_caches(value=3, sync=True) start_time = time.time() result = function(*args, **kwargs) duration = time.time() - start_time @@ -823,6 +951,10 @@ def _decorator(*args, **kwargs): if __name__ == '__main__': + IS_ADMIN = get_admin() + IS_LINUX, IS_WINDOWS, IS_DARWIN = get_oses() args = parse_args() loggeria.setup_conductor_logging(logger_level=loggeria.LEVEL_MAP.get(args.log_level)) + logger.info("admin: %s", IS_ADMIN) + logger.info("os: %s", platform.system()) args.func(args) diff --git a/tests/windows/Clear-StandByMemory.ps1 b/tests/windows/Clear-StandByMemory.ps1 new file mode 100644 index 00000000..75dd8a9c --- /dev/null +++ b/tests/windows/Clear-StandByMemory.ps1 @@ -0,0 +1,147 @@ +#Requires -RunAsAdministrator + +# This script originates from https://gallery.technet.microsoft.com/scriptcenter/c-PowerShell-wrapper-6465e028 + +$Source = @" +using System; +using System.ComponentModel; +using System.Runtime.InteropServices; +using System.Security.Principal; + +namespace ClearStandbyList +{ + public class Program + { + const int SE_PRIVILEGE_ENABLED = 2; + const string SE_INCREASE_QUOTA_NAME = "SeIncreaseQuotaPrivilege"; + const string SE_PROFILE_SINGLE_PROCESS_NAME = "SeProfileSingleProcessPrivilege"; + const int SystemFileCacheInformation = 0x0015; + const int SystemMemoryListInformation = 0x0050; + const int MemoryPurgeStandbyList = 4; + + [DllImport("advapi32.dll", SetLastError = true)] + internal static extern bool LookupPrivilegeValue(string host, string name, ref long pluid); + + [DllImport("advapi32.dll", SetLastError = true)] + internal static extern bool AdjustTokenPrivileges(IntPtr htok, bool disall, ref TokPriv1Luid newst, int len, IntPtr prev, IntPtr relen); + + [DllImport("ntdll.dll")] + public static extern UInt32 NtSetSystemInformation(int InfoClass, IntPtr Info, int Length); + public static bool Is64BitMode() + { + return Marshal.SizeOf(typeof(IntPtr)) == 8; + } + + static void Main(string[] args) + { + ClearFileSystemCache(true); + } + + public static void ClearFileSystemCache(bool ClearStandbyCache) + { + try + { + if (SetIncreasePrivilege(SE_INCREASE_QUOTA_NAME)) + { + uint num1; + int SystemInfoLength; + GCHandle gcHandle; + if (!Is64BitMode()) + { + SYSTEM_CACHE_INFORMATION cacheInformation = new SYSTEM_CACHE_INFORMATION(); + cacheInformation.MinimumWorkingSet = uint.MaxValue; + cacheInformation.MaximumWorkingSet = uint.MaxValue; + SystemInfoLength = Marshal.SizeOf(cacheInformation); + gcHandle = GCHandle.Alloc(cacheInformation, GCHandleType.Pinned); + num1 = NtSetSystemInformation(SystemFileCacheInformation, gcHandle.AddrOfPinnedObject(), SystemInfoLength); + gcHandle.Free(); + } + else + { + SYSTEM_CACHE_INFORMATION_64_BIT information64Bit = new SYSTEM_CACHE_INFORMATION_64_BIT(); + information64Bit.MinimumWorkingSet = -1L; + information64Bit.MaximumWorkingSet = -1L; + SystemInfoLength = Marshal.SizeOf(information64Bit); + gcHandle = GCHandle.Alloc(information64Bit, GCHandleType.Pinned); + num1 = NtSetSystemInformation(SystemFileCacheInformation, gcHandle.AddrOfPinnedObject(), SystemInfoLength); + gcHandle.Free(); + } + if (num1 != 0) + throw new Exception("NtSetSystemInformation(SYSTEMCACHEINFORMATION) error: ", new Win32Exception(Marshal.GetLastWin32Error())); + } + if (ClearStandbyCache && SetIncreasePrivilege(SE_PROFILE_SINGLE_PROCESS_NAME)) + { + + int SystemInfoLength = Marshal.SizeOf(MemoryPurgeStandbyList); + GCHandle gcHandle = GCHandle.Alloc(MemoryPurgeStandbyList, GCHandleType.Pinned); + uint num2 = NtSetSystemInformation(SystemMemoryListInformation, gcHandle.AddrOfPinnedObject(), SystemInfoLength); + gcHandle.Free(); + if (num2 != 0) + throw new Exception("NtSetSystemInformation(SYSTEMMEMORYLISTINFORMATION) error: ", new Win32Exception(Marshal.GetLastWin32Error())); + } + } + catch (Exception ex) + { + Console.Write(ex.ToString()); + } + } + + private static bool SetIncreasePrivilege(string privilegeName) + { + using (WindowsIdentity current = WindowsIdentity.GetCurrent(TokenAccessLevels.Query | TokenAccessLevels.AdjustPrivileges)) + { + TokPriv1Luid newst; + newst.Count = 1; + newst.Luid = 0L; + newst.Attr = SE_PRIVILEGE_ENABLED; + if (!LookupPrivilegeValue(null, privilegeName, ref newst.Luid)) + throw new Exception("Error in LookupPrivilegeValue: ", new Win32Exception(Marshal.GetLastWin32Error())); + int num = AdjustTokenPrivileges(current.Token, false, ref newst, 0, IntPtr.Zero, IntPtr.Zero) ? 1 : 0; + if (num == 0) + throw new Exception("Error in AdjustTokenPrivileges: ", new Win32Exception(Marshal.GetLastWin32Error())); + return num != 0; + } + } + } + + [StructLayout(LayoutKind.Sequential, Pack = 1)] + struct SYSTEM_CACHE_INFORMATION + { + public uint CurrentSize; + public uint PeakSize; + public uint PageFaultCount; + public uint MinimumWorkingSet; + public uint MaximumWorkingSet; + public uint Unused1; + public uint Unused2; + public uint Unused3; + public uint Unused4; + } + + [StructLayout(LayoutKind.Sequential, Pack = 1)] + struct SYSTEM_CACHE_INFORMATION_64_BIT + { + public long CurrentSize; + public long PeakSize; + public long PageFaultCount; + public long MinimumWorkingSet; + public long MaximumWorkingSet; + public long Unused1; + public long Unused2; + public long Unused3; + public long Unused4; + } + + [StructLayout(LayoutKind.Sequential, Pack = 1)] + internal struct TokPriv1Luid + { + public int Count; + public long Luid; + public int Attr; + } +} +"@ + +Add-Type -TypeDefinition $Source -Language CSharp + +[ClearStandbyList.Program]::ClearFileSystemCache($true) \ No newline at end of file