Class Bunch

Lib/test/lock_tests.py:28–92

A bunch of threads.

class Bunch(object):
    """
    A bunch of threads.
    """
    def __init__(self, func, nthread, wait_before_exit=False):
        """
        Construct a bunch of `nthread` threads running the same function `func`.
        If `wait_before_exit` is True, the threads won't terminate until
        do_finish() is called.
        """
        self.func = func
        self.nthread = nthread
        self.started = []
        self.finished = []
        self.exceptions = []
        self._can_exit = not wait_before_exit
        self._wait_thread = None

    def task(self):
        tid = threading.get_ident()
        self.started.append(tid)
        try:
            self.func()
        except BaseException as exc:
            self.exceptions.append(exc)
        finally:
            self.finished.append(tid)
            for _ in support.sleeping_retry(support.SHORT_TIMEOUT):
                if self._can_exit:
                    break

    def __enter__(self):
        self._wait_thread = threading_helper.wait_threads_exit(support.SHORT_TIMEOUT)
        self._wait_thread.__enter__()

        try:
            for _ in range(self.nthread):
                start_new_thread(self.task, ())
        except:
            self._can_exit = True
            raise

        for _ in support.sleeping_retry(support.SHORT_TIMEOUT):
            if len(self.started) >= self.nthread:
                break

        return self

    def __exit__(self, exc_type, exc_value, traceback):
        for _ in support.sleeping_retry(support.SHORT_TIMEOUT):
            if len(self.finished) >= self.nthread:
                break

        # Wait until threads completely exit according to _thread._count()
        self._wait_thread.__exit__(None, None, None)

        # Break reference cycle
        exceptions = self.exceptions