yumoqing/llmage
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: 后台任务引用保持+异常处理

Browse Source 
- 保存asyncio.create_task引用防止GC回收
- query_task_status添加try/except和CancelledError处理
- 确保连接在任务异常时正确释放
This commit is contained in:
yumoqing 2026-06-27 12:54:10 +08:00
parent 0fccc895ce
commit f6d4462ed8
1 changed files with 57 additions and 43 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

100
llmage/asyncinference.py
View File

@ -15,6 +15,9 @@ from ahserver.filestorage import FileStorage
from .accounting import llm_accounting, llm_charging from .accounting import llm_accounting, llm_charging
from .utils import * from .utils import *
# Global set to keep references to background tasks
_background_tasks = set()
async def get_today_asynctask_list(userid): async def get_today_asynctask_list(userid):
env = ServerEnv() env = ServerEnv()
async with get_sor_context(env, 'llmage') as sor: async with get_sor_context(env, 'llmage') as sor:
@ -39,7 +42,9 @@ async def get_asynctask_status(request, taskid):
t = timestampAdd(r.use_time, 600) t = timestampAdd(r.use_time, 600)
now = time.time() now = time.time()
if r.status not in ['UNKNOWN', 'FAILED', 'SUCCEEDED'] and now > t: if r.status not in ['UNKNOWN', 'FAILED', 'SUCCEEDED'] and now > t:
asyncio.create_task(query_task_status(request, r.id)) task = asyncio.create_task(query_task_status(request, r.id))
_background_tasks.add(task)
task.add_done_callback(_background_tasks.discard)
return output return output
return { return {
'taskid': taskid, 'taskid': taskid,
@ -112,7 +117,9 @@ async def async_uapi_request(request, llm,
if d.status == 'FAILED': if d.status == 'FAILED':
e = Exception(f'resp={d} FFAILED') e = Exception(f'resp={d} FFAILED')
return return
asyncio.create_task(query_task_status(request, luid)) task = asyncio.create_task(query_task_status(request, luid))
_background_tasks.add(task)
task.add_done_callback(_background_tasks.discard)
except Exception as e: except Exception as e:
ed = {"error": f"ERROR:{e}", "status": "FAILED"} ed = {"error": f"ERROR:{e}", "status": "FAILED"}
@ -165,48 +172,55 @@ async def query_task_status(request, luid, onetime=False):
upappid = llm.upappid upappid = llm.upappid
apinames = llm.query_apiname.split(',') apinames = llm.query_apiname.split(',')
for apiname in apinames: try:
while True: for apiname in apinames:
lastoutout = await get_lastoutput(llmusage.ioinfo) while True:
if lastoutout['status'] in ['UNKNOWN', 'FAILED', 'SUCCEEDED']: lastoutout = await get_lastoutput(llmusage.ioinfo)
critical(f"{lastoutout['status']=}") if lastoutout['status'] in ['UNKNOWN', 'FAILED', 'SUCCEEDED']:
return critical(f"{lastoutout['status']=}")
ns = {'taskid': taskid} return
new_output = b = d = None ns = {'taskid': taskid}
try: new_output = b = d = None
b = await uapi.call(upappid, apiname, userid, params=ns) try:
if isinstance(b, bytes): b = await uapi.call(upappid, apiname, userid, params=ns)
b = b.decode('utf-8') if isinstance(b, bytes):
new_output = json.loads(b) b = b.decode('utf-8')
except Exception as e: new_output = json.loads(b)
exception(f'{e}, {b=}') except Exception as e:
new_output = { exception(f'{e}, {b=}')
'status': 'FAILED', new_output = {
'error': f'{b},{e}' 'status': 'FAILED',
} 'error': f'{b},{e}'
if not new_output.get('status'): }
e = Exception(f"{new_output=} {upappid=}, {apiname=} has not status field") if not new_output.get('status'):
critical(f'{e}') e = Exception(f"{new_output=} {upappid=}, {apiname=} has not status field")
raise e critical(f'{e}')
if lastoutout['status'] != new_output.get('status'): raise e
llmusage.status = new_output['status'] if lastoutout['status'] != new_output.get('status'):
ns = { llmusage.status = new_output['status']
'id': llmusage.id, ns = {
'status': llmusage.status 'id': llmusage.id,
} 'status': llmusage.status
if 'usage' in new_output.keys(): }
ns['usages'] = json.dumps(new_output['usage']) if 'usage' in new_output.keys():
await append_new_llmoutput(llmusage.ioinfo, new_output) ns['usages'] = json.dumps(new_output['usage'])
await modify_llmusage(ns) await append_new_llmoutput(llmusage.ioinfo, new_output)
if llmusage.status in ['UNKNOWN', 'FAILED', 'SUCCEEDED']: await modify_llmusage(ns)
critical(f'finished .. {llmusage.status=}') if llmusage.status in ['UNKNOWN', 'FAILED', 'SUCCEEDED']:
return critical(f'finished .. {llmusage.status=}')
return
if onetime: if onetime:
critical(f'onetime is true, returned') critical(f'onetime is true, returned')
return return
await asyncio.sleep(llm.query_period or 30) await asyncio.sleep(llm.query_period or 30)
critical(f'{llm.query_period=} seconds will retry, {new_output["status"]=}') critical(f'{llm.query_period=} seconds will retry, {new_output["status"]=}')
except asyncio.CancelledError:
critical(f'query_task_status cancelled for {luid=}')
raise
except Exception as e:
exception(f'query_task_status error for {luid=}: {e}')
raise
async def async_uapi_request_product(llm, api_userid, user_id, user_org_id, params_kw, luid): async def async_uapi_request_product(llm, api_userid, user_id, user_org_id, params_kw, luid):