bazarr/libs/tqdm/tests/tests_synchronisation.py

from __future__ import division
from tqdm import tqdm, trange, TMonitor
from tests_tqdm import with_setup, pretest, posttest, SkipTest, \
    StringIO, closing
from tests_tqdm import DiscreteTimer, cpu_timify
from tests_perf import retry_on_except

import sys
from time import sleep
from threading import Event


class FakeSleep(object):
    """Wait until the discrete timer reached the required time"""
    def __init__(self, dtimer):
        self.dtimer = dtimer

    def sleep(self, t):
        end = t + self.dtimer.t
        while self.dtimer.t < end:
            sleep(0.0000001)  # sleep a bit to interrupt (instead of pass)


class FakeTqdm(object):
    _instances = []


def make_create_fake_sleep_event(sleep):
    def wait(self, timeout=None):
        if timeout is not None:
            sleep(timeout)
        return self.is_set()

    def create_fake_sleep_event():
        event = Event()
        event.wait = wait
        return event

    return create_fake_sleep_event


def incr(x):
    return x + 1


def incr_bar(x):
    with closing(StringIO()) as our_file:
        for _ in trange(x, lock_args=(False,), file=our_file):
            pass
    return incr(x)


@with_setup(pretest, posttest)
def test_monitor_thread():
    """Test dummy monitoring thread"""
    maxinterval = 10

    # Setup a discrete timer
    timer = DiscreteTimer()
    TMonitor._time = timer.time
    # And a fake sleeper
    sleeper = FakeSleep(timer)
    TMonitor._event = make_create_fake_sleep_event(sleeper.sleep)

    # Instanciate the monitor
    monitor = TMonitor(FakeTqdm, maxinterval)
    # Test if alive, then killed
    assert monitor.report()
    monitor.exit()
    timer.sleep(maxinterval * 2)  # need to go out of the sleep to die
    assert not monitor.report()
    # assert not monitor.is_alive()  # not working dunno why, thread not killed
    del monitor


@with_setup(pretest, posttest)
def test_monitoring_and_cleanup():
    """Test for stalled tqdm instance and monitor deletion"""
    # Note: should fix miniters for these tests, else with dynamic_miniters
    # it's too complicated to handle with monitoring update and maxinterval...
    maxinterval = 2

    total = 1000
    # Setup a discrete timer
    timer = DiscreteTimer()
    # And a fake sleeper
    sleeper = FakeSleep(timer)
    # Setup TMonitor to use the timer
    TMonitor._time = timer.time
    TMonitor._event = make_create_fake_sleep_event(sleeper.sleep)
    # Set monitor interval
    tqdm.monitor_interval = maxinterval
    with closing(StringIO()) as our_file:
        with tqdm(total=total, file=our_file, miniters=500, mininterval=0.1,
                  maxinterval=maxinterval) as t:
            cpu_timify(t, timer)
            # Do a lot of iterations in a small timeframe
            # (smaller than monitor interval)
            timer.sleep(maxinterval / 2)  # monitor won't wake up
            t.update(500)
            # check that our fixed miniters is still there
            assert t.miniters == 500
            # Then do 1 it after monitor interval, so that monitor kicks in
            timer.sleep(maxinterval * 2)
            t.update(1)
            # Wait for the monitor to get out of sleep's loop and update tqdm..
            timeend = timer.time()
            while not (t.monitor.woken >= timeend and t.miniters == 1):
                timer.sleep(1)  # Force monitor to wake up if it woken too soon
                sleep(0.000001)  # sleep to allow interrupt (instead of pass)
            assert t.miniters == 1  # check that monitor corrected miniters
            # Note: at this point, there may be a race condition: monitor saved
            # current woken time but timer.sleep() happen just before monitor
            # sleep. To fix that, either sleep here or increase time in a loop
            # to ensure that monitor wakes up at some point.

            # Try again but already at miniters = 1 so nothing will be done
            timer.sleep(maxinterval * 2)
            t.update(2)
            timeend = timer.time()
            while t.monitor.woken < timeend:
                timer.sleep(1)  # Force monitor to wake up if it woken too soon
                sleep(0.000001)
            # Wait for the monitor to get out of sleep's loop and update tqdm..
            assert t.miniters == 1  # check that monitor corrected miniters

    # Check that class var monitor is deleted if no instance left
    tqdm.monitor_interval = 10
    assert tqdm.monitor is None


@with_setup(pretest, posttest)
def test_monitoring_multi():
    """Test on multiple bars, one not needing miniters adjustment"""
    # Note: should fix miniters for these tests, else with dynamic_miniters
    # it's too complicated to handle with monitoring update and maxinterval...
    maxinterval = 2

    total = 1000
    # Setup a discrete timer
    timer = DiscreteTimer()
    # And a fake sleeper
    sleeper = FakeSleep(timer)
    # Setup TMonitor to use the timer
    TMonitor._time = timer.time
    TMonitor._event = make_create_fake_sleep_event(sleeper.sleep)
    # Set monitor interval
    tqdm.monitor_interval = maxinterval
    with closing(StringIO()) as our_file:
        with tqdm(total=total, file=our_file, miniters=500, mininterval=0.1,
                  maxinterval=maxinterval) as t1:
            # Set high maxinterval for t2 so monitor does not need to adjust it
            with tqdm(total=total, file=our_file, miniters=500, mininterval=0.1,
                      maxinterval=1E5) as t2:
                cpu_timify(t1, timer)
                cpu_timify(t2, timer)
                # Do a lot of iterations in a small timeframe
                timer.sleep(maxinterval / 2)
                t1.update(500)
                t2.update(500)
                assert t1.miniters == 500
                assert t2.miniters == 500
                # Then do 1 it after monitor interval, so that monitor kicks in
                timer.sleep(maxinterval * 2)
                t1.update(1)
                t2.update(1)
                # Wait for the monitor to get out of sleep and update tqdm
                timeend = timer.time()
                while not (t1.monitor.woken >= timeend and t1.miniters == 1):
                    timer.sleep(1)
                    sleep(0.000001)
                assert t1.miniters == 1  # check that monitor corrected miniters
                assert t2.miniters == 500  # check that t2 was not adjusted

    # Check that class var monitor is deleted if no instance left
    tqdm.monitor_interval = 10
    assert tqdm.monitor is None


@with_setup(pretest, posttest)
def test_imap():
    """Test multiprocessing.Pool"""
    try:
        from multiprocessing import Pool
    except ImportError:
        raise SkipTest

    pool = Pool()
    res = list(tqdm(pool.imap(incr, range(100)), disable=True))
    assert res[-1] == 100


# py2: locks won't propagate to incr_bar so may cause `AttributeError`
@retry_on_except(n=3 if sys.version_info < (3,) else 1)
@with_setup(pretest, posttest)
def test_threadpool():
    """Test concurrent.futures.ThreadPoolExecutor"""
    try:
        from concurrent.futures import ThreadPoolExecutor
        from threading import RLock
    except ImportError:
        raise SkipTest

    tqdm.set_lock(RLock())
    with ThreadPoolExecutor(8) as pool:
        try:
            res = list(tqdm(pool.map(incr_bar, range(100)), disable=True))
        except AttributeError:
            if sys.version_info < (3,):
                raise SkipTest
            else:
                raise
    assert sum(res) == sum(range(1, 101))