使用Cvxpy结合Mosek求解优化问题时如何捕获超时异常

方案1：捕获Mosek超时异常+判断求解状态（推荐优先使用） #

Mosek触发你设置的optimizer_max_time阈值后，会主动终止求解过程，此时要么抛出cvxpy.SolverError异常，要么返回的问题状态为cvxpy.TIME_LIMIT，你只需要通过try-except块捕获异常，同时判断求解后的状态即可识别超时场景，也能保证后续代码正常执行。

修改后的可运行代码如下：

import cvxpy as cp
import numpy as np
import mosek

m = 15
n = 10
np.random.seed(1)
s0 = np.random.randn(m)
lamb0 = np.maximum(-s0, 0)
s0 = np.maximum(s0, 0)
x0 = np.random.randn(n)
A = np.random.randn(m, n)
b = A @ x0 + s0
c = -A.T @ lamb0

# Define and solve the CVXPY problem.
x = cp.Variable(n)
prob = cp.Problem(cp.Minimize(c.T@x),
                 [A @ x <= b])
timeout_flag = False
try:
    # 注意：optimizer_max_time单位为秒，2小时对应设置为7200
    prob.solve(cp.MOSEK, mosek_params={mosek.dparam.optimizer_max_time: 0.01})
except cp.SolverError as e:
    if "time limit" in str(e).lower():
        timeout_flag = True
        print('Timeout occured')
    else:
        # 其他求解错误逻辑可自行补充
        raise e

# 额外判断求解状态是否为超时
if prob.status == cp.TIME_LIMIT:
    timeout_flag = True
    print('Timeout occured from status check')

print(prob.value)

def execute_other_important_stuff():
    print("Hello world")

execute_other_important_stuff()  # 现在可以正常执行

注意：optimizer_max_time参数的单位为秒，2小时的超时阈值需要设置为7200，请不要填错单位。

方案2：进程级超时兜底（应对极端卡死场景） #

如果遇到Mosek本身异常卡死、没有触发自带超时逻辑的极端情况，可以用多进程封装求解逻辑，在进程层设置超时时间，到点直接终止求解进程，完全避免主程序被卡住：

import cvxpy as cp
import numpy as np
import mosek
from multiprocessing import Process, Queue

def solve_process(queue):
    m = 15
    n = 10
    np.random.seed(1)
    s0 = np.random.randn(m)
    lamb0 = np.maximum(-s0, 0)
    s0 = np.maximum(s0, 0)
    x0 = np.random.randn(n)
    A = np.random.randn(m, n)
    b = A @ x0 + s0
    c = -A.T @ lamb0

    x = cp.Variable(n)
    prob = cp.Problem(cp.Minimize(c.T@x), [A @ x <= b])
    try:
        prob.solve(cp.MOSEK, mosek_params={mosek.dparam.optimizer_max_time: 0.01})
        queue.put((prob.status, prob.value))
    except Exception as e:
        queue.put(("error", str(e)))

if __name__ == "__main__":
    q = Queue()
    p = Process(target=solve_process, args=(q,))
    p.start()
    # 进程层超时设置，单位为秒，2小时设为7200，可略大于Mosek自带超时做兜底
    p.join(timeout=1)
    
    if p.is_alive():
        p.terminate()
        p.join()
        print("Timeout occured from process level")
        timeout_flag = True
    else:
        res = q.get()
        print(f"Solve finished, status: {res[0]}, value: {res[1]}")
    
    def execute_other_important_stuff():
        print("Hello world")
    execute_other_important_stuff()

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