bazarr/libs/tqdm/tests/tests_perf.py

from __future__ import print_function, division

from nose.plugins.skip import SkipTest

from contextlib import contextmanager

import sys
from time import sleep, time

from tqdm import trange
from tqdm import tqdm

from tests_tqdm import with_setup, pretest, posttest, StringIO, closing, _range

# Use relative/cpu timer to have reliable timings when there is a sudden load
try:
    from time import process_time
except ImportError:
    from time import clock
    process_time = clock


def get_relative_time(prevtime=0):
    return process_time() - prevtime


def cpu_sleep(t):
    """Sleep the given amount of cpu time"""
    start = process_time()
    while (process_time() - start) < t:
        pass


def checkCpuTime(sleeptime=0.2):
    """Check if cpu time works correctly"""
    if checkCpuTime.passed:
        return True
    # First test that sleeping does not consume cputime
    start1 = process_time()
    sleep(sleeptime)
    t1 = process_time() - start1

    # secondly check by comparing to cpusleep (where we actually do something)
    start2 = process_time()
    cpu_sleep(sleeptime)
    t2 = process_time() - start2

    if abs(t1) < 0.0001 and (t1 < t2 / 10):
        return True
    raise SkipTest


checkCpuTime.passed = False


@contextmanager
def relative_timer():
    start = process_time()

    def elapser():
        return process_time() - start

    yield lambda: elapser()
    spent = process_time() - start

    def elapser():  # NOQA
        return spent


def retry_on_except(n=3):
    def wrapper(fn):
        def test_inner():
            for i in range(1, n + 1):
                try:
                    checkCpuTime()
                    fn()
                except SkipTest:
                    if i >= n:
                        raise
                else:
                    return

        test_inner.__doc__ = fn.__doc__
        return test_inner

    return wrapper


class MockIO(StringIO):
    """Wraps StringIO to mock a file with no I/O"""

    def write(self, data):
        return


def simple_progress(iterable=None, total=None, file=sys.stdout, desc='',
                    leave=False, miniters=1, mininterval=0.1, width=60):
    """Simple progress bar reproducing tqdm's major features"""
    n = [0]  # use a closure
    start_t = [time()]
    last_n = [0]
    last_t = [0]
    if iterable is not None:
        total = len(iterable)

    def format_interval(t):
        mins, s = divmod(int(t), 60)
        h, m = divmod(mins, 60)
        if h:
            return '{0:d}:{1:02d}:{2:02d}'.format(h, m, s)
        else:
            return '{0:02d}:{1:02d}'.format(m, s)

    def update_and_print(i=1):
        n[0] += i
        if (n[0] - last_n[0]) >= miniters:
            last_n[0] = n[0]

            if (time() - last_t[0]) >= mininterval:
                last_t[0] = time()  # last_t[0] == current time

                spent = last_t[0] - start_t[0]
                spent_fmt = format_interval(spent)
                rate = n[0] / spent if spent > 0 else 0
                if 0.0 < rate < 1.0:
                    rate_fmt = "%.2fs/it" % (1.0 / rate)
                else:
                    rate_fmt = "%.2fit/s" % rate

                frac = n[0] / total
                percentage = int(frac * 100)
                eta = (total - n[0]) / rate if rate > 0 else 0
                eta_fmt = format_interval(eta)

                # bar = "#" * int(frac * width)
                barfill = " " * int((1.0 - frac) * width)
                bar_length, frac_bar_length = divmod(int(frac * width * 10), 10)
                bar = '#' * bar_length
                frac_bar = chr(48 + frac_bar_length) if frac_bar_length \
                    else ' '

                file.write("\r%s %i%%|%s%s%s| %i/%i [%s<%s, %s]" %
                           (desc, percentage, bar, frac_bar, barfill, n[0],
                            total, spent_fmt, eta_fmt, rate_fmt))

                if n[0] == total and leave:
                    file.write("\n")
                file.flush()

    def update_and_yield():
        for elt in iterable:
            yield elt
            update_and_print()

    update_and_print(0)
    if iterable is not None:
        return update_and_yield()
    else:
        return update_and_print


def assert_performance(thresh, name_left, time_left, name_right, time_right):
    """raises if time_left > thresh * time_right"""
    if time_left > thresh * time_right:
        raise ValueError(
            ('{name[0]}: {time[0]:f}, '
             '{name[1]}: {time[1]:f}, '
             'ratio {ratio:f} > {thresh:f}').format(
                name=(name_left, name_right),
                time=(time_left, time_right),
                ratio=time_left / time_right, thresh=thresh))


@with_setup(pretest, posttest)
@retry_on_except()
def test_iter_overhead():
    """Test overhead of iteration based tqdm"""

    total = int(1e6)

    with closing(MockIO()) as our_file:
        a = 0
        with trange(total, file=our_file) as t:
            with relative_timer() as time_tqdm:
                for i in t:
                    a += i
        assert a == (total * total - total) / 2.0

        a = 0
        with relative_timer() as time_bench:
            for i in _range(total):
                a += i
                our_file.write(a)

    assert_performance(6, 'trange', time_tqdm(), 'range', time_bench())


@with_setup(pretest, posttest)
@retry_on_except()
def test_manual_overhead():
    """Test overhead of manual tqdm"""

    total = int(1e6)

    with closing(MockIO()) as our_file:
        with tqdm(total=total * 10, file=our_file, leave=True) as t:
            a = 0
            with relative_timer() as time_tqdm:
                for i in _range(total):
                    a += i
                    t.update(10)

        a = 0
        with relative_timer() as time_bench:
            for i in _range(total):
                a += i
                our_file.write(a)

    assert_performance(6, 'tqdm', time_tqdm(), 'range', time_bench())


def worker(total, blocking=True):
    def incr_bar(x):
        with closing(StringIO()) as our_file:
            for _ in trange(
                    total, file=our_file,
                    lock_args=None if blocking else (False,),
                    miniters=1, mininterval=0, maxinterval=0):
                pass
        return x + 1
    return incr_bar


@with_setup(pretest, posttest)
@retry_on_except()
def test_lock_args():
    """Test overhead of nonblocking threads"""
    try:
        from concurrent.futures import ThreadPoolExecutor
        from threading import RLock
    except ImportError:
        raise SkipTest
    import sys

    total = 8
    subtotal = 1000

    tqdm.set_lock(RLock())
    with ThreadPoolExecutor(total) as pool:
        sys.stderr.write('block ... ')
        sys.stderr.flush()
        with relative_timer() as time_tqdm:
            res = list(pool.map(worker(subtotal, True), range(total)))
            assert sum(res) == sum(range(total)) + total
        sys.stderr.write('noblock ... ')
        sys.stderr.flush()
        with relative_timer() as time_noblock:
            res = list(pool.map(worker(subtotal, False), range(total)))
            assert sum(res) == sum(range(total)) + total

    assert_performance(0.2, 'noblock', time_noblock(), 'tqdm', time_tqdm())


@with_setup(pretest, posttest)
@retry_on_except()
def test_iter_overhead_hard():
    """Test overhead of iteration based tqdm (hard)"""

    total = int(1e5)

    with closing(MockIO()) as our_file:
        a = 0
        with trange(total, file=our_file, leave=True, miniters=1,
                    mininterval=0, maxinterval=0) as t:
            with relative_timer() as time_tqdm:
                for i in t:
                    a += i
        assert a == (total * total - total) / 2.0

        a = 0
        with relative_timer() as time_bench:
            for i in _range(total):
                a += i
                our_file.write(("%i" % a) * 40)

    assert_performance(85, 'trange', time_tqdm(), 'range', time_bench())


@with_setup(pretest, posttest)
@retry_on_except()
def test_manual_overhead_hard():
    """Test overhead of manual tqdm (hard)"""

    total = int(1e5)

    with closing(MockIO()) as our_file:
        t = tqdm(total=total * 10, file=our_file, leave=True, miniters=1,
                 mininterval=0, maxinterval=0)
        a = 0
        with relative_timer() as time_tqdm:
            for i in _range(total):
                a += i
                t.update(10)

        a = 0
        with relative_timer() as time_bench:
            for i in _range(total):
                a += i
                our_file.write(("%i" % a) * 40)

    assert_performance(85, 'tqdm', time_tqdm(), 'range', time_bench())


@with_setup(pretest, posttest)
@retry_on_except()
def test_iter_overhead_simplebar_hard():
    """Test overhead of iteration based tqdm vs simple progress bar (hard)"""

    total = int(1e4)

    with closing(MockIO()) as our_file:
        a = 0
        with trange(total, file=our_file, leave=True, miniters=1,
                    mininterval=0, maxinterval=0) as t:
            with relative_timer() as time_tqdm:
                for i in t:
                    a += i
        assert a == (total * total - total) / 2.0

        a = 0
        s = simple_progress(_range(total), file=our_file, leave=True,
                            miniters=1, mininterval=0)
        with relative_timer() as time_bench:
            for i in s:
                a += i

    assert_performance(
        5, 'trange', time_tqdm(), 'simple_progress', time_bench())


@with_setup(pretest, posttest)
@retry_on_except()
def test_manual_overhead_simplebar_hard():
    """Test overhead of manual tqdm vs simple progress bar (hard)"""

    total = int(1e4)

    with closing(MockIO()) as our_file:
        t = tqdm(total=total * 10, file=our_file, leave=True, miniters=1,
                 mininterval=0, maxinterval=0)
        a = 0
        with relative_timer() as time_tqdm:
            for i in _range(total):
                a += i
                t.update(10)

        simplebar_update = simple_progress(
            total=total * 10, file=our_file, leave=True, miniters=1,
            mininterval=0)
        a = 0
        with relative_timer() as time_bench:
            for i in _range(total):
                a += i
                simplebar_update(10)

    assert_performance(
        5, 'tqdm', time_tqdm(), 'simple_progress', time_bench())