LU-1061 agl: cl_locks_prune() waits for the last user

[fs/lustre-release.git] / lustre / osc / osc_lock.c

diff --git a/lustre/osc/osc_lock.c b/lustre/osc/osc_lock.c
index 9aa7f59..a1dc1bf 100644 (file)
--- a/lustre/osc/osc_lock.c
+++ b/lustre/osc/osc_lock.c
@@ -535,13 +535,15 @@ static int osc_lock_upcall(void *cookie, int errcode)
                                 dlmlock->l_ast_data = NULL;
                                 olck->ols_handle.cookie = 0ULL;
                                 cfs_spin_unlock(&osc_ast_guard);
-                                ldlm_lock_fail_match_locked(dlmlock, rc);
+                                ldlm_lock_fail_match_locked(dlmlock);
                                 unlock_res_and_lock(dlmlock);
                                 LDLM_LOCK_PUT(dlmlock);
                         }
                 } else {
-                        if (olck->ols_glimpse)
+                        if (olck->ols_glimpse) {
                                 olck->ols_glimpse = 0;
+                                olck->ols_agl = 0 ;
+                        }
                         osc_lock_upcall0(env, olck);
                 }
 
@@ -1390,11 +1392,27 @@ static void osc_lock_cancel(const struct lu_env *env,
         osc_lock_detach(env, olck);
 }
 
-void cl_lock_page_list_fixup(const struct lu_env *env,
-                             struct cl_io *io, struct cl_lock *lock,
-                             struct cl_page_list *queue);
-
 #ifdef INVARIANT_CHECK
+static int check_cb(const struct lu_env *env, struct cl_io *io,
+                    struct cl_page *page, void *cbdata)
+{
+        struct cl_lock *lock = cbdata;
+
+        if (lock->cll_descr.cld_mode == CLM_READ) {
+                struct cl_lock *tmp;
+                tmp = cl_lock_at_page(env, lock->cll_descr.cld_obj,
+                                     page, lock, 1, 0);
+                if (tmp != NULL) {
+                        cl_lock_put(env, tmp);
+                        return CLP_GANG_OKAY;
+                }
+        }
+
+        CL_LOCK_DEBUG(D_ERROR, env, lock, "still has pages\n");
+        CL_PAGE_DEBUG(D_ERROR, env, page, "\n");
+        return CLP_GANG_ABORT;
+}
+
 /**
  * Returns true iff there are pages under \a olck not protected by other
  * locks.
@@ -1405,44 +1423,39 @@ static int osc_lock_has_pages(struct osc_lock *olck)
         struct cl_lock_descr *descr;
         struct cl_object     *obj;
         struct osc_object    *oob;
-        struct cl_page_list  *plist;
-        struct cl_page       *page;
         struct cl_env_nest    nest;
         struct cl_io         *io;
         struct lu_env        *env;
         int                   result;
 
         env = cl_env_nested_get(&nest);
-        if (!IS_ERR(env)) {
-                obj   = olck->ols_cl.cls_obj;
-                oob   = cl2osc(obj);
-                io    = &oob->oo_debug_io;
-                lock  = olck->ols_cl.cls_lock;
-                descr = &lock->cll_descr;
-                plist = &osc_env_info(env)->oti_plist;
-                cl_page_list_init(plist);
-
-                cfs_mutex_lock(&oob->oo_debug_mutex);
-
-                io->ci_obj = cl_object_top(obj);
-                cl_io_init(env, io, CIT_MISC, io->ci_obj);
-                cl_page_gang_lookup(env, obj, io,
-                                    descr->cld_start, descr->cld_end, plist);
-                cl_lock_page_list_fixup(env, io, lock, plist);
-                if (plist->pl_nr > 0) {
-                        CL_LOCK_DEBUG(D_ERROR, env, lock, "still has pages\n");
-                        cl_page_list_for_each(page, plist)
-                                CL_PAGE_DEBUG(D_ERROR, env, page, "\n");
-                }
-                result = plist->pl_nr > 0;
-                cl_page_list_disown(env, io, plist);
-                cl_page_list_fini(env, plist);
-                cl_io_fini(env, io);
-                cfs_mutex_unlock(&oob->oo_debug_mutex);
-                cl_env_nested_put(&nest, env);
-        } else
-                result = 0;
-        return result;
+        if (IS_ERR(env))
+                return 0;
+
+        obj   = olck->ols_cl.cls_obj;
+        oob   = cl2osc(obj);
+        io    = &oob->oo_debug_io;
+        lock  = olck->ols_cl.cls_lock;
+        descr = &lock->cll_descr;
+
+        cfs_mutex_lock(&oob->oo_debug_mutex);
+
+        io->ci_obj = cl_object_top(obj);
+        cl_io_init(env, io, CIT_MISC, io->ci_obj);
+        do {
+                result = cl_page_gang_lookup(env, obj, io,
+                                             descr->cld_start, descr->cld_end,
+                                             check_cb, (void *)lock);
+                if (result == CLP_GANG_ABORT)
+                        break;
+                if (result == CLP_GANG_RESCHED)
+                        cfs_cond_resched();
+        } while (result != CLP_GANG_OKAY);
+        cl_io_fini(env, io);
+        cfs_mutex_unlock(&oob->oo_debug_mutex);
+        cl_env_nested_put(&nest, env);
+
+        return (result == CLP_GANG_ABORT);
 }
 #else
 static int osc_lock_has_pages(struct osc_lock *olck)