chore(utils): use threads replace asyncio

Author: duanhongyi

rootfs/api/models/app.py

@@ -26,7 +26,7 @@
 from api.models.tls import TLS
 from api.models.appsettings import AppSettings
 from api.models.volume import Volume
-from api.utils import generate_app_name, async_run
+from api.utils import generate_app_name, apply_tasks
 from scheduler import KubeHTTPException, KubeException
 
 logger = logging.getLogger(__name__)
@@ -383,7 +383,7 @@ def restart(self, **kwargs):  # noqa
                 ) for pod in self.list_pods(**kwargs)
             ]
 
-            async_run(tasks)
+            apply_tasks(tasks)
         except Exception as e:
             err = "warning, some pods failed to stop:\n{}".format(str(e))
             self.log(err, logging.WARNING)
@@ -528,7 +528,7 @@ def _scale_pods(self, scale_types):
             # create the application config in k8s (secret in this case) for all deploy objects
             self.set_application_config(release)
 
-            async_run(tasks)
+            apply_tasks(tasks)
         except Exception as e:
             err = '(scale): {}'.format(e)
             self.log(err, logging.ERROR)
@@ -619,7 +619,7 @@ def deploy(self, release, force_deploy=False, rollback_on_failure=True):  # noqa
             ]
 
             try:
-                async_run(tasks)
+                apply_tasks(tasks)
             except KubeException as e:
                 # Don't rollback if the previous release doesn't have a build which means
                 # this is the first build and all the previous releases are just config changes.

rootfs/api/utils.py

@@ -1,7 +1,6 @@
 """
 Helper functions used by the Drycc server.
 """
-import asyncio
 import base64
 import concurrent
 import hashlib
@@ -145,43 +144,22 @@ def dict_merge(origin, merge):
     return result
 
 
-def async_run(tasks):
+def apply_tasks(tasks):
     """
     run a group of tasks async
     Requires the tasks arg to be a list of functools.partial()
     """
     if not tasks:
         return
 
-    # start a new async event loop
-    loop = asyncio.get_event_loop()
-    # https://github.com/python/asyncio/issues/258
     executor = concurrent.futures.ThreadPoolExecutor(5)
-    loop.set_default_executor(executor)
-
-    async_tasks = [asyncio.ensure_future(async_task(task, loop)) for task in tasks]
-    # run tasks in parallel
-    loop.run_until_complete(asyncio.wait(async_tasks))
-    # deal with errors (exceptions, etc)
-    for task in async_tasks:
-        error = task.exception()
+    for future in [executor.submit(task) for task in tasks]:
+        error = future.exception()
         if error is not None:
             raise error
-
     executor.shutdown(wait=True)
 
 
-async def async_task(params, loop):
-    """
-    Perform a task asynchronously.
-    """
-    # get the calling function
-    logger.debug('Running {}'.format(params))
-    # This executes a task in its own thread (in parallel)
-    await loop.run_in_executor(None, params)
-    logger.debug('Finished running {}'.format(params))
-
-
 if __name__ == "__main__":
     import doctest
     doctest.testmod()