test_ioloop_construction_and_retrieve_in_multithread.py

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()