event loop and server optimizations

Author: DevAlone

main.py

@@ -1,16 +1,23 @@
 #!/usr/bin/env python3
-import materialized_view_updater
+import uvloop
+
+uvloop.install()
+
 from proxy_py import settings
 from processor import Processor
 from server.proxy_provider_server import ProxyProviderServer
 from statistics import statistics
 from checkers.base_checker import BaseChecker
+from tools import test_collector
 
+import materialized_view_updater
 import asyncio
 import logging
 import argparse
+import subprocess
+import sys
+
 
-from tools import test_collector
 
 test_collector_path = None
 main_logger = None
@@ -84,32 +91,24 @@ def prepare_loggers():
     main_logger.addHandler(logger_file_handler)
 
 
-def main():
+async def core():
+
     process_cmd_arguments()
     prepare_loggers()
 
-    loop = asyncio.get_event_loop()
-
     if test_collector_path is not None:
-        return loop.run_until_complete(test_collector.run(test_collector_path))
+        return await test_collector.run(test_collector_path)
 
     proxy_processor = Processor.get_instance()
 
-    proxy_provider_server = ProxyProviderServer(
-        settings.PROXY_PROVIDER_SERVER_ADDRESS['HOST'],
-        settings.PROXY_PROVIDER_SERVER_ADDRESS['PORT'],
-        proxy_processor,
-    )
-
-    loop.run_until_complete(proxy_provider_server.start(loop))
-
     try:
-        loop.run_until_complete(asyncio.gather(*[
+        code = await asyncio.gather(*[
             proxy_processor.worker(),
             statistics.worker(),
             materialized_view_updater.worker(),
-        ]))
+        ])
         BaseChecker.clean()
+        return code
     except KeyboardInterrupt:
         pass
     except BaseException as ex:
@@ -120,5 +119,31 @@ def main():
     return 0
 
 
+def server():
+    proxy_provider_server = ProxyProviderServer(
+        settings.PROXY_PROVIDER_SERVER_ADDRESS['HOST'],
+        settings.PROXY_PROVIDER_SERVER_ADDRESS['PORT'],
+    )
+
+    return proxy_provider_server.start(asyncio.get_event_loop())
+
+
+def main():
+    if len(sys.argv) < 2:
+        # run default configuration
+        p = subprocess.Popen(["python", sys.argv[0], 'server'])
+
+        code = asyncio.get_event_loop().run_until_complete(core())
+        p.wait()
+        return code
+
+    command = sys.argv[1].strip()
+    sys.argv = sys.argv[1:]
+    return {
+        'core': lambda: asyncio.get_event_loop().run_until_complete(core()),
+        'server': server,
+    }[command]()
+
+
 if __name__ == "__main__":
     exit(main())

requirements.txt

@@ -12,3 +12,4 @@ py_mini_racer
 pytest
 pytest-asyncio
 termcolor
+uvloop

server/api_v1/api_request_handler.py

@@ -42,6 +42,7 @@ async def handle(self, request: aiohttp.ClientRequest, post_data: dict):
                     ex
                 )
             )
+            self.app.log_exception(request, ex)
 
             response = {
                 'status': 'error',

server/api_v1/requests_to_models/request_executor.py

@@ -1,3 +1,5 @@
+import peewee_async
+
 from proxy_py import settings
 from models import db
 from ..requests_to_models.request import Request, GetRequest, CountRequest, FetchRequest

server/base_app.py

@@ -37,6 +37,9 @@ def log_critical(self, *args, **kwargs):
     def log_error(self, *args, **kwargs):
         self.log('error', *args, **kwargs)
 
+    def log_exception(self, *args, **kwargs):
+        self.log('exception', *args, **kwargs)
+
     def log_warning(self, *args, **kwargs):
         self.log('warning', *args, **kwargs)
 
@@ -47,15 +50,11 @@ def log_debug(self, *args, **kwargs):
         self.log('debug', *args, **kwargs)
 
     def log(self, level, request, message):
-        if False and settings.DEBUG:
-            client_host, client_port = request.transport.get_extra_info("peername")
-            client_ip = "{}:{}".format(client_host, client_port)
-        else:
-            # behind nginx or other reverse proxy
-            client_ip = str(request.headers.get("X-Real-IP", "None"))
-
-            if client_ip == "None" or client_ip.startswith("127.0.0.1"):
-                self.logger.error("Your reverse proxy doesn't present user's IP", extra={"client_ip": client_ip})
+        # behind nginx or other reverse proxy
+        client_ip = str(request.headers.get("X-Real-IP", "None"))
+
+        if client_ip == "None" or client_ip.startswith("127.0.0.1"):
+            self.logger.error("Your reverse proxy doesn't present user's IP", extra={"client_ip": client_ip})
 
         getattr(self.logger, level)(message, extra={
             "client_ip": client_ip,

server/proxy_provider_server.py

@@ -1,3 +1,5 @@
+import asyncio
+
 from proxy_py import settings
 from .base_app import BaseApp
 from .api_v1.app import App as ApiV1App
@@ -12,7 +14,7 @@
 
 
 class ProxyProviderServer(BaseApp):
-    def __init__(self, host, port, processor):
+    def __init__(self, host, port):
         logger = logging.getLogger("proxy_py/server")
 
         if settings.DEBUG:
@@ -30,21 +32,22 @@ def __init__(self, host, port, processor):
 
         super(ProxyProviderServer, self).__init__(logger)
 
-        self._processor = processor
         self.host = host
         self.port = port
         self._request_number = 0
 
-    async def start(self, loop):
-        await self.init()
+    def start(self, loop):
+        loop.run_until_complete(self.init())
 
+        '''
         server = await loop.create_server(
             self._app.make_handler(),
             self.host,
             self.port
         )
+        '''
 
-        return server
+        return web.run_app(self._app, host=self.host, port=self.port, loop=loop)
 
     async def setup_router(self):
         api_v1_app = ApiV1App(logger=self.logger)