AI/dify
1
0
Fork 0
mirror of https://git.mirrors.martin98.com/https://github.com/langgenius/dify.git synced 2025-08-15 20:15:52 +08:00
Code Issues Packages Projects Releases Wiki Activity

fix: same chunk insert deadlock (#12502)

Browse Source 
Co-authored-by: huangzhuo <huangzhuo1@xiaomi.com>
This commit is contained in:
huangzhuo1949 2025-01-09 15:16:41 +08:00 committed by GitHub
parent 20f090537f
commit e84bf35e2a
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
1 changed files with 14 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

18
api/core/indexing_runner.py
View File

@ -530,7 +530,6 @@ class IndexingRunner:
# chunk nodes by chunk size # chunk nodes by chunk size
indexing_start_at = time.perf_counter() indexing_start_at = time.perf_counter()
tokens = 0 tokens = 0
chunk_size = 10
if dataset_document.doc_form != IndexType.PARENT_CHILD_INDEX: if dataset_document.doc_form != IndexType.PARENT_CHILD_INDEX:
# create keyword index # create keyword index
create_keyword_thread = threading.Thread( create_keyword_thread = threading.Thread(
@ -539,11 +538,22 @@ class IndexingRunner:
) )
create_keyword_thread.start() create_keyword_thread.start()
max_workers = 10
if dataset.indexing_technique == "high_quality": if dataset.indexing_technique == "high_quality":
with concurrent.futures.ThreadPoolExecutor(max_workers=10) as executor: with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor:
futures = [] futures = []
for i in range(0, len(documents), chunk_size):
chunk_documents = documents[i : i + chunk_size] # Distribute documents into multiple groups based on the hash values of page_content
# This is done to prevent multiple threads from processing the same document,
# Thereby avoiding potential database insertion deadlocks
document_groups: list[list[Document]] = [[] for _ in range(max_workers)]
for document in documents:
hash = helper.generate_text_hash(document.page_content)
group_index = int(hash, 16) % max_workers
document_groups[group_index].append(document)
for chunk_documents in document_groups:
if len(chunk_documents) == 0:
continue
futures.append( futures.append(
executor.submit( executor.submit(
self._process_chunk, self._process_chunk,