From f707cca074809954771cdf2e41b78a15e566cf8c Mon Sep 17 00:00:00 2001 From: Wang Qi Date: Tue, 4 Aug 2026 19:04:16 +0800 Subject: [PATCH] Fix: Let delete dataset/document to a dedicated thread to avoid blocking othe APIs (#17800) --- api/apps/restful_apis/document_api.py | 4 +-- api/apps/services/dataset_api_service.py | 5 ++- common/misc_utils.py | 31 +++++++++++++++++++ .../test_dataset_api_service_list_datasets.py | 1 + 4 files changed, 36 insertions(+), 5 deletions(-) diff --git a/api/apps/restful_apis/document_api.py b/api/apps/restful_apis/document_api.py index 195fcd72d1..e2931c911e 100644 --- a/api/apps/restful_apis/document_api.py +++ b/api/apps/restful_apis/document_api.py @@ -71,7 +71,7 @@ from api.utils.validation_utils import ( from common import settings from common.constants import ParserType, RetCode, TaskStatus, SANDBOX_ARTIFACT_BUCKET from common.metadata_utils import convert_conditions, meta_filter, turn2jsonschema -from common.misc_utils import get_uuid, thread_pool_exec +from common.misc_utils import get_uuid, thread_pool_exec, thread_pool_exec_long_time from api.utils.file_utils import filename_type, thumbnail from api.utils.file_response import apply_preview_file_response_headers from api.utils.web_utils import CONTENT_TYPE_MAP, html2pdf, is_valid_url, apply_safe_file_response_headers @@ -1203,7 +1203,7 @@ async def delete_documents(tenant_id, dataset_id): doc_ids = unique_doc_ids # Delete documents using existing FileService.delete_docs - errors = await thread_pool_exec(FileService.delete_docs, doc_ids, tenant_id) + errors = await thread_pool_exec_long_time(FileService.delete_docs, doc_ids, tenant_id) if errors: return get_error_data_result(message=str(errors)) diff --git a/api/apps/services/dataset_api_service.py b/api/apps/services/dataset_api_service.py index 112ce20db4..1f0a1626ad 100644 --- a/api/apps/services/dataset_api_service.py +++ b/api/apps/services/dataset_api_service.py @@ -13,7 +13,6 @@ # See the License for the specific language governing permissions and # limitations under the License. # -import asyncio import logging import json import os @@ -33,7 +32,7 @@ from api.db.services.tenant_model_service import TenantModelService from api.db.services.user_service import TenantService, UserService, UserTenantService from common.constants import FileSource, StatusEnum from api.utils.api_utils import deep_merge, get_parser_config, remap_dictionary_keys, verify_embedding_availability -from common.misc_utils import thread_pool_exec +from common.misc_utils import thread_pool_exec, thread_pool_exec_long_time from rag.advanced_rag.knowlege_compile.wiki import WIKI_PAGE_COMPILE_KWD # KB-wide structure-graph merge index types. Each (re)builds the ``dataset_graph`` @@ -222,7 +221,7 @@ def _delete_datasets_sync(tenant_id: str, ids: list = None, delete_all: bool = F async def delete_datasets(tenant_id: str, ids: list = None, delete_all: bool = False): - return await asyncio.to_thread(_delete_datasets_sync, tenant_id, ids, delete_all) + return await thread_pool_exec_long_time(_delete_datasets_sync, tenant_id, ids, delete_all) def get_dataset(dataset_id: str, tenant_id: str): diff --git a/common/misc_utils.py b/common/misc_utils.py index 673efb210e..d9f6b4dbec 100644 --- a/common/misc_utils.py +++ b/common/misc_utils.py @@ -30,6 +30,7 @@ from urllib.parse import urljoin from concurrent.futures import ThreadPoolExecutor logger = logging.getLogger(__name__) +_LONG_TIME_THREAD_POOL_EXECUTOR = ThreadPoolExecutor(max_workers=int(os.getenv("LONG_TIME_THREAD_POOL_WORKERS", "1")), thread_name_prefix="long-time") def get_uuid(): @@ -257,3 +258,33 @@ async def thread_pool_exec(func, *args, **kwargs): inner = functools.partial(func, *args, **kwargs) return await loop.run_in_executor(executor, ctx.run, inner) return await loop.run_in_executor(executor, ctx.run, func, *args) + + +async def thread_pool_exec_long_time(func, *args, **kwargs): + """Run long blocking work in a shared bounded executor. + + Use this for synchronous work that can outlive the HTTP request, such as + large document or dataset cleanup. Do not use ``thread_pool_exec`` for + those paths: it creates a temporary executor with a ``with`` block, and + leaving that block calls ``shutdown(wait=True)``. If the client disconnects + or the HTTP request times out while the worker is still running, request + cancellation can unwind the coroutine into that shutdown path and wait for + the long worker to finish anyway. + + This helper uses a process-level executor instead, so there is no per-call + executor shutdown during request cancellation. The running sync callable is + still not force-cancelled by Python; it continues in the long-task pool. The + important behavior is that the Quart event loop/request task can be released + and continue serving other API calls. The pool is bounded by + ``LONG_TIME_THREAD_POOL_WORKERS`` (default 1), so multiple expensive jobs + queue instead of spawning unbounded cleanup threads or competing with the + event loop's default executor. + + ContextVars are copied into the worker thread, matching ``thread_pool_exec``. + """ + loop = asyncio.get_running_loop() + ctx = contextvars.copy_context() + if kwargs: + inner = functools.partial(func, *args, **kwargs) + return await loop.run_in_executor(_LONG_TIME_THREAD_POOL_EXECUTOR, ctx.run, inner) + return await loop.run_in_executor(_LONG_TIME_THREAD_POOL_EXECUTOR, ctx.run, func, *args) diff --git a/test/unit_test/api/apps/services/test_dataset_api_service_list_datasets.py b/test/unit_test/api/apps/services/test_dataset_api_service_list_datasets.py index f7bedd44d5..bc81ff1de2 100644 --- a/test/unit_test/api/apps/services/test_dataset_api_service_list_datasets.py +++ b/test/unit_test/api/apps/services/test_dataset_api_service_list_datasets.py @@ -169,6 +169,7 @@ def _load_list_datasets_module(monkeypatch, *, kbs, parsing_status_by_kb): monkeypatch, "common.misc_utils", thread_pool_exec=MagicMock(), + thread_pool_exec_long_time=MagicMock(), ) _stub( monkeypatch,