Now that the minimum version is 3.7, drop some of the 3.6-specific hacks
we've been carrying. A single remaining compatibility hack concerning
3.6's lack of @asynccontextmanager is addressed in the following commit.
Signed-off-by: John Snow <js...@redhat.com>
cherry picked from commit 3e8e34e594cfc6b707e6f67959166acde4b421b8
---
python/qemu/qmp/protocol.py | 48 +++++++---------
python/qemu/qmp/qmp_tui.py | 8 +--
python/qemu/qmp/util.py | 107 ++----------------------------------
python/tests/protocol.py | 8 +--
4 files changed, 33 insertions(+), 138 deletions(-)
diff --git a/python/qemu/qmp/protocol.py b/python/qemu/qmp/protocol.py
index ec4762c567b..958aeca08ac 100644
--- a/python/qemu/qmp/protocol.py
+++ b/python/qemu/qmp/protocol.py
@@ -15,6 +15,7 @@
import asyncio
from asyncio import StreamReader, StreamWriter
+from contextlib import asynccontextmanager
from enum import Enum
from functools import wraps
import logging
@@ -22,6 +23,7 @@
from ssl import SSLContext
from typing import (
Any,
+ AsyncGenerator,
Awaitable,
Callable,
Generic,
@@ -36,13 +38,10 @@
from .error import QMPError
from .util import (
bottom_half,
- create_task,
exception_summary,
flush,
- is_closing,
pretty_traceback,
upper_half,
- wait_closed,
)
@@ -340,9 +339,8 @@ async def start_server(self, address: SocketAddrT,
This exception will wrap a more concrete one. In most cases,
the wrapped exception will be `OSError`.
"""
- await self._session_guard(
- self._do_start_server(address, ssl),
- 'Failed to establish connection')
+ async with self._session_guard('Failed to establish connection'):
+ await self._do_start_server(address, ssl)
assert self.runstate == Runstate.CONNECTING
@upper_half
@@ -365,12 +363,10 @@ async def accept(self) -> None:
"""
if self._accepted is None:
raise QMPError("Cannot call accept() before start_server().")
- await self._session_guard(
- self._do_accept(),
- 'Failed to establish connection')
- await self._session_guard(
- self._establish_session(),
- 'Failed to establish session')
+ async with self._session_guard('Failed to establish connection'):
+ await self._do_accept()
+ async with self._session_guard('Failed to establish session'):
+ await self._establish_session()
assert self.runstate == Runstate.RUNNING
@upper_half
@@ -395,12 +391,10 @@ async def connect(self, address: Union[SocketAddrT,
socket.socket],
protocol-level failure occurs while establishing a new
session, the wrapped error may also be an `QMPError`.
"""
- await self._session_guard(
- self._do_connect(address, ssl),
- 'Failed to establish connection')
- await self._session_guard(
- self._establish_session(),
- 'Failed to establish session')
+ async with self._session_guard('Failed to establish connection'):
+ await self._do_connect(address, ssl)
+ async with self._session_guard('Failed to establish session'):
+ await self._establish_session()
assert self.runstate == Runstate.RUNNING
@upper_half
@@ -421,7 +415,8 @@ async def disconnect(self) -> None:
# Section: Session machinery
# --------------------------
- async def _session_guard(self, coro: Awaitable[None], emsg: str) -> None:
+ @asynccontextmanager
+ async def _session_guard(self, emsg: str) -> AsyncGenerator[None, None]:
"""
Async guard function used to roll back to `IDLE` on any error.
@@ -438,10 +433,9 @@ async def _session_guard(self, coro: Awaitable[None],
emsg: str) -> None:
:raise ConnectError:
When any other error is encountered in the guarded block.
"""
- # Note: After Python 3.6 support is removed, this should be an
- # @asynccontextmanager instead of accepting a callback.
try:
- await coro
+ # Caller's code runs here.
+ yield
except BaseException as err:
self.logger.error("%s: %s", emsg, exception_summary(err))
self.logger.debug("%s:\n%s\n", emsg, pretty_traceback())
@@ -682,8 +676,8 @@ async def _establish_session(self) -> None:
reader_coro = self._bh_loop_forever(self._bh_recv_message, 'Reader')
writer_coro = self._bh_loop_forever(self._bh_send_message, 'Writer')
- self._reader_task = create_task(reader_coro)
- self._writer_task = create_task(writer_coro)
+ self._reader_task = asyncio.create_task(reader_coro)
+ self._writer_task = asyncio.create_task(writer_coro)
self._bh_tasks = asyncio.gather(
self._reader_task,
@@ -708,7 +702,7 @@ def _schedule_disconnect(self) -> None:
if not self._dc_task:
self._set_state(Runstate.DISCONNECTING)
self.logger.debug("Scheduling disconnect.")
- self._dc_task = create_task(self._bh_disconnect())
+ self._dc_task = asyncio.create_task(self._bh_disconnect())
@upper_half
async def _wait_disconnect(self) -> None:
@@ -844,13 +838,13 @@ async def _bh_close_stream(self, error_pathway: bool =
False) -> None:
if not self._writer:
return
- if not is_closing(self._writer):
+ if not self._writer.is_closing():
self.logger.debug("Closing StreamWriter.")
self._writer.close()
self.logger.debug("Waiting for StreamWriter to close ...")
try:
- await wait_closed(self._writer)
+ await self._writer.wait_closed()
except Exception: # pylint: disable=broad-except
# It's hard to tell if the Stream is already closed or
# not. Even if one of the tasks has failed, it may have
diff --git a/python/qemu/qmp/qmp_tui.py b/python/qemu/qmp/qmp_tui.py
index 2d9ebbd20bc..562be008d5e 100644
--- a/python/qemu/qmp/qmp_tui.py
+++ b/python/qemu/qmp/qmp_tui.py
@@ -40,7 +40,7 @@
from .message import DeserializationError, Message, UnexpectedTypeError
from .protocol import ConnectError, Runstate
from .qmp_client import ExecInterruptedError, QMPClient
-from .util import create_task, pretty_traceback
+from .util import pretty_traceback
# The name of the signal that is used to update the history list
@@ -225,7 +225,7 @@ def cb_send_to_server(self, raw_msg: str) -> None:
"""
try:
msg = Message(bytes(raw_msg, encoding='utf-8'))
- create_task(self._send_to_server(msg))
+ asyncio.create_task(self._send_to_server(msg))
except (DeserializationError, UnexpectedTypeError) as err:
raw_msg = format_json(raw_msg)
logging.info('Invalid message: %s', err.error_message)
@@ -246,7 +246,7 @@ def kill_app(self) -> None:
Initiates killing of app. A bridge between asynchronous and synchronous
code.
"""
- create_task(self._kill_app())
+ asyncio.create_task(self._kill_app())
async def _kill_app(self) -> None:
"""
@@ -393,7 +393,7 @@ def run(self, debug: bool = False) -> None:
handle_mouse=True,
event_loop=event_loop)
- create_task(self.manage_connection(), self.aloop)
+ self.aloop.create_task(self.manage_connection())
try:
main_loop.run()
except Exception as err:
diff --git a/python/qemu/qmp/util.py b/python/qemu/qmp/util.py
index ca6225e9cda..0b3e781373d 100644
--- a/python/qemu/qmp/util.py
+++ b/python/qemu/qmp/util.py
@@ -1,25 +1,15 @@
"""
Miscellaneous Utilities
-This module provides asyncio utilities and compatibility wrappers for
-Python 3.6 to provide some features that otherwise become available in
-Python 3.7+.
-
-Various logging and debugging utilities are also provided, such as
-`exception_summary()` and `pretty_traceback()`, used primarily for
-adding information into the logging stream.
+This module provides asyncio and various logging and debugging
+utilities, such as `exception_summary()` and `pretty_traceback()`, used
+primarily for adding information into the logging stream.
"""
import asyncio
import sys
import traceback
-from typing import (
- Any,
- Coroutine,
- Optional,
- TypeVar,
- cast,
-)
+from typing import TypeVar, cast
T = TypeVar('T')
@@ -79,95 +69,6 @@ def bottom_half(func: T) -> T:
return func
-# -------------------------------
-# Section: Compatibility Wrappers
-# -------------------------------
-
-
-def create_task(coro: Coroutine[Any, Any, T],
- loop: Optional[asyncio.AbstractEventLoop] = None
- ) -> 'asyncio.Future[T]':
- """
- Python 3.6-compatible `asyncio.create_task` wrapper.
-
- :param coro: The coroutine to execute in a task.
- :param loop: Optionally, the loop to create the task in.
-
- :return: An `asyncio.Future` object.
- """
- if sys.version_info >= (3, 7):
- if loop is not None:
- return loop.create_task(coro)
- return asyncio.create_task(coro) # pylint: disable=no-member
-
- # Python 3.6:
- return asyncio.ensure_future(coro, loop=loop)
-
-
-def is_closing(writer: asyncio.StreamWriter) -> bool:
- """
- Python 3.6-compatible `asyncio.StreamWriter.is_closing` wrapper.
-
- :param writer: The `asyncio.StreamWriter` object.
- :return: `True` if the writer is closing, or closed.
- """
- if sys.version_info >= (3, 7):
- return writer.is_closing()
-
- # Python 3.6:
- transport = writer.transport
- assert isinstance(transport, asyncio.WriteTransport)
- return transport.is_closing()
-
-
-async def wait_closed(writer: asyncio.StreamWriter) -> None:
- """
- Python 3.6-compatible `asyncio.StreamWriter.wait_closed` wrapper.
-
- :param writer: The `asyncio.StreamWriter` to wait on.
- """
- if sys.version_info >= (3, 7):
- await writer.wait_closed()
- return
-
- # Python 3.6
- transport = writer.transport
- assert isinstance(transport, asyncio.WriteTransport)
-
- while not transport.is_closing():
- await asyncio.sleep(0)
-
- # This is an ugly workaround, but it's the best I can come up with.
- sock = transport.get_extra_info('socket')
-
- if sock is None:
- # Our transport doesn't have a socket? ...
- # Nothing we can reasonably do.
- return
-
- while sock.fileno() != -1:
- await asyncio.sleep(0)
-
-
-def asyncio_run(coro: Coroutine[Any, Any, T], *, debug: bool = False) -> T:
- """
- Python 3.6-compatible `asyncio.run` wrapper.
-
- :param coro: A coroutine to execute now.
- :return: The return value from the coroutine.
- """
- if sys.version_info >= (3, 7):
- return asyncio.run(coro, debug=debug)
-
- # Python 3.6
- loop = asyncio.get_event_loop()
- loop.set_debug(debug)
- ret = loop.run_until_complete(coro)
- loop.close()
-
- return ret
-
-
# ----------------------------
# Section: Logging & Debugging
# ----------------------------
diff --git a/python/tests/protocol.py b/python/tests/protocol.py
index 56c4d441f9c..c254c77b176 100644
--- a/python/tests/protocol.py
+++ b/python/tests/protocol.py
@@ -8,7 +8,6 @@
from qemu.qmp import ConnectError, Runstate
from qemu.qmp.protocol import AsyncProtocol, StateError
-from qemu.qmp.util import asyncio_run, create_task
class NullProtocol(AsyncProtocol[None]):
@@ -124,7 +123,7 @@ async def _runner():
if allow_cancellation:
return
raise
- return create_task(_runner())
+ return asyncio.create_task(_runner())
@contextmanager
@@ -271,7 +270,7 @@ async def _watcher():
msg=f"Expected state '{state.name}'",
)
- self.runstate_watcher = create_task(_watcher())
+ self.runstate_watcher = asyncio.create_task(_watcher())
# Kick the loop and force the task to block on the event.
await asyncio.sleep(0)
@@ -589,7 +588,8 @@ async def _asyncTearDown(self):
async def testSmoke(self):
with TemporaryDirectory(suffix='.qmp') as tmpdir:
sock = os.path.join(tmpdir, type(self.proto).__name__ + ".sock")
- server_task =
create_task(self.server.start_server_and_accept(sock))
+ server_task = asyncio.create_task(
+ self.server.start_server_and_accept(sock))
# give the server a chance to start listening [...]
await asyncio.sleep(0)
--
2.50.1
