validate_pihole.py 文件源码

def h0_opt_test_stresstest(IP=None, stress_count=2000, stress_threads=50, **kwargs):
    '''
    Throw {stress_count} domains at the pihole via {stress_threads} threads
    '''

    from joblib import Parallel, delayed

    top_array = open('topsites.txt').read().splitlines()
    random.shuffle(top_array)

    results = Parallel(n_jobs=stress_threads, backend='threading' )(delayed(dns_stress)(IP, site) for site in top_array[:stress_count])
    good = sum( 1 for (a,b) in results if a == 'good' )
    numbers = [ b for (a,b) in results if a == 'good' ]
    bad = sum( 1 for (a,b) in results if a == 'bad' )
    vmin = min(numbers)*1000
    vmax = max(numbers)*1000
    vavg = sum(numbers)*1000//len(numbers)
    vstd = (sum(((n*1000) - vavg) ** 2 for n in numbers) / len(numbers)) ** .5

    return not bad or (good/bad)>0.05, "{good}/{bad} min {vmin:.2f}ms avg {vavg:.2f}ms max {vmax:.2f}ms std {vstd:.2f}ms".format(**locals())