Python中Minizinc求解器自定义超时停止（基于上次解）问题

我需要实现一个优化问题的超时机制：当自上次找到解决方案后经过特定时长时，停止Minizinc求解器。尝试用Python的asyncio库实现，但实际运行中求解器的fzn进程会在后台持续运行，必须手动从任务管理器杀死才能让函数返回值。

我的思路是每次找到解就启动计时器，超时则停止求解器并返回最后一个解，但未达到预期效果。原代码如下：

from minizinc import Solver, Model, Instance
import asyncio
import pandas as pd
import pymzn
import os
import sys
from pymzn.aio import minizinc
import time

async def timed_solver():
    optimizer = Model("TestAsync/optim.mzn")
    solver = Solver.lookup("com.google.ortools.sat")
    instance = Instance(solver, optimizer)
    res = []
    timeout = 180
    solver_task = None
    
    async def monitor_timeout(last_solution):
        i = 0
        while True:
            i += 1
            print(i)
            current = time.perf_counter()
            if current - last_solution >= timeout:
                print("Timeout reached, stopping the process.")
                solver_task.cancel()
                return None
            try:
                await asyncio.sleep(1)
            except asyncio.exceptions.CancelledError:
                break
        return None
    
    async def getSolutions():
        timeout_task = None
        try:
            async for result in instance.solutions(optimisation_level=1, processes=8, free_search=True):
            
                if result.solution is None:
                    continue
                else:
                
                    res.append(result)
                    last_solution = time.perf_counter()
                    print("###################")
                    print(res[-1]["objective"])
                    if timeout_task:
                        timeout_task.cancel()
                    timeout_task = asyncio.create_task(monitor_timeout(last_solution))
        except asyncio.exceptions.CancelledError:
            print("Timeout reached")
        return res[-1] if res else None
    
    solver_task = asyncio.create_task(getSolutions())

    try:
        await solver_task    
    
    except asyncio.exceptions.CancelledError:
        return res[-1] if res else None
    
    return res[-1] if res else None

async def main():
    res = await timed_solver()
    if res:
        print(res)
    else:
        print("No solution found or timeout occurred.")

asyncio.run(main())

问题原因 #

1. 仅调用solver_task.cancel()只会取消asyncio层面的任务，无法直接终止底层的Minizinc求解器进程。
2. instance.solutions()的异步迭代器未被正确关闭，导致子进程持续运行。
3. 超时逻辑没有绑定到求解器进程的终止操作。

修复方案 #

需要在超时触发时，同时取消asyncio任务并显式终止Minizinc实例的底层进程。同时优化超时监控逻辑，减少不必要的循环检查。

修复后的代码：

from minizinc import Solver, Model, Instance
import asyncio
import time

async def timed_solver():
    optimizer = Model("TestAsync/optim.mzn")
    solver = Solver.lookup("com.google.ortools.sat")
    instance = Instance(solver, optimizer)
    res = []
    timeout = 180
    timeout_task = None

    async def monitor_timeout():
        await asyncio.sleep(timeout)
        print("Timeout reached, stopping the process.")
        # 直接终止求解器进程
        instance.stop()
        # 取消求解任务
        if solver_task:
            solver_task.cancel()

    async def get_solutions():
        nonlocal timeout_task
        try:
            async for result in instance.solutions(optimisation_level=1, processes=8, free_search=True):
                if result.solution is None:
                    continue
                res.append(result)
                print("###################")
                print(res[-1]["objective"])
                # 重置超时任务
                if timeout_task:
                    timeout_task.cancel()
                timeout_task = asyncio.create_task(monitor_timeout())
        except asyncio.CancelledError:
            print("Solver task cancelled due to timeout")
        except Exception as e:
            print(f"Error during solving: {e}")
        finally:
            # 确保进程被终止
            instance.stop()
        return res[-1] if res else None

    solver_task = asyncio.create_task(get_solutions())
    
    try:
        return await solver_task
    except asyncio.CancelledError:
        return res[-1] if res else None

async def main():
    res = await timed_solver()
    if res:
        print(res)
    else:
        print("No solution found or timeout occurred.")

asyncio.run(main())

关键修改点 #

• 使用instance.stop()直接终止Minizinc底层进程，这是确保进程退出的核心操作。
• 简化超时监控逻辑，用asyncio.sleep(timeout)替代循环计时，减少资源消耗。
• 在get_solutions的finally块中添加instance.stop()，确保无论是否异常，进程都会被清理。
• 移除了未使用的依赖（pandas、pymzn等），精简代码。

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