Merge pull request #4530 from jlaw/develop

Create event loop manually if uvloop is available

freqtrade/rpc/api_server/uvicorn_threaded.py

@@ -8,12 +8,33 @@ import uvicorn
 class UvicornServer(uvicorn.Server):
     """
     Multithreaded server - as found in https://github.com/encode/uvicorn/issues/742
+
+    Removed install_signal_handlers() override based on changes from this commit:
+        https://github.com/encode/uvicorn/commit/ce2ef45a9109df8eae038c0ec323eb63d644cbc6
+
+    Cannot rely on asyncio.get_event_loop() to create new event loop because of this check:
+        https://github.com/python/cpython/blob/4d7f11e05731f67fd2c07ec2972c6cb9861d52be/Lib/asyncio/events.py#L638
+
+    Fix by overriding run() and forcing creation of new event loop if uvloop is available
     """
-    def install_signal_handlers(self):
+
+    def run(self, sockets=None):
+        import asyncio
+
         """
-        In the parent implementation, this starts the thread, therefore we must patch it away here.
+        Parent implementation calls self.config.setup_event_loop(),
+            but we need to create uvloop event loop manually
         """
-        pass
+        try:
+            import uvloop  # noqa
+        except ImportError:  # pragma: no cover
+            from uvicorn.loops.asyncio import asyncio_setup
+            asyncio_setup()
+        else:
+            asyncio.set_event_loop(uvloop.new_event_loop())
+
+        loop = asyncio.get_event_loop()
+        loop.run_until_complete(self.serve(sockets=sockets))
 
     @contextlib.contextmanager
     def run_in_thread(self):

tests/conftest.py

@@ -6,7 +6,7 @@ from copy import deepcopy
 from datetime import datetime
 from functools import reduce
 from pathlib import Path
-from unittest.mock import MagicMock, PropertyMock
+from unittest.mock import MagicMock, Mock, PropertyMock
 
 import arrow
 import numpy as np
@@ -64,6 +64,14 @@ def get_args(args):
     return Arguments(args).get_parsed_arg()
 
 
+# Source: https://stackoverflow.com/questions/29881236/how-to-mock-asyncio-coroutines
+def get_mock_coro(return_value):
+    async def mock_coro(*args, **kwargs):
+        return return_value
+
+    return Mock(wraps=mock_coro)
+
+
 def patched_configuration_load_config_file(mocker, config) -> None:
     mocker.patch(
         'freqtrade.configuration.configuration.load_config_file',
@@ -1736,7 +1744,7 @@ def import_fails() -> None:
     realimport = builtins.__import__
 
     def mockedimport(name, *args, **kwargs):
-        if name in ["filelock", 'systemd.journal']:
+        if name in ["filelock", 'systemd.journal', 'uvloop']:
             raise ImportError(f"No module named '{name}'")
         return realimport(name, *args, **kwargs)
 

tests/exchange/test_exchange.py

@@ -18,21 +18,13 @@ from freqtrade.exchange.exchange import (market_is_active, timeframe_to_minutes,
                                          timeframe_to_next_date, timeframe_to_prev_date,
                                          timeframe_to_seconds)
 from freqtrade.resolvers.exchange_resolver import ExchangeResolver
-from tests.conftest import get_patched_exchange, log_has, log_has_re
+from tests.conftest import get_mock_coro, get_patched_exchange, log_has, log_has_re
 
 
 # Make sure to always keep one exchange here which is NOT subclassed!!
 EXCHANGES = ['bittrex', 'binance', 'kraken', 'ftx']
 
 
-# Source: https://stackoverflow.com/questions/29881236/how-to-mock-asyncio-coroutines
-def get_mock_coro(return_value):
-    async def mock_coro(*args, **kwargs):
-        return return_value
-
-    return Mock(wraps=mock_coro)
-
-
 def ccxt_exceptionhandlers(mocker, default_conf, api_mock, exchange_name,
                            fun, mock_ccxt_fun, retries=API_RETRY_COUNT + 1, **kwargs):
 

tests/rpc/test_rpc_apiserver.py

@@ -23,8 +23,8 @@ from freqtrade.rpc.api_server import ApiServer
 from freqtrade.rpc.api_server.api_auth import create_token, get_user_from_token
 from freqtrade.rpc.api_server.uvicorn_threaded import UvicornServer
 from freqtrade.state import RunMode, State
-from tests.conftest import (create_mock_trades, get_patched_freqtradebot, log_has, log_has_re,
-                            patch_get_signal)
+from tests.conftest import (create_mock_trades, get_mock_coro, get_patched_freqtradebot, log_has,
+                            log_has_re, patch_get_signal)
 
 
 BASE_URI = "/api/v1"
@@ -230,7 +230,7 @@ def test_api__init__(default_conf, mocker):
     assert apiserver._config == default_conf
 
 
-def test_api_UvicornServer(default_conf, mocker):
+def test_api_UvicornServer(mocker):
     thread_mock = mocker.patch('freqtrade.rpc.api_server.uvicorn_threaded.threading.Thread')
     s = UvicornServer(uvicorn.Config(MagicMock(), port=8080, host='127.0.0.1'))
     assert thread_mock.call_count == 0
@@ -248,6 +248,38 @@ def test_api_UvicornServer(default_conf, mocker):
     assert s.should_exit is True
 
 
+def test_api_UvicornServer_run(mocker):
+    serve_mock = mocker.patch('freqtrade.rpc.api_server.uvicorn_threaded.UvicornServer.serve',
+                              get_mock_coro(None))
+    s = UvicornServer(uvicorn.Config(MagicMock(), port=8080, host='127.0.0.1'))
+    assert serve_mock.call_count == 0
+
+    s.install_signal_handlers()
+    # Original implementation starts a thread - make sure that's not the case
+    assert serve_mock.call_count == 0
+
+    # Fake started to avoid sleeping forever
+    s.started = True
+    s.run()
+    assert serve_mock.call_count == 1
+
+
+def test_api_UvicornServer_run_no_uvloop(mocker, import_fails):
+    serve_mock = mocker.patch('freqtrade.rpc.api_server.uvicorn_threaded.UvicornServer.serve',
+                              get_mock_coro(None))
+    s = UvicornServer(uvicorn.Config(MagicMock(), port=8080, host='127.0.0.1'))
+    assert serve_mock.call_count == 0
+
+    s.install_signal_handlers()
+    # Original implementation starts a thread - make sure that's not the case
+    assert serve_mock.call_count == 0
+
+    # Fake started to avoid sleeping forever
+    s.started = True
+    s.run()
+    assert serve_mock.call_count == 1
+
+
 def test_api_run(default_conf, mocker, caplog):
     default_conf.update({"api_server": {"enabled": True,
                                         "listen_ip_address": "127.0.0.1",