文档建议反馈控制台
首页
学习
活动
专区
圈层
工具
MCP广场
文章/答案/技术大牛
发布
社区首页 >问答首页 >使用websockets将夸脱项目从异步迁移到三重奏

问使用websockets将夸脱项目从异步迁移到三重奏
EN

Stack Overflow用户
提问于 2021-07-19 18:27:35
回答 1查看 328关注 0票数 1

我正在尝试把我的异步项目转换成三重奏。我知道我必须使用内存通道而不是队列，但由于某种原因，我没有预期的结果。

我的主要问题是，当我运行两个客户端时，如果第二个客户机离开(从服务器广播“部件”消息会引发错误)，第一个客户端将不会收到通知。另一个问题是，有时客户端在打开websocket时立即退出。当我使用异步时，一切都很好。

下面是当第二个客户端断开连接时得到的堆栈跟踪：

代码语言：javascript
复制
[2021-07-30 18:39:51,899] ERROR in app: Exception on websocket /ws
Traceback (most recent call last):
  File "/tmp/debug/venv/lib/python3.9/site-packages/quart_trio/app.py", line 175, in handle_websocket
    return await self.full_dispatch_websocket(websocket_context)
  File "/tmp/debug/venv/lib/python3.9/site-packages/quart_trio/app.py", line 197, in full_dispatch_websocket
    result = await self.handle_user_exception(error)
  File "/tmp/debug/venv/lib/python3.9/site-packages/quart_trio/app.py", line 166, in handle_user_exception
    raise error
  File "/tmp/debug/venv/lib/python3.9/site-packages/quart_trio/app.py", line 195, in full_dispatch_websocket
    result = await self.dispatch_websocket(websocket_context)
  File "/tmp/debug/venv/lib/python3.9/site-packages/quart/app.py", line 1651, in dispatch_websocket
    return await self.ensure_async(handler)(**websocket_.view_args)
  File "/tmp/debug/server.py", line 103, in wsocket
    nursery.start_soon(receiving, u)
  File "/tmp/debug/venv/lib/python3.9/site-packages/trio/_core/_run.py", line 815, in __aexit__
    raise combined_error_from_nursery
trio.MultiError: Cancelled(), Cancelled(), Cancelled()

Details of embedded exception 1:

  Traceback (most recent call last):
    File "/tmp/debug/venv/lib/python3.9/site-packages/trio/_core/_run.py", line 1172, in raise_cancel
      raise Cancelled._create()
  trio.Cancelled: Cancelled

Details of embedded exception 2:

  Traceback (most recent call last):
    File "/tmp/debug/server.py", line 68, in receiving
      data = await websocket.receive_json()
    File "/tmp/debug/venv/lib/python3.9/site-packages/quart/wrappers/websocket.py", line 68, in receive_json
      data = await self.receive()
    File "/tmp/debug/venv/lib/python3.9/site-packages/quart/wrappers/websocket.py", line 57, in receive
      return await self._receive()
    File "/tmp/debug/venv/lib/python3.9/site-packages/trio/_channel.py", line 314, in receive
      return await trio.lowlevel.wait_task_rescheduled(abort_fn)
    File "/tmp/debug/venv/lib/python3.9/site-packages/trio/_core/_traps.py", line 166, in wait_task_rescheduled
      return (await _async_yield(WaitTaskRescheduled(abort_func))).unwrap()
    File "/tmp/debug/venv/lib/python3.9/site-packages/outcome/_impl.py", line 138, in unwrap
      raise captured_error
    File "/tmp/debug/venv/lib/python3.9/site-packages/trio/_core/_run.py", line 1172, in raise_cancel
      raise Cancelled._create()
  trio.Cancelled: Cancelled

Details of embedded exception 3:

  Traceback (most recent call last):
    File "/tmp/debug/server.py", line 54, in sending
      data = await u.queue_recv.receive()
    File "/tmp/debug/venv/lib/python3.9/site-packages/trio/_channel.py", line 314, in receive
      return await trio.lowlevel.wait_task_rescheduled(abort_fn)
    File "/tmp/debug/venv/lib/python3.9/site-packages/trio/_core/_traps.py", line 166, in wait_task_rescheduled
      return (await _async_yield(WaitTaskRescheduled(abort_func))).unwrap()
    File "/tmp/debug/venv/lib/python3.9/site-packages/outcome/_impl.py", line 138, in unwrap
      raise captured_error
    File "/tmp/debug/venv/lib/python3.9/site-packages/trio/_core/_run.py", line 1172, in raise_cancel
      raise Cancelled._create()
  trio.Cancelled: Cancelled

  During handling of the above exception, another exception occurred:

  Traceback (most recent call last):
    File "/tmp/debug/server.py", line 63, in sending
      await broadcast({'type': 'part', 'data': u.name})
    File "/tmp/debug/server.py", line 75, in broadcast
      await user.queue_send.send(message)
    File "/tmp/debug/venv/lib/python3.9/site-packages/trio/_channel.py", line 159, in send
      await trio.lowlevel.checkpoint_if_cancelled()
    File "/tmp/debug/venv/lib/python3.9/site-packages/trio/_core/_run.py", line 2361, in checkpoint_if_cancelled
      await _core.checkpoint()
    File "/tmp/debug/venv/lib/python3.9/site-packages/trio/_core/_run.py", line 2339, in checkpoint
      await _core.wait_task_rescheduled(lambda _: _core.Abort.SUCCEEDED)
    File "/tmp/debug/venv/lib/python3.9/site-packages/trio/_core/_traps.py", line 166, in wait_task_rescheduled
      return (await _async_yield(WaitTaskRescheduled(abort_func))).unwrap()
    File "/tmp/debug/venv/lib/python3.9/site-packages/outcome/_impl.py", line 138, in unwrap
      raise captured_error
    File "/tmp/debug/venv/lib/python3.9/site-packages/trio/_core/_run.py", line 1172, in raise_cancel
      raise Cancelled._create()
  trio.Cancelled: Cancelled

下面是代码(将TRIO设置为False以使用异步)：

server.py

代码语言：javascript
复制
#!/usr/bin/env python
from quart import Quart, websocket, request, jsonify, json
from quart_trio import QuartTrio
from functools import wraps
import uuid
import trio
import asyncio
from quart_auth import AuthUser, AuthManager, login_user, _AuthSerializer

TRIO = True

if TRIO:
    app = QuartTrio(__name__)
else:
    app = Quart(__name__)
app.secret_key = '**changeme**'

authorized_users = set()

class User(AuthUser):
    @staticmethod
    def current():
        token = websocket.cookies['QUART_AUTH']
        serializer = _AuthSerializer('**changeme**', 'quart auth salt')
        user_id = serializer.loads(token)
        for u in authorized_users:
            if u.auth_id == user_id:
                return u
        return None

    def __init__(self, auth_id):
        super().__init__(auth_id)
        self.name = None
        self.queue = None # asyncio
        self.queue_send = None #trio
        self.queue_recv = None #trio
        self.connected = False
        self.websockets = set()    

    def to_dict(self):
        return {
            'id': self.auth_id,
            'name': self.name
        }

auth_manager = AuthManager()
auth_manager.user_class = User

async def sending(u: User):
    await broadcast({'type': 'join', 'data': u.name})
    try:
        while True:
            if TRIO:
                data = await u.queue_recv.receive()
            else:
                data = await u.queue.get()
            for s in u.websockets:
                await s.send_json(data)
    finally:
        u.websockets.remove(websocket._get_current_object())
        if len(u.websockets) == 0:
            u.connected = False
            await broadcast({'type': 'part', 'data': u.name})


async def receiving(u: User):
    while True:
        data = await websocket.receive_json()
        if data['type'] == 'msg':
            await broadcast({'type': 'msg', 'user': u.name, 'data': data['data']})

async def broadcast(message):
    for user in [u for u in authorized_users if u.connected]:
        if TRIO:
            await user.queue_send.send(message)
        else:
            await user.queue.put(message)

@app.route('/api/v1/auth', methods=['POST'])
async def auth_login():
    data = await request.json
    user_id = str(uuid.uuid4())[:8]
    u = User(user_id)
    u.name = data['login'] or 'Anonymous'+user_id
    if TRIO:
        u.queue_send, u.queue_recv = trio.open_memory_channel(float('inf'))
    else:
        u.queue = asyncio.Queue()
    login_user(u, True)
    authorized_users.add(u)
    return jsonify({'id': user_id, 'name': u.name}), 200

@app.websocket('/ws')
async def wsocket():
    u = User.current()
    if u is None:
        return
    u.websockets.add(websocket._get_current_object())
    u.connected = True
    if TRIO:
        async with trio.open_nursery() as nursery:
            nursery.start_soon(sending, u)
            nursery.start_soon(receiving, u)
    else:
        producer = asyncio.create_task(sending(u))
        consumer = asyncio.create_task(receiving(u))
        await asyncio.gather(producer, consumer)


auth_manager.init_app(app)

if __name__ == "__main__":
    app.run(host='localhost', port=8080)

client.py

代码语言：javascript
复制
#!/usr/bin/env python

import asks
import trio
import trio_websocket
import json

asks.init(trio)

class User:
    def __init__(self, name: str="") -> None:
        self.name = name

class Client(User):
    def __init__(self) -> None:
        super(Client, self).__init__()
        self.web_url = 'http://localhost:8080/api/v1'
        self.ws_url = 'ws://localhost:8080/ws'
        self.ws = None
        self.nursery = None
        self.cookiejar = {}
    
    async def send(self, msg: dict) -> None:
        if self.ws is not None:
            await self.ws.send_message(json.dumps(msg))

    async def reader(self, websocket) -> None:
        while True:
            try:
                message_raw = await websocket.get_message()
                msg = json.loads(message_raw)
                if msg['type'] == 'msg':
                    print(f"<{msg['user']}> {msg['data']}")
                elif msg['type'] == 'join':
                    print(f"* {msg['data']} joined")
                elif msg['type'] == 'part':
                    print(f"* {msg['data']} left")
            except trio_websocket.ConnectionClosed:
                break

    async def login(self) -> None:
        rlogin = await asks.post(self.web_url + '/auth', json={'login': self.name, 'password': 'password'})
        for c in rlogin.cookies:
            if c.name == 'QUART_AUTH':
                self.cookiejar = {'QUART_AUTH': c.value}

    async def connect(self) -> None:
        await self.login()
        async with trio_websocket.open_websocket_url(self.ws_url, extra_headers=[('Cookie', 'QUART_AUTH'+'='+self.cookiejar['QUART_AUTH'])]) as websocket:
            self.ws = websocket
            await self.send({'type': 'msg', 'data': 'hello'})
            async with trio.open_nursery() as nursery:
                self.nursery = nursery
                nursery.start_soon(self.reader, websocket)

    def run(self) -> None:
        trio.run(self.connect)

c = Client()
c.name = 'clientA'
c.run()

编辑:我使用anyio进行了测试，虽然anyio+trio也是这样做的，但是anyio+asyncio会重复这个问题(无一例外)。所以我想它来自队列替换。

python
websocket
quart
python-trio
EN

回答 1

Stack Overflow用户

回答已采纳

发布于 2021-07-31 10:21:30

“好吧，”蒂布斯，我想我找到了这个问题。问题在于三人组处理取消的方式。要获得完整的文档，请阅读以下文档：

https://trio.readthedocs.io/en/stable/reference-core.html#cancellation-and-timeouts

然而，要解释这里发生了什么，当用户断开连接时，夸脱-Trio所做的是在每个运行/等待websocket的协同线中引发一个Cancelled异常。对于websocket用户，目前有两个位置正在等待：

在async def sending(u: User):中

代码语言：javascript
复制
async def sending(u: User):
    await broadcast({'type': 'join', 'data': u.name})
    try:
        while True:
            if TRIO:
                data = await u.queue_recv.receive()  <--- Code is waiting here, Cancelled is raised here
            else:
                data = await u.queue.get()
            for s in u.websockets:
                await s.send_json(data)
    finally:
        u.websockets.remove(websocket._get_current_object())
        if len(u.websockets) == 0:
            u.connected = False
            await broadcast({'type': 'part', 'data': u.name})

在async def receiving(u: User):中

代码语言：javascript
复制
async def receiving(u: User):
    while True:
        data = await websocket.receive_json()   <--- Code is waiting here, Cancelled is raised here
        if data['type'] == 'msg':
            await broadcast({'type': 'msg', 'user': u.name, 'data': data['data']})

好吧，那么接下来会发生什么呢？在sending()函数中，我们向下移动到finally块，它开始执行，但接着我们调用另一个可访问的函数：

代码语言：javascript
复制
    finally:
        u.websockets.remove(websocket._get_current_object())
        if len(u.websockets) == 0:
            u.connected = False
            await broadcast({'type': 'part', 'data': u.name})  <--- we call an awaitable here

来自三重奏的医生：

三重奏中的

取消是“级别触发”，这意味着一旦一个块被取消，该块中所有可取消的操作都会一直被取消。

因此，当await broadcast(...)被调用时，它立即是Cancelled，与asyncio不同的是，它的行为不同。这就解释了为什么您的“部件”消息从未发送过。因此，如果您想在被取消时执行一些清理工作，则应该打开一个新的cancel作用域，并保护它不被取消，如下所示：

代码语言：javascript
复制
async def sending(u: User):
    await broadcast({'type': 'join', 'data': u.name})
    try:
        while True:
            if TRIO:
                data = await u.queue_recv.receive()  <--- Code is waiting here, Cancelled is raised here
            else:
                data = await u.queue.get()
            for s in u.websockets:
                await s.send_json(data)
    finally:
        u.websockets.remove(websocket._get_current_object())
        if len(u.websockets) == 0:
            u.connected = False
            with trio.move_on_after(5) as leaving_cancel_scope:
                # Shield from the cancellation for 5s to run the broadcast of leaving
                leaving_cancel_scope.shield = True
                await broadcast({'type': 'part', 'data': u.name})

或者，您也可以在应用程序托儿所上启动广播协同。请注意，如果broadcast(...)崩溃，则整个运行中的应用程序都会崩溃，除非您在broadcast(...)函数中添加了try/ put：

代码语言：javascript
复制
async def sending(u: User):
    await broadcast({'type': 'join', 'data': u.name})
    try:
        while True:
            if TRIO:
                data = await u.queue_recv.receive()
            else:
                data = await u.queue.get()
            for s in u.websockets:
                await s.send_json(data)
    finally:
        u.websockets.remove(websocket._get_current_object())
        if len(u.websockets) == 0:
            u.connected = False
            app.nursery.start_soon(broadcast, {'type': 'part', 'data': u.name})

在此之后，您仍然会看到Cancelled异常流到您的websocket函数中，因此您可能希望在那里捕获它们。请注意，需要捕获BaseException才能捕获错误，如下所示：

代码语言：javascript
复制
@app.websocket('/ws')
async def wsocket():
    u = User.current()
    if u is None:
        return
    u.websockets.add(websocket._get_current_object())
    u.connected = True
    if TRIO:
        try:
            async with trio.open_nursery() as nursery:
                nursery.start_soon(sending, u)
                nursery.start_soon(receiving, u)
        except BaseException as e:
            print(f'websocket funcs crashed with exception: {e}')

特别是这是因为trio不允许您悄悄地删除异常，您需要捕获它们或崩溃。我希望这足以让你开始解决你所看到的问题。

票数 0
EN
页面原文内容由Stack Overflow提供。腾讯云小微IT领域专用引擎提供翻译支持
原文链接：

https://stackoverflow.com/questions/68445354

复制
相关文章

相似问题

领券

腾讯云开发者

扫码关注腾讯云开发者

扫码关注腾讯云开发者

领取腾讯云代金券

热门产品

热门推荐

更多推荐

Copyright © 2013 - 2026 Tencent Cloud. All Rights Reserved. 腾讯云 版权所有 

深圳市腾讯计算机系统有限公司 ICP备案/许可证号：粤B2-20090059 粤公网安备44030502008569号

腾讯云计算（北京）有限责任公司 京ICP证150476号 |  京ICP备11018762号

问题归档专栏文章快讯文章归档关键词归档开发者手册归档开发者手册 Section 归档