test_threadpool.py 文件源码

def testPersistence(self):
        tp = threadpool.ThreadPool(7, 20)
        tp.start()

        # XXX Sigh - race condition: start should return a Deferred
        # which fires when all the workers it started have fully
        # started up.
        time.sleep(0.1)

        self.assertEquals(len(tp.threads), 7)
        self.assertEquals(tp.min, 7)
        self.assertEquals(tp.max, 20)

        # check that unpickled threadpool has same number of threads
        s = pickle.dumps(tp)
        tp2 = pickle.loads(s)
        tp2.start()

        # XXX As above
        time.sleep(0.1)

        self.assertEquals(len(tp2.threads), 7)
        self.assertEquals(tp2.min, 7)
        self.assertEquals(tp2.max, 20)

        tp.stop()
        tp2.stop()