Improve merge lock handling Don't check the merge lock with Is_Locked in FilePurger_Purge_Aborted_Merge. This prevented background mergers from cleaning up because they already held the merge lock. Require that callers hold the merge lock instead.
Don't check the merge lock with Is_Locked in Indexer_init. Since we hold the write lock, and the merge.json file from aborted merge has been removed, the presence of a merge.json file is a reliable indicator that it belongs to an active background merger. Move the merge lock request form S_maybe_merge to Indexer_init. Also, S_maybe_merge used to Obtain the merge lock instead of merely Requesting it. Waiting for a potentially long-running background merger slows things down unnecessarily and increases the chance for write lock contention. This didn't affect the default configuration which has a zero merge lock timeout. Project: http://git-wip-us.apache.org/repos/asf/lucy/repo Commit: http://git-wip-us.apache.org/repos/asf/lucy/commit/4e6e5bbf Tree: http://git-wip-us.apache.org/repos/asf/lucy/tree/4e6e5bbf Diff: http://git-wip-us.apache.org/repos/asf/lucy/diff/4e6e5bbf Branch: refs/heads/master Commit: 4e6e5bbf2814e8e2a824b5ae1b91d068233cd248 Parents: 5270b98 Author: Nick Wellnhofer <wellnho...@aevum.de> Authored: Fri Feb 17 18:25:35 2017 +0100 Committer: Nick Wellnhofer <wellnho...@aevum.de> Committed: Mon Feb 20 16:26:22 2017 +0100 ---------------------------------------------------------------------- core/Lucy/Index/FilePurger.c | 53 ++++++++++++++++--------------------- core/Lucy/Index/FilePurger.cfh | 6 +++-- core/Lucy/Index/Indexer.c | 39 ++++++++++++--------------- 3 files changed, 44 insertions(+), 54 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/lucy/blob/4e6e5bbf/core/Lucy/Index/FilePurger.c ---------------------------------------------------------------------- diff --git a/core/Lucy/Index/FilePurger.c b/core/Lucy/Index/FilePurger.c index 433dec3..5795bb0 100644 --- a/core/Lucy/Index/FilePurger.c +++ b/core/Lucy/Index/FilePurger.c @@ -139,44 +139,37 @@ FilePurger_Purge_Snapshots_IMP(FilePurger *self, Snapshot *current) { void FilePurger_Purge_Aborted_Merge_IMP(FilePurger *self) { FilePurgerIVARS *const ivars = FilePurger_IVARS(self); - IndexManager *manager = ivars->manager; - Lock *merge_lock = IxManager_Make_Merge_Lock(manager); - - if (!Lock_Is_Locked_Exclusive(merge_lock)) { - Hash *merge_data = IxManager_Read_Merge_Data(manager); - Obj *cutoff = merge_data - ? Hash_Fetch_Utf8(merge_data, "cutoff", 6) - : NULL; - - if (cutoff) { - Folder *folder = ivars->folder; - - String *cutoff_seg = Seg_num_to_name(Json_obj_to_i64(cutoff)); - if (Folder_Local_Exists(folder, cutoff_seg)) { - if (!S_delete_entry(folder, cutoff_seg)) { - if (Folder_Local_Exists(folder, cutoff_seg)) { - WARN("Couldn't delete '%o' from aborted merge", - cutoff_seg); - } - } - } - - String *merge_json = SSTR_WRAP_C("merge.json"); - if (!Folder_Local_Delete(folder, merge_json)) { - if (Folder_Local_Exists(folder, merge_json)) { + IndexManager *manager = ivars->manager; + Hash *merge_data = IxManager_Read_Merge_Data(manager); + Obj *cutoff = merge_data + ? Hash_Fetch_Utf8(merge_data, "cutoff", 6) + : NULL; + + if (cutoff) { + Folder *folder = ivars->folder; + + String *cutoff_seg = Seg_num_to_name(Json_obj_to_i64(cutoff)); + if (Folder_Local_Exists(folder, cutoff_seg)) { + if (!S_delete_entry(folder, cutoff_seg)) { + if (Folder_Local_Exists(folder, cutoff_seg)) { WARN("Couldn't delete '%o' from aborted merge", - merge_json); + cutoff_seg); } } + } - DECREF(cutoff_seg); + String *merge_json = SSTR_WRAP_C("merge.json"); + if (!Folder_Local_Delete(folder, merge_json)) { + if (Folder_Local_Exists(folder, merge_json)) { + WARN("Couldn't delete '%o' from aborted merge", + merge_json); + } } - DECREF(merge_data); + DECREF(cutoff_seg); } - DECREF(merge_lock); - return; + DECREF(merge_data); } static void http://git-wip-us.apache.org/repos/asf/lucy/blob/4e6e5bbf/core/Lucy/Index/FilePurger.cfh ---------------------------------------------------------------------- diff --git a/core/Lucy/Index/FilePurger.cfh b/core/Lucy/Index/FilePurger.cfh index 4f3923d..c781170 100644 --- a/core/Lucy/Index/FilePurger.cfh +++ b/core/Lucy/Index/FilePurger.cfh @@ -30,12 +30,14 @@ class Lucy::Index::FilePurger inherits Clownfish::Obj { inert FilePurger* init(FilePurger *self, Folder *folder, IndexManager *manager = NULL); - /** Purge obsolete files from the index. + /** Purge obsolete files from the index. The caller must hold the + * write lock. */ void Purge_Snapshots(FilePurger *self, Snapshot *current); - /** Purge files left behind by an aborted background merge. + /** Purge files left behind by an aborted background merge. The caller + * must hold the merge lock and the write lock. */ void Purge_Aborted_Merge(FilePurger *self); http://git-wip-us.apache.org/repos/asf/lucy/blob/4e6e5bbf/core/Lucy/Index/Indexer.c ---------------------------------------------------------------------- diff --git a/core/Lucy/Index/Indexer.c b/core/Lucy/Index/Indexer.c index be5519a..9190b93 100644 --- a/core/Lucy/Index/Indexer.c +++ b/core/Lucy/Index/Indexer.c @@ -132,6 +132,7 @@ Indexer_init(Indexer *self, Schema *schema, Obj *index, schema_file = NULL; } else { + S_release_write_lock(self); THROW(ERR, "Failed to parse %o", schema_file); } } @@ -163,25 +164,25 @@ Indexer_init(Indexer *self, Schema *schema, Obj *index, // Create new FilePurger and zap detritus from an aborted background // merge. ivars->file_purger = FilePurger_new(folder, ivars->manager); - FilePurger_Purge_Aborted_Merge(ivars->file_purger); + Lock *merge_lock = IxManager_Make_Merge_Lock(ivars->manager); + if (Lock_Request_Exclusive(merge_lock)) { + // No background merger is running. + ivars->merge_lock = merge_lock; + FilePurger_Purge_Aborted_Merge(ivars->file_purger); + } + else { + DECREF(merge_lock); + } // Create a new segment. int64_t new_seg_num = IxManager_Highest_Seg_Num(ivars->manager, latest_snapshot) + 1; - Lock *merge_lock = IxManager_Make_Merge_Lock(ivars->manager); - if (Lock_Is_Locked_Exclusive(merge_lock)) { - // If there's a background merge process going on, stay out of its - // way. - Hash *merge_data = IxManager_Read_Merge_Data(ivars->manager); - Obj *cutoff_obj = merge_data - ? Hash_Fetch_Utf8(merge_data, "cutoff", 6) - : NULL; - if (!cutoff_obj) { - DECREF(merge_lock); - DECREF(merge_data); - THROW(ERR, "Background merge detected, but can't read merge data"); - } - else { + // If there's a background merge process going on, stay out of its + // way. + Hash *merge_data = IxManager_Read_Merge_Data(ivars->manager); + if (merge_data) { + Obj *cutoff_obj = Hash_Fetch_Utf8(merge_data, "cutoff", 6); + if (cutoff_obj) { int64_t cutoff = Json_obj_to_i64(cutoff_obj); if (cutoff >= new_seg_num) { new_seg_num = cutoff + 1; @@ -198,8 +199,6 @@ Indexer_init(Indexer *self, Schema *schema, Obj *index, } DECREF(fields); - DECREF(merge_lock); - // Create new SegWriter. ivars->seg_writer = SegWriter_new(ivars->schema, ivars->snapshot, ivars->segment, ivars->polyreader); @@ -396,12 +395,9 @@ S_maybe_merge(Indexer *self, Vector *seg_readers) { IndexerIVARS *const ivars = Indexer_IVARS(self); bool merge_happened = false; size_t num_seg_readers = Vec_Get_Size(seg_readers); - Lock *merge_lock = IxManager_Make_Merge_Lock(ivars->manager); - bool got_merge_lock = Lock_Obtain_Exclusive(merge_lock); int64_t cutoff; - if (got_merge_lock) { - ivars->merge_lock = merge_lock; + if (ivars->merge_lock) { cutoff = 0; } else { @@ -420,7 +416,6 @@ S_maybe_merge(Indexer *self, Vector *seg_readers) { else { cutoff = INT64_MAX; } - DECREF(merge_lock); } // Get a list of segments to recycle. Validate and confirm that there are
