Advertisement
Not a member of Pastebin yet?
Sign Up,
it unlocks many cool features!
- from twisted.internet import reactor
- from twisted.internet.defer import gatherResults
- from twisted.internet.threads import deferToThread
- from twisted.trial.unittest import TestCase
- from time import sleep
- class Case(TestCase):
- def test_threads(self):
- return start()
- def counter(x):
- c = 0
- while True:
- c += x
- print c
- sleep(1)
- def start():
- print reactor.getThreadPool().started
- d1 = deferToThread(counter, 2)
- d2 = deferToThread(counter, 5)
- return gatherResults([d1, d2], consumeErrors=True)
- if __name__ == '__main__':
- reactor.callWhenRunning(start)
- reactor.run()
Advertisement
Add Comment
Please, Sign In to add comment
Advertisement