xdelta3-regtest.py

Linux下一个可以比较二进制文件的工具xdelta3.0u的源码。

    print "rcsfiles: rcsfiles %d; subdirs %d; others %d; skipped %d" % (len(rcsf.rcsfiles),
                                                                        len(rcsf.subdirs),
                                                                        len(rcsf.others),
                                                                        len(rcsf.skipped))
    print StatList([x.rcssize for x in rcsf.rcsfiles], "rcssize").str
    print StatList([x.totrev for x in rcsf.rcsfiles], "totrev").str
    return rcsf
#end

class SampleDataTest:
    def __init__(self, dirs):
        self.pairs = []
        while dirs:
            d = dirs[0]
            dirs = dirs[1:]
            l = os.listdir(d)
            files = []
            for e in l:
                p = os.path.join(d, e)
                if os.path.isdir(p):
                    dirs.append(p)
                else:
                    files.append(p)
                #end
            #end
            if len(files) > 1:
                files.sort()
                for x in xrange(len(files) - 1):
                    self.pairs.append((files[x], files[x+1],
                                       '%s-%s' % (files[x], files[x+1])))
                #end
            #end
        #end
    #end

    def Generator(self):
        return lambda rand: rand.choice(self.pairs)
    #end
#end

# configs are represented as a list of values,
# program takes a list of strings:
def ConfigToArgs(config):
    args = [ '-C',
             ','.join([str(x) for x in config[0:SOFT_CONFIG_CNT]])]
    for i in range(SOFT_CONFIG_CNT, len(CONFIG_ORDER)):
        key = CONFIG_ARGMAP[CONFIG_ORDER[i]]
        val = config[i]
        if val == 'true' or val == 'false':
            if val == 'true':
                args.append('%s' % key)
            #end
        else:
            args.append('%s=%s' % (key, val))
        #end
    #end
    return args
#end

#
class RandomTest:
    def __init__(self, tnum, tinput, config, syntuple = None):
        self.mytinput = tinput[2]
        self.myconfig = config
        self.tnum = tnum

        if syntuple != None:
            self.runtime = syntuple[0]
            self.compsize = syntuple[1]
            self.decodetime = None
        else:
            args = ConfigToArgs(config)
            result = TimedTest(tinput[1], tinput[0], Xdelta3Runner(args))

            self.runtime = result.encode_time.mean
            self.compsize = result.encode_size
            self.decodetime = result.decode_time.mean
        #end

        self.score = None
        self.time_pos = None
        self.size_pos = None
        self.score_pos = None
    #end

    def __str__(self):
        decodestr = ' %.6f' % self.decodetime
        return 'time %.6f%s size %d%s << %s >>%s' % (
            self.time(), ((self.time_pos != None) and (" (%s)" % self.time_pos) or ""),
            self.size(), ((self.size_pos != None) and (" (%s)" % self.size_pos) or ""),
            c2str(self.config()),
            decodestr)
    #end

    def time(self):
        return self.runtime
    #end

    def size(self):
        return self.compsize
    #end

    def config(self):
        return self.myconfig
    #end

    def score(self):
        return self.score
    #end

    def tinput(self):
        return self.mytinput
    #end
#end

def PosInAlist(l, e):
    for i in range(0, len(l)):
        if l[i][1] == e:
            return i;
        #end
    #end
    return -1
#end

# Generates a set of num_results test configurations, given the list of
# retest-configs.
def RandomTestConfigs(rand, input_configs, num_results):

    outputs = input_configs[:]
    have_set = dict([(c,c) for c in input_configs])

    # Compute a random configuration
    def RandomConfig():
        config = []
        cmap = {}
        for key in CONFIG_ORDER:
            val = cmap[key] = (INPUT_SPEC(rand)[key])(cmap)
            config.append(val)
        #end
        return tuple(config)
    #end

    while len(outputs) < num_results:
        newc = None
        for i in xrange(100):
            c = RandomConfig()
            if have_set.has_key(c):
                continue
            #end
            have_set[c] = c
            newc = c
            break
        if newc is None:
            print 'stopped looking for configs at %d' % len(outputs)
            break
        #end
        outputs.append(c)
    #end
    outputs.sort()
    return outputs
#end

def RunTestLoop(rand, generator, rounds):
    configs = []
    for rnum in xrange(rounds):
        configs = RandomTestConfigs(rand, configs, MAX_RESULTS)
        tinput = generator(rand)
        tests = []
        for x in xrange(len(configs)):
            t = RandomTest(x, tinput, configs[x])
            print 'Round %d test %d: %s' % (rnum, x, t)
            tests.append(t)
        #end
        results = ScoreTests(tests)

        for r in results:
            c = r.config()
            if not test_all_config_results.has_key(c):
                test_all_config_results[c] = [r]
            else:
                test_all_config_results[c].append(r)
            #end
        #end

        GraphResults('expt%d' % rnum, results)
        GraphSummary('sum%d' % rnum, results)

        # re-test some fraction
        configs = [r.config() for r in results[0:int(MAX_RESULTS * KEEP_P)]]
    #end
#end

# TODO: cleanup
test_all_config_results = {}

def ScoreTests(results):
    scored = []
    timed = []
    sized = []

    t_min = float(min([test.time() for test in results]))
    #t_max = float(max([test.time() for test in results]))
    s_min = float(min([test.size() for test in results]))
    #s_max = float(max([test.size() for test in results]))

    for test in results:

        # Hyperbolic function. Smaller scores still better
        red = 0.999  # minimum factors for each dimension are 1/1000
        test.score = ((test.size() - s_min * red) *
                      (test.time() - t_min * red))

        scored.append((test.score, test))
        timed.append((test.time(), test))
        sized.append((test.size(), test))
    #end

    scored.sort()
    timed.sort()
    sized.sort()

    best_by_size = []
    best_by_time = []

    pos = 0
    for (score, test) in scored:
        pos += 1
        test.score_pos = pos
    #end

    scored = [x[1] for x in scored]

    for test in scored:
        test.size_pos = PosInAlist(sized, test)
        test.time_pos = PosInAlist(timed, test)
    #end

    for test in scored:
        c = test.config()
        s = 0.0
        print 'H-Score: %0.9f %s' % (test.score, test)
    #end

    return scored
#end

def GraphResults(desc, results):
    f = open("data-%s.csv" % desc, "w")
    for r in results:
        f.write("%0.9f\t%d\t# %s\n" % (r.time(), r.size(), r))
    #end
    f.close()
    os.system("./plot.sh data-%s.csv plot-%s.jpg" % (desc, desc))
#end

def GraphSummary(desc, results_ignore):
    test_population = 0
    config_ordered = []

    # drops duplicate test/config pairs (TODO: don't retest them)
    for config, cresults in test_all_config_results.items():
        input_config_map = {}
        uniq = []
        for test in cresults:
            assert test.config() == config
            test_population += 1
            key = test.tinput()
            if not input_config_map.has_key(key):
                input_config_map[key] = {}
            #end
            if input_config_map[key].has_key(config):
                print 'skipping repeat test %s vs. %s' % (input_config_map[key][config], test)
                continue
            #end
            input_config_map[key][config] = test
            uniq.append(test)
        #end
        config_ordered.append(uniq)
    #end

    # sort configs descending by number of tests
    config_ordered.sort(lambda x, y: len(y) - len(x))

    print 'population %d: %d configs %d results' % \
          (test_population,
           len(config_ordered),
           len(config_ordered[0]))

    if config_ordered[0] == 1:
        return
    #end

    # a map from test-key to test-list w/ various configs
    input_set = {}
    osize = len(config_ordered)

    for i in xrange(len(config_ordered)):
        config = config_ordered[i][0].config()
        config_tests = config_ordered[i]

        #print '%s has %d tested inputs' % (config, len(config_tests))

        if len(input_set) == 0:
            input_set = dict([(t.tinput(), [t]) for t in config_tests])
            continue
        #end

        # a map from test-key to test-list w/ various configs
        update_set = {}
        for r in config_tests:
            t = r.tinput()
            if input_set.has_key(t):
                update_set[t] = input_set[t] + [r]
            else:
                #print 'config %s does not have test %s' % (config, t)
                pass
            #end
        #end

        if len(update_set) <= 1:
            break
        #end

        input_set = update_set

        # continue if there are more w/ the same number of inputs
        if i < (len(config_ordered) - 1) and \
           len(config_ordered[i + 1]) == len(config_tests):
            continue
        #end

        # synthesize results for multi-test inputs
        config_num = None

        # map of config to sum(various test-keys)
        smap = {}
        for (key, tests) in input_set.items():
            if config_num == None:
                # config_num should be the same in all elements
                config_num = len(tests)
                smap = dict([(r.config(),
                              (r.time(),
                               r.size()))
                             for r in tests])
            else:
                # compuate the per-config sum of time/size
                assert config_num == len(tests)
                smap = dict([(r.config(),
                              (smap[r.config()][0] + r.time(),
                               smap[r.config()][1] + r.size()))
                             for r in tests])
            #end
        #end

        if config_num == 1:
            continue
        #end

        if len(input_set) == osize:
            break
        #end

        summary = '%s-%d' % (desc, len(input_set))
        osize = len(input_set)

        print 'generate %s w/ %d configs' % (summary, config_num)
        syn = [RandomTest(0, (None, None, summary), config,
                          syntuple = (smap[config][0], smap[config][1]))
               for config in smap.keys()]
        syn = ScoreTests(syn)
        #print 'smap is %s' % (smap,)
        #print 'syn is %s' % (' and '.join([str(x) for x in syn]))
        GraphResults(summary, syn)
    #end
#end

if __name__ == "__main__":
    try:
        RunCommand(['rm', '-rf', TMPDIR])
        os.mkdir(TMPDIR)

        rcsf = GetTestRcsFiles()
        #generator = rcsf.Generator()

        #sample = SampleDataTest([SAMPLEDIR])
        #generator = sample.Generator()

        #rand = random.Random(135135135135135)
        #RunTestLoop(rand, generator, TEST_ROUNDS)

        #RunSpeedTest()

        #x3r = rcsf.AllPairsByDate(Xdelta3RunClass(['-9']))
        x3r = rcsf.AllPairsByDate(Xdelta3RunClass(['-9', '-S', 'djw']))
        x3r = rcsf.AllPairsByDate(Xdelta3RunClass(['-1', '-S', 'djw']))
        #x3r = rcsf.AllPairsByDate(Xdelta3RunClass(['-9', '-T']))

        #x1r = rcsf.AllPairsByDate(Xdelta1RunClass())

    except CommandError:
        pass
    else:
        RunCommand(['rm', '-rf', TMPDIR])
        pass
    #end
#end