fjdriver.py

#!/usr/bin/python3

#
# fjdriver.py for Fork/Join pool projects
#
# Written by Godmar Back and Scott Pruett for CS 3214
#
# https://git-scm.com/docs/pretty-formats
version = "$Format:%H committed by %cn$"

#
import getopt, sys, os, subprocess, signal, re, json, resource, time, socket
from datetime import datetime
from collections import namedtuple, defaultdict

# add location of this script to python sys.path
# add directory in which script is located to python path
script_dir = "/".join(__file__.split("/")[:-1])
if script_dir == "":
    script_dir = "."

script_dir = os.path.realpath(script_dir)
if script_dir not in sys.path:
    sys.path.append(script_dir)

from fjtests import load_test_module

src_dir = os.path.dirname(os.path.dirname(os.path.realpath(__file__))) + "/tests"
results_file = "full-results.json"
filelist = src_dir + "/FILELIST"
timestamp = str(datetime.now()).replace(" ", "_")
workdir = "fj_testdir_" + timestamp
poolfile = "./threadpool.c"
verbose = False
silent = False
list_tests = False
grade_mode = False
benchmark_runs = 1

# Benchmark info

# Times of serial runs for benchmarked tests
# XXX these are out of date currently and ultimately not used.
benchmark_times = {
    'nqueens 14':      17.158,
    'quicksort large': 19.81442,
    'mergesort large': 22.33417
}

tests = load_test_module('standard')

#
# getopt
#
# look for threadpool.c in current dir, or point at with flag
#
def usage():
    print ("""
Usage: %s [options]
    -v              Verbose
    -V              Print Version and exit
    -a              Run benchmark anyway even if machine is not idle
    -r              Only run required tests.
    -h              Show help 
    -p              <file> - Location of threadpool implementation, default ./threadpool.c
    -l              List available tests
    -t              Filter test by name, given as a comma separated list.
                    e.g.: -t basic1,psum
    """ % (sys.argv[0]))

try:
    opts, args = getopt.getopt(sys.argv[1:], "Varvhlp:t:o:B:gL", ["verbose", "help", "list-tests"])
except getopt.GetoptError as err:
    print (str(err)) # will print something like "option -a not recognized"
    usage()
    sys.exit(2)

runfilter = lambda test : True
ignore_if_not_idle = False

for opt, arg in opts: 
    if opt == "-r":
        oldrunfilter = runfilter
        runfilter = lambda test: test.is_required and oldrunfilter(test)

    elif opt == "-V":
        print ("Version", version)
        sys.exit(0)
    elif opt == "-a":
        ignore_if_not_idle = True
    elif opt == "-v":
        verbose = True
    elif opt in ("-h", "--help"):
        usage()
        sys.exit()
    elif opt == '-p':
        poolfile = arg
    elif opt == '-l':
        list_tests = True
    elif opt == '-o':
        results_file = arg
    elif opt == '-B':
        benchmark_runs = int(arg)
    elif opt == '-g':
        grade_mode = True
    elif opt == '-t':
        filtered = arg.split(',')
        for _filter in filtered:
            for test in tests:
                if _filter == test.name:
                    break
            else:
                print ('Unknown test: %s. Use -l to list test names.' % _filter)
                usage()
                sys.exit()
        oldrunfilter = runfilter
        runfilter = lambda test: test.name in filtered and oldrunfilter(test)
    else:
        assert False, "unhandled option"

if list_tests:
    print ('Available tests (with applied filters):')
    print (80 * '=')
    for test in tests:
        if runfilter(test):
            print ('%s: %s' % (test.name, test.description))
    sys.exit()

def copyfile(src, dst):
    cmd = "cp %s %s" % (src, dst)
    if verbose:
        print (cmd)
    ex = os.system(cmd)
    if ex:
        sys.exit(ex)

def setup_working_directory():
    if verbose:
        print ("Creating working directory",  workdir)

    os.mkdir(workdir)

    if verbose:
        print ("Copying files")

    if not os.access(poolfile, os.R_OK):
        print ()
        print ("I cannot find %s" % poolfile)
        usage()
        sys.exit(2)

    copyfile(poolfile, workdir + "/threadpool.c")

    flist = open(filelist, 'r')
    for file in flist:
        if file.startswith("#"):
            continue
        file = file.strip()
        copyfile(src_dir + "/" + file, workdir)

    flist.close()
    if verbose:
        print ("Copying %s" % poolfile)

    os.chdir(workdir)

    if os.system("make -j 20"):
        if grade_mode:
            op = open(results_file, 'w')
            op.write(json.dumps({'error': 'did not compile'}))
            op.close()
        raise Exception("make failed, run 'make' in %s to see why" % workdir)

def check_software_engineering(objfile, allowedsymbols):
    hex = "[0-9A-Fa-f]{8,16}"
    if verbose:
        print ("Performing some checks that %s conforms to accepted software engineering practice..." % objfile)

    symbols = subprocess.Popen(["nm", objfile], stdout=subprocess.PIPE)\
        .communicate()[0].decode().split("\n")

    for sym in symbols:
        if sym == "" or re.match("\s+U (\S+)", sym):
            continue

        m = re.match(hex + " (\S) (\S+)", sym)
        if not m:
            raise Exception("unexpected line in nm:\n" + sym)

        if m.group(1).islower():    # local symbols are fine
            continue

        if m.group(1) == "T":
            if m.group(2) in allowedsymbols:
                continue
            
            if grade_mode:
                op = open(results_file, 'w')
                op.write(json.dumps({'error': 'defines global function %s' % m.group(2)}))
                op.close()
            raise Exception(("%s defines global function '%s'\n"
                +"allowed functions are: %s") % (objfile, m.group(2), str(allowedsymbols)))

        if grade_mode:
            op = open(results_file, 'w')
            op.write(json.dumps({'error': 'defines global symbol %s' % m.group(2)}))
            op.close()
        raise Exception(("%s must not define any global or static variables"
                +", but you define: %s") % (objfile, m.group(2)))

allowedsymbols = [ "future_free", "future_get",
                   "thread_pool_new", "thread_pool_shutdown_and_destroy", 
                   "thread_pool_submit" ] 

#
# build it (like check.py)
#

def count_number_of_processes():
    proc = subprocess.Popen(["ps", "ux", "-L"], stdout=subprocess.PIPE, stderr=subprocess.PIPE)

    stdout, stderr = proc.communicate()
    # -2 for ps and header
    return len(stdout.decode().strip().split('\n')) - 2

def get_load_average():
    """
    Returns tuple nproc, loadavg where nproc is the current number of
    running threads (minus 1) and loadavg is the load average
    """
    # 0.57 0.65 0.54 1/426 28121
    with open("/proc/loadavg") as f:
        c = f.read().strip()
        m = re.match(r'(\S+) \S+ \S+ (\d+)/\d+ \d+', c)
        load = float(m.groups()[0])
        nprocs = int(m.groups()[1])

    return nprocs - 1, load

def wait_for_load_to_go_down():
    while True:
        time.sleep(0.2)
        nprocs, load = get_load_average()
        if nprocs == 0 and load < 1.0:
            break

        print ("Warning. There are other %d processes running on this machine, loadavg %f." % (nprocs, load))
        print ("Sleeping for 1 second.  Use the -a switch to run the benchmarks regardless.")
        time.sleep(0.8)

# run tests
#
# echo test:
#  progname  + set of inputs/cmdline parameters
#
# each should support -n <thread> for number of threads
#

#
# result: expected output + exit code(?)
#

def set_threadlimit(nthreads):
    def limit_threads():
        resource.setrlimit(resource.RLIMIT_NPROC, (nthreads, nthreads))
    return limit_threads

def run_single_test(test, run, threads):
    cmdline = ['timeout', str(run.timeout), test.command, '-n', str(threads)] + run.args
    rundata = {
        'command' : ' '.join(cmdline),
        'nthreads' : threads
    }
    def addrundata(d):
        for k, v in d.items():
            rundata[k] = v

    if not silent:
        print ('Running:', ' '.join(cmdline), end=' ')
        sys.stdout.flush()
    infile = None
    if run.input_file:
        infile = open(run.input_file, 'r')
    # preexec_fn sets the system-wide NPROC for this user.
    # we set it to #threads + 1 (for the main thread)
    # plus existing procs
    starttime = time.time()
    preexec_fn=set_threadlimit(threads + 2 + number_of_existing_processes) if test.limit_threads \
                else (lambda : None)
    proc = subprocess.Popen(cmdline, stdout=subprocess.PIPE,
                            stderr=subprocess.PIPE, stdin=infile,
                            preexec_fn=preexec_fn)

    stdout, stderr = proc.communicate()
    if grade_mode:
        stdout = stdout[:100]
        stderr = stderr[:100]
    runningtime = time.time() - starttime

    if infile:
        infile.close()

    signames = dict((k, v) for v, k in signal.__dict__.items() if v.startswith('SIG'))
    signum = proc.returncode - 128
    if proc.returncode < 0:
        signum = -proc.returncode
    if proc.returncode >= 128 or proc.returncode < 0 and signum in signames:
        timeoutmsg = ''
        if runningtime >= run.timeout:
            timeoutmsg = '\nProgram ran and most likely timed out at %.3fs' % (runningtime)
        error = """\
        Program terminated with signal %d (%s) %s
        --------------------------------------------
        Program output:
        %s
        StdErr output:
        %s
        """ % (signum, signames[signum], timeoutmsg, stdout.decode(), stderr.decode())
        addrundata({
            'error': error
        })
        if not silent:
            print ('[ ]')
            if verbose or grade_mode:
                print (error)

    elif proc.returncode > 0:
        # non-zero exit code
        timeoutmsg = ''
        if proc.returncode == 124:
            timeoutmsg = '\nProgram ran and most likely timed out at %.3fs' % (runningtime)

        error = """\
        Program exited with non-zero exit status %d. %s
        --------------------------------------------
        Program output:
        %s
        StdErr output:
        %s
        """ % (proc.returncode, timeoutmsg, stdout.decode(), stderr.decode())

        addrundata({
            'error': error
        })
        if not silent:
            print ('[ ]')
            if verbose or grade_mode:
                print (error)

    else:
        if not silent:
            print ('[+]')

        outfile = 'runresult.%d.json' % (proc.pid)
        addrundata({'stdout': stdout.decode()})
        if len(stderr) > 0: 
            addrundata({'stderr': stderr.decode()})

        if not os.access(outfile, os.R_OK):
            addrundata({
                'error': 'The benchmark did not create the expected result file %s' % outfile
            })
        else:
            with open(outfile, 'r') as f:
                data = f.read()

            addrundata(json.loads(data))
            os.unlink(outfile)
    return rundata
    
def average_run(runs):
    data = {
        'nthreads': runs[0]['nthreads'],
        'command': runs[0]['command']
    }
    totalrtime = 0.0
    totalstime = 0.0
    totalutime = 0.0
    error = None
    passed = 0
    for run in runs:
        if 'error' in run:
            error = run['error']
        else:
            passed += 1
            totalrtime += run['realtime']
            totalstime += run['ru_stime']
            totalutime += run['ru_utime']

    if error or passed != len(runs):
        data['error'] = error
        data['run_count'] = len(runs)
        data['passed'] = passed
    else:
        data['realtime'] = totalrtime / len(runs)
        data['cputime'] = (totalstime + totalutime) / len(runs)
        data['ru_stime'] = totalstime / len(runs)
        data['ru_utime'] = totalutime / len(runs)
        data['run_count'] = len(runs)
        data['passed'] = passed

    return data

def benchmark_speedup(data, testname):
    if 'realtime' not in data or 'cputime' not in data:
        return
    serial_time = benchmark_times[testname]
    data['speedup'] = serial_time / data['realtime']
    data['cpu_overuse'] = data['cputime'] / data['realtime']

def run_tests(tests):
    results = defaultdict(dict)

    summary = {}
    for test in tests:
        if not runfilter(test):
            if verbose:
                print ('Skipping test: ' + test.description)
            continue

        if not silent:
            print ('')
            print ('Starting test: ' + test.description)
            print ('=' * 80)

        results[test.name] = {}
        for run in test.runs:
            perthreadresults = []
            results[test.name][run.name] = perthreadresults

            for threads in run.thread_count:
                if grade_mode:
                    repeats = benchmark_runs if test.is_required or run.is_benchmarked else 1
                    runs = []
                    for repeat in range(repeats):
                        runs.append(run_single_test(test, run, threads))
                    rundata = average_run(runs)
                    rundata['runs'] = runs
                    if run.is_benchmarked:
                        benchmark_speedup(rundata, run.name)
                    perthreadresults.append(rundata)
                else:
                    runs = [run_single_test(test, run, threads)]
                    rundata = average_run(runs)
                    rundata['runs'] = runs
                    perthreadresults.append(rundata)
    return results

def print_results(results):
    print (json.dumps(results, indent = 4, sort_keys = True, separators = (',', ': ')))
    # try pretting printing errors for better readability
    try:
        def print_info(o):
            for k in ['error', 'stderr', 'stdout']:
                if k in o:
                    print (o[k])

        for test, r in results.items():
            for rr in r.values():
                for o in rr:
                    print_info(o)
                    if 'runs' in o:
                        map(print_info, o['runs'])

    except Exception as e:
        print(e)
        pass


def write_results_to_json(filename):
    jfile = open(results_file, "w")
    print (json.dumps(results, indent = 4, sort_keys = True, separators = (',', ': ')), file=jfile)
    jfile.close()

def find_thread_run(perthreadresults, threadcount):
    return next(filter(lambda result: result['nthreads'] == threadcount, perthreadresults), None)

def print_grade_table(results, tests):
    # report the results of running each tests with 
    thread_headers = [1, 2, 4, 8, 16, 32]
    print ('')
    print ('Test name:' + (16 * ' ') + ''.join(map(lambda x: '%-10s' % str(x), thread_headers)))
    print ('='*80)
    minimum_requirements = True
    for test in tests:
        if not runfilter(test) and test.is_required:
            if not silent:
                print ('WARNING: Skipping minimum requirement test (%s), will not say you passed!' % test.name)
            minimum_requirements = False

        if not runfilter(test):
            continue
        if not test.name in results:
            print ('%s: could not find test data!' % test.name)
        res = results[test.name]
        print ('%s:' % test.name.upper() + '  ' + test.description)
    
        passed = True
        for run in test.runs:
            statuses = []
            for threads in thread_headers:
                thread_run = find_thread_run(res[run.name], threads)
                if not thread_run:
                    statuses.append('')  # MISSING
                elif 'error' in thread_run:
                    passed = False
                    statuses.append('[ ]')
                elif run.is_benchmarked:
                    statuses.append('[%.3fs]' % thread_run['realtime'])
                else:
                    statuses.append('[X]')

            print ('  %-23s' % (run.name) + ''.join(map(lambda x: '%-10s' % x, statuses)))
        
        if not passed and test.is_required:
            minimum_requirements = False

    print ('='*80)
    if minimum_requirements:
        print ('You have met minimum requirements, your performance score will count.')
    else:
        print ('You did not meet minimum requirements; your performance score will be zero.')


setup_working_directory()
check_software_engineering("threadpool.o", allowedsymbols)
number_of_existing_processes = count_number_of_processes()
if verbose:
    print ("There are %d process currently running for user %s" % (number_of_existing_processes, os.getlogin()))

if not ignore_if_not_idle:
    wait_for_load_to_go_down()
    
results = run_tests(tests)
if verbose:
    print_results(results)
if not silent:
    print_grade_table(results, tests)

write_results_to_json(results_file)
print ("Wrote full results to %s/%s" % (workdir, results_file))