在Flask中用async为ble.Peripheral.connect设置自定义超时

问题描述 #

在Python3.6中使用btle.Peripheral()时，需要为connect函数设置1秒自定义超时，但该库本身不支持此功能。尝试用asyncio实现超时控制，但实际运行时仍会等待库自带的20秒超时。即使将Flask的app.run设置为threaded=True，问题依旧存在。

以下是复现问题的相关代码：

最初的尝试代码 #

def myfunction():
    # ...其他逻辑

    async def coroutine(future):
        try:
            await asyncio.wait_for(peripheral.connect(ble_mac_address), timeout=1)
            future.set_result("OK")
        except asyncio.TimeoutError:
            future.set_result("NOK")

    loop = asyncio.new_event_loop()
    future = loop.create_future()
    asyncio.set_event_loop(loop)
    ret = asyncio.run(coroutine(future))
    loop.close()

服务端复现代码 #

import asyncio
import json
import time

from flask import Flask, render_template
from flask_socketio import SocketIO, emit

from flask_cors import CORS

from gevent.pywsgi import WSGIServer
from geventwebsocket.handler import WebSocketHandler

app = Flask(__name__)
app.config['SECRET_KEY'] = 'secret!'

@app.route('/new_recipe', methods=["POST"])
def new_recipe():
    print('MdtServer             :')
    print('MdtServer             : new_recipe')

    async def do_io(loop):
        print('io start')
        # await asyncio.gather(loop.run_in_executor(None, self.peripheral.connect(ble_mac_address)))
        await asyncio.gather(loop.run_in_executor(None, time.sleep(10)))
        print('io end')

    async def do_other_things(future: asyncio.Future, loop):
        print('doing other things')
        try:
            res = await asyncio.wait_for(do_io(loop), timeout=1)
            future.set_result('OK')

        except Exception as err:
            print('err {}'.format(err))
            future.set_result('NOK')

    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)
    future = loop.create_future()
    ret = loop.run_until_complete(do_other_things(future, loop))
    print('ret {}'.format(ret))
    print('ret future {}'.format(future))
    loop.close()
    # Generate JSON return
    answer = dict()

    return json.dumps(answer)

if __name__ == '__main__':
    ## With Only flask
    app.run(threaded=True)
    ### WIth WSGIServer
    # http_server = WSGIServer(('localhost', 5000), app, handler_class=WebSocketHandler)
    # http_server.serve_forever()

客户端复现代码 #

import json
import requests

step_0 = dict()

request = f'Post on /new_recipe with data {json.dumps(step_0)}'
print(f" request > {request}")
get_request = requests.post('http://localhost:5000' + '/new_recipe',
                            data=json.dumps(step_0), headers={'content-type': 'application/json'})
response = get_request.json()
print(f" response > {response}")

解决方案 #

问题根源 #

asyncio.wait_for只能取消异步任务的future，但无法终止底层线程中运行的同步阻塞函数（比如btle.Peripheral.connect或示例中的time.sleep）。即使超时触发，同步函数所在的线程仍会继续执行直到完成，导致看起来还是等待了默认的20秒。

实现步骤 #

1. 自定义可终止线程：创建支持强制终止的线程类，用于执行同步阻塞的BLE连接操作。
2. 异步包装与超时控制：在异步函数中启动该线程，用循环+短间隔sleep的方式检查线程状态，超时后主动终止线程并清理BLE资源。
3. Flask中适配：在Flask路由中正确整合异步逻辑，避免线程阻塞影响请求响应。

修改后的服务端代码（适配BLE场景） #

import asyncio
import json
import threading
import ctypes
from threading import Thread

from flask import Flask
from btle import Peripheral, BTLEException

app = Flask(__name__)
app.config['SECRET_KEY'] = 'secret!'

# 自定义可终止线程类
class TerminateableThread(Thread):
    def __init__(self, target, args=()):
        super().__init__(target=target, args=args)
        self._terminated = False
        self.result = None
        self.error = None

    def run(self):
        try:
            self.result = self._target(*self._args)
        except Exception as e:
            self.error = e
        finally:
            self._terminated = True

    def terminate(self):
        if not self._terminated and self.is_alive():
            # 强制终止线程
            tid = self.ident
            res = ctypes.pythonapi.PyThreadState_SetAsyncExc(tid, ctypes.py_object(SystemExit))
            if res > 1:
                ctypes.pythonapi.PyThreadState_SetAsyncExc(tid, 0)

async def ble_connect_task(ble_mac_address):
    peripheral = Peripheral()
    connection_success = False

    def connect_sync():
        nonlocal connection_success
        try:
            peripheral.connect(ble_mac_address)
            connection_success = True
        except BTLEException:
            pass

    thread = TerminateableThread(target=connect_sync)
    thread.start()

    # 等待线程完成或超时（1秒）
    try:
        for _ in range(10):  # 10*0.1秒=1秒超时窗口
            if not thread.is_alive():
                break
            await asyncio.sleep(0.1)
        else:
            # 超时，终止线程并清理资源
            thread.terminate()
            thread.join(timeout=0.5)
            if peripheral:
                try:
                    peripheral.disconnect()
                except:
                    pass
            raise asyncio.TimeoutError("BLE连接超时")
    finally:
        # 确保线程被终止
        if thread.is_alive():
            thread.terminate()
            thread.join(timeout=0.5)

    if thread.error:
        raise thread.error
    return connection_success

@app.route('/new_recipe', methods=["POST"])
def new_recipe():
    print('MdtServer             : new_recipe')
    ble_mac_address = "你的BLE设备MAC地址"  # 替换为实际设备MAC

    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)
    result = "NOK"

    try:
        loop.run_until_complete(asyncio.wait_for(ble_connect_task(ble_mac_address), timeout=1.5))
        result = "OK"
    except asyncio.TimeoutError:
        print("连接超时，返回NOK")
    except Exception as e:
        print(f"连接出错: {e}")
    finally:
        loop.close()

    answer = {"result": result}
    return json.dumps(answer)

if __name__ == '__main__':
    app.run(threaded=True)

关键说明 #

• 线程终止：使用ctypes强制终止线程是非常规操作，需配合资源清理步骤，避免BLE连接资源泄漏。
• 超时检查：通过循环+短间隔sleep的方式，确保异步事件循环不会被长时间阻塞，超时触发更及时。
• 资源清理：无论连接成功、超时还是出错，都要主动断开BLE设备连接，避免影响后续操作。

内容的提问来源于stack exchange，提问作者macchina001