Function test_concurrent

tests/ce/server/test_update_weight.py:171–192

高并发场景： 1. 并发 infer 5 个 2. pause -> 正在推理请求被中断 3. 查询状态 -> True 4. resume -> 查询状态 -> False

()

# 测试用例4：高并发pause/resume
# ---------------------------
def test_concurrent():
    """
    高并发场景：
    1. 并发 infer 5 个
    2. pause -> 正在推理请求被中断
    3. 查询状态 -> True
    4. resume -> 查询状态 -> False
    """
    threads = []
    for _ in range(5):
        t = threading.Thread(target=lambda: infer(expect_success=False))
        threads.append(t)
        t.start()

    time.sleep(0.1)
    pause()
    is_paused(expected=True)
    resume()
    is_paused(expected=False)

    for t in threads:
        t.join()


# ---------------------------