use ThreadPoolExecutor

oml/item/icons.py

@@ -8,8 +8,10 @@ import tornado.concurrent
 import tornado.gen
 import tornado.ioloop
 import tornado.web
+from concurrent.futures import ThreadPoolExecutor
+from tornado.concurrent import run_on_executor
+
 
-from oxtornado import run_async
 from settings import icons_db_path, static_path
 from utils import resize_image, is_svg
 import db
@@ -18,6 +20,9 @@ import db
 import logging
 logger = logging.getLogger(__name__)
 
+MAX_WORKERS = 4
+
+
 
 class Icons(dict):
     def __init__(self, db):
@@ -174,53 +179,20 @@ def get_icon_sync(id, type_, size):
         data = ''
     return data
 
-@run_async
-def get_icon(id, type_, size, callback):
-    callback(get_icon_sync(id, type_, size))
-
 def clear_default_cover_cache():
     icons.clear('default:cover:')
 
-@run_async
-def get_icon_app(id, type_, size, callback):
-    with db.session():
-        from item.models import Item
-        item = Item.get(id)
-        if not item:
-            data = ''
-        else:
-            if type_ == 'cover' and not item.meta.get('cover'):
-                type_ = 'preview'
-            if type_ == 'preview' and not item.files.count():
-                type_ = 'cover'
-            if size:
-                skey = '%s:%s:%s' % (type_, id, size)
-            key = '%s:%s' % (type_, id)
-            data = None
-            if size:
-                data = icons[skey]
-                if data:
-                    size = None
-            if not data:
-                data = icons[key]
-            if not data:
-                data = icons.default_cover()
-                size = None
-            if size:
-                try:
-                    data = resize_image(data, size=size)
-                    icons[skey] = data
-                except:
-                    pass
-            data = bytes(data) or ''
-    callback(data)
 
 class IconHandler(tornado.web.RequestHandler):
+    executor = ThreadPoolExecutor(max_workers=MAX_WORKERS)
 
     def initialize(self):
         pass
 
-    @tornado.web.asynchronous
+    @run_on_executor
+    def get_icon(self, id, type_, size):
+        return get_icon_sync(id, type_, size)
+
     @tornado.gen.coroutine
     def get(self, id, type_, size=None):
 
@@ -232,7 +204,7 @@ class IconHandler(tornado.web.RequestHandler):
 
         self.set_header('Content-Type', 'image/jpeg')
 
-        response = yield tornado.gen.Task(get_icon, id, type_, size)
+        response = yield self.get_icon(id, type_, size)
         if not response:
             self.set_status(404)
             return

oml/oxtornado.py

@@ -12,13 +12,15 @@ import tornado.ioloop
 import tornado.web
 import tornado.gen
 import tornado.concurrent
-from threading import Thread
+from concurrent.futures import ThreadPoolExecutor
-from functools import wraps
+from tornado.concurrent import run_on_executor
+
 
 import logging
 logger = logging.getLogger(__name__)
 
 
+MAX_WORKERS = 4
 
 def json_response(data=None, status=200, text='ok'):
     if not data:
@@ -37,15 +39,6 @@ def json_dumps(obj):
     indent = 2
     return json.dumps(obj, indent=indent, default=_to_json, ensure_ascii=False).encode()
 
-def run_async(func):
-    @wraps(func)
-    def async_func(*args, **kwargs):
-        func_hl = Thread(target=func, args=args, kwargs=kwargs)
-        func_hl.start()
-        return func_hl
-
-    return async_func
-
 def trim(docstring):
     if not docstring:
         return ''
@@ -75,44 +68,46 @@ def trim(docstring):
 def defaultcontext():
     yield
 
-@run_async
-def api_task(context, request, callback):
-    import settings
-    if context == None:
-        context = defaultcontext
-    action = request.arguments.get('action', [None])[0].decode('utf-8')
-    data = request.arguments.get('data', [b'{}'])[0]
-    data = json.loads(data.decode('utf-8')) if data else {}
-    if not action:
-        methods = list(actions.keys())
-        api = []
-        for f in sorted(methods):
-            api.append({'name': f,
-                        'doc': actions.doc(f).replace('\n', '<br>\n')})
-        response = json_response(api)
-    else:
-        if settings.DEBUG_API:
-            logger.debug('API %s %s', action, data)
-        f = actions.get(action)
-        if f:
-            with context():
-                try:
-                    response = f(data)
-                except:
-                    logger.debug('FAILED %s %s', action, data, exc_info=True)
-                    response = json_response(status=500, text='%s failed' % action)
-        else:
-            response = json_response(status=400, text='Unknown action %s' % action)
-    callback(response)
-
 class ApiHandler(tornado.web.RequestHandler):
+    executor = ThreadPoolExecutor(max_workers=MAX_WORKERS)
+
     def initialize(self, context=None):
         self._context = context
 
     def get(self):
         self.write('use POST')
 
-    @tornado.web.asynchronous
+    @run_on_executor
+    def api_task(self, request):
+        import settings
+        context = self._context
+        if context == None:
+            context = defaultcontext
+        action = request.arguments.get('action', [None])[0].decode('utf-8')
+        data = request.arguments.get('data', [b'{}'])[0]
+        data = json.loads(data.decode('utf-8')) if data else {}
+        if not action:
+            methods = list(actions.keys())
+            api = []
+            for f in sorted(methods):
+                api.append({'name': f,
+                            'doc': actions.doc(f).replace('\n', '<br>\n')})
+            response = json_response(api)
+        else:
+            if settings.DEBUG_API:
+                logger.debug('API %s %s', action, data)
+            f = actions.get(action)
+            if f:
+                with context():
+                    try:
+                        response = f(data)
+                    except:
+                        logger.debug('FAILED %s %s', action, data, exc_info=True)
+                        response = json_response(status=500, text='%s failed' % action)
+            else:
+                response = json_response(status=400, text='Unknown action %s' % action)
+        return response
+
     @tornado.gen.coroutine
     def post(self):
         if 'origin' in self.request.headers and self.request.host not in self.request.headers['origin']:
@@ -121,7 +116,7 @@ class ApiHandler(tornado.web.RequestHandler):
             self.write('')
             return
 
-        response = yield tornado.gen.Task(api_task, self._context, self.request)
+        response = yield self.api_task(self.request)
         if not 'status' in response:
             response = json_response(response)
         response = json_dumps(response)