diff --git a/src/dfm-search/dfm-search-lib/contentsearch/contentstrategies/indexedstrategy.cpp b/src/dfm-search/dfm-search-lib/contentsearch/contentstrategies/indexedstrategy.cpp index 36bc1a81..b6ee45ff 100644 --- a/src/dfm-search/dfm-search-lib/contentsearch/contentstrategies/indexedstrategy.cpp +++ b/src/dfm-search/dfm-search-lib/contentsearch/contentstrategies/indexedstrategy.cpp @@ -15,6 +15,7 @@ #include #include #include +#include #include #include @@ -22,6 +23,7 @@ #include "3rdparty/fulltext/chineseanalyzer.h" #include "utils/cancellablecollector.h" #include "utils/contenthighlighter.h" +#include "utils/lucenecommitlockguard.h" #include "utils/lucenequeryutils.h" #include "utils/searchutility.h" #include "utils/lucene_cancellation_compat.h" @@ -419,16 +421,29 @@ void ContentIndexedStrategy::performContentSearch(const SearchQuery &query) SearchCancellationGuard guard(cancelledFlag); try { - // 获取索引目录 - FSDirectoryPtr directory = FSDirectory::open(m_indexDir.toStdWString()); + // 获取索引目录；显式传入 NativeFSLockFactory， + // 让本进程的 commit.lock 与写端（另一个进程的 IndexWriter） + // 共用同一把 fcntl 锁，从而同步 writer 的 commit 窗口。 + // 缓存 FSDirectory 避免每次搜索都重建对象。 + { + QMutexLocker locker(&m_readerMutex); + if (!m_cachedDirectory) { + m_cachedDirectory = FSDirectory::open( + m_indexDir.toStdWString(), + newLucene(m_indexDir.toStdWString())); + } + } + FSDirectoryPtr directory = m_cachedDirectory; if (!directory) { qWarning() << "Failed to open index directory:" << m_indexDir; emit errorOccurred(SearchError(ContentSearchErrorCode::ContentIndexNotFound)); return; } - // 获取索引读取器 - IndexReaderPtr reader = IndexReader::open(directory, true); + // 获取索引读取器（带 commit.lock 同步 + 版本缓存）。 + // IndexReader::open() 必须运行在 commit.lock 持有期间， + // 否则可能读到 writer commit() 中间状态而触发段解析错误。 + IndexReaderPtr reader = getOrCreateReader(directory); if (!reader || reader->numDocs() == 0) { qWarning() << "Index is empty or cannot be opened"; emit errorOccurred(SearchError(ContentSearchErrorCode::ContentIndexNotFound)); @@ -509,4 +524,84 @@ void ContentIndexedStrategy::cancel() m_cancelledRef->store(true); } +Lucene::IndexReaderPtr ContentIndexedStrategy::getOrCreateReader(const Lucene::FSDirectoryPtr &directory) +{ + // Fast path: cached reader 仍反映最新 commit。 + // 注意：fast path 不持 commit.lock，因为 IndexReader 在 open() 期间 + // 已经固定了 segments 引用的所有段文件 fd；后续 writer commit() 只会 + // 创建新 segments_N + 新段文件，老 reader 持有的 fd 不会被 writer 触碰， + // 所以 read-only 操作与并发 commit 安全共存。 + // isCurrent() 内部仅读 segments_N 的版本号，不读取段数据，代价极低。 + { + QMutexLocker locker(&m_readerMutex); + if (m_cachedReader) { + try { + if (m_cachedReader->isCurrent()) { + return m_cachedReader; + } + } catch (const Lucene::LuceneException &e) { + qWarning() << "IndexReader::isCurrent() failed:" << QString::fromStdWString(e.getError()); + } catch (const std::exception &e) { + qWarning() << "IndexReader::isCurrent() std exception:" << e.what(); + } + } + } + // Mutex released; concurrent searches can proceed. + + // Slow path: acquire commit.lock before any open/reopen so we don't observe + // an in-flight commit(). open() reads segments_N and segment files; if a + // commit is in progress, the resulting reader can reference torn segment + // data and later corrupt the heap inside weight->scorer(). + LuceneCommitLockGuard commitLock(directory, /*timeoutMs=*/1000, /*maxAttempts=*/3); + if (!commitLock.acquired()) { + qWarning() << "Cannot acquire commit.lock, abort search this round"; + return nullptr; + } + + // Double-check: another thread may have already reopened while we + // were waiting for commit.lock. + IndexReaderPtr cachedSnapshot; + { + QMutexLocker locker(&m_readerMutex); + cachedSnapshot = m_cachedReader; + if (cachedSnapshot) { + try { + if (cachedSnapshot->isCurrent()) { + return cachedSnapshot; + } + } catch (...) { + // isCurrent() failed; proceed with reopen below + } + } + } + + IndexReaderPtr result; + try { + if (cachedSnapshot) { + result = cachedSnapshot->reopen(true); + if (!result) { + result = IndexReader::open(directory, true); + } + } else { + result = IndexReader::open(directory, true); + } + } catch (const Lucene::LuceneException &e) { + qWarning() << "IndexReader open/reopen failed:" << QString::fromStdWString(e.getError()); + QMutexLocker locker(&m_readerMutex); + m_cachedReader.reset(); + return nullptr; + } catch (const std::exception &e) { + qWarning() << "IndexReader open/reopen std exception:" << e.what(); + QMutexLocker locker(&m_readerMutex); + m_cachedReader.reset(); + return nullptr; + } + + if (result) { + QMutexLocker locker(&m_readerMutex); + m_cachedReader = result; + } + return result; +} + DFM_SEARCH_END_NS diff --git a/src/dfm-search/dfm-search-lib/contentsearch/contentstrategies/indexedstrategy.h b/src/dfm-search/dfm-search-lib/contentsearch/contentstrategies/indexedstrategy.h index 724474ed..83af22c3 100644 --- a/src/dfm-search/dfm-search-lib/contentsearch/contentstrategies/indexedstrategy.h +++ b/src/dfm-search/dfm-search-lib/contentsearch/contentstrategies/indexedstrategy.h @@ -6,7 +6,11 @@ #include "basestrategy.h" +#include + #include +#include +#include #include #include #include @@ -60,9 +64,15 @@ class ContentIndexedStrategy : public ContentBaseStrategy void processSearchResults(const Lucene::IndexSearcherPtr &searcher, const Lucene::Collection &scoreDocs); + // 获取/创建 IndexReader；带 commit.lock 同步 + 版本缓存 + Lucene::IndexReaderPtr getOrCreateReader(const Lucene::FSDirectoryPtr &directory); + QString m_indexDir; Lucene::QueryPtr m_currentQuery; // 存储当前查询 QStringList m_keywords; + QMutex m_readerMutex; // 保护 m_cachedReader 的并发访问 + Lucene::IndexReaderPtr m_cachedReader; // 复用的 reader，由 m_readerMutex 守护 + Lucene::FSDirectoryPtr m_cachedDirectory; // 缓存的 FSDirectory，避免每次搜索重建 }; DFM_SEARCH_END_NS diff --git a/src/dfm-search/dfm-search-lib/ocrtextsearch/ocrtextstrategies/indexedstrategy.cpp b/src/dfm-search/dfm-search-lib/ocrtextsearch/ocrtextstrategies/indexedstrategy.cpp index adbf595a..ff854c51 100644 --- a/src/dfm-search/dfm-search-lib/ocrtextsearch/ocrtextstrategies/indexedstrategy.cpp +++ b/src/dfm-search/dfm-search-lib/ocrtextsearch/ocrtextstrategies/indexedstrategy.cpp @@ -13,6 +13,7 @@ #include #include #include +#include #include #include @@ -23,6 +24,7 @@ #include "utils/contenthighlighter.h" #include "utils/lucenequeryutils.h" #include "utils/searchutility.h" +#include "utils/lucenecommitlockguard.h" #include "utils/lucene_cancellation_compat.h" #include "utils/timerangeutils.h" @@ -418,16 +420,28 @@ void OcrTextIndexedStrategy::performOcrTextSearch(const SearchQuery &query) SearchCancellationGuard guard(cancelledFlag); try { - // Get index directory - FSDirectoryPtr directory = FSDirectory::open(m_indexDir.toStdWString()); + // 获取索引目录；显式传入 NativeFSLockFactory， + // 让本进程的 commit.lock 与写端（另一个进程的 IndexWriter） + // 共用同一把 fcntl 锁，从而同步 writer 的 commit 窗口。 + { + QMutexLocker locker(&m_readerMutex); + if (!m_cachedDirectory) { + m_cachedDirectory = FSDirectory::open( + m_indexDir.toStdWString(), + newLucene(m_indexDir.toStdWString())); + } + } + FSDirectoryPtr directory = m_cachedDirectory; if (!directory) { qWarning() << "Failed to open OCR text index directory:" << m_indexDir; emit errorOccurred(SearchError(OcrTextSearchErrorCode::OcrTextIndexNotFound)); return; } - // Get index reader - IndexReaderPtr reader = IndexReader::open(directory, true); + // 获取索引读取器（带 commit.lock 同步 + 版本缓存）。 + // IndexReader::open() 必须运行在 commit.lock 持有期间， + // 否则可能读到 writer commit() 中间状态而触发段解析错误。 + IndexReaderPtr reader = getOrCreateReader(directory); if (!reader || reader->numDocs() == 0) { qWarning() << "OCR text index is empty or cannot be opened"; emit errorOccurred(SearchError(OcrTextSearchErrorCode::OcrTextIndexNotFound)); @@ -506,4 +520,84 @@ void OcrTextIndexedStrategy::cancel() m_cancelledRef->store(true); } +Lucene::IndexReaderPtr OcrTextIndexedStrategy::getOrCreateReader(const Lucene::FSDirectoryPtr &directory) +{ + // Fast path: cached reader 仍反映最新 commit。 + // 注意：fast path 不持 commit.lock，因为 IndexReader 在 open() 期间 + // 已经固定了 segments 引用的所有段文件 fd；后续 writer commit() 只会 + // 创建新 segments_N + 新段文件，老 reader 持有的 fd 不会被 writer 触碰， + // 所以 read-only 操作与并发 commit 安全共存。 + // isCurrent() 内部仅读 segments_N 的版本号，不读取段数据，代价极低。 + { + QMutexLocker locker(&m_readerMutex); + if (m_cachedReader) { + try { + if (m_cachedReader->isCurrent()) { + return m_cachedReader; + } + } catch (const Lucene::LuceneException &e) { + qWarning() << "IndexReader::isCurrent() failed:" << QString::fromStdWString(e.getError()); + } catch (const std::exception &e) { + qWarning() << "IndexReader::isCurrent() std exception:" << e.what(); + } + } + } + // Mutex released; concurrent searches can proceed. + + // Slow path: acquire commit.lock before any open/reopen so we don't observe + // an in-flight commit(). open() reads segments_N and segment files; if a + // commit is in progress, the resulting reader can reference torn segment + // data and later corrupt the heap inside weight->scorer(). + LuceneCommitLockGuard commitLock(directory, /*timeoutMs=*/1000, /*maxAttempts=*/3); + if (!commitLock.acquired()) { + qWarning() << "Cannot acquire commit.lock, abort OCR text search this round"; + return nullptr; + } + + // Double-check: another thread may have already reopened while we + // were waiting for commit.lock. + IndexReaderPtr cachedSnapshot; + { + QMutexLocker locker(&m_readerMutex); + cachedSnapshot = m_cachedReader; + if (cachedSnapshot) { + try { + if (cachedSnapshot->isCurrent()) { + return cachedSnapshot; + } + } catch (...) { + // isCurrent() failed; proceed with reopen below + } + } + } + + IndexReaderPtr result; + try { + if (cachedSnapshot) { + result = cachedSnapshot->reopen(true); + if (!result) { + result = IndexReader::open(directory, true); + } + } else { + result = IndexReader::open(directory, true); + } + } catch (const Lucene::LuceneException &e) { + qWarning() << "IndexReader open/reopen failed:" << QString::fromStdWString(e.getError()); + QMutexLocker locker(&m_readerMutex); + m_cachedReader.reset(); + return nullptr; + } catch (const std::exception &e) { + qWarning() << "IndexReader open/reopen std exception:" << e.what(); + QMutexLocker locker(&m_readerMutex); + m_cachedReader.reset(); + return nullptr; + } + + if (result) { + QMutexLocker locker(&m_readerMutex); + m_cachedReader = result; + } + return result; +} + DFM_SEARCH_END_NS diff --git a/src/dfm-search/dfm-search-lib/ocrtextsearch/ocrtextstrategies/indexedstrategy.h b/src/dfm-search/dfm-search-lib/ocrtextsearch/ocrtextstrategies/indexedstrategy.h index b09aead1..84fb020e 100644 --- a/src/dfm-search/dfm-search-lib/ocrtextsearch/ocrtextstrategies/indexedstrategy.h +++ b/src/dfm-search/dfm-search-lib/ocrtextsearch/ocrtextstrategies/indexedstrategy.h @@ -6,7 +6,11 @@ #include "basestrategy.h" +#include + #include +#include +#include #include #include #include @@ -62,9 +66,15 @@ class OcrTextIndexedStrategy : public OcrTextBaseStrategy void processSearchResults(const Lucene::IndexSearcherPtr &searcher, const Lucene::Collection &scoreDocs); + // 获取/创建 IndexReader；带 commit.lock 同步 + 版本缓存 + Lucene::IndexReaderPtr getOrCreateReader(const Lucene::FSDirectoryPtr &directory); + QString m_indexDir; Lucene::QueryPtr m_currentQuery; QStringList m_keywords; + QMutex m_readerMutex; // 保护 m_cachedReader 的并发访问 + Lucene::IndexReaderPtr m_cachedReader; // 复用的 reader，由 m_readerMutex 守护 + Lucene::FSDirectoryPtr m_cachedDirectory; // 缓存的 FSDirectory，避免每次搜索重建 }; DFM_SEARCH_END_NS diff --git a/src/dfm-search/dfm-search-lib/utils/lucenecommitlockguard.cpp b/src/dfm-search/dfm-search-lib/utils/lucenecommitlockguard.cpp new file mode 100644 index 00000000..017038cf --- /dev/null +++ b/src/dfm-search/dfm-search-lib/utils/lucenecommitlockguard.cpp @@ -0,0 +1,57 @@ +// SPDX-FileCopyrightText: 2025 - 2026 UnionTech Software Technology Co., Ltd. +// +// SPDX-License-Identifier: GPL-3.0-or-later +#include "lucenecommitlockguard.h" + +#include + +using namespace Lucene; + +DFM_SEARCH_BEGIN_NS + +LuceneCommitLockGuard::LuceneCommitLockGuard(const FSDirectoryPtr &dir, int timeoutMs, int maxAttempts) + : m_lock(dir ? dir->makeLock(L"commit.lock") : nullptr) + , m_acquired(false) +{ + if (!m_lock) { + qWarning() << "LuceneCommitLockGuard: failed to create commit.lock handle"; + return; + } + + for (int attempt = 0; attempt < maxAttempts; ++attempt) { + try { + if (m_lock->obtain(timeoutMs)) { + m_acquired = true; + return; + } + } catch (const Lucene::LockObtainFailedException &e) { + qWarning() << "LuceneCommitLockGuard: commit.lock acquire timeout, attempt" + << (attempt + 1) << "of" << maxAttempts + << ":" << QString::fromStdWString(e.getError()); + } catch (const Lucene::LuceneException &e) { + qWarning() << "LuceneCommitLockGuard: commit.lock acquire failed, attempt" + << (attempt + 1) << "of" << maxAttempts + << ":" << QString::fromStdWString(e.getError()); + break; + } + } +} + +LuceneCommitLockGuard::~LuceneCommitLockGuard() +{ + if (m_acquired && m_lock) { + try { + m_lock->release(); + } catch (const Lucene::LuceneException &e) { + qWarning() << "LuceneCommitLockGuard: failed to release commit.lock:" + << QString::fromStdWString(e.getError()); + } catch (const std::exception &e) { + qWarning() << "LuceneCommitLockGuard: failed to release commit.lock:" << e.what(); + } catch (...) { + qWarning() << "LuceneCommitLockGuard: unknown exception while releasing commit.lock"; + } + m_acquired = false; + } +} + +DFM_SEARCH_END_NS diff --git a/src/dfm-search/dfm-search-lib/utils/lucenecommitlockguard.h b/src/dfm-search/dfm-search-lib/utils/lucenecommitlockguard.h new file mode 100644 index 00000000..03efc695 --- /dev/null +++ b/src/dfm-search/dfm-search-lib/utils/lucenecommitlockguard.h @@ -0,0 +1,56 @@ +// SPDX-FileCopyrightText: 2025 - 2026 UnionTech Software Technology Co., Ltd. +// +// SPDX-License-Identifier: GPL-3.0-or-later +#ifndef LUCENE_COMMIT_LOCK_GUARD_H +#define LUCENE_COMMIT_LOCK_GUARD_H + +#include +#include +#include + +#include + +DFM_SEARCH_BEGIN_NS + +/** + * @brief RAII 守护：在持有期间获取 Lucene 索引的 commit.lock， + * 防止 IndexWriter commit() 与 IndexReader::open() 并发造成的 + * 段文件不一致读取（表现为 boost::make_shared 内部 + * "unsorted double linked list corrupted" 崩溃）。 + * + * 写端 IndexWriter 在 commit() 期间默认会获取 commit.lock； + * 读端 IndexReader::open(readOnly=true) 不会自动获取任何锁， + * 故需读端手动 acquire 以同步 commit 窗口。 + * + * 构造时立即尝试获取锁；析构时若已获取则释放。 + * 失败时 acquired() 返回 false，调用方应放弃本轮读操作而不是阻塞或崩溃。 + */ +class LuceneCommitLockGuard +{ +public: + /** + * @param dir 已带 NativeFSLockFactory 的 FSDirectory + * @param timeoutMs 每次 attempt 内阻塞等待锁的最长时间 + * @param maxAttempts 总尝试次数；超时后会再次重试直到耗尽 + */ + explicit LuceneCommitLockGuard(const Lucene::FSDirectoryPtr &dir, + int timeoutMs = 1000, + int maxAttempts = 3); + + ~LuceneCommitLockGuard(); + + bool acquired() const { return m_acquired; } + + LuceneCommitLockGuard(const LuceneCommitLockGuard &) = delete; + LuceneCommitLockGuard &operator=(const LuceneCommitLockGuard &) = delete; + LuceneCommitLockGuard(LuceneCommitLockGuard &&) = delete; + LuceneCommitLockGuard &operator=(LuceneCommitLockGuard &&) = delete; + +private: + Lucene::LockPtr m_lock; + bool m_acquired { false }; +}; + +DFM_SEARCH_END_NS + +#endif // LUCENE_COMMIT_LOCK_GUARD_H