Skip to content

Instantly share code, notes, and snippets.

Show Gist options
  • Save justdoit0823/91a0d3599a8aceb10367728a3665e2a3 to your computer and use it in GitHub Desktop.
Save justdoit0823/91a0d3599a8aceb10367728a3665e2a3 to your computer and use it in GitHub Desktop.
import asyncio
import threading
from tornado.ioloop import IOLoop
def non_main_thread_task(g_io_loop):
assert IOLoop.current(False) is None, "invalid current thread's ioloop object."
assert IOLoop.current() is g_io_loop, "invalid current thread's ioloop object."
io_loop = IOLoop()
assert g_io_loop is not io_loop, "invalid new ioloop object."
assert IOLoop.current() is io_loop, "IOLoop.current returns invalid ioloop object."
try:
asyncio.get_event_loop()
except RuntimeError:
pass
else:
assert False, "invalid current's asyncio ioloop object."
a_io_loop = asyncio.new_event_loop()
try:
asyncio.get_event_loop()
except RuntimeError:
pass
else:
assert False, "invalid current's asyncio ioloop object."
asyncio.set_event_loop(a_io_loop)
try:
ag_io_loop = asyncio.get_event_loop()
except RuntimeError:
assert False, "get_event_loop returns invalid ioloop object."
else:
assert a_io_loop is ag_io_loop, "invalid current's asyncio ioloop object."
print('non main thread is ok...')
def main_thread_task():
assert IOLoop.current(False) is None, "invalid current thread's ioloop object."
io_loop = IOLoop()
g_io_loop = IOLoop.instance()
assert io_loop is not g_io_loop, "current ioloop object is the global ioloop object."
try:
asyncio.get_event_loop()
except RuntimeError:
assert False, "incorrect asyncio event loop policy."
return g_io_loop
def main():
g_io_loop = main_thread_task()
thread_list = []
for idx in range(4):
non_main_thread = threading.Thread(target=non_main_thread_task, args=(g_io_loop,))
thread_list.append(non_main_thread)
non_main_thread.start()
for thread in thread_list:
thread.join()
print('Everything is ok...')
if __name__ == '__main__':
main()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment