From 46bf723e55c17f52bcc834bdeda33b77644d32ce Mon Sep 17 00:00:00 2001 From: jasinluo <1127097451@qq.com> Date: Tue, 10 Nov 2026 10:57:59 +0800 Subject: [PATCH] CI: fix code lint problem, ci.yml --- .flake8 | 2 ++ .github/workflows/ci.yml | 1 + tests/http/test_http_server_transport.py | 10 +++++++ tests/transport/test_thread_model.py | 8 +++++- trpc_sdk/runtime/__init__.py | 13 +++++---- trpc_sdk/runtime/_threadmodel/_loop_thread.py | 10 +++---- .../runtime/_threadmodel/_separate_thread.py | 6 ++-- trpc_sdk/transport/__init__.py | 1 - trpc_sdk/transport/_server/__init__.py | 1 - .../transport/_server/_datagram_connection.py | 15 ++++------ .../transport/_server/_stream_transport.py | 1 - trpc_sdk/transport/_server/_trpc_transport.py | 28 +++++++++++-------- 12 files changed, 57 insertions(+), 39 deletions(-) create mode 100644 .flake8 diff --git a/.flake8 b/.flake8 new file mode 100644 index 0000000..6deafc2 --- /dev/null +++ b/.flake8 @@ -0,0 +1,2 @@ +[flake8] +max-line-length = 120 diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 95f05e2..591aaeb 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -71,6 +71,7 @@ jobs: - name: Install package and dependencies run: | + pip install -e ".[dev,fast]" "protobuf>=3.19.4,<3.20" pip install -r requirements-test.txt - name: Run tests with coverage diff --git a/tests/http/test_http_server_transport.py b/tests/http/test_http_server_transport.py index 7129c4d..16bce5c 100644 --- a/tests/http/test_http_server_transport.py +++ b/tests/http/test_http_server_transport.py @@ -34,6 +34,7 @@ from trpc_sdk.http import get_http_response from trpc_sdk.http import register_user_http_decode_func from trpc_sdk.http import _routes +from trpc_sdk.utils import set_default_close_wait def test_http_server_transport(): @@ -58,6 +59,15 @@ def test_http_listen_serves_the_runner_on_one_loop(monkeypatch): transport = HttpServerTransportFactory().create() transport.bind(opts) + real_create_socket = transport.create_socket + + def _create_socket(reuse_port, stream=True): + sock = real_create_socket(reuse_port, stream) + set_default_close_wait(sock) + return sock + + monkeypatch.setattr(transport, "create_socket", _create_socket) + def _app(): app = web.Application() app.router.add_get("/", lambda request: web.Response(text="ok")) diff --git a/tests/transport/test_thread_model.py b/tests/transport/test_thread_model.py index 8f34fdf..907346f 100644 --- a/tests/transport/test_thread_model.py +++ b/tests/transport/test_thread_model.py @@ -30,6 +30,7 @@ from trpc_sdk.transport._server import serve_stream from trpc_sdk.transport._server import _trpc_transport from trpc_sdk.utils import create_new_socket +from trpc_sdk.utils import set_default_close_wait @pytest.fixture @@ -67,7 +68,12 @@ def attach_socket(self, sock, protocol_factory, ssl_context=None): def _listen_socket() -> socket.socket: - return create_new_socket(("127.0.0.1", 0), "", False, True, 16) + # Linux inherits SO_LINGER(1, 0) from the listen socket, so closing an + # accepted fd sends RST and can fail the client's connect. Restore the + # default linger so this test sees a normal FIN. + sock = create_new_socket(("127.0.0.1", 0), "", False, True, 16) + set_default_close_wait(sock) + return sock def test_default_model_is_rebuilt_on_the_calling_loop(loop): diff --git a/trpc_sdk/runtime/__init__.py b/trpc_sdk/runtime/__init__.py index 69497e4..01a27cf 100644 --- a/trpc_sdk/runtime/__init__.py +++ b/trpc_sdk/runtime/__init__.py @@ -3,7 +3,6 @@ # Copyright (C) 2026 Tencent. All rights reserved. # # tRPC-Python is licensed under Apache-2.0. - """tRPC-Python runtime: the process and thread model behind a server. There are two kinds of process. The **worker** is the process the application @@ -30,7 +29,8 @@ what a CPU-bound service needs; ``merge`` will not scale it across cores. Only tRPC-over-TCP addresses pass through the worker; UDP, HTTP and user-protocol addresses are bound by every rpc task with ``SO_REUSEPORT``. -- ``peer`` (``multi_process`` with ``io_handle_type: merge``) — the worker forks ``worker_num - 1`` equal workers; every one, +- ``peer`` (``multi_process`` with ``io_handle_type: merge``) — the worker + forks ``worker_num - 1`` equal workers; every one, itself included, binds every address with ``SO_REUSEPORT`` and serves it like a single-loop ``merge`` worker. Nothing is forwarded between processes; the kernel spreads connections. Tasks are not supported. @@ -56,7 +56,8 @@ ``runtime/threadmodel`` and ``separate_runtime``); responses cross back to the I/O loops through :class:`~trpc_sdk.transport.ConnWriter`, re-exported here -- :mod:`~trpc_sdk.runtime._common` — SIGTERM callbacks, process metadata and the exit template every mode shares (trpc-cpp +- :mod:`~trpc_sdk.runtime._common` — SIGTERM callbacks, process metadata + and the exit template every mode shares (trpc-cpp ``runtime/common``) - :mod:`~trpc_sdk.runtime._process` — task children and the pre-fork socketpair mesh (trpc-cpp ``runtime/task``), plus the multi-process @@ -116,9 +117,9 @@ # Called by Server.serve. Absent from __all__ on purpose: application code # uses set_signal_handle, register_handler_thread_hook, and create_server. -from ._common import run_server_signal_handle -from ._modes import create_worker -from ._threadmodel import freeze_handler_thread_hooks +from ._common import run_server_signal_handle # noqa: F401 +from ._modes import create_worker # noqa: F401 +from ._threadmodel import freeze_handler_thread_hooks # noqa: F401 __all__ = [ # Process model diff --git a/trpc_sdk/runtime/_threadmodel/_loop_thread.py b/trpc_sdk/runtime/_threadmodel/_loop_thread.py index c7e32ea..4bb518f 100644 --- a/trpc_sdk/runtime/_threadmodel/_loop_thread.py +++ b/trpc_sdk/runtime/_threadmodel/_loop_thread.py @@ -3,7 +3,6 @@ # Copyright (C) 2026 Tencent. All rights reserved. # # tRPC-Python is licensed under Apache-2.0. - """Event-loop threads that a merge-mode worker spreads its work over. In ``merge`` mode one process both accepts connections and runs handlers. @@ -331,8 +330,7 @@ def _join(self, timeout: float) -> bool: return not self._thread.is_alive() -def stop_loop_threads(threads: Sequence[LoopThread], - before_stop: Optional[Callable[[int], Awaitable[None]]], +def stop_loop_threads(threads: Sequence[LoopThread], before_stop: Optional[Callable[[int], Awaitable[None]]], timeout: float) -> bool: """Stop several loop threads in parallel, under one shared deadline. @@ -360,8 +358,10 @@ def stop_loop_threads(threads: Sequence[LoopThread], if before_stop is None or thread.loop is None or thread.loop.is_closed(): continue try: - pending.append((thread, thread._schedule( # pylint: disable=protected-access - lambda index=thread.index: before_stop(index)))) + pending.append(( + thread, + thread._schedule( # pylint: disable=protected-access + lambda index=thread.index: before_stop(index)))) except RuntimeError: continue for thread, future in pending: diff --git a/trpc_sdk/runtime/_threadmodel/_separate_thread.py b/trpc_sdk/runtime/_threadmodel/_separate_thread.py index 441fdea..f1d7b91 100644 --- a/trpc_sdk/runtime/_threadmodel/_separate_thread.py +++ b/trpc_sdk/runtime/_threadmodel/_separate_thread.py @@ -3,7 +3,6 @@ # Copyright (C) 2026 Tencent. All rights reserved. # # tRPC-Python is licensed under Apache-2.0. - """Worker of ``server.mode: merge`` with ``handle_thread_num > 0`` (``io_handle_type: separate``). inside one process; not to be confused with ``server.mode: separate`` in @@ -115,8 +114,9 @@ def _warn_io_threads_without_tls(self) -> None: return if any(t.ipc_forwardable and t.opts.tls_cert and t.opts.tls_key for t in self.transports): return - logger.warning("[separate thread worker] io_thread_num=%s, but no tRPC TCP endpoint uses TLS; " - "extra I/O loops only pay off for TLS", self.io_thread_num) + logger.warning( + "[separate thread worker] io_thread_num=%s, but no tRPC TCP endpoint uses TLS; " + "extra I/O loops only pay off for TLS", self.io_thread_num) def _build_thread_model(self) -> ThreadModel: """I/O on the main loop plus ``io_thread_num - 1`` I/O threads; handlers on the handle threads. diff --git a/trpc_sdk/transport/__init__.py b/trpc_sdk/transport/__init__.py index e95f512..33789a8 100644 --- a/trpc_sdk/transport/__init__.py +++ b/trpc_sdk/transport/__init__.py @@ -3,7 +3,6 @@ # Copyright (C) 2026 Tencent. All rights reserved. # # tRPC-Python is licensed under Apache-2.0. - """tRPC-Python transport package: how bytes travel, not what they mean. The package holds both halves of the wire path. diff --git a/trpc_sdk/transport/_server/__init__.py b/trpc_sdk/transport/_server/__init__.py index ca4be9e..0890ac7 100644 --- a/trpc_sdk/transport/_server/__init__.py +++ b/trpc_sdk/transport/_server/__init__.py @@ -3,7 +3,6 @@ # Copyright (C) 2026 Tencent. All rights reserved. # # tRPC-Python is licensed under Apache-2.0. - """Server half of the transport package: the per-service listen transports. A server transport is constructed and bound in the parent (``bind`` only diff --git a/trpc_sdk/transport/_server/_datagram_connection.py b/trpc_sdk/transport/_server/_datagram_connection.py index c6928fb..0ecea4c 100644 --- a/trpc_sdk/transport/_server/_datagram_connection.py +++ b/trpc_sdk/transport/_server/_datagram_connection.py @@ -3,7 +3,6 @@ # Copyright (C) 2026 Tencent. All rights reserved. # # tRPC-Python is licensed under Apache-2.0. - """Datagram protocol of a tRPC-over-UDP listen socket. :meth:`~trpc_sdk.transport.TrpcServerTransport.datagram_protocol_factory` @@ -50,12 +49,8 @@ class DatagramPeer: __slots__ = ("aio_conn", "local_addr", "remote_addr", "_writer") - def __init__(self, - loop: asyncio.AbstractEventLoop, - aio_conn: asyncio.DatagramTransport, - local_addr: Tuple[str, int], - remote_addr: Tuple[str, int], - source: Any) -> None: + def __init__(self, loop: asyncio.AbstractEventLoop, aio_conn: asyncio.DatagramTransport, + local_addr: Tuple[str, int], remote_addr: Tuple[str, int], source: Any) -> None: """Bind the peer to one datagram's source. Args: @@ -141,8 +136,10 @@ def datagram_received(self, data: bytes, addr: Any) -> None: return except Exception as err: # pylint: disable=broad-except report_attr("trpc_datagram_connection_read_fatal_err", 1) - logger.error(f"[datagram_connection] fatal err while decoding datagram, remote_addr: {addr}, " - f"err: {err}", exc_info=True) + logger.error( + f"[datagram_connection] fatal err while decoding datagram, remote_addr: {addr}, " + f"err: {err}", + exc_info=True) self.parser.reset() return if not frames: diff --git a/trpc_sdk/transport/_server/_stream_transport.py b/trpc_sdk/transport/_server/_stream_transport.py index ea623d0..af538e0 100644 --- a/trpc_sdk/transport/_server/_stream_transport.py +++ b/trpc_sdk/transport/_server/_stream_transport.py @@ -3,7 +3,6 @@ # Copyright (C) 2026 Tencent. All rights reserved. # # tRPC-Python is licensed under Apache-2.0. - """Streaming server transport that maps (session, stream) to a write handle. Inherits :class:`TrpcServerTransport`, so ``bind`` (pre-fork, options only) diff --git a/trpc_sdk/transport/_server/_trpc_transport.py b/trpc_sdk/transport/_server/_trpc_transport.py index 12cf045..e1e5523 100644 --- a/trpc_sdk/transport/_server/_trpc_transport.py +++ b/trpc_sdk/transport/_server/_trpc_transport.py @@ -3,7 +3,6 @@ # Copyright (C) 2026 Tencent. All rights reserved. # # tRPC-Python is licensed under Apache-2.0. - """tRPC server transport: record options before fork, listen after it. :meth:`TrpcServerTransport.bind` runs in ``ServiceAdapter.init_transport`` and only keeps the @@ -113,7 +112,8 @@ class TrpcServerTransport(BaseServerTransport): default. A subclass serving another protocol over the same listen machinery calls :meth:`set_io_connection`, overrides :meth:`_msg_handler` and, when the framing differs, sets another ``opts.parser``; its - connections are then :class:`DefaultIoConnection` objects. Datagrams are always served by :class:`DefaultDatagramConnection`, which + connections are then :class:`DefaultIoConnection` objects. Datagrams + are always served by :class:`DefaultDatagramConnection`, which hands every frame to :meth:`_msg_handler`. Attributes: @@ -248,11 +248,12 @@ def protocol_factory(self, shared_pool: Optional[Any] = None) -> Callable[[], An """ opts = self.opts if self._io_connection is None: - return partial(TCPConnection, - TcpConnectionOption(parser=opts.parser, - rpc_handler=opts.rpc_handler, - timeout=opts.idle_timeout, - shared_pool=shared_pool)) + return partial( + TCPConnection, + TcpConnectionOption(parser=opts.parser, + rpc_handler=opts.rpc_handler, + timeout=opts.idle_timeout, + shared_pool=shared_pool)) return partial(self._io_connection, ConnectionOption(parser=opts.parser, msg_handler=self._msg_handler)) async def _msg_handler(self, conn: ServerConnection, data: bytes) -> None: @@ -309,9 +310,7 @@ def datagram_protocol_factory(self, shared_pool: Optional[Any] = None) -> Callab """ opts = self.opts return partial(DefaultDatagramConnection, - ConnectionOption(parser=opts.parser, - msg_handler=self._msg_handler, - shared_pool=shared_pool)) + ConnectionOption(parser=opts.parser, msg_handler=self._msg_handler, shared_pool=shared_pool)) def create_socket(self, reuse_port: bool, stream: bool = True) -> socket.socket: """Create and bind one socket on the configured address. @@ -347,8 +346,13 @@ def _serve_stream(self, """ sock = self.create_socket(model.reuse_port) try: - listener = serve_stream(model, group, sock, factory_for, self.opts.backlog, - ssl_context=ssl_context, address=self.opts.address) + listener = serve_stream(model, + group, + sock, + factory_for, + self.opts.backlog, + ssl_context=ssl_context, + address=self.opts.address) except BaseException: sock.close() raise