diff --git a/fs/ocfs2/dlm/dlmmaster.c b/fs/ocfs2/dlm/dlmmaster.c index 612969867ff9..ea48c013f7fb 100644 --- a/fs/ocfs2/dlm/dlmmaster.c +++ b/fs/ocfs2/dlm/dlmmaster.c @@ -2525,6 +2525,52 @@ static int dlm_is_lockres_migratable(struct dlm_ctxt *dlm, */ +/* + * Drop the sender's references on a migration MLE after the attempt has + * failed. The finishing assert from the target may already have unlinked + * the MLE and consumed its base reference; decide under the locks the + * assert handler holds, and only drop the base reference while the MLE is + * still hashed. Returns true when the assert got there first: the target + * has become the master, so the caller must complete the migration. + */ +static bool dlm_put_migration_mle(struct dlm_ctxt *dlm, + struct dlm_master_list_entry *mle) +{ + bool asserted; + + spin_lock(&dlm->spinlock); + spin_lock(&dlm->master_lock); + __dlm_mle_detach_hb_events(dlm, mle); + asserted = hlist_unhashed(&mle->master_hash_node); + if (!asserted) { + __dlm_unlink_mle(dlm, mle); + __dlm_put_mle(mle); + } + spin_unlock(&dlm->master_lock); + spin_unlock(&dlm->spinlock); + + dlm_put_mle_inuse(mle); + return asserted; +} + +/* + * The finishing assert got there first: it switched the owner to the + * target and cleared DLM_LOCK_RES_MIGRATING. Drop the other nodes' locks + * as a completed migration does, unless recovery has taken the lockres + * over since, and let the usage be recalculated. + */ +static void dlm_migrate_lockres_asserted(struct dlm_ctxt *dlm, + struct dlm_lock_resource *res, + u8 target) +{ + spin_lock(&res->spinlock); + if (res->owner == target) + dlm_remove_nonlocal_locks(dlm, res); + spin_unlock(&res->spinlock); + + dlm_lockres_calc_usage(dlm, res); +} + static int dlm_migrate_lockres(struct dlm_ctxt *dlm, struct dlm_lock_resource *res, u8 target) { @@ -2609,9 +2655,10 @@ static int dlm_migrate_lockres(struct dlm_ctxt *dlm, if (ret < 0) { if (mle_added) { - dlm_mle_detach_hb_events(dlm, mle); - dlm_put_mle(mle); - dlm_put_mle_inuse(mle); + if (dlm_put_migration_mle(dlm, mle)) { + dlm_migrate_lockres_asserted(dlm, res, target); + ret = 0; + } } else if (mle) { kmem_cache_free(dlm_mle_cache, mle); mle = NULL; @@ -2640,9 +2687,11 @@ static int dlm_migrate_lockres(struct dlm_ctxt *dlm, mlog(0, "migration to node %u failed with %d\n", target, ret); /* migration failed, detach and clean up mle */ - dlm_mle_detach_hb_events(dlm, mle); - dlm_put_mle(mle); - dlm_put_mle_inuse(mle); + if (dlm_put_migration_mle(dlm, mle)) { + dlm_migrate_lockres_asserted(dlm, res, target); + ret = 0; + goto leave; + } spin_lock(&res->spinlock); res->state &= ~DLM_LOCK_RES_MIGRATING; wake = 1; @@ -2686,9 +2735,12 @@ static int dlm_migrate_lockres(struct dlm_ctxt *dlm, res->lockname.name, target); ret = -EINVAL; /* migration failed, detach and clean up mle */ - dlm_mle_detach_hb_events(dlm, mle); - dlm_put_mle(mle); - dlm_put_mle_inuse(mle); + if (dlm_put_migration_mle(dlm, mle)) { + dlm_migrate_lockres_asserted(dlm, res, + target); + ret = 0; + goto leave; + } spin_lock(&res->spinlock); res->state &= ~DLM_LOCK_RES_MIGRATING; wake = 1;