test_threadpool.py 文件源码

def __init__(self, testCase, *args, **kwargs):
        """
        Create a L{PoolHelper}.

        @param testCase: a test case attached to this helper.

        @type args: The arguments passed to a L{threadpool.ThreadPool}.

        @type kwargs: The arguments passed to a L{threadpool.ThreadPool}
        """
        coordinator, self.performCoordination = createMemoryWorker()
        self.workers = []
        def newWorker():
            self.workers.append(createMemoryWorker())
            return self.workers[-1][0]
        self.threadpool = MemoryPool(coordinator, testCase.fail, newWorker,
                                     *args, **kwargs)