threading_latency.py 文件源码

def testCallFromThread(self):
        for i in range(self.numRounds):
            reactor.callInThread(self.tcmf_2, time.time())
            self.wait()
        assert len(self.in_times) == len(self.from_times)
        assert len(self.in_times) == self.numRounds
        self.printResult()