coalib/: Support Python 3.4.2 and 3.4.3

Python 3.4.3 is the default Python version available on
all non-Python Travis images.  Not supporting it makes
using non-Python languages on Travis slower as an additional
Python would need to be installed.

Use compatibility layer imported from `homeassistant`,
to support 3.4.2 and 3.4.3.

Closes #5891

.coafile

@@ -3,6 +3,7 @@ files = *.py, coalib/**/*.py, ./coala, tests/**/*.py, docs/conf.py
 ignore =
     tests/bearlib/languages/documentation/documentation_extraction_testdata/*.py,
     tests/collecting/collectors_test_dir/bears/incorrect_bear.py,
+    coalib/misc/Asyncio.py,
 
 max_line_length = 80
 use_spaces = True

.misc/check_unsupported.sh

@@ -30,6 +30,6 @@ fi
 set -e
 
 # The following is emitted on stdout
-grep -q 'coala supports only python 3.4.4 or later' setup.log
+grep -q 'coala supports only Python 3.4.2 or later' setup.log
 
 echo "Unsupported check completed successfully"

.moban.yaml

@@ -45,11 +45,13 @@ gitignore_extra_rulesets:
 
 requires:
   - https://gitlab.com/coala/mobans.git
+  - https://github.com/NAStools/homeassistant.git
 configuration:
   template_dir:
     - .moban.dt/
     - mobans:templates/
     - mobans:assets/
+    - homeassistant:homeassistant/util/
   configuration: .moban.yaml
   configuration_dir: 'mobans:'
 targets:
@@ -65,3 +67,6 @@ targets:
   - netlify.toml: docs/netlify.toml
   - .misc/appveyor.yml: ci/appveyor.yml.jj2
   - .misc/run_with_env.cmd: run_with_env.cmd
+copy:
+  # homeassistant license MIT
+  - coalib/misc/Asyncio.py: async.py

.travis.yml

@@ -3,6 +3,7 @@ language: python
 python:
   - 3.4.4
   - 3.5
+  - 3.4.2
 
 stages:
   - name: sentinel

coalib/__init__.py

@@ -22,6 +22,6 @@ def get_version():
 
 
 def assert_supported_version():
-    if sys.version_info < (3, 4, 4):
-        print('coala supports only python 3.4.4 or later.')
+    if sys.version_info < (3, 4, 2):
+        print('coala supports only Python 3.4.2 or later.')
         exit(4)

coalib/core/Core.py

@@ -6,6 +6,7 @@
 from coalib.core.DependencyTracker import DependencyTracker
 from coalib.core.Graphs import traverse_graph
 from coalib.core.PersistentHash import persistent_hash
+from coalib.misc.Compatibility import run_coroutine_threadsafe
 
 
 def group(iterable, key=lambda x: x):
@@ -339,7 +340,7 @@ def _execute_task_with_cache(self, bear, task):
         else:
             bear_args, bear_kwargs = task
 
-            future = asyncio.run_coroutine_threadsafe(
+            future = run_coroutine_threadsafe(
                 asyncio.wait_for(
                     self.event_loop.run_in_executor(
                         self.executor, bear.execute_task,

coalib/misc/Asyncio.py

@@ -0,0 +1,176 @@
+"""Asyncio backports for Python 3.4.3 compatibility."""
+import concurrent.futures
+import threading
+import logging
+from asyncio import coroutines
+from asyncio.futures import Future
+
+try:
+    # pylint: disable=ungrouped-imports
+    from asyncio import ensure_future
+except ImportError:
+    # Python 3.4.3 and earlier has this as async
+    # pylint: disable=unused-import
+    from asyncio import async
+    ensure_future = async
+
+
+_LOGGER = logging.getLogger(__name__)
+
+
+def _set_result_unless_cancelled(fut, result):
+    """Helper setting the result only if the future was not cancelled."""
+    if fut.cancelled():
+        return
+    fut.set_result(result)
+
+
+def _set_concurrent_future_state(concurr, source):
+    """Copy state from a future to a concurrent.futures.Future."""
+    assert source.done()
+    if source.cancelled():
+        concurr.cancel()
+    if not concurr.set_running_or_notify_cancel():
+        return
+    exception = source.exception()
+    if exception is not None:
+        concurr.set_exception(exception)
+    else:
+        result = source.result()
+        concurr.set_result(result)
+
+
+def _copy_future_state(source, dest):
+    """Internal helper to copy state from another Future.
+
+    The other Future may be a concurrent.futures.Future.
+    """
+    assert source.done()
+    if dest.cancelled():
+        return
+    assert not dest.done()
+    if source.cancelled():
+        dest.cancel()
+    else:
+        exception = source.exception()
+        if exception is not None:
+            dest.set_exception(exception)
+        else:
+            result = source.result()
+            dest.set_result(result)
+
+
+def _chain_future(source, destination):
+    """Chain two futures so that when one completes, so does the other.
+
+    The result (or exception) of source will be copied to destination.
+    If destination is cancelled, source gets cancelled too.
+    Compatible with both asyncio.Future and concurrent.futures.Future.
+    """
+    if not isinstance(source, (Future, concurrent.futures.Future)):
+        raise TypeError('A future is required for source argument')
+    if not isinstance(destination, (Future, concurrent.futures.Future)):
+        raise TypeError('A future is required for destination argument')
+    # pylint: disable=protected-access
+    source_loop = source._loop if isinstance(source, Future) else None
+    dest_loop = destination._loop if isinstance(destination, Future) else None
+
+    def _set_state(future, other):
+        if isinstance(future, Future):
+            _copy_future_state(other, future)
+        else:
+            _set_concurrent_future_state(future, other)
+
+    def _call_check_cancel(destination):
+        if destination.cancelled():
+            if source_loop is None or source_loop is dest_loop:
+                source.cancel()
+            else:
+                source_loop.call_soon_threadsafe(source.cancel)
+
+    def _call_set_state(source):
+        if dest_loop is None or dest_loop is source_loop:
+            _set_state(destination, source)
+        else:
+            dest_loop.call_soon_threadsafe(_set_state, destination, source)
+
+    destination.add_done_callback(_call_check_cancel)
+    source.add_done_callback(_call_set_state)
+
+
+def run_coroutine_threadsafe(coro, loop):
+    """Submit a coroutine object to a given event loop.
+
+    Return a concurrent.futures.Future to access the result.
+    """
+    ident = loop.__dict__.get("_thread_ident")
+    if ident is not None and ident == threading.get_ident():
+        raise RuntimeError('Cannot be called from within the event loop')
+
+    if not coroutines.iscoroutine(coro):
+        raise TypeError('A coroutine object is required')
+    future = concurrent.futures.Future()
+
+    def callback():
+        """Callback to call the coroutine."""
+        try:
+            # pylint: disable=deprecated-method
+            _chain_future(ensure_future(coro, loop=loop), future)
+        # pylint: disable=broad-except
+        except Exception as exc:
+            if future.set_running_or_notify_cancel():
+                future.set_exception(exc)
+            else:
+                _LOGGER.warning("Exception on lost future: ", exc_info=True)
+
+    loop.call_soon_threadsafe(callback)
+    return future
+
+
+def fire_coroutine_threadsafe(coro, loop):
+    """Submit a coroutine object to a given event loop.
+
+    This method does not provide a way to retrieve the result and
+    is intended for fire-and-forget use. This reduces the
+    work involved to fire the function on the loop.
+    """
+    ident = loop.__dict__.get("_thread_ident")
+    if ident is not None and ident == threading.get_ident():
+        raise RuntimeError('Cannot be called from within the event loop')
+
+    if not coroutines.iscoroutine(coro):
+        raise TypeError('A coroutine object is required: %s' % coro)
+
+    def callback():
+        """Callback to fire coroutine."""
+        # pylint: disable=deprecated-method
+        ensure_future(coro, loop=loop)
+
+    loop.call_soon_threadsafe(callback)
+    return
+
+
+def run_callback_threadsafe(loop, callback, *args):
+    """Submit a callback object to a given event loop.
+
+    Return a concurrent.futures.Future to access the result.
+    """
+    ident = loop.__dict__.get("_thread_ident")
+    if ident is not None and ident == threading.get_ident():
+        raise RuntimeError('Cannot be called from within the event loop')
+
+    future = concurrent.futures.Future()
+
+    def run_callback():
+        """Run callback and store result."""
+        try:
+            future.set_result(callback(*args))
+        # pylint: disable=broad-except
+        except Exception as exc:
+            if future.set_running_or_notify_cancel():
+                future.set_exception(exc)
+            else:
+                _LOGGER.warning("Exception on lost future: ", exc_info=True)
+
+    loop.call_soon_threadsafe(run_callback)
+    return future

coalib/misc/Compatibility.py

@@ -4,3 +4,13 @@
     JSONDecodeError = json.decoder.JSONDecodeError
 except AttributeError:  # pragma Python 3.5,3.6: no cover
     JSONDecodeError = ValueError
+
+try:
+    from asyncio import run_coroutine_threadsafe
+except ImportError:  # pragma: no cover
+    from coalib.misc.Asyncio import run_coroutine_threadsafe
+
+__all__ = [
+    'JSONDecodeError',
+    'run_coroutine_threadsafe',
+]

setup.cfg

@@ -61,6 +61,7 @@ source =
 omit =
   tests/*
   setup.py
+  coalib/misc/Asyncio.py
 
 [coverage:report]
 fail_under = 100