mirror of
https://gitee.com/kekingcn/file-online-preview.git
synced 2026-06-19 04:17:07 +00:00
集成OpenOffice替换为LibreOffice
This commit is contained in:
@@ -0,0 +1,45 @@
|
||||
"""The asyncio package, tracking PEP 3156."""
|
||||
|
||||
# flake8: noqa
|
||||
|
||||
import sys
|
||||
|
||||
# This relies on each of the submodules having an __all__ variable.
|
||||
from .base_events import *
|
||||
from .coroutines import *
|
||||
from .events import *
|
||||
from .exceptions import *
|
||||
from .futures import *
|
||||
from .locks import *
|
||||
from .protocols import *
|
||||
from .runners import *
|
||||
from .queues import *
|
||||
from .streams import *
|
||||
from .subprocess import *
|
||||
from .tasks import *
|
||||
from .transports import *
|
||||
|
||||
# Exposed for _asynciomodule.c to implement now deprecated
|
||||
# Task.all_tasks() method. This function will be removed in 3.9.
|
||||
from .tasks import _all_tasks_compat # NoQA
|
||||
|
||||
__all__ = (base_events.__all__ +
|
||||
coroutines.__all__ +
|
||||
events.__all__ +
|
||||
exceptions.__all__ +
|
||||
futures.__all__ +
|
||||
locks.__all__ +
|
||||
protocols.__all__ +
|
||||
runners.__all__ +
|
||||
queues.__all__ +
|
||||
streams.__all__ +
|
||||
subprocess.__all__ +
|
||||
tasks.__all__ +
|
||||
transports.__all__)
|
||||
|
||||
if sys.platform == 'win32': # pragma: no cover
|
||||
from .windows_events import *
|
||||
__all__ += windows_events.__all__
|
||||
else:
|
||||
from .unix_events import * # pragma: no cover
|
||||
__all__ += unix_events.__all__
|
||||
@@ -0,0 +1,125 @@
|
||||
import ast
|
||||
import asyncio
|
||||
import code
|
||||
import concurrent.futures
|
||||
import inspect
|
||||
import sys
|
||||
import threading
|
||||
import types
|
||||
import warnings
|
||||
|
||||
from . import futures
|
||||
|
||||
|
||||
class AsyncIOInteractiveConsole(code.InteractiveConsole):
|
||||
|
||||
def __init__(self, locals, loop):
|
||||
super().__init__(locals)
|
||||
self.compile.compiler.flags |= ast.PyCF_ALLOW_TOP_LEVEL_AWAIT
|
||||
|
||||
self.loop = loop
|
||||
|
||||
def runcode(self, code):
|
||||
future = concurrent.futures.Future()
|
||||
|
||||
def callback():
|
||||
global repl_future
|
||||
global repl_future_interrupted
|
||||
|
||||
repl_future = None
|
||||
repl_future_interrupted = False
|
||||
|
||||
func = types.FunctionType(code, self.locals)
|
||||
try:
|
||||
coro = func()
|
||||
except SystemExit:
|
||||
raise
|
||||
except KeyboardInterrupt as ex:
|
||||
repl_future_interrupted = True
|
||||
future.set_exception(ex)
|
||||
return
|
||||
except BaseException as ex:
|
||||
future.set_exception(ex)
|
||||
return
|
||||
|
||||
if not inspect.iscoroutine(coro):
|
||||
future.set_result(coro)
|
||||
return
|
||||
|
||||
try:
|
||||
repl_future = self.loop.create_task(coro)
|
||||
futures._chain_future(repl_future, future)
|
||||
except BaseException as exc:
|
||||
future.set_exception(exc)
|
||||
|
||||
loop.call_soon_threadsafe(callback)
|
||||
|
||||
try:
|
||||
return future.result()
|
||||
except SystemExit:
|
||||
raise
|
||||
except BaseException:
|
||||
if repl_future_interrupted:
|
||||
self.write("\nKeyboardInterrupt\n")
|
||||
else:
|
||||
self.showtraceback()
|
||||
|
||||
|
||||
class REPLThread(threading.Thread):
|
||||
|
||||
def run(self):
|
||||
try:
|
||||
banner = (
|
||||
f'asyncio REPL {sys.version} on {sys.platform}\n'
|
||||
f'Use "await" directly instead of "asyncio.run()".\n'
|
||||
f'Type "help", "copyright", "credits" or "license" '
|
||||
f'for more information.\n'
|
||||
f'{getattr(sys, "ps1", ">>> ")}import asyncio'
|
||||
)
|
||||
|
||||
console.interact(
|
||||
banner=banner,
|
||||
exitmsg='exiting asyncio REPL...')
|
||||
finally:
|
||||
warnings.filterwarnings(
|
||||
'ignore',
|
||||
message=r'^coroutine .* was never awaited$',
|
||||
category=RuntimeWarning)
|
||||
|
||||
loop.call_soon_threadsafe(loop.stop)
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
loop = asyncio.new_event_loop()
|
||||
asyncio.set_event_loop(loop)
|
||||
|
||||
repl_locals = {'asyncio': asyncio}
|
||||
for key in {'__name__', '__package__',
|
||||
'__loader__', '__spec__',
|
||||
'__builtins__', '__file__'}:
|
||||
repl_locals[key] = locals()[key]
|
||||
|
||||
console = AsyncIOInteractiveConsole(repl_locals, loop)
|
||||
|
||||
repl_future = None
|
||||
repl_future_interrupted = False
|
||||
|
||||
try:
|
||||
import readline # NoQA
|
||||
except ImportError:
|
||||
pass
|
||||
|
||||
repl_thread = REPLThread()
|
||||
repl_thread.daemon = True
|
||||
repl_thread.start()
|
||||
|
||||
while True:
|
||||
try:
|
||||
loop.run_forever()
|
||||
except KeyboardInterrupt:
|
||||
if repl_future and not repl_future.done():
|
||||
repl_future.cancel()
|
||||
repl_future_interrupted = True
|
||||
continue
|
||||
else:
|
||||
break
|
||||
File diff suppressed because it is too large
Load Diff
@@ -0,0 +1,80 @@
|
||||
__all__ = ()
|
||||
|
||||
import reprlib
|
||||
from _thread import get_ident
|
||||
|
||||
from . import format_helpers
|
||||
|
||||
# States for Future.
|
||||
_PENDING = 'PENDING'
|
||||
_CANCELLED = 'CANCELLED'
|
||||
_FINISHED = 'FINISHED'
|
||||
|
||||
|
||||
def isfuture(obj):
|
||||
"""Check for a Future.
|
||||
|
||||
This returns True when obj is a Future instance or is advertising
|
||||
itself as duck-type compatible by setting _asyncio_future_blocking.
|
||||
See comment in Future for more details.
|
||||
"""
|
||||
return (hasattr(obj.__class__, '_asyncio_future_blocking') and
|
||||
obj._asyncio_future_blocking is not None)
|
||||
|
||||
|
||||
def _format_callbacks(cb):
|
||||
"""helper function for Future.__repr__"""
|
||||
size = len(cb)
|
||||
if not size:
|
||||
cb = ''
|
||||
|
||||
def format_cb(callback):
|
||||
return format_helpers._format_callback_source(callback, ())
|
||||
|
||||
if size == 1:
|
||||
cb = format_cb(cb[0][0])
|
||||
elif size == 2:
|
||||
cb = '{}, {}'.format(format_cb(cb[0][0]), format_cb(cb[1][0]))
|
||||
elif size > 2:
|
||||
cb = '{}, <{} more>, {}'.format(format_cb(cb[0][0]),
|
||||
size - 2,
|
||||
format_cb(cb[-1][0]))
|
||||
return f'cb=[{cb}]'
|
||||
|
||||
|
||||
# bpo-42183: _repr_running is needed for repr protection
|
||||
# when a Future or Task result contains itself directly or indirectly.
|
||||
# The logic is borrowed from @reprlib.recursive_repr decorator.
|
||||
# Unfortunately, the direct decorator usage is impossible because of
|
||||
# AttributeError: '_asyncio.Task' object has no attribute '__module__' error.
|
||||
#
|
||||
# After fixing this thing we can return to the decorator based approach.
|
||||
_repr_running = set()
|
||||
|
||||
|
||||
def _future_repr_info(future):
|
||||
# (Future) -> str
|
||||
"""helper function for Future.__repr__"""
|
||||
info = [future._state.lower()]
|
||||
if future._state == _FINISHED:
|
||||
if future._exception is not None:
|
||||
info.append(f'exception={future._exception!r}')
|
||||
else:
|
||||
key = id(future), get_ident()
|
||||
if key in _repr_running:
|
||||
result = '...'
|
||||
else:
|
||||
_repr_running.add(key)
|
||||
try:
|
||||
# use reprlib to limit the length of the output, especially
|
||||
# for very long strings
|
||||
result = reprlib.repr(future._result)
|
||||
finally:
|
||||
_repr_running.discard(key)
|
||||
info.append(f'result={result}')
|
||||
if future._callbacks:
|
||||
info.append(_format_callbacks(future._callbacks))
|
||||
if future._source_traceback:
|
||||
frame = future._source_traceback[-1]
|
||||
info.append(f'created at {frame[0]}:{frame[1]}')
|
||||
return info
|
||||
@@ -0,0 +1,285 @@
|
||||
import collections
|
||||
import subprocess
|
||||
import warnings
|
||||
|
||||
from . import protocols
|
||||
from . import transports
|
||||
from .log import logger
|
||||
|
||||
|
||||
class BaseSubprocessTransport(transports.SubprocessTransport):
|
||||
|
||||
def __init__(self, loop, protocol, args, shell,
|
||||
stdin, stdout, stderr, bufsize,
|
||||
waiter=None, extra=None, **kwargs):
|
||||
super().__init__(extra)
|
||||
self._closed = False
|
||||
self._protocol = protocol
|
||||
self._loop = loop
|
||||
self._proc = None
|
||||
self._pid = None
|
||||
self._returncode = None
|
||||
self._exit_waiters = []
|
||||
self._pending_calls = collections.deque()
|
||||
self._pipes = {}
|
||||
self._finished = False
|
||||
|
||||
if stdin == subprocess.PIPE:
|
||||
self._pipes[0] = None
|
||||
if stdout == subprocess.PIPE:
|
||||
self._pipes[1] = None
|
||||
if stderr == subprocess.PIPE:
|
||||
self._pipes[2] = None
|
||||
|
||||
# Create the child process: set the _proc attribute
|
||||
try:
|
||||
self._start(args=args, shell=shell, stdin=stdin, stdout=stdout,
|
||||
stderr=stderr, bufsize=bufsize, **kwargs)
|
||||
except:
|
||||
self.close()
|
||||
raise
|
||||
|
||||
self._pid = self._proc.pid
|
||||
self._extra['subprocess'] = self._proc
|
||||
|
||||
if self._loop.get_debug():
|
||||
if isinstance(args, (bytes, str)):
|
||||
program = args
|
||||
else:
|
||||
program = args[0]
|
||||
logger.debug('process %r created: pid %s',
|
||||
program, self._pid)
|
||||
|
||||
self._loop.create_task(self._connect_pipes(waiter))
|
||||
|
||||
def __repr__(self):
|
||||
info = [self.__class__.__name__]
|
||||
if self._closed:
|
||||
info.append('closed')
|
||||
if self._pid is not None:
|
||||
info.append(f'pid={self._pid}')
|
||||
if self._returncode is not None:
|
||||
info.append(f'returncode={self._returncode}')
|
||||
elif self._pid is not None:
|
||||
info.append('running')
|
||||
else:
|
||||
info.append('not started')
|
||||
|
||||
stdin = self._pipes.get(0)
|
||||
if stdin is not None:
|
||||
info.append(f'stdin={stdin.pipe}')
|
||||
|
||||
stdout = self._pipes.get(1)
|
||||
stderr = self._pipes.get(2)
|
||||
if stdout is not None and stderr is stdout:
|
||||
info.append(f'stdout=stderr={stdout.pipe}')
|
||||
else:
|
||||
if stdout is not None:
|
||||
info.append(f'stdout={stdout.pipe}')
|
||||
if stderr is not None:
|
||||
info.append(f'stderr={stderr.pipe}')
|
||||
|
||||
return '<{}>'.format(' '.join(info))
|
||||
|
||||
def _start(self, args, shell, stdin, stdout, stderr, bufsize, **kwargs):
|
||||
raise NotImplementedError
|
||||
|
||||
def set_protocol(self, protocol):
|
||||
self._protocol = protocol
|
||||
|
||||
def get_protocol(self):
|
||||
return self._protocol
|
||||
|
||||
def is_closing(self):
|
||||
return self._closed
|
||||
|
||||
def close(self):
|
||||
if self._closed:
|
||||
return
|
||||
self._closed = True
|
||||
|
||||
for proto in self._pipes.values():
|
||||
if proto is None:
|
||||
continue
|
||||
proto.pipe.close()
|
||||
|
||||
if (self._proc is not None and
|
||||
# has the child process finished?
|
||||
self._returncode is None and
|
||||
# the child process has finished, but the
|
||||
# transport hasn't been notified yet?
|
||||
self._proc.poll() is None):
|
||||
|
||||
if self._loop.get_debug():
|
||||
logger.warning('Close running child process: kill %r', self)
|
||||
|
||||
try:
|
||||
self._proc.kill()
|
||||
except ProcessLookupError:
|
||||
pass
|
||||
|
||||
# Don't clear the _proc reference yet: _post_init() may still run
|
||||
|
||||
def __del__(self, _warn=warnings.warn):
|
||||
if not self._closed:
|
||||
_warn(f"unclosed transport {self!r}", ResourceWarning, source=self)
|
||||
self.close()
|
||||
|
||||
def get_pid(self):
|
||||
return self._pid
|
||||
|
||||
def get_returncode(self):
|
||||
return self._returncode
|
||||
|
||||
def get_pipe_transport(self, fd):
|
||||
if fd in self._pipes:
|
||||
return self._pipes[fd].pipe
|
||||
else:
|
||||
return None
|
||||
|
||||
def _check_proc(self):
|
||||
if self._proc is None:
|
||||
raise ProcessLookupError()
|
||||
|
||||
def send_signal(self, signal):
|
||||
self._check_proc()
|
||||
self._proc.send_signal(signal)
|
||||
|
||||
def terminate(self):
|
||||
self._check_proc()
|
||||
self._proc.terminate()
|
||||
|
||||
def kill(self):
|
||||
self._check_proc()
|
||||
self._proc.kill()
|
||||
|
||||
async def _connect_pipes(self, waiter):
|
||||
try:
|
||||
proc = self._proc
|
||||
loop = self._loop
|
||||
|
||||
if proc.stdin is not None:
|
||||
_, pipe = await loop.connect_write_pipe(
|
||||
lambda: WriteSubprocessPipeProto(self, 0),
|
||||
proc.stdin)
|
||||
self._pipes[0] = pipe
|
||||
|
||||
if proc.stdout is not None:
|
||||
_, pipe = await loop.connect_read_pipe(
|
||||
lambda: ReadSubprocessPipeProto(self, 1),
|
||||
proc.stdout)
|
||||
self._pipes[1] = pipe
|
||||
|
||||
if proc.stderr is not None:
|
||||
_, pipe = await loop.connect_read_pipe(
|
||||
lambda: ReadSubprocessPipeProto(self, 2),
|
||||
proc.stderr)
|
||||
self._pipes[2] = pipe
|
||||
|
||||
assert self._pending_calls is not None
|
||||
|
||||
loop.call_soon(self._protocol.connection_made, self)
|
||||
for callback, data in self._pending_calls:
|
||||
loop.call_soon(callback, *data)
|
||||
self._pending_calls = None
|
||||
except (SystemExit, KeyboardInterrupt):
|
||||
raise
|
||||
except BaseException as exc:
|
||||
if waiter is not None and not waiter.cancelled():
|
||||
waiter.set_exception(exc)
|
||||
else:
|
||||
if waiter is not None and not waiter.cancelled():
|
||||
waiter.set_result(None)
|
||||
|
||||
def _call(self, cb, *data):
|
||||
if self._pending_calls is not None:
|
||||
self._pending_calls.append((cb, data))
|
||||
else:
|
||||
self._loop.call_soon(cb, *data)
|
||||
|
||||
def _pipe_connection_lost(self, fd, exc):
|
||||
self._call(self._protocol.pipe_connection_lost, fd, exc)
|
||||
self._try_finish()
|
||||
|
||||
def _pipe_data_received(self, fd, data):
|
||||
self._call(self._protocol.pipe_data_received, fd, data)
|
||||
|
||||
def _process_exited(self, returncode):
|
||||
assert returncode is not None, returncode
|
||||
assert self._returncode is None, self._returncode
|
||||
if self._loop.get_debug():
|
||||
logger.info('%r exited with return code %r', self, returncode)
|
||||
self._returncode = returncode
|
||||
if self._proc.returncode is None:
|
||||
# asyncio uses a child watcher: copy the status into the Popen
|
||||
# object. On Python 3.6, it is required to avoid a ResourceWarning.
|
||||
self._proc.returncode = returncode
|
||||
self._call(self._protocol.process_exited)
|
||||
self._try_finish()
|
||||
|
||||
# wake up futures waiting for wait()
|
||||
for waiter in self._exit_waiters:
|
||||
if not waiter.cancelled():
|
||||
waiter.set_result(returncode)
|
||||
self._exit_waiters = None
|
||||
|
||||
async def _wait(self):
|
||||
"""Wait until the process exit and return the process return code.
|
||||
|
||||
This method is a coroutine."""
|
||||
if self._returncode is not None:
|
||||
return self._returncode
|
||||
|
||||
waiter = self._loop.create_future()
|
||||
self._exit_waiters.append(waiter)
|
||||
return await waiter
|
||||
|
||||
def _try_finish(self):
|
||||
assert not self._finished
|
||||
if self._returncode is None:
|
||||
return
|
||||
if all(p is not None and p.disconnected
|
||||
for p in self._pipes.values()):
|
||||
self._finished = True
|
||||
self._call(self._call_connection_lost, None)
|
||||
|
||||
def _call_connection_lost(self, exc):
|
||||
try:
|
||||
self._protocol.connection_lost(exc)
|
||||
finally:
|
||||
self._loop = None
|
||||
self._proc = None
|
||||
self._protocol = None
|
||||
|
||||
|
||||
class WriteSubprocessPipeProto(protocols.BaseProtocol):
|
||||
|
||||
def __init__(self, proc, fd):
|
||||
self.proc = proc
|
||||
self.fd = fd
|
||||
self.pipe = None
|
||||
self.disconnected = False
|
||||
|
||||
def connection_made(self, transport):
|
||||
self.pipe = transport
|
||||
|
||||
def __repr__(self):
|
||||
return f'<{self.__class__.__name__} fd={self.fd} pipe={self.pipe!r}>'
|
||||
|
||||
def connection_lost(self, exc):
|
||||
self.disconnected = True
|
||||
self.proc._pipe_connection_lost(self.fd, exc)
|
||||
self.proc = None
|
||||
|
||||
def pause_writing(self):
|
||||
self.proc._protocol.pause_writing()
|
||||
|
||||
def resume_writing(self):
|
||||
self.proc._protocol.resume_writing()
|
||||
|
||||
|
||||
class ReadSubprocessPipeProto(WriteSubprocessPipeProto,
|
||||
protocols.Protocol):
|
||||
|
||||
def data_received(self, data):
|
||||
self.proc._pipe_data_received(self.fd, data)
|
||||
@@ -0,0 +1,85 @@
|
||||
import linecache
|
||||
import traceback
|
||||
|
||||
from . import base_futures
|
||||
from . import coroutines
|
||||
|
||||
|
||||
def _task_repr_info(task):
|
||||
info = base_futures._future_repr_info(task)
|
||||
|
||||
if task._must_cancel:
|
||||
# replace status
|
||||
info[0] = 'cancelling'
|
||||
|
||||
info.insert(1, 'name=%r' % task.get_name())
|
||||
|
||||
coro = coroutines._format_coroutine(task._coro)
|
||||
info.insert(2, f'coro=<{coro}>')
|
||||
|
||||
if task._fut_waiter is not None:
|
||||
info.insert(3, f'wait_for={task._fut_waiter!r}')
|
||||
return info
|
||||
|
||||
|
||||
def _task_get_stack(task, limit):
|
||||
frames = []
|
||||
if hasattr(task._coro, 'cr_frame'):
|
||||
# case 1: 'async def' coroutines
|
||||
f = task._coro.cr_frame
|
||||
elif hasattr(task._coro, 'gi_frame'):
|
||||
# case 2: legacy coroutines
|
||||
f = task._coro.gi_frame
|
||||
elif hasattr(task._coro, 'ag_frame'):
|
||||
# case 3: async generators
|
||||
f = task._coro.ag_frame
|
||||
else:
|
||||
# case 4: unknown objects
|
||||
f = None
|
||||
if f is not None:
|
||||
while f is not None:
|
||||
if limit is not None:
|
||||
if limit <= 0:
|
||||
break
|
||||
limit -= 1
|
||||
frames.append(f)
|
||||
f = f.f_back
|
||||
frames.reverse()
|
||||
elif task._exception is not None:
|
||||
tb = task._exception.__traceback__
|
||||
while tb is not None:
|
||||
if limit is not None:
|
||||
if limit <= 0:
|
||||
break
|
||||
limit -= 1
|
||||
frames.append(tb.tb_frame)
|
||||
tb = tb.tb_next
|
||||
return frames
|
||||
|
||||
|
||||
def _task_print_stack(task, limit, file):
|
||||
extracted_list = []
|
||||
checked = set()
|
||||
for f in task.get_stack(limit=limit):
|
||||
lineno = f.f_lineno
|
||||
co = f.f_code
|
||||
filename = co.co_filename
|
||||
name = co.co_name
|
||||
if filename not in checked:
|
||||
checked.add(filename)
|
||||
linecache.checkcache(filename)
|
||||
line = linecache.getline(filename, lineno, f.f_globals)
|
||||
extracted_list.append((filename, lineno, name, line))
|
||||
|
||||
exc = task._exception
|
||||
if not extracted_list:
|
||||
print(f'No stack for {task!r}', file=file)
|
||||
elif exc is not None:
|
||||
print(f'Traceback for {task!r} (most recent call last):', file=file)
|
||||
else:
|
||||
print(f'Stack for {task!r} (most recent call last):', file=file)
|
||||
|
||||
traceback.print_list(extracted_list, file=file)
|
||||
if exc is not None:
|
||||
for line in traceback.format_exception_only(exc.__class__, exc):
|
||||
print(line, file=file, end='')
|
||||
@@ -0,0 +1,27 @@
|
||||
import enum
|
||||
|
||||
# After the connection is lost, log warnings after this many write()s.
|
||||
LOG_THRESHOLD_FOR_CONNLOST_WRITES = 5
|
||||
|
||||
# Seconds to wait before retrying accept().
|
||||
ACCEPT_RETRY_DELAY = 1
|
||||
|
||||
# Number of stack entries to capture in debug mode.
|
||||
# The larger the number, the slower the operation in debug mode
|
||||
# (see extract_stack() in format_helpers.py).
|
||||
DEBUG_STACK_DEPTH = 10
|
||||
|
||||
# Number of seconds to wait for SSL handshake to complete
|
||||
# The default timeout matches that of Nginx.
|
||||
SSL_HANDSHAKE_TIMEOUT = 60.0
|
||||
|
||||
# Used in sendfile fallback code. We use fallback for platforms
|
||||
# that don't support sendfile, or for TLS connections.
|
||||
SENDFILE_FALLBACK_READBUFFER_SIZE = 1024 * 256
|
||||
|
||||
# The enum should be here to break circular dependencies between
|
||||
# base_events and sslproto
|
||||
class _SendfileMode(enum.Enum):
|
||||
UNSUPPORTED = enum.auto()
|
||||
TRY_NATIVE = enum.auto()
|
||||
FALLBACK = enum.auto()
|
||||
@@ -0,0 +1,269 @@
|
||||
__all__ = 'coroutine', 'iscoroutinefunction', 'iscoroutine'
|
||||
|
||||
import collections.abc
|
||||
import functools
|
||||
import inspect
|
||||
import os
|
||||
import sys
|
||||
import traceback
|
||||
import types
|
||||
import warnings
|
||||
|
||||
from . import base_futures
|
||||
from . import constants
|
||||
from . import format_helpers
|
||||
from .log import logger
|
||||
|
||||
|
||||
def _is_debug_mode():
|
||||
# If you set _DEBUG to true, @coroutine will wrap the resulting
|
||||
# generator objects in a CoroWrapper instance (defined below). That
|
||||
# instance will log a message when the generator is never iterated
|
||||
# over, which may happen when you forget to use "await" or "yield from"
|
||||
# with a coroutine call.
|
||||
# Note that the value of the _DEBUG flag is taken
|
||||
# when the decorator is used, so to be of any use it must be set
|
||||
# before you define your coroutines. A downside of using this feature
|
||||
# is that tracebacks show entries for the CoroWrapper.__next__ method
|
||||
# when _DEBUG is true.
|
||||
return sys.flags.dev_mode or (not sys.flags.ignore_environment and
|
||||
bool(os.environ.get('PYTHONASYNCIODEBUG')))
|
||||
|
||||
|
||||
_DEBUG = _is_debug_mode()
|
||||
|
||||
|
||||
class CoroWrapper:
|
||||
# Wrapper for coroutine object in _DEBUG mode.
|
||||
|
||||
def __init__(self, gen, func=None):
|
||||
assert inspect.isgenerator(gen) or inspect.iscoroutine(gen), gen
|
||||
self.gen = gen
|
||||
self.func = func # Used to unwrap @coroutine decorator
|
||||
self._source_traceback = format_helpers.extract_stack(sys._getframe(1))
|
||||
self.__name__ = getattr(gen, '__name__', None)
|
||||
self.__qualname__ = getattr(gen, '__qualname__', None)
|
||||
|
||||
def __repr__(self):
|
||||
coro_repr = _format_coroutine(self)
|
||||
if self._source_traceback:
|
||||
frame = self._source_traceback[-1]
|
||||
coro_repr += f', created at {frame[0]}:{frame[1]}'
|
||||
|
||||
return f'<{self.__class__.__name__} {coro_repr}>'
|
||||
|
||||
def __iter__(self):
|
||||
return self
|
||||
|
||||
def __next__(self):
|
||||
return self.gen.send(None)
|
||||
|
||||
def send(self, value):
|
||||
return self.gen.send(value)
|
||||
|
||||
def throw(self, type, value=None, traceback=None):
|
||||
return self.gen.throw(type, value, traceback)
|
||||
|
||||
def close(self):
|
||||
return self.gen.close()
|
||||
|
||||
@property
|
||||
def gi_frame(self):
|
||||
return self.gen.gi_frame
|
||||
|
||||
@property
|
||||
def gi_running(self):
|
||||
return self.gen.gi_running
|
||||
|
||||
@property
|
||||
def gi_code(self):
|
||||
return self.gen.gi_code
|
||||
|
||||
def __await__(self):
|
||||
return self
|
||||
|
||||
@property
|
||||
def gi_yieldfrom(self):
|
||||
return self.gen.gi_yieldfrom
|
||||
|
||||
def __del__(self):
|
||||
# Be careful accessing self.gen.frame -- self.gen might not exist.
|
||||
gen = getattr(self, 'gen', None)
|
||||
frame = getattr(gen, 'gi_frame', None)
|
||||
if frame is not None and frame.f_lasti == -1:
|
||||
msg = f'{self!r} was never yielded from'
|
||||
tb = getattr(self, '_source_traceback', ())
|
||||
if tb:
|
||||
tb = ''.join(traceback.format_list(tb))
|
||||
msg += (f'\nCoroutine object created at '
|
||||
f'(most recent call last, truncated to '
|
||||
f'{constants.DEBUG_STACK_DEPTH} last lines):\n')
|
||||
msg += tb.rstrip()
|
||||
logger.error(msg)
|
||||
|
||||
|
||||
def coroutine(func):
|
||||
"""Decorator to mark coroutines.
|
||||
|
||||
If the coroutine is not yielded from before it is destroyed,
|
||||
an error message is logged.
|
||||
"""
|
||||
warnings.warn('"@coroutine" decorator is deprecated since Python 3.8, use "async def" instead',
|
||||
DeprecationWarning,
|
||||
stacklevel=2)
|
||||
if inspect.iscoroutinefunction(func):
|
||||
# In Python 3.5 that's all we need to do for coroutines
|
||||
# defined with "async def".
|
||||
return func
|
||||
|
||||
if inspect.isgeneratorfunction(func):
|
||||
coro = func
|
||||
else:
|
||||
@functools.wraps(func)
|
||||
def coro(*args, **kw):
|
||||
res = func(*args, **kw)
|
||||
if (base_futures.isfuture(res) or inspect.isgenerator(res) or
|
||||
isinstance(res, CoroWrapper)):
|
||||
res = yield from res
|
||||
else:
|
||||
# If 'res' is an awaitable, run it.
|
||||
try:
|
||||
await_meth = res.__await__
|
||||
except AttributeError:
|
||||
pass
|
||||
else:
|
||||
if isinstance(res, collections.abc.Awaitable):
|
||||
res = yield from await_meth()
|
||||
return res
|
||||
|
||||
coro = types.coroutine(coro)
|
||||
if not _DEBUG:
|
||||
wrapper = coro
|
||||
else:
|
||||
@functools.wraps(func)
|
||||
def wrapper(*args, **kwds):
|
||||
w = CoroWrapper(coro(*args, **kwds), func=func)
|
||||
if w._source_traceback:
|
||||
del w._source_traceback[-1]
|
||||
# Python < 3.5 does not implement __qualname__
|
||||
# on generator objects, so we set it manually.
|
||||
# We use getattr as some callables (such as
|
||||
# functools.partial may lack __qualname__).
|
||||
w.__name__ = getattr(func, '__name__', None)
|
||||
w.__qualname__ = getattr(func, '__qualname__', None)
|
||||
return w
|
||||
|
||||
wrapper._is_coroutine = _is_coroutine # For iscoroutinefunction().
|
||||
return wrapper
|
||||
|
||||
|
||||
# A marker for iscoroutinefunction.
|
||||
_is_coroutine = object()
|
||||
|
||||
|
||||
def iscoroutinefunction(func):
|
||||
"""Return True if func is a decorated coroutine function."""
|
||||
return (inspect.iscoroutinefunction(func) or
|
||||
getattr(func, '_is_coroutine', None) is _is_coroutine)
|
||||
|
||||
|
||||
# Prioritize native coroutine check to speed-up
|
||||
# asyncio.iscoroutine.
|
||||
_COROUTINE_TYPES = (types.CoroutineType, types.GeneratorType,
|
||||
collections.abc.Coroutine, CoroWrapper)
|
||||
_iscoroutine_typecache = set()
|
||||
|
||||
|
||||
def iscoroutine(obj):
|
||||
"""Return True if obj is a coroutine object."""
|
||||
if type(obj) in _iscoroutine_typecache:
|
||||
return True
|
||||
|
||||
if isinstance(obj, _COROUTINE_TYPES):
|
||||
# Just in case we don't want to cache more than 100
|
||||
# positive types. That shouldn't ever happen, unless
|
||||
# someone stressing the system on purpose.
|
||||
if len(_iscoroutine_typecache) < 100:
|
||||
_iscoroutine_typecache.add(type(obj))
|
||||
return True
|
||||
else:
|
||||
return False
|
||||
|
||||
|
||||
def _format_coroutine(coro):
|
||||
assert iscoroutine(coro)
|
||||
|
||||
is_corowrapper = isinstance(coro, CoroWrapper)
|
||||
|
||||
def get_name(coro):
|
||||
# Coroutines compiled with Cython sometimes don't have
|
||||
# proper __qualname__ or __name__. While that is a bug
|
||||
# in Cython, asyncio shouldn't crash with an AttributeError
|
||||
# in its __repr__ functions.
|
||||
if is_corowrapper:
|
||||
return format_helpers._format_callback(coro.func, (), {})
|
||||
|
||||
if hasattr(coro, '__qualname__') and coro.__qualname__:
|
||||
coro_name = coro.__qualname__
|
||||
elif hasattr(coro, '__name__') and coro.__name__:
|
||||
coro_name = coro.__name__
|
||||
else:
|
||||
# Stop masking Cython bugs, expose them in a friendly way.
|
||||
coro_name = f'<{type(coro).__name__} without __name__>'
|
||||
return f'{coro_name}()'
|
||||
|
||||
def is_running(coro):
|
||||
try:
|
||||
return coro.cr_running
|
||||
except AttributeError:
|
||||
try:
|
||||
return coro.gi_running
|
||||
except AttributeError:
|
||||
return False
|
||||
|
||||
coro_code = None
|
||||
if hasattr(coro, 'cr_code') and coro.cr_code:
|
||||
coro_code = coro.cr_code
|
||||
elif hasattr(coro, 'gi_code') and coro.gi_code:
|
||||
coro_code = coro.gi_code
|
||||
|
||||
coro_name = get_name(coro)
|
||||
|
||||
if not coro_code:
|
||||
# Built-in types might not have __qualname__ or __name__.
|
||||
if is_running(coro):
|
||||
return f'{coro_name} running'
|
||||
else:
|
||||
return coro_name
|
||||
|
||||
coro_frame = None
|
||||
if hasattr(coro, 'gi_frame') and coro.gi_frame:
|
||||
coro_frame = coro.gi_frame
|
||||
elif hasattr(coro, 'cr_frame') and coro.cr_frame:
|
||||
coro_frame = coro.cr_frame
|
||||
|
||||
# If Cython's coroutine has a fake code object without proper
|
||||
# co_filename -- expose that.
|
||||
filename = coro_code.co_filename or '<empty co_filename>'
|
||||
|
||||
lineno = 0
|
||||
if (is_corowrapper and
|
||||
coro.func is not None and
|
||||
not inspect.isgeneratorfunction(coro.func)):
|
||||
source = format_helpers._get_function_source(coro.func)
|
||||
if source is not None:
|
||||
filename, lineno = source
|
||||
if coro_frame is None:
|
||||
coro_repr = f'{coro_name} done, defined at {filename}:{lineno}'
|
||||
else:
|
||||
coro_repr = f'{coro_name} running, defined at {filename}:{lineno}'
|
||||
|
||||
elif coro_frame is not None:
|
||||
lineno = coro_frame.f_lineno
|
||||
coro_repr = f'{coro_name} running at {filename}:{lineno}'
|
||||
|
||||
else:
|
||||
lineno = coro_code.co_firstlineno
|
||||
coro_repr = f'{coro_name} done, defined at {filename}:{lineno}'
|
||||
|
||||
return coro_repr
|
||||
File diff suppressed because it is too large
Load Diff
@@ -0,0 +1,58 @@
|
||||
"""asyncio exceptions."""
|
||||
|
||||
|
||||
__all__ = ('CancelledError', 'InvalidStateError', 'TimeoutError',
|
||||
'IncompleteReadError', 'LimitOverrunError',
|
||||
'SendfileNotAvailableError')
|
||||
|
||||
|
||||
class CancelledError(BaseException):
|
||||
"""The Future or Task was cancelled."""
|
||||
|
||||
|
||||
class TimeoutError(Exception):
|
||||
"""The operation exceeded the given deadline."""
|
||||
|
||||
|
||||
class InvalidStateError(Exception):
|
||||
"""The operation is not allowed in this state."""
|
||||
|
||||
|
||||
class SendfileNotAvailableError(RuntimeError):
|
||||
"""Sendfile syscall is not available.
|
||||
|
||||
Raised if OS does not support sendfile syscall for given socket or
|
||||
file type.
|
||||
"""
|
||||
|
||||
|
||||
class IncompleteReadError(EOFError):
|
||||
"""
|
||||
Incomplete read error. Attributes:
|
||||
|
||||
- partial: read bytes string before the end of stream was reached
|
||||
- expected: total number of expected bytes (or None if unknown)
|
||||
"""
|
||||
def __init__(self, partial, expected):
|
||||
r_expected = 'undefined' if expected is None else repr(expected)
|
||||
super().__init__(f'{len(partial)} bytes read on a total of '
|
||||
f'{r_expected} expected bytes')
|
||||
self.partial = partial
|
||||
self.expected = expected
|
||||
|
||||
def __reduce__(self):
|
||||
return type(self), (self.partial, self.expected)
|
||||
|
||||
|
||||
class LimitOverrunError(Exception):
|
||||
"""Reached the buffer limit while looking for a separator.
|
||||
|
||||
Attributes:
|
||||
- consumed: total number of to be consumed bytes.
|
||||
"""
|
||||
def __init__(self, message, consumed):
|
||||
super().__init__(message)
|
||||
self.consumed = consumed
|
||||
|
||||
def __reduce__(self):
|
||||
return type(self), (self.args[0], self.consumed)
|
||||
@@ -0,0 +1,76 @@
|
||||
import functools
|
||||
import inspect
|
||||
import reprlib
|
||||
import sys
|
||||
import traceback
|
||||
|
||||
from . import constants
|
||||
|
||||
|
||||
def _get_function_source(func):
|
||||
func = inspect.unwrap(func)
|
||||
if inspect.isfunction(func):
|
||||
code = func.__code__
|
||||
return (code.co_filename, code.co_firstlineno)
|
||||
if isinstance(func, functools.partial):
|
||||
return _get_function_source(func.func)
|
||||
if isinstance(func, functools.partialmethod):
|
||||
return _get_function_source(func.func)
|
||||
return None
|
||||
|
||||
|
||||
def _format_callback_source(func, args):
|
||||
func_repr = _format_callback(func, args, None)
|
||||
source = _get_function_source(func)
|
||||
if source:
|
||||
func_repr += f' at {source[0]}:{source[1]}'
|
||||
return func_repr
|
||||
|
||||
|
||||
def _format_args_and_kwargs(args, kwargs):
|
||||
"""Format function arguments and keyword arguments.
|
||||
|
||||
Special case for a single parameter: ('hello',) is formatted as ('hello').
|
||||
"""
|
||||
# use reprlib to limit the length of the output
|
||||
items = []
|
||||
if args:
|
||||
items.extend(reprlib.repr(arg) for arg in args)
|
||||
if kwargs:
|
||||
items.extend(f'{k}={reprlib.repr(v)}' for k, v in kwargs.items())
|
||||
return '({})'.format(', '.join(items))
|
||||
|
||||
|
||||
def _format_callback(func, args, kwargs, suffix=''):
|
||||
if isinstance(func, functools.partial):
|
||||
suffix = _format_args_and_kwargs(args, kwargs) + suffix
|
||||
return _format_callback(func.func, func.args, func.keywords, suffix)
|
||||
|
||||
if hasattr(func, '__qualname__') and func.__qualname__:
|
||||
func_repr = func.__qualname__
|
||||
elif hasattr(func, '__name__') and func.__name__:
|
||||
func_repr = func.__name__
|
||||
else:
|
||||
func_repr = repr(func)
|
||||
|
||||
func_repr += _format_args_and_kwargs(args, kwargs)
|
||||
if suffix:
|
||||
func_repr += suffix
|
||||
return func_repr
|
||||
|
||||
|
||||
def extract_stack(f=None, limit=None):
|
||||
"""Replacement for traceback.extract_stack() that only does the
|
||||
necessary work for asyncio debug mode.
|
||||
"""
|
||||
if f is None:
|
||||
f = sys._getframe().f_back
|
||||
if limit is None:
|
||||
# Limit the amount of work to a reasonable amount, as extract_stack()
|
||||
# can be called for each coroutine and future in debug mode.
|
||||
limit = constants.DEBUG_STACK_DEPTH
|
||||
stack = traceback.StackSummary.extract(traceback.walk_stack(f),
|
||||
limit=limit,
|
||||
lookup_lines=False)
|
||||
stack.reverse()
|
||||
return stack
|
||||
@@ -0,0 +1,399 @@
|
||||
"""A Future class similar to the one in PEP 3148."""
|
||||
|
||||
__all__ = (
|
||||
'Future', 'wrap_future', 'isfuture',
|
||||
)
|
||||
|
||||
import concurrent.futures
|
||||
import contextvars
|
||||
import logging
|
||||
import sys
|
||||
|
||||
from . import base_futures
|
||||
from . import events
|
||||
from . import exceptions
|
||||
from . import format_helpers
|
||||
|
||||
|
||||
isfuture = base_futures.isfuture
|
||||
|
||||
|
||||
_PENDING = base_futures._PENDING
|
||||
_CANCELLED = base_futures._CANCELLED
|
||||
_FINISHED = base_futures._FINISHED
|
||||
|
||||
|
||||
STACK_DEBUG = logging.DEBUG - 1 # heavy-duty debugging
|
||||
|
||||
|
||||
class Future:
|
||||
"""This class is *almost* compatible with concurrent.futures.Future.
|
||||
|
||||
Differences:
|
||||
|
||||
- This class is not thread-safe.
|
||||
|
||||
- result() and exception() do not take a timeout argument and
|
||||
raise an exception when the future isn't done yet.
|
||||
|
||||
- Callbacks registered with add_done_callback() are always called
|
||||
via the event loop's call_soon().
|
||||
|
||||
- This class is not compatible with the wait() and as_completed()
|
||||
methods in the concurrent.futures package.
|
||||
|
||||
(In Python 3.4 or later we may be able to unify the implementations.)
|
||||
"""
|
||||
|
||||
# Class variables serving as defaults for instance variables.
|
||||
_state = _PENDING
|
||||
_result = None
|
||||
_exception = None
|
||||
_loop = None
|
||||
_source_traceback = None
|
||||
|
||||
# This field is used for a dual purpose:
|
||||
# - Its presence is a marker to declare that a class implements
|
||||
# the Future protocol (i.e. is intended to be duck-type compatible).
|
||||
# The value must also be not-None, to enable a subclass to declare
|
||||
# that it is not compatible by setting this to None.
|
||||
# - It is set by __iter__() below so that Task._step() can tell
|
||||
# the difference between
|
||||
# `await Future()` or`yield from Future()` (correct) vs.
|
||||
# `yield Future()` (incorrect).
|
||||
_asyncio_future_blocking = False
|
||||
|
||||
__log_traceback = False
|
||||
|
||||
def __init__(self, *, loop=None):
|
||||
"""Initialize the future.
|
||||
|
||||
The optional event_loop argument allows explicitly setting the event
|
||||
loop object used by the future. If it's not provided, the future uses
|
||||
the default event loop.
|
||||
"""
|
||||
if loop is None:
|
||||
self._loop = events.get_event_loop()
|
||||
else:
|
||||
self._loop = loop
|
||||
self._callbacks = []
|
||||
if self._loop.get_debug():
|
||||
self._source_traceback = format_helpers.extract_stack(
|
||||
sys._getframe(1))
|
||||
|
||||
_repr_info = base_futures._future_repr_info
|
||||
|
||||
def __repr__(self):
|
||||
return '<{} {}>'.format(self.__class__.__name__,
|
||||
' '.join(self._repr_info()))
|
||||
|
||||
def __del__(self):
|
||||
if not self.__log_traceback:
|
||||
# set_exception() was not called, or result() or exception()
|
||||
# has consumed the exception
|
||||
return
|
||||
exc = self._exception
|
||||
context = {
|
||||
'message':
|
||||
f'{self.__class__.__name__} exception was never retrieved',
|
||||
'exception': exc,
|
||||
'future': self,
|
||||
}
|
||||
if self._source_traceback:
|
||||
context['source_traceback'] = self._source_traceback
|
||||
self._loop.call_exception_handler(context)
|
||||
|
||||
@property
|
||||
def _log_traceback(self):
|
||||
return self.__log_traceback
|
||||
|
||||
@_log_traceback.setter
|
||||
def _log_traceback(self, val):
|
||||
if bool(val):
|
||||
raise ValueError('_log_traceback can only be set to False')
|
||||
self.__log_traceback = False
|
||||
|
||||
def get_loop(self):
|
||||
"""Return the event loop the Future is bound to."""
|
||||
loop = self._loop
|
||||
if loop is None:
|
||||
raise RuntimeError("Future object is not initialized.")
|
||||
return loop
|
||||
|
||||
def cancel(self):
|
||||
"""Cancel the future and schedule callbacks.
|
||||
|
||||
If the future is already done or cancelled, return False. Otherwise,
|
||||
change the future's state to cancelled, schedule the callbacks and
|
||||
return True.
|
||||
"""
|
||||
self.__log_traceback = False
|
||||
if self._state != _PENDING:
|
||||
return False
|
||||
self._state = _CANCELLED
|
||||
self.__schedule_callbacks()
|
||||
return True
|
||||
|
||||
def __schedule_callbacks(self):
|
||||
"""Internal: Ask the event loop to call all callbacks.
|
||||
|
||||
The callbacks are scheduled to be called as soon as possible. Also
|
||||
clears the callback list.
|
||||
"""
|
||||
callbacks = self._callbacks[:]
|
||||
if not callbacks:
|
||||
return
|
||||
|
||||
self._callbacks[:] = []
|
||||
for callback, ctx in callbacks:
|
||||
self._loop.call_soon(callback, self, context=ctx)
|
||||
|
||||
def cancelled(self):
|
||||
"""Return True if the future was cancelled."""
|
||||
return self._state == _CANCELLED
|
||||
|
||||
# Don't implement running(); see http://bugs.python.org/issue18699
|
||||
|
||||
def done(self):
|
||||
"""Return True if the future is done.
|
||||
|
||||
Done means either that a result / exception are available, or that the
|
||||
future was cancelled.
|
||||
"""
|
||||
return self._state != _PENDING
|
||||
|
||||
def result(self):
|
||||
"""Return the result this future represents.
|
||||
|
||||
If the future has been cancelled, raises CancelledError. If the
|
||||
future's result isn't yet available, raises InvalidStateError. If
|
||||
the future is done and has an exception set, this exception is raised.
|
||||
"""
|
||||
if self._state == _CANCELLED:
|
||||
raise exceptions.CancelledError
|
||||
if self._state != _FINISHED:
|
||||
raise exceptions.InvalidStateError('Result is not ready.')
|
||||
self.__log_traceback = False
|
||||
if self._exception is not None:
|
||||
raise self._exception
|
||||
return self._result
|
||||
|
||||
def exception(self):
|
||||
"""Return the exception that was set on this future.
|
||||
|
||||
The exception (or None if no exception was set) is returned only if
|
||||
the future is done. If the future has been cancelled, raises
|
||||
CancelledError. If the future isn't done yet, raises
|
||||
InvalidStateError.
|
||||
"""
|
||||
if self._state == _CANCELLED:
|
||||
raise exceptions.CancelledError
|
||||
if self._state != _FINISHED:
|
||||
raise exceptions.InvalidStateError('Exception is not set.')
|
||||
self.__log_traceback = False
|
||||
return self._exception
|
||||
|
||||
def add_done_callback(self, fn, *, context=None):
|
||||
"""Add a callback to be run when the future becomes done.
|
||||
|
||||
The callback is called with a single argument - the future object. If
|
||||
the future is already done when this is called, the callback is
|
||||
scheduled with call_soon.
|
||||
"""
|
||||
if self._state != _PENDING:
|
||||
self._loop.call_soon(fn, self, context=context)
|
||||
else:
|
||||
if context is None:
|
||||
context = contextvars.copy_context()
|
||||
self._callbacks.append((fn, context))
|
||||
|
||||
# New method not in PEP 3148.
|
||||
|
||||
def remove_done_callback(self, fn):
|
||||
"""Remove all instances of a callback from the "call when done" list.
|
||||
|
||||
Returns the number of callbacks removed.
|
||||
"""
|
||||
filtered_callbacks = [(f, ctx)
|
||||
for (f, ctx) in self._callbacks
|
||||
if f != fn]
|
||||
removed_count = len(self._callbacks) - len(filtered_callbacks)
|
||||
if removed_count:
|
||||
self._callbacks[:] = filtered_callbacks
|
||||
return removed_count
|
||||
|
||||
# So-called internal methods (note: no set_running_or_notify_cancel()).
|
||||
|
||||
def set_result(self, result):
|
||||
"""Mark the future done and set its result.
|
||||
|
||||
If the future is already done when this method is called, raises
|
||||
InvalidStateError.
|
||||
"""
|
||||
if self._state != _PENDING:
|
||||
raise exceptions.InvalidStateError(f'{self._state}: {self!r}')
|
||||
self._result = result
|
||||
self._state = _FINISHED
|
||||
self.__schedule_callbacks()
|
||||
|
||||
def set_exception(self, exception):
|
||||
"""Mark the future done and set an exception.
|
||||
|
||||
If the future is already done when this method is called, raises
|
||||
InvalidStateError.
|
||||
"""
|
||||
if self._state != _PENDING:
|
||||
raise exceptions.InvalidStateError(f'{self._state}: {self!r}')
|
||||
if isinstance(exception, type):
|
||||
exception = exception()
|
||||
if type(exception) is StopIteration:
|
||||
raise TypeError("StopIteration interacts badly with generators "
|
||||
"and cannot be raised into a Future")
|
||||
self._exception = exception
|
||||
self._state = _FINISHED
|
||||
self.__schedule_callbacks()
|
||||
self.__log_traceback = True
|
||||
|
||||
def __await__(self):
|
||||
if not self.done():
|
||||
self._asyncio_future_blocking = True
|
||||
yield self # This tells Task to wait for completion.
|
||||
if not self.done():
|
||||
raise RuntimeError("await wasn't used with future")
|
||||
return self.result() # May raise too.
|
||||
|
||||
__iter__ = __await__ # make compatible with 'yield from'.
|
||||
|
||||
|
||||
# Needed for testing purposes.
|
||||
_PyFuture = Future
|
||||
|
||||
|
||||
def _get_loop(fut):
|
||||
# Tries to call Future.get_loop() if it's available.
|
||||
# Otherwise fallbacks to using the old '_loop' property.
|
||||
try:
|
||||
get_loop = fut.get_loop
|
||||
except AttributeError:
|
||||
pass
|
||||
else:
|
||||
return get_loop()
|
||||
return fut._loop
|
||||
|
||||
|
||||
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 _convert_future_exc(exc):
|
||||
exc_class = type(exc)
|
||||
if exc_class is concurrent.futures.CancelledError:
|
||||
return exceptions.CancelledError(*exc.args)
|
||||
elif exc_class is concurrent.futures.TimeoutError:
|
||||
return exceptions.TimeoutError(*exc.args)
|
||||
elif exc_class is concurrent.futures.InvalidStateError:
|
||||
return exceptions.InvalidStateError(*exc.args)
|
||||
else:
|
||||
return exc
|
||||
|
||||
|
||||
def _set_concurrent_future_state(concurrent, source):
|
||||
"""Copy state from a future to a concurrent.futures.Future."""
|
||||
assert source.done()
|
||||
if source.cancelled():
|
||||
concurrent.cancel()
|
||||
if not concurrent.set_running_or_notify_cancel():
|
||||
return
|
||||
exception = source.exception()
|
||||
if exception is not None:
|
||||
concurrent.set_exception(_convert_future_exc(exception))
|
||||
else:
|
||||
result = source.result()
|
||||
concurrent.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(_convert_future_exc(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 isfuture(source) and not isinstance(source,
|
||||
concurrent.futures.Future):
|
||||
raise TypeError('A future is required for source argument')
|
||||
if not isfuture(destination) and not isinstance(destination,
|
||||
concurrent.futures.Future):
|
||||
raise TypeError('A future is required for destination argument')
|
||||
source_loop = _get_loop(source) if isfuture(source) else None
|
||||
dest_loop = _get_loop(destination) if isfuture(destination) else None
|
||||
|
||||
def _set_state(future, other):
|
||||
if isfuture(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 (destination.cancelled() and
|
||||
dest_loop is not None and dest_loop.is_closed()):
|
||||
return
|
||||
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 wrap_future(future, *, loop=None):
|
||||
"""Wrap concurrent.futures.Future object."""
|
||||
if isfuture(future):
|
||||
return future
|
||||
assert isinstance(future, concurrent.futures.Future), \
|
||||
f'concurrent.futures.Future is expected, got {future!r}'
|
||||
if loop is None:
|
||||
loop = events.get_event_loop()
|
||||
new_future = loop.create_future()
|
||||
_chain_future(future, new_future)
|
||||
return new_future
|
||||
|
||||
|
||||
try:
|
||||
import _asyncio
|
||||
except ImportError:
|
||||
pass
|
||||
else:
|
||||
# _CFuture is needed for tests.
|
||||
Future = _CFuture = _asyncio.Future
|
||||
File diff suppressed because it is too large
Load Diff
@@ -0,0 +1,7 @@
|
||||
"""Logging configuration."""
|
||||
|
||||
import logging
|
||||
|
||||
|
||||
# Name the logger after the package.
|
||||
logger = logging.getLogger(__package__)
|
||||
File diff suppressed because it is too large
Load Diff
@@ -0,0 +1,220 @@
|
||||
"""Abstract Protocol base classes."""
|
||||
|
||||
__all__ = (
|
||||
'BaseProtocol', 'Protocol', 'DatagramProtocol',
|
||||
'SubprocessProtocol', 'BufferedProtocol',
|
||||
)
|
||||
|
||||
|
||||
class BaseProtocol:
|
||||
"""Common base class for protocol interfaces.
|
||||
|
||||
Usually user implements protocols that derived from BaseProtocol
|
||||
like Protocol or ProcessProtocol.
|
||||
|
||||
The only case when BaseProtocol should be implemented directly is
|
||||
write-only transport like write pipe
|
||||
"""
|
||||
|
||||
__slots__ = ()
|
||||
|
||||
def connection_made(self, transport):
|
||||
"""Called when a connection is made.
|
||||
|
||||
The argument is the transport representing the pipe connection.
|
||||
To receive data, wait for data_received() calls.
|
||||
When the connection is closed, connection_lost() is called.
|
||||
"""
|
||||
|
||||
def connection_lost(self, exc):
|
||||
"""Called when the connection is lost or closed.
|
||||
|
||||
The argument is an exception object or None (the latter
|
||||
meaning a regular EOF is received or the connection was
|
||||
aborted or closed).
|
||||
"""
|
||||
|
||||
def pause_writing(self):
|
||||
"""Called when the transport's buffer goes over the high-water mark.
|
||||
|
||||
Pause and resume calls are paired -- pause_writing() is called
|
||||
once when the buffer goes strictly over the high-water mark
|
||||
(even if subsequent writes increases the buffer size even
|
||||
more), and eventually resume_writing() is called once when the
|
||||
buffer size reaches the low-water mark.
|
||||
|
||||
Note that if the buffer size equals the high-water mark,
|
||||
pause_writing() is not called -- it must go strictly over.
|
||||
Conversely, resume_writing() is called when the buffer size is
|
||||
equal or lower than the low-water mark. These end conditions
|
||||
are important to ensure that things go as expected when either
|
||||
mark is zero.
|
||||
|
||||
NOTE: This is the only Protocol callback that is not called
|
||||
through EventLoop.call_soon() -- if it were, it would have no
|
||||
effect when it's most needed (when the app keeps writing
|
||||
without yielding until pause_writing() is called).
|
||||
"""
|
||||
|
||||
def resume_writing(self):
|
||||
"""Called when the transport's buffer drains below the low-water mark.
|
||||
|
||||
See pause_writing() for details.
|
||||
"""
|
||||
|
||||
|
||||
class Protocol(BaseProtocol):
|
||||
"""Interface for stream protocol.
|
||||
|
||||
The user should implement this interface. They can inherit from
|
||||
this class but don't need to. The implementations here do
|
||||
nothing (they don't raise exceptions).
|
||||
|
||||
When the user wants to requests a transport, they pass a protocol
|
||||
factory to a utility function (e.g., EventLoop.create_connection()).
|
||||
|
||||
When the connection is made successfully, connection_made() is
|
||||
called with a suitable transport object. Then data_received()
|
||||
will be called 0 or more times with data (bytes) received from the
|
||||
transport; finally, connection_lost() will be called exactly once
|
||||
with either an exception object or None as an argument.
|
||||
|
||||
State machine of calls:
|
||||
|
||||
start -> CM [-> DR*] [-> ER?] -> CL -> end
|
||||
|
||||
* CM: connection_made()
|
||||
* DR: data_received()
|
||||
* ER: eof_received()
|
||||
* CL: connection_lost()
|
||||
"""
|
||||
|
||||
__slots__ = ()
|
||||
|
||||
def data_received(self, data):
|
||||
"""Called when some data is received.
|
||||
|
||||
The argument is a bytes object.
|
||||
"""
|
||||
|
||||
def eof_received(self):
|
||||
"""Called when the other end calls write_eof() or equivalent.
|
||||
|
||||
If this returns a false value (including None), the transport
|
||||
will close itself. If it returns a true value, closing the
|
||||
transport is up to the protocol.
|
||||
"""
|
||||
|
||||
|
||||
class BufferedProtocol(BaseProtocol):
|
||||
"""Interface for stream protocol with manual buffer control.
|
||||
|
||||
Important: this has been added to asyncio in Python 3.7
|
||||
*on a provisional basis*! Consider it as an experimental API that
|
||||
might be changed or removed in Python 3.8.
|
||||
|
||||
Event methods, such as `create_server` and `create_connection`,
|
||||
accept factories that return protocols that implement this interface.
|
||||
|
||||
The idea of BufferedProtocol is that it allows to manually allocate
|
||||
and control the receive buffer. Event loops can then use the buffer
|
||||
provided by the protocol to avoid unnecessary data copies. This
|
||||
can result in noticeable performance improvement for protocols that
|
||||
receive big amounts of data. Sophisticated protocols can allocate
|
||||
the buffer only once at creation time.
|
||||
|
||||
State machine of calls:
|
||||
|
||||
start -> CM [-> GB [-> BU?]]* [-> ER?] -> CL -> end
|
||||
|
||||
* CM: connection_made()
|
||||
* GB: get_buffer()
|
||||
* BU: buffer_updated()
|
||||
* ER: eof_received()
|
||||
* CL: connection_lost()
|
||||
"""
|
||||
|
||||
__slots__ = ()
|
||||
|
||||
def get_buffer(self, sizehint):
|
||||
"""Called to allocate a new receive buffer.
|
||||
|
||||
*sizehint* is a recommended minimal size for the returned
|
||||
buffer. When set to -1, the buffer size can be arbitrary.
|
||||
|
||||
Must return an object that implements the
|
||||
:ref:`buffer protocol <bufferobjects>`.
|
||||
It is an error to return a zero-sized buffer.
|
||||
"""
|
||||
|
||||
def buffer_updated(self, nbytes):
|
||||
"""Called when the buffer was updated with the received data.
|
||||
|
||||
*nbytes* is the total number of bytes that were written to
|
||||
the buffer.
|
||||
"""
|
||||
|
||||
def eof_received(self):
|
||||
"""Called when the other end calls write_eof() or equivalent.
|
||||
|
||||
If this returns a false value (including None), the transport
|
||||
will close itself. If it returns a true value, closing the
|
||||
transport is up to the protocol.
|
||||
"""
|
||||
|
||||
|
||||
class DatagramProtocol(BaseProtocol):
|
||||
"""Interface for datagram protocol."""
|
||||
|
||||
__slots__ = ()
|
||||
|
||||
def datagram_received(self, data, addr):
|
||||
"""Called when some datagram is received."""
|
||||
|
||||
def error_received(self, exc):
|
||||
"""Called when a send or receive operation raises an OSError.
|
||||
|
||||
(Other than BlockingIOError or InterruptedError.)
|
||||
"""
|
||||
|
||||
|
||||
class SubprocessProtocol(BaseProtocol):
|
||||
"""Interface for protocol for subprocess calls."""
|
||||
|
||||
__slots__ = ()
|
||||
|
||||
def pipe_data_received(self, fd, data):
|
||||
"""Called when the subprocess writes data into stdout/stderr pipe.
|
||||
|
||||
fd is int file descriptor.
|
||||
data is bytes object.
|
||||
"""
|
||||
|
||||
def pipe_connection_lost(self, fd, exc):
|
||||
"""Called when a file descriptor associated with the child process is
|
||||
closed.
|
||||
|
||||
fd is the int file descriptor that was closed.
|
||||
"""
|
||||
|
||||
def process_exited(self):
|
||||
"""Called when subprocess has exited."""
|
||||
|
||||
|
||||
def _feed_data_to_buffered_proto(proto, data):
|
||||
data_len = len(data)
|
||||
while data_len:
|
||||
buf = proto.get_buffer(data_len)
|
||||
buf_len = len(buf)
|
||||
if not buf_len:
|
||||
raise RuntimeError('get_buffer() returned an empty buffer')
|
||||
|
||||
if buf_len >= data_len:
|
||||
buf[:data_len] = data
|
||||
proto.buffer_updated(data_len)
|
||||
return
|
||||
else:
|
||||
buf[:buf_len] = data[:buf_len]
|
||||
proto.buffer_updated(buf_len)
|
||||
data = data[buf_len:]
|
||||
data_len = len(data)
|
||||
@@ -0,0 +1,249 @@
|
||||
__all__ = ('Queue', 'PriorityQueue', 'LifoQueue', 'QueueFull', 'QueueEmpty')
|
||||
|
||||
import collections
|
||||
import heapq
|
||||
import warnings
|
||||
|
||||
from . import events
|
||||
from . import locks
|
||||
|
||||
|
||||
class QueueEmpty(Exception):
|
||||
"""Raised when Queue.get_nowait() is called on an empty Queue."""
|
||||
pass
|
||||
|
||||
|
||||
class QueueFull(Exception):
|
||||
"""Raised when the Queue.put_nowait() method is called on a full Queue."""
|
||||
pass
|
||||
|
||||
|
||||
class Queue:
|
||||
"""A queue, useful for coordinating producer and consumer coroutines.
|
||||
|
||||
If maxsize is less than or equal to zero, the queue size is infinite. If it
|
||||
is an integer greater than 0, then "await put()" will block when the
|
||||
queue reaches maxsize, until an item is removed by get().
|
||||
|
||||
Unlike the standard library Queue, you can reliably know this Queue's size
|
||||
with qsize(), since your single-threaded asyncio application won't be
|
||||
interrupted between calling qsize() and doing an operation on the Queue.
|
||||
"""
|
||||
|
||||
def __init__(self, maxsize=0, *, loop=None):
|
||||
if loop is None:
|
||||
self._loop = events.get_event_loop()
|
||||
else:
|
||||
self._loop = loop
|
||||
warnings.warn("The loop argument is deprecated since Python 3.8, "
|
||||
"and scheduled for removal in Python 3.10.",
|
||||
DeprecationWarning, stacklevel=2)
|
||||
self._maxsize = maxsize
|
||||
|
||||
# Futures.
|
||||
self._getters = collections.deque()
|
||||
# Futures.
|
||||
self._putters = collections.deque()
|
||||
self._unfinished_tasks = 0
|
||||
self._finished = locks.Event(loop=loop)
|
||||
self._finished.set()
|
||||
self._init(maxsize)
|
||||
|
||||
# These three are overridable in subclasses.
|
||||
|
||||
def _init(self, maxsize):
|
||||
self._queue = collections.deque()
|
||||
|
||||
def _get(self):
|
||||
return self._queue.popleft()
|
||||
|
||||
def _put(self, item):
|
||||
self._queue.append(item)
|
||||
|
||||
# End of the overridable methods.
|
||||
|
||||
def _wakeup_next(self, waiters):
|
||||
# Wake up the next waiter (if any) that isn't cancelled.
|
||||
while waiters:
|
||||
waiter = waiters.popleft()
|
||||
if not waiter.done():
|
||||
waiter.set_result(None)
|
||||
break
|
||||
|
||||
def __repr__(self):
|
||||
return f'<{type(self).__name__} at {id(self):#x} {self._format()}>'
|
||||
|
||||
def __str__(self):
|
||||
return f'<{type(self).__name__} {self._format()}>'
|
||||
|
||||
def _format(self):
|
||||
result = f'maxsize={self._maxsize!r}'
|
||||
if getattr(self, '_queue', None):
|
||||
result += f' _queue={list(self._queue)!r}'
|
||||
if self._getters:
|
||||
result += f' _getters[{len(self._getters)}]'
|
||||
if self._putters:
|
||||
result += f' _putters[{len(self._putters)}]'
|
||||
if self._unfinished_tasks:
|
||||
result += f' tasks={self._unfinished_tasks}'
|
||||
return result
|
||||
|
||||
def qsize(self):
|
||||
"""Number of items in the queue."""
|
||||
return len(self._queue)
|
||||
|
||||
@property
|
||||
def maxsize(self):
|
||||
"""Number of items allowed in the queue."""
|
||||
return self._maxsize
|
||||
|
||||
def empty(self):
|
||||
"""Return True if the queue is empty, False otherwise."""
|
||||
return not self._queue
|
||||
|
||||
def full(self):
|
||||
"""Return True if there are maxsize items in the queue.
|
||||
|
||||
Note: if the Queue was initialized with maxsize=0 (the default),
|
||||
then full() is never True.
|
||||
"""
|
||||
if self._maxsize <= 0:
|
||||
return False
|
||||
else:
|
||||
return self.qsize() >= self._maxsize
|
||||
|
||||
async def put(self, item):
|
||||
"""Put an item into the queue.
|
||||
|
||||
Put an item into the queue. If the queue is full, wait until a free
|
||||
slot is available before adding item.
|
||||
"""
|
||||
while self.full():
|
||||
putter = self._loop.create_future()
|
||||
self._putters.append(putter)
|
||||
try:
|
||||
await putter
|
||||
except:
|
||||
putter.cancel() # Just in case putter is not done yet.
|
||||
try:
|
||||
# Clean self._putters from canceled putters.
|
||||
self._putters.remove(putter)
|
||||
except ValueError:
|
||||
# The putter could be removed from self._putters by a
|
||||
# previous get_nowait call.
|
||||
pass
|
||||
if not self.full() and not putter.cancelled():
|
||||
# We were woken up by get_nowait(), but can't take
|
||||
# the call. Wake up the next in line.
|
||||
self._wakeup_next(self._putters)
|
||||
raise
|
||||
return self.put_nowait(item)
|
||||
|
||||
def put_nowait(self, item):
|
||||
"""Put an item into the queue without blocking.
|
||||
|
||||
If no free slot is immediately available, raise QueueFull.
|
||||
"""
|
||||
if self.full():
|
||||
raise QueueFull
|
||||
self._put(item)
|
||||
self._unfinished_tasks += 1
|
||||
self._finished.clear()
|
||||
self._wakeup_next(self._getters)
|
||||
|
||||
async def get(self):
|
||||
"""Remove and return an item from the queue.
|
||||
|
||||
If queue is empty, wait until an item is available.
|
||||
"""
|
||||
while self.empty():
|
||||
getter = self._loop.create_future()
|
||||
self._getters.append(getter)
|
||||
try:
|
||||
await getter
|
||||
except:
|
||||
getter.cancel() # Just in case getter is not done yet.
|
||||
try:
|
||||
# Clean self._getters from canceled getters.
|
||||
self._getters.remove(getter)
|
||||
except ValueError:
|
||||
# The getter could be removed from self._getters by a
|
||||
# previous put_nowait call.
|
||||
pass
|
||||
if not self.empty() and not getter.cancelled():
|
||||
# We were woken up by put_nowait(), but can't take
|
||||
# the call. Wake up the next in line.
|
||||
self._wakeup_next(self._getters)
|
||||
raise
|
||||
return self.get_nowait()
|
||||
|
||||
def get_nowait(self):
|
||||
"""Remove and return an item from the queue.
|
||||
|
||||
Return an item if one is immediately available, else raise QueueEmpty.
|
||||
"""
|
||||
if self.empty():
|
||||
raise QueueEmpty
|
||||
item = self._get()
|
||||
self._wakeup_next(self._putters)
|
||||
return item
|
||||
|
||||
def task_done(self):
|
||||
"""Indicate that a formerly enqueued task is complete.
|
||||
|
||||
Used by queue consumers. For each get() used to fetch a task,
|
||||
a subsequent call to task_done() tells the queue that the processing
|
||||
on the task is complete.
|
||||
|
||||
If a join() is currently blocking, it will resume when all items have
|
||||
been processed (meaning that a task_done() call was received for every
|
||||
item that had been put() into the queue).
|
||||
|
||||
Raises ValueError if called more times than there were items placed in
|
||||
the queue.
|
||||
"""
|
||||
if self._unfinished_tasks <= 0:
|
||||
raise ValueError('task_done() called too many times')
|
||||
self._unfinished_tasks -= 1
|
||||
if self._unfinished_tasks == 0:
|
||||
self._finished.set()
|
||||
|
||||
async def join(self):
|
||||
"""Block until all items in the queue have been gotten and processed.
|
||||
|
||||
The count of unfinished tasks goes up whenever an item is added to the
|
||||
queue. The count goes down whenever a consumer calls task_done() to
|
||||
indicate that the item was retrieved and all work on it is complete.
|
||||
When the count of unfinished tasks drops to zero, join() unblocks.
|
||||
"""
|
||||
if self._unfinished_tasks > 0:
|
||||
await self._finished.wait()
|
||||
|
||||
|
||||
class PriorityQueue(Queue):
|
||||
"""A subclass of Queue; retrieves entries in priority order (lowest first).
|
||||
|
||||
Entries are typically tuples of the form: (priority number, data).
|
||||
"""
|
||||
|
||||
def _init(self, maxsize):
|
||||
self._queue = []
|
||||
|
||||
def _put(self, item, heappush=heapq.heappush):
|
||||
heappush(self._queue, item)
|
||||
|
||||
def _get(self, heappop=heapq.heappop):
|
||||
return heappop(self._queue)
|
||||
|
||||
|
||||
class LifoQueue(Queue):
|
||||
"""A subclass of Queue that retrieves most recently added entries first."""
|
||||
|
||||
def _init(self, maxsize):
|
||||
self._queue = []
|
||||
|
||||
def _put(self, item):
|
||||
self._queue.append(item)
|
||||
|
||||
def _get(self):
|
||||
return self._queue.pop()
|
||||
@@ -0,0 +1,73 @@
|
||||
__all__ = 'run',
|
||||
|
||||
from . import coroutines
|
||||
from . import events
|
||||
from . import tasks
|
||||
|
||||
|
||||
def run(main, *, debug=None):
|
||||
"""Execute the coroutine and return the result.
|
||||
|
||||
This function runs the passed coroutine, taking care of
|
||||
managing the asyncio event loop and finalizing asynchronous
|
||||
generators.
|
||||
|
||||
This function cannot be called when another asyncio event loop is
|
||||
running in the same thread.
|
||||
|
||||
If debug is True, the event loop will be run in debug mode.
|
||||
|
||||
This function always creates a new event loop and closes it at the end.
|
||||
It should be used as a main entry point for asyncio programs, and should
|
||||
ideally only be called once.
|
||||
|
||||
Example:
|
||||
|
||||
async def main():
|
||||
await asyncio.sleep(1)
|
||||
print('hello')
|
||||
|
||||
asyncio.run(main())
|
||||
"""
|
||||
if events._get_running_loop() is not None:
|
||||
raise RuntimeError(
|
||||
"asyncio.run() cannot be called from a running event loop")
|
||||
|
||||
if not coroutines.iscoroutine(main):
|
||||
raise ValueError("a coroutine was expected, got {!r}".format(main))
|
||||
|
||||
loop = events.new_event_loop()
|
||||
try:
|
||||
events.set_event_loop(loop)
|
||||
if debug is not None:
|
||||
loop.set_debug(debug)
|
||||
return loop.run_until_complete(main)
|
||||
finally:
|
||||
try:
|
||||
_cancel_all_tasks(loop)
|
||||
loop.run_until_complete(loop.shutdown_asyncgens())
|
||||
finally:
|
||||
events.set_event_loop(None)
|
||||
loop.close()
|
||||
|
||||
|
||||
def _cancel_all_tasks(loop):
|
||||
to_cancel = tasks.all_tasks(loop)
|
||||
if not to_cancel:
|
||||
return
|
||||
|
||||
for task in to_cancel:
|
||||
task.cancel()
|
||||
|
||||
loop.run_until_complete(
|
||||
tasks.gather(*to_cancel, loop=loop, return_exceptions=True))
|
||||
|
||||
for task in to_cancel:
|
||||
if task.cancelled():
|
||||
continue
|
||||
if task.exception() is not None:
|
||||
loop.call_exception_handler({
|
||||
'message': 'unhandled exception during asyncio.run() shutdown',
|
||||
'exception': task.exception(),
|
||||
'task': task,
|
||||
})
|
||||
File diff suppressed because it is too large
Load Diff
File diff suppressed because it is too large
Load Diff
@@ -0,0 +1,149 @@
|
||||
"""Support for running coroutines in parallel with staggered start times."""
|
||||
|
||||
__all__ = 'staggered_race',
|
||||
|
||||
import contextlib
|
||||
import typing
|
||||
|
||||
from . import events
|
||||
from . import exceptions as exceptions_mod
|
||||
from . import locks
|
||||
from . import tasks
|
||||
|
||||
|
||||
async def staggered_race(
|
||||
coro_fns: typing.Iterable[typing.Callable[[], typing.Awaitable]],
|
||||
delay: typing.Optional[float],
|
||||
*,
|
||||
loop: events.AbstractEventLoop = None,
|
||||
) -> typing.Tuple[
|
||||
typing.Any,
|
||||
typing.Optional[int],
|
||||
typing.List[typing.Optional[Exception]]
|
||||
]:
|
||||
"""Run coroutines with staggered start times and take the first to finish.
|
||||
|
||||
This method takes an iterable of coroutine functions. The first one is
|
||||
started immediately. From then on, whenever the immediately preceding one
|
||||
fails (raises an exception), or when *delay* seconds has passed, the next
|
||||
coroutine is started. This continues until one of the coroutines complete
|
||||
successfully, in which case all others are cancelled, or until all
|
||||
coroutines fail.
|
||||
|
||||
The coroutines provided should be well-behaved in the following way:
|
||||
|
||||
* They should only ``return`` if completed successfully.
|
||||
|
||||
* They should always raise an exception if they did not complete
|
||||
successfully. In particular, if they handle cancellation, they should
|
||||
probably reraise, like this::
|
||||
|
||||
try:
|
||||
# do work
|
||||
except asyncio.CancelledError:
|
||||
# undo partially completed work
|
||||
raise
|
||||
|
||||
Args:
|
||||
coro_fns: an iterable of coroutine functions, i.e. callables that
|
||||
return a coroutine object when called. Use ``functools.partial`` or
|
||||
lambdas to pass arguments.
|
||||
|
||||
delay: amount of time, in seconds, between starting coroutines. If
|
||||
``None``, the coroutines will run sequentially.
|
||||
|
||||
loop: the event loop to use.
|
||||
|
||||
Returns:
|
||||
tuple *(winner_result, winner_index, exceptions)* where
|
||||
|
||||
- *winner_result*: the result of the winning coroutine, or ``None``
|
||||
if no coroutines won.
|
||||
|
||||
- *winner_index*: the index of the winning coroutine in
|
||||
``coro_fns``, or ``None`` if no coroutines won. If the winning
|
||||
coroutine may return None on success, *winner_index* can be used
|
||||
to definitively determine whether any coroutine won.
|
||||
|
||||
- *exceptions*: list of exceptions returned by the coroutines.
|
||||
``len(exceptions)`` is equal to the number of coroutines actually
|
||||
started, and the order is the same as in ``coro_fns``. The winning
|
||||
coroutine's entry is ``None``.
|
||||
|
||||
"""
|
||||
# TODO: when we have aiter() and anext(), allow async iterables in coro_fns.
|
||||
loop = loop or events.get_running_loop()
|
||||
enum_coro_fns = enumerate(coro_fns)
|
||||
winner_result = None
|
||||
winner_index = None
|
||||
exceptions = []
|
||||
running_tasks = []
|
||||
|
||||
async def run_one_coro(
|
||||
previous_failed: typing.Optional[locks.Event]) -> None:
|
||||
# Wait for the previous task to finish, or for delay seconds
|
||||
if previous_failed is not None:
|
||||
with contextlib.suppress(exceptions_mod.TimeoutError):
|
||||
# Use asyncio.wait_for() instead of asyncio.wait() here, so
|
||||
# that if we get cancelled at this point, Event.wait() is also
|
||||
# cancelled, otherwise there will be a "Task destroyed but it is
|
||||
# pending" later.
|
||||
await tasks.wait_for(previous_failed.wait(), delay)
|
||||
# Get the next coroutine to run
|
||||
try:
|
||||
this_index, coro_fn = next(enum_coro_fns)
|
||||
except StopIteration:
|
||||
return
|
||||
# Start task that will run the next coroutine
|
||||
this_failed = locks.Event()
|
||||
next_task = loop.create_task(run_one_coro(this_failed))
|
||||
running_tasks.append(next_task)
|
||||
assert len(running_tasks) == this_index + 2
|
||||
# Prepare place to put this coroutine's exceptions if not won
|
||||
exceptions.append(None)
|
||||
assert len(exceptions) == this_index + 1
|
||||
|
||||
try:
|
||||
result = await coro_fn()
|
||||
except (SystemExit, KeyboardInterrupt):
|
||||
raise
|
||||
except BaseException as e:
|
||||
exceptions[this_index] = e
|
||||
this_failed.set() # Kickstart the next coroutine
|
||||
else:
|
||||
# Store winner's results
|
||||
nonlocal winner_index, winner_result
|
||||
assert winner_index is None
|
||||
winner_index = this_index
|
||||
winner_result = result
|
||||
# Cancel all other tasks. We take care to not cancel the current
|
||||
# task as well. If we do so, then since there is no `await` after
|
||||
# here and CancelledError are usually thrown at one, we will
|
||||
# encounter a curious corner case where the current task will end
|
||||
# up as done() == True, cancelled() == False, exception() ==
|
||||
# asyncio.CancelledError. This behavior is specified in
|
||||
# https://bugs.python.org/issue30048
|
||||
for i, t in enumerate(running_tasks):
|
||||
if i != this_index:
|
||||
t.cancel()
|
||||
|
||||
first_task = loop.create_task(run_one_coro(None))
|
||||
running_tasks.append(first_task)
|
||||
try:
|
||||
# Wait for a growing list of tasks to all finish: poor man's version of
|
||||
# curio's TaskGroup or trio's nursery
|
||||
done_count = 0
|
||||
while done_count != len(running_tasks):
|
||||
done, _ = await tasks.wait(running_tasks)
|
||||
done_count = len(done)
|
||||
# If run_one_coro raises an unhandled exception, it's probably a
|
||||
# programming error, and I want to see it.
|
||||
if __debug__:
|
||||
for d in done:
|
||||
if d.done() and not d.cancelled() and d.exception():
|
||||
raise d.exception()
|
||||
return winner_result, winner_index, exceptions
|
||||
finally:
|
||||
# Make sure no tasks are left running if we leave this function
|
||||
for t in running_tasks:
|
||||
t.cancel()
|
||||
File diff suppressed because it is too large
Load Diff
@@ -0,0 +1,241 @@
|
||||
__all__ = 'create_subprocess_exec', 'create_subprocess_shell'
|
||||
|
||||
import subprocess
|
||||
import warnings
|
||||
|
||||
from . import events
|
||||
from . import protocols
|
||||
from . import streams
|
||||
from . import tasks
|
||||
from .log import logger
|
||||
|
||||
|
||||
PIPE = subprocess.PIPE
|
||||
STDOUT = subprocess.STDOUT
|
||||
DEVNULL = subprocess.DEVNULL
|
||||
|
||||
|
||||
class SubprocessStreamProtocol(streams.FlowControlMixin,
|
||||
protocols.SubprocessProtocol):
|
||||
"""Like StreamReaderProtocol, but for a subprocess."""
|
||||
|
||||
def __init__(self, limit, loop):
|
||||
super().__init__(loop=loop)
|
||||
self._limit = limit
|
||||
self.stdin = self.stdout = self.stderr = None
|
||||
self._transport = None
|
||||
self._process_exited = False
|
||||
self._pipe_fds = []
|
||||
self._stdin_closed = self._loop.create_future()
|
||||
|
||||
def __repr__(self):
|
||||
info = [self.__class__.__name__]
|
||||
if self.stdin is not None:
|
||||
info.append(f'stdin={self.stdin!r}')
|
||||
if self.stdout is not None:
|
||||
info.append(f'stdout={self.stdout!r}')
|
||||
if self.stderr is not None:
|
||||
info.append(f'stderr={self.stderr!r}')
|
||||
return '<{}>'.format(' '.join(info))
|
||||
|
||||
def connection_made(self, transport):
|
||||
self._transport = transport
|
||||
|
||||
stdout_transport = transport.get_pipe_transport(1)
|
||||
if stdout_transport is not None:
|
||||
self.stdout = streams.StreamReader(limit=self._limit,
|
||||
loop=self._loop)
|
||||
self.stdout.set_transport(stdout_transport)
|
||||
self._pipe_fds.append(1)
|
||||
|
||||
stderr_transport = transport.get_pipe_transport(2)
|
||||
if stderr_transport is not None:
|
||||
self.stderr = streams.StreamReader(limit=self._limit,
|
||||
loop=self._loop)
|
||||
self.stderr.set_transport(stderr_transport)
|
||||
self._pipe_fds.append(2)
|
||||
|
||||
stdin_transport = transport.get_pipe_transport(0)
|
||||
if stdin_transport is not None:
|
||||
self.stdin = streams.StreamWriter(stdin_transport,
|
||||
protocol=self,
|
||||
reader=None,
|
||||
loop=self._loop)
|
||||
|
||||
def pipe_data_received(self, fd, data):
|
||||
if fd == 1:
|
||||
reader = self.stdout
|
||||
elif fd == 2:
|
||||
reader = self.stderr
|
||||
else:
|
||||
reader = None
|
||||
if reader is not None:
|
||||
reader.feed_data(data)
|
||||
|
||||
def pipe_connection_lost(self, fd, exc):
|
||||
if fd == 0:
|
||||
pipe = self.stdin
|
||||
if pipe is not None:
|
||||
pipe.close()
|
||||
self.connection_lost(exc)
|
||||
if exc is None:
|
||||
self._stdin_closed.set_result(None)
|
||||
else:
|
||||
self._stdin_closed.set_exception(exc)
|
||||
return
|
||||
if fd == 1:
|
||||
reader = self.stdout
|
||||
elif fd == 2:
|
||||
reader = self.stderr
|
||||
else:
|
||||
reader = None
|
||||
if reader is not None:
|
||||
if exc is None:
|
||||
reader.feed_eof()
|
||||
else:
|
||||
reader.set_exception(exc)
|
||||
|
||||
if fd in self._pipe_fds:
|
||||
self._pipe_fds.remove(fd)
|
||||
self._maybe_close_transport()
|
||||
|
||||
def process_exited(self):
|
||||
self._process_exited = True
|
||||
self._maybe_close_transport()
|
||||
|
||||
def _maybe_close_transport(self):
|
||||
if len(self._pipe_fds) == 0 and self._process_exited:
|
||||
self._transport.close()
|
||||
self._transport = None
|
||||
|
||||
def _get_close_waiter(self, stream):
|
||||
if stream is self.stdin:
|
||||
return self._stdin_closed
|
||||
|
||||
|
||||
class Process:
|
||||
def __init__(self, transport, protocol, loop):
|
||||
self._transport = transport
|
||||
self._protocol = protocol
|
||||
self._loop = loop
|
||||
self.stdin = protocol.stdin
|
||||
self.stdout = protocol.stdout
|
||||
self.stderr = protocol.stderr
|
||||
self.pid = transport.get_pid()
|
||||
|
||||
def __repr__(self):
|
||||
return f'<{self.__class__.__name__} {self.pid}>'
|
||||
|
||||
@property
|
||||
def returncode(self):
|
||||
return self._transport.get_returncode()
|
||||
|
||||
async def wait(self):
|
||||
"""Wait until the process exit and return the process return code."""
|
||||
return await self._transport._wait()
|
||||
|
||||
def send_signal(self, signal):
|
||||
self._transport.send_signal(signal)
|
||||
|
||||
def terminate(self):
|
||||
self._transport.terminate()
|
||||
|
||||
def kill(self):
|
||||
self._transport.kill()
|
||||
|
||||
async def _feed_stdin(self, input):
|
||||
debug = self._loop.get_debug()
|
||||
self.stdin.write(input)
|
||||
if debug:
|
||||
logger.debug(
|
||||
'%r communicate: feed stdin (%s bytes)', self, len(input))
|
||||
try:
|
||||
await self.stdin.drain()
|
||||
except (BrokenPipeError, ConnectionResetError) as exc:
|
||||
# communicate() ignores BrokenPipeError and ConnectionResetError
|
||||
if debug:
|
||||
logger.debug('%r communicate: stdin got %r', self, exc)
|
||||
|
||||
if debug:
|
||||
logger.debug('%r communicate: close stdin', self)
|
||||
self.stdin.close()
|
||||
|
||||
async def _noop(self):
|
||||
return None
|
||||
|
||||
async def _read_stream(self, fd):
|
||||
transport = self._transport.get_pipe_transport(fd)
|
||||
if fd == 2:
|
||||
stream = self.stderr
|
||||
else:
|
||||
assert fd == 1
|
||||
stream = self.stdout
|
||||
if self._loop.get_debug():
|
||||
name = 'stdout' if fd == 1 else 'stderr'
|
||||
logger.debug('%r communicate: read %s', self, name)
|
||||
output = await stream.read()
|
||||
if self._loop.get_debug():
|
||||
name = 'stdout' if fd == 1 else 'stderr'
|
||||
logger.debug('%r communicate: close %s', self, name)
|
||||
transport.close()
|
||||
return output
|
||||
|
||||
async def communicate(self, input=None):
|
||||
if input is not None:
|
||||
stdin = self._feed_stdin(input)
|
||||
else:
|
||||
stdin = self._noop()
|
||||
if self.stdout is not None:
|
||||
stdout = self._read_stream(1)
|
||||
else:
|
||||
stdout = self._noop()
|
||||
if self.stderr is not None:
|
||||
stderr = self._read_stream(2)
|
||||
else:
|
||||
stderr = self._noop()
|
||||
stdin, stdout, stderr = await tasks.gather(stdin, stdout, stderr,
|
||||
loop=self._loop)
|
||||
await self.wait()
|
||||
return (stdout, stderr)
|
||||
|
||||
|
||||
async def create_subprocess_shell(cmd, stdin=None, stdout=None, stderr=None,
|
||||
loop=None, limit=streams._DEFAULT_LIMIT,
|
||||
**kwds):
|
||||
if loop is None:
|
||||
loop = events.get_event_loop()
|
||||
else:
|
||||
warnings.warn("The loop argument is deprecated since Python 3.8 "
|
||||
"and scheduled for removal in Python 3.10.",
|
||||
DeprecationWarning,
|
||||
stacklevel=2
|
||||
)
|
||||
|
||||
protocol_factory = lambda: SubprocessStreamProtocol(limit=limit,
|
||||
loop=loop)
|
||||
transport, protocol = await loop.subprocess_shell(
|
||||
protocol_factory,
|
||||
cmd, stdin=stdin, stdout=stdout,
|
||||
stderr=stderr, **kwds)
|
||||
return Process(transport, protocol, loop)
|
||||
|
||||
|
||||
async def create_subprocess_exec(program, *args, stdin=None, stdout=None,
|
||||
stderr=None, loop=None,
|
||||
limit=streams._DEFAULT_LIMIT, **kwds):
|
||||
if loop is None:
|
||||
loop = events.get_event_loop()
|
||||
else:
|
||||
warnings.warn("The loop argument is deprecated since Python 3.8 "
|
||||
"and scheduled for removal in Python 3.10.",
|
||||
DeprecationWarning,
|
||||
stacklevel=2
|
||||
)
|
||||
protocol_factory = lambda: SubprocessStreamProtocol(limit=limit,
|
||||
loop=loop)
|
||||
transport, protocol = await loop.subprocess_exec(
|
||||
protocol_factory,
|
||||
program, *args,
|
||||
stdin=stdin, stdout=stdout,
|
||||
stderr=stderr, **kwds)
|
||||
return Process(transport, protocol, loop)
|
||||
File diff suppressed because it is too large
Load Diff
@@ -0,0 +1,329 @@
|
||||
"""Abstract Transport class."""
|
||||
|
||||
__all__ = (
|
||||
'BaseTransport', 'ReadTransport', 'WriteTransport',
|
||||
'Transport', 'DatagramTransport', 'SubprocessTransport',
|
||||
)
|
||||
|
||||
|
||||
class BaseTransport:
|
||||
"""Base class for transports."""
|
||||
|
||||
__slots__ = ('_extra',)
|
||||
|
||||
def __init__(self, extra=None):
|
||||
if extra is None:
|
||||
extra = {}
|
||||
self._extra = extra
|
||||
|
||||
def get_extra_info(self, name, default=None):
|
||||
"""Get optional transport information."""
|
||||
return self._extra.get(name, default)
|
||||
|
||||
def is_closing(self):
|
||||
"""Return True if the transport is closing or closed."""
|
||||
raise NotImplementedError
|
||||
|
||||
def close(self):
|
||||
"""Close the transport.
|
||||
|
||||
Buffered data will be flushed asynchronously. No more data
|
||||
will be received. After all buffered data is flushed, the
|
||||
protocol's connection_lost() method will (eventually) be
|
||||
called with None as its argument.
|
||||
"""
|
||||
raise NotImplementedError
|
||||
|
||||
def set_protocol(self, protocol):
|
||||
"""Set a new protocol."""
|
||||
raise NotImplementedError
|
||||
|
||||
def get_protocol(self):
|
||||
"""Return the current protocol."""
|
||||
raise NotImplementedError
|
||||
|
||||
|
||||
class ReadTransport(BaseTransport):
|
||||
"""Interface for read-only transports."""
|
||||
|
||||
__slots__ = ()
|
||||
|
||||
def is_reading(self):
|
||||
"""Return True if the transport is receiving."""
|
||||
raise NotImplementedError
|
||||
|
||||
def pause_reading(self):
|
||||
"""Pause the receiving end.
|
||||
|
||||
No data will be passed to the protocol's data_received()
|
||||
method until resume_reading() is called.
|
||||
"""
|
||||
raise NotImplementedError
|
||||
|
||||
def resume_reading(self):
|
||||
"""Resume the receiving end.
|
||||
|
||||
Data received will once again be passed to the protocol's
|
||||
data_received() method.
|
||||
"""
|
||||
raise NotImplementedError
|
||||
|
||||
|
||||
class WriteTransport(BaseTransport):
|
||||
"""Interface for write-only transports."""
|
||||
|
||||
__slots__ = ()
|
||||
|
||||
def set_write_buffer_limits(self, high=None, low=None):
|
||||
"""Set the high- and low-water limits for write flow control.
|
||||
|
||||
These two values control when to call the protocol's
|
||||
pause_writing() and resume_writing() methods. If specified,
|
||||
the low-water limit must be less than or equal to the
|
||||
high-water limit. Neither value can be negative.
|
||||
|
||||
The defaults are implementation-specific. If only the
|
||||
high-water limit is given, the low-water limit defaults to an
|
||||
implementation-specific value less than or equal to the
|
||||
high-water limit. Setting high to zero forces low to zero as
|
||||
well, and causes pause_writing() to be called whenever the
|
||||
buffer becomes non-empty. Setting low to zero causes
|
||||
resume_writing() to be called only once the buffer is empty.
|
||||
Use of zero for either limit is generally sub-optimal as it
|
||||
reduces opportunities for doing I/O and computation
|
||||
concurrently.
|
||||
"""
|
||||
raise NotImplementedError
|
||||
|
||||
def get_write_buffer_size(self):
|
||||
"""Return the current size of the write buffer."""
|
||||
raise NotImplementedError
|
||||
|
||||
def write(self, data):
|
||||
"""Write some data bytes to the transport.
|
||||
|
||||
This does not block; it buffers the data and arranges for it
|
||||
to be sent out asynchronously.
|
||||
"""
|
||||
raise NotImplementedError
|
||||
|
||||
def writelines(self, list_of_data):
|
||||
"""Write a list (or any iterable) of data bytes to the transport.
|
||||
|
||||
The default implementation concatenates the arguments and
|
||||
calls write() on the result.
|
||||
"""
|
||||
data = b''.join(list_of_data)
|
||||
self.write(data)
|
||||
|
||||
def write_eof(self):
|
||||
"""Close the write end after flushing buffered data.
|
||||
|
||||
(This is like typing ^D into a UNIX program reading from stdin.)
|
||||
|
||||
Data may still be received.
|
||||
"""
|
||||
raise NotImplementedError
|
||||
|
||||
def can_write_eof(self):
|
||||
"""Return True if this transport supports write_eof(), False if not."""
|
||||
raise NotImplementedError
|
||||
|
||||
def abort(self):
|
||||
"""Close the transport immediately.
|
||||
|
||||
Buffered data will be lost. No more data will be received.
|
||||
The protocol's connection_lost() method will (eventually) be
|
||||
called with None as its argument.
|
||||
"""
|
||||
raise NotImplementedError
|
||||
|
||||
|
||||
class Transport(ReadTransport, WriteTransport):
|
||||
"""Interface representing a bidirectional transport.
|
||||
|
||||
There may be several implementations, but typically, the user does
|
||||
not implement new transports; rather, the platform provides some
|
||||
useful transports that are implemented using the platform's best
|
||||
practices.
|
||||
|
||||
The user never instantiates a transport directly; they call a
|
||||
utility function, passing it a protocol factory and other
|
||||
information necessary to create the transport and protocol. (E.g.
|
||||
EventLoop.create_connection() or EventLoop.create_server().)
|
||||
|
||||
The utility function will asynchronously create a transport and a
|
||||
protocol and hook them up by calling the protocol's
|
||||
connection_made() method, passing it the transport.
|
||||
|
||||
The implementation here raises NotImplemented for every method
|
||||
except writelines(), which calls write() in a loop.
|
||||
"""
|
||||
|
||||
__slots__ = ()
|
||||
|
||||
|
||||
class DatagramTransport(BaseTransport):
|
||||
"""Interface for datagram (UDP) transports."""
|
||||
|
||||
__slots__ = ()
|
||||
|
||||
def sendto(self, data, addr=None):
|
||||
"""Send data to the transport.
|
||||
|
||||
This does not block; it buffers the data and arranges for it
|
||||
to be sent out asynchronously.
|
||||
addr is target socket address.
|
||||
If addr is None use target address pointed on transport creation.
|
||||
"""
|
||||
raise NotImplementedError
|
||||
|
||||
def abort(self):
|
||||
"""Close the transport immediately.
|
||||
|
||||
Buffered data will be lost. No more data will be received.
|
||||
The protocol's connection_lost() method will (eventually) be
|
||||
called with None as its argument.
|
||||
"""
|
||||
raise NotImplementedError
|
||||
|
||||
|
||||
class SubprocessTransport(BaseTransport):
|
||||
|
||||
__slots__ = ()
|
||||
|
||||
def get_pid(self):
|
||||
"""Get subprocess id."""
|
||||
raise NotImplementedError
|
||||
|
||||
def get_returncode(self):
|
||||
"""Get subprocess returncode.
|
||||
|
||||
See also
|
||||
http://docs.python.org/3/library/subprocess#subprocess.Popen.returncode
|
||||
"""
|
||||
raise NotImplementedError
|
||||
|
||||
def get_pipe_transport(self, fd):
|
||||
"""Get transport for pipe with number fd."""
|
||||
raise NotImplementedError
|
||||
|
||||
def send_signal(self, signal):
|
||||
"""Send signal to subprocess.
|
||||
|
||||
See also:
|
||||
docs.python.org/3/library/subprocess#subprocess.Popen.send_signal
|
||||
"""
|
||||
raise NotImplementedError
|
||||
|
||||
def terminate(self):
|
||||
"""Stop the subprocess.
|
||||
|
||||
Alias for close() method.
|
||||
|
||||
On Posix OSs the method sends SIGTERM to the subprocess.
|
||||
On Windows the Win32 API function TerminateProcess()
|
||||
is called to stop the subprocess.
|
||||
|
||||
See also:
|
||||
http://docs.python.org/3/library/subprocess#subprocess.Popen.terminate
|
||||
"""
|
||||
raise NotImplementedError
|
||||
|
||||
def kill(self):
|
||||
"""Kill the subprocess.
|
||||
|
||||
On Posix OSs the function sends SIGKILL to the subprocess.
|
||||
On Windows kill() is an alias for terminate().
|
||||
|
||||
See also:
|
||||
http://docs.python.org/3/library/subprocess#subprocess.Popen.kill
|
||||
"""
|
||||
raise NotImplementedError
|
||||
|
||||
|
||||
class _FlowControlMixin(Transport):
|
||||
"""All the logic for (write) flow control in a mix-in base class.
|
||||
|
||||
The subclass must implement get_write_buffer_size(). It must call
|
||||
_maybe_pause_protocol() whenever the write buffer size increases,
|
||||
and _maybe_resume_protocol() whenever it decreases. It may also
|
||||
override set_write_buffer_limits() (e.g. to specify different
|
||||
defaults).
|
||||
|
||||
The subclass constructor must call super().__init__(extra). This
|
||||
will call set_write_buffer_limits().
|
||||
|
||||
The user may call set_write_buffer_limits() and
|
||||
get_write_buffer_size(), and their protocol's pause_writing() and
|
||||
resume_writing() may be called.
|
||||
"""
|
||||
|
||||
__slots__ = ('_loop', '_protocol_paused', '_high_water', '_low_water')
|
||||
|
||||
def __init__(self, extra=None, loop=None):
|
||||
super().__init__(extra)
|
||||
assert loop is not None
|
||||
self._loop = loop
|
||||
self._protocol_paused = False
|
||||
self._set_write_buffer_limits()
|
||||
|
||||
def _maybe_pause_protocol(self):
|
||||
size = self.get_write_buffer_size()
|
||||
if size <= self._high_water:
|
||||
return
|
||||
if not self._protocol_paused:
|
||||
self._protocol_paused = True
|
||||
try:
|
||||
self._protocol.pause_writing()
|
||||
except (SystemExit, KeyboardInterrupt):
|
||||
raise
|
||||
except BaseException as exc:
|
||||
self._loop.call_exception_handler({
|
||||
'message': 'protocol.pause_writing() failed',
|
||||
'exception': exc,
|
||||
'transport': self,
|
||||
'protocol': self._protocol,
|
||||
})
|
||||
|
||||
def _maybe_resume_protocol(self):
|
||||
if (self._protocol_paused and
|
||||
self.get_write_buffer_size() <= self._low_water):
|
||||
self._protocol_paused = False
|
||||
try:
|
||||
self._protocol.resume_writing()
|
||||
except (SystemExit, KeyboardInterrupt):
|
||||
raise
|
||||
except BaseException as exc:
|
||||
self._loop.call_exception_handler({
|
||||
'message': 'protocol.resume_writing() failed',
|
||||
'exception': exc,
|
||||
'transport': self,
|
||||
'protocol': self._protocol,
|
||||
})
|
||||
|
||||
def get_write_buffer_limits(self):
|
||||
return (self._low_water, self._high_water)
|
||||
|
||||
def _set_write_buffer_limits(self, high=None, low=None):
|
||||
if high is None:
|
||||
if low is None:
|
||||
high = 64 * 1024
|
||||
else:
|
||||
high = 4 * low
|
||||
if low is None:
|
||||
low = high // 4
|
||||
|
||||
if not high >= low >= 0:
|
||||
raise ValueError(
|
||||
f'high ({high!r}) must be >= low ({low!r}) must be >= 0')
|
||||
|
||||
self._high_water = high
|
||||
self._low_water = low
|
||||
|
||||
def set_write_buffer_limits(self, high=None, low=None):
|
||||
self._set_write_buffer_limits(high=high, low=low)
|
||||
self._maybe_pause_protocol()
|
||||
|
||||
def get_write_buffer_size(self):
|
||||
raise NotImplementedError
|
||||
@@ -0,0 +1,206 @@
|
||||
import socket
|
||||
import warnings
|
||||
|
||||
|
||||
class TransportSocket:
|
||||
|
||||
"""A socket-like wrapper for exposing real transport sockets.
|
||||
|
||||
These objects can be safely returned by APIs like
|
||||
`transport.get_extra_info('socket')`. All potentially disruptive
|
||||
operations (like "socket.close()") are banned.
|
||||
"""
|
||||
|
||||
__slots__ = ('_sock',)
|
||||
|
||||
def __init__(self, sock: socket.socket):
|
||||
self._sock = sock
|
||||
|
||||
def _na(self, what):
|
||||
warnings.warn(
|
||||
f"Using {what} on sockets returned from get_extra_info('socket') "
|
||||
f"will be prohibited in asyncio 3.9. Please report your use case "
|
||||
f"to bugs.python.org.",
|
||||
DeprecationWarning, source=self)
|
||||
|
||||
@property
|
||||
def family(self):
|
||||
return self._sock.family
|
||||
|
||||
@property
|
||||
def type(self):
|
||||
return self._sock.type
|
||||
|
||||
@property
|
||||
def proto(self):
|
||||
return self._sock.proto
|
||||
|
||||
def __repr__(self):
|
||||
s = (
|
||||
f"<asyncio.TransportSocket fd={self.fileno()}, "
|
||||
f"family={self.family!s}, type={self.type!s}, "
|
||||
f"proto={self.proto}"
|
||||
)
|
||||
|
||||
if self.fileno() != -1:
|
||||
try:
|
||||
laddr = self.getsockname()
|
||||
if laddr:
|
||||
s = f"{s}, laddr={laddr}"
|
||||
except socket.error:
|
||||
pass
|
||||
try:
|
||||
raddr = self.getpeername()
|
||||
if raddr:
|
||||
s = f"{s}, raddr={raddr}"
|
||||
except socket.error:
|
||||
pass
|
||||
|
||||
return f"{s}>"
|
||||
|
||||
def __getstate__(self):
|
||||
raise TypeError("Cannot serialize asyncio.TransportSocket object")
|
||||
|
||||
def fileno(self):
|
||||
return self._sock.fileno()
|
||||
|
||||
def dup(self):
|
||||
return self._sock.dup()
|
||||
|
||||
def get_inheritable(self):
|
||||
return self._sock.get_inheritable()
|
||||
|
||||
def shutdown(self, how):
|
||||
# asyncio doesn't currently provide a high-level transport API
|
||||
# to shutdown the connection.
|
||||
self._sock.shutdown(how)
|
||||
|
||||
def getsockopt(self, *args, **kwargs):
|
||||
return self._sock.getsockopt(*args, **kwargs)
|
||||
|
||||
def setsockopt(self, *args, **kwargs):
|
||||
self._sock.setsockopt(*args, **kwargs)
|
||||
|
||||
def getpeername(self):
|
||||
return self._sock.getpeername()
|
||||
|
||||
def getsockname(self):
|
||||
return self._sock.getsockname()
|
||||
|
||||
def getsockbyname(self):
|
||||
return self._sock.getsockbyname()
|
||||
|
||||
def accept(self):
|
||||
self._na('accept() method')
|
||||
return self._sock.accept()
|
||||
|
||||
def connect(self, *args, **kwargs):
|
||||
self._na('connect() method')
|
||||
return self._sock.connect(*args, **kwargs)
|
||||
|
||||
def connect_ex(self, *args, **kwargs):
|
||||
self._na('connect_ex() method')
|
||||
return self._sock.connect_ex(*args, **kwargs)
|
||||
|
||||
def bind(self, *args, **kwargs):
|
||||
self._na('bind() method')
|
||||
return self._sock.bind(*args, **kwargs)
|
||||
|
||||
def ioctl(self, *args, **kwargs):
|
||||
self._na('ioctl() method')
|
||||
return self._sock.ioctl(*args, **kwargs)
|
||||
|
||||
def listen(self, *args, **kwargs):
|
||||
self._na('listen() method')
|
||||
return self._sock.listen(*args, **kwargs)
|
||||
|
||||
def makefile(self):
|
||||
self._na('makefile() method')
|
||||
return self._sock.makefile()
|
||||
|
||||
def sendfile(self, *args, **kwargs):
|
||||
self._na('sendfile() method')
|
||||
return self._sock.sendfile(*args, **kwargs)
|
||||
|
||||
def close(self):
|
||||
self._na('close() method')
|
||||
return self._sock.close()
|
||||
|
||||
def detach(self):
|
||||
self._na('detach() method')
|
||||
return self._sock.detach()
|
||||
|
||||
def sendmsg_afalg(self, *args, **kwargs):
|
||||
self._na('sendmsg_afalg() method')
|
||||
return self._sock.sendmsg_afalg(*args, **kwargs)
|
||||
|
||||
def sendmsg(self, *args, **kwargs):
|
||||
self._na('sendmsg() method')
|
||||
return self._sock.sendmsg(*args, **kwargs)
|
||||
|
||||
def sendto(self, *args, **kwargs):
|
||||
self._na('sendto() method')
|
||||
return self._sock.sendto(*args, **kwargs)
|
||||
|
||||
def send(self, *args, **kwargs):
|
||||
self._na('send() method')
|
||||
return self._sock.send(*args, **kwargs)
|
||||
|
||||
def sendall(self, *args, **kwargs):
|
||||
self._na('sendall() method')
|
||||
return self._sock.sendall(*args, **kwargs)
|
||||
|
||||
def set_inheritable(self, *args, **kwargs):
|
||||
self._na('set_inheritable() method')
|
||||
return self._sock.set_inheritable(*args, **kwargs)
|
||||
|
||||
def share(self, process_id):
|
||||
self._na('share() method')
|
||||
return self._sock.share(process_id)
|
||||
|
||||
def recv_into(self, *args, **kwargs):
|
||||
self._na('recv_into() method')
|
||||
return self._sock.recv_into(*args, **kwargs)
|
||||
|
||||
def recvfrom_into(self, *args, **kwargs):
|
||||
self._na('recvfrom_into() method')
|
||||
return self._sock.recvfrom_into(*args, **kwargs)
|
||||
|
||||
def recvmsg_into(self, *args, **kwargs):
|
||||
self._na('recvmsg_into() method')
|
||||
return self._sock.recvmsg_into(*args, **kwargs)
|
||||
|
||||
def recvmsg(self, *args, **kwargs):
|
||||
self._na('recvmsg() method')
|
||||
return self._sock.recvmsg(*args, **kwargs)
|
||||
|
||||
def recvfrom(self, *args, **kwargs):
|
||||
self._na('recvfrom() method')
|
||||
return self._sock.recvfrom(*args, **kwargs)
|
||||
|
||||
def recv(self, *args, **kwargs):
|
||||
self._na('recv() method')
|
||||
return self._sock.recv(*args, **kwargs)
|
||||
|
||||
def settimeout(self, value):
|
||||
if value == 0:
|
||||
return
|
||||
raise ValueError(
|
||||
'settimeout(): only 0 timeout is allowed on transport sockets')
|
||||
|
||||
def gettimeout(self):
|
||||
return 0
|
||||
|
||||
def setblocking(self, flag):
|
||||
if not flag:
|
||||
return
|
||||
raise ValueError(
|
||||
'setblocking(): transport sockets cannot be blocking')
|
||||
|
||||
def __enter__(self):
|
||||
self._na('context manager protocol')
|
||||
return self._sock.__enter__()
|
||||
|
||||
def __exit__(self, *err):
|
||||
self._na('context manager protocol')
|
||||
return self._sock.__exit__(*err)
|
||||
File diff suppressed because it is too large
Load Diff
File diff suppressed because it is too large
Load Diff
@@ -0,0 +1,173 @@
|
||||
"""Various Windows specific bits and pieces."""
|
||||
|
||||
import sys
|
||||
|
||||
if sys.platform != 'win32': # pragma: no cover
|
||||
raise ImportError('win32 only')
|
||||
|
||||
import _winapi
|
||||
import itertools
|
||||
import msvcrt
|
||||
import os
|
||||
import subprocess
|
||||
import tempfile
|
||||
import warnings
|
||||
|
||||
|
||||
__all__ = 'pipe', 'Popen', 'PIPE', 'PipeHandle'
|
||||
|
||||
|
||||
# Constants/globals
|
||||
|
||||
|
||||
BUFSIZE = 8192
|
||||
PIPE = subprocess.PIPE
|
||||
STDOUT = subprocess.STDOUT
|
||||
_mmap_counter = itertools.count()
|
||||
|
||||
|
||||
# Replacement for os.pipe() using handles instead of fds
|
||||
|
||||
|
||||
def pipe(*, duplex=False, overlapped=(True, True), bufsize=BUFSIZE):
|
||||
"""Like os.pipe() but with overlapped support and using handles not fds."""
|
||||
address = tempfile.mktemp(
|
||||
prefix=r'\\.\pipe\python-pipe-{:d}-{:d}-'.format(
|
||||
os.getpid(), next(_mmap_counter)))
|
||||
|
||||
if duplex:
|
||||
openmode = _winapi.PIPE_ACCESS_DUPLEX
|
||||
access = _winapi.GENERIC_READ | _winapi.GENERIC_WRITE
|
||||
obsize, ibsize = bufsize, bufsize
|
||||
else:
|
||||
openmode = _winapi.PIPE_ACCESS_INBOUND
|
||||
access = _winapi.GENERIC_WRITE
|
||||
obsize, ibsize = 0, bufsize
|
||||
|
||||
openmode |= _winapi.FILE_FLAG_FIRST_PIPE_INSTANCE
|
||||
|
||||
if overlapped[0]:
|
||||
openmode |= _winapi.FILE_FLAG_OVERLAPPED
|
||||
|
||||
if overlapped[1]:
|
||||
flags_and_attribs = _winapi.FILE_FLAG_OVERLAPPED
|
||||
else:
|
||||
flags_and_attribs = 0
|
||||
|
||||
h1 = h2 = None
|
||||
try:
|
||||
h1 = _winapi.CreateNamedPipe(
|
||||
address, openmode, _winapi.PIPE_WAIT,
|
||||
1, obsize, ibsize, _winapi.NMPWAIT_WAIT_FOREVER, _winapi.NULL)
|
||||
|
||||
h2 = _winapi.CreateFile(
|
||||
address, access, 0, _winapi.NULL, _winapi.OPEN_EXISTING,
|
||||
flags_and_attribs, _winapi.NULL)
|
||||
|
||||
ov = _winapi.ConnectNamedPipe(h1, overlapped=True)
|
||||
ov.GetOverlappedResult(True)
|
||||
return h1, h2
|
||||
except:
|
||||
if h1 is not None:
|
||||
_winapi.CloseHandle(h1)
|
||||
if h2 is not None:
|
||||
_winapi.CloseHandle(h2)
|
||||
raise
|
||||
|
||||
|
||||
# Wrapper for a pipe handle
|
||||
|
||||
|
||||
class PipeHandle:
|
||||
"""Wrapper for an overlapped pipe handle which is vaguely file-object like.
|
||||
|
||||
The IOCP event loop can use these instead of socket objects.
|
||||
"""
|
||||
def __init__(self, handle):
|
||||
self._handle = handle
|
||||
|
||||
def __repr__(self):
|
||||
if self._handle is not None:
|
||||
handle = f'handle={self._handle!r}'
|
||||
else:
|
||||
handle = 'closed'
|
||||
return f'<{self.__class__.__name__} {handle}>'
|
||||
|
||||
@property
|
||||
def handle(self):
|
||||
return self._handle
|
||||
|
||||
def fileno(self):
|
||||
if self._handle is None:
|
||||
raise ValueError("I/O operation on closed pipe")
|
||||
return self._handle
|
||||
|
||||
def close(self, *, CloseHandle=_winapi.CloseHandle):
|
||||
if self._handle is not None:
|
||||
CloseHandle(self._handle)
|
||||
self._handle = None
|
||||
|
||||
def __del__(self, _warn=warnings.warn):
|
||||
if self._handle is not None:
|
||||
_warn(f"unclosed {self!r}", ResourceWarning, source=self)
|
||||
self.close()
|
||||
|
||||
def __enter__(self):
|
||||
return self
|
||||
|
||||
def __exit__(self, t, v, tb):
|
||||
self.close()
|
||||
|
||||
|
||||
# Replacement for subprocess.Popen using overlapped pipe handles
|
||||
|
||||
|
||||
class Popen(subprocess.Popen):
|
||||
"""Replacement for subprocess.Popen using overlapped pipe handles.
|
||||
|
||||
The stdin, stdout, stderr are None or instances of PipeHandle.
|
||||
"""
|
||||
def __init__(self, args, stdin=None, stdout=None, stderr=None, **kwds):
|
||||
assert not kwds.get('universal_newlines')
|
||||
assert kwds.get('bufsize', 0) == 0
|
||||
stdin_rfd = stdout_wfd = stderr_wfd = None
|
||||
stdin_wh = stdout_rh = stderr_rh = None
|
||||
if stdin == PIPE:
|
||||
stdin_rh, stdin_wh = pipe(overlapped=(False, True), duplex=True)
|
||||
stdin_rfd = msvcrt.open_osfhandle(stdin_rh, os.O_RDONLY)
|
||||
else:
|
||||
stdin_rfd = stdin
|
||||
if stdout == PIPE:
|
||||
stdout_rh, stdout_wh = pipe(overlapped=(True, False))
|
||||
stdout_wfd = msvcrt.open_osfhandle(stdout_wh, 0)
|
||||
else:
|
||||
stdout_wfd = stdout
|
||||
if stderr == PIPE:
|
||||
stderr_rh, stderr_wh = pipe(overlapped=(True, False))
|
||||
stderr_wfd = msvcrt.open_osfhandle(stderr_wh, 0)
|
||||
elif stderr == STDOUT:
|
||||
stderr_wfd = stdout_wfd
|
||||
else:
|
||||
stderr_wfd = stderr
|
||||
try:
|
||||
super().__init__(args, stdin=stdin_rfd, stdout=stdout_wfd,
|
||||
stderr=stderr_wfd, **kwds)
|
||||
except:
|
||||
for h in (stdin_wh, stdout_rh, stderr_rh):
|
||||
if h is not None:
|
||||
_winapi.CloseHandle(h)
|
||||
raise
|
||||
else:
|
||||
if stdin_wh is not None:
|
||||
self.stdin = PipeHandle(stdin_wh)
|
||||
if stdout_rh is not None:
|
||||
self.stdout = PipeHandle(stdout_rh)
|
||||
if stderr_rh is not None:
|
||||
self.stderr = PipeHandle(stderr_rh)
|
||||
finally:
|
||||
if stdin == PIPE:
|
||||
os.close(stdin_rfd)
|
||||
if stdout == PIPE:
|
||||
os.close(stdout_wfd)
|
||||
if stderr == PIPE:
|
||||
os.close(stderr_wfd)
|
||||
Reference in New Issue
Block a user