convert threading processes to asyncio

README.md

@@ -10,4 +10,4 @@ Python packages: `gunicorn aiohttp aiohttp_jinja2`
 2. Walk the dinosaur  
 
 ## Usage  
-`gunicorn aberrant:app --bind localhost:5250 --worker-class aiohttp.GunicornWebWorker`  
+`gunicorn aberrant:init_app --bind localhost:5250 --worker-class aiohttp.GunicornWebWorker`  

aberrant.py

@@ -3,6 +3,7 @@
 The primary module for serving the Aberrant application.
 """
 import json
+import asyncio
 
 import jinja2
 import aiohttp_jinja2
@@ -14,14 +15,10 @@ import events
 import rtorrent
 import buckler_aiohttp
 
-app = web.Application(middlewares=[buckler_aiohttp.buckler_session])
-app.on_shutdown.append(rtorrent.stop_watch)
-aiohttp_jinja2.setup(app, loader=jinja2.FileSystemLoader('templates'))
-rtorrent.init()
 
 routes = web.RouteTableDef()
 
-@routes.get(config.prefix + "/", name='index')
+@routes.get('/', name='index')
 async def index(request):
 	"""The index page."""
 	torrents = rtorrent.get_active()
@@ -29,7 +26,7 @@ async def index(request):
 	return render_template("index.html", request, locals())
 
 
-@routes.get(config.prefix + '/ws', name='ws')
+@routes.get('/ws', name='ws')
 async def websocket_handler(request):
 	"""The websocket endpoint."""
 	ws = web.WebSocketResponse(heartbeat=30)
@@ -56,7 +53,39 @@ async def websocket_handler(request):
 	return ws
 
 
+async def watcher_loop():
+	try:
+		while True:
+			await rtorrent.update_torrents()
+			await asyncio.sleep(10)
+	except asyncio.CancelledError:
+		return
+
+
+async def start_background_tasks(app):
+	rtorrent.init()
+	app['watcher'] = asyncio.create_task(watcher_loop())
+
+
+async def cleanup_background_tasks(app):
+	app['watcher'].cancel()
+	await app['watcher']
+
+
+async def init_app():
+	"""Initializes the application."""
+	app = web.Application(middlewares=[buckler_aiohttp.buckler_session])
+	aiohttp_jinja2.setup(app, loader=jinja2.FileSystemLoader('templates'))
+	app.on_startup.append(start_background_tasks)
+	app.on_cleanup.append(cleanup_background_tasks)
+
 	app.router.add_routes(routes)
 
+	app_wrap = web.Application()
+	app_wrap.add_subapp(config.url_prefix, app)
+	return app_wrap
+
+
 if __name__ == "__main__":
-	aiohttp.web.run_app(app, host='0.0.0.0', port=5250)
+	app = init_app()
+	web.run_app(app, host='0.0.0.0', port=5250)

rtorrent.py

@@ -4,15 +4,14 @@ This module handles the interface with rTorrent via XMLRPC.
 """
 import re
 import time
-import threading
+import asyncio
 from collections import defaultdict
 
 import config
 import rtorrent_xmlrpc
 
-WATCH_HANDLE = None
 sp = []
-torrents = [[]] * len(config.rtorrent_insts)
+torrents = [[] for _ in range(len(config.rtorrent_insts))]
 
 class Torrent:
 	def __init__(self, raw, rt_id):
@@ -55,26 +54,12 @@ class Torrent:
 		self.hashing = raw[10]
 
 
-class Watch(threading.Thread):
-	"""A thread class that continously queries the rTorrent instances."""
-	def __init__(self):
-		super(Watch, self).__init__()
-		self._stop_event = threading.Event()
-
-	def stop(self):
-		self._stop_event.set()
-
-	def stopped(self):
-		return self._stop_event.is_set()
-
-	def run(self):
-		global torrents
-		while not self.stopped():
+async def update_torrents():
 	for n in range(len(torrents)):
-				if self.stopped():
-					break
-				torrents[n] = get_all(n)
-				self._stop_event.wait(0.5)
+		try:
+			torrents[n] = await asyncio.wait_for(get_all(n), 5)
+		except asyncio.TimeoutError:
+			continue
 
 
 def size_units(rate):
@@ -126,7 +111,7 @@ def all_torrents():
 		res += item
 	return res
 
-def get_all(n):
+async def get_all(n):
 	"""Gets all torrent information from a instance and returns it."""
 	res = sp[n].d.multicall2('', 'main',
 		'd.hash=',
@@ -145,18 +130,10 @@ def get_all(n):
 
 def init():
 	"""Initializes the rTorrent interface."""
-	global WATCH_HANDLE
 	global sp
 	for rt_port in config.rtorrent_insts:
 		s = rtorrent_xmlrpc.SCGIServerProxy(rt_port)
 		sp.append(s)
-	WATCH_HANDLE = Watch()
-	WATCH_HANDLE.start()
-
-async def stop_watch(*args, **kwargs):
-	"""Stops the watch thread."""
-	global WATCH_HANDLE
-	WATCH_HANDLE.stop()
 
 def get_active():
 	"""Returns all actively seeding or leeching torrents."""