Enqueue glimpse locks with CEF_MUST flag so that they are not accidentaly

converted into a lockless mode.
b=18104
i=oleg.drokin@sun.com
i=jinshan.xiong@sun.com

diff --git a/lustre/include/cl_object.h b/lustre/include/cl_object.h
index ad4b468..5b922bd 100644 (file)
--- a/lustre/include/cl_object.h
+++ b/lustre/include/cl_object.h
@@ -2100,16 +2100,22 @@ enum cl_enq_flags {
          */
         CEF_DISCARD_DATA = 0x00000004,
         /**
          */
         CEF_DISCARD_DATA = 0x00000004,
         /**

-         * tell the sub layers that it must be a `real' lock.
+         * tell the sub layers that it must be a `real' lock. This is used for
+         * mmapped-buffer locks and glimpse locks that must be never converted
+         * into lockless mode.
+         *
+         * \see vvp_mmap_locks(), cl_glimpse_lock().

          */
         CEF_MUST         = 0x00000008,
         /**
          */
         CEF_MUST         = 0x00000008,
         /**

-         * tell the sub layers that never request a `real' lock.
-         * currently, the CEF_MUST & CEF_NEVER are only used for mmap locks.
-         * cl_io::ci_lockreq and these two flags: ci_lockreq just describes
-         * generic information of lock requirement for this IO, especially for
-         * locks which belong to the object doing IO; however, lock itself may
-         * have precise requirements, this is described by the latter.
+         * tell the sub layers that never request a `real' lock. This flag is
+         * not used currently.
+         *
+         * cl_io::ci_lockreq and CEF_{MUST,NEVER} flags specify lockless
+         * conversion policy: ci_lockreq describes generic information of lock
+         * requirement for this IO, especially for locks which belong to the
+         * object doing IO; however, lock itself may have precise requirements
+         * that are described by the enqueue flags.

          */
         CEF_NEVER        = 0x00000010,
         /**
          */
         CEF_NEVER        = 0x00000010,
         /**

diff --git a/lustre/lclient/glimpse.c b/lustre/lclient/glimpse.c
index 78acee6..ed81f15 100644 (file)
--- a/lustre/lclient/glimpse.c
+++ b/lustre/lclient/glimpse.c
@@ -118,11 +118,17 @@ int cl_glimpse_lock(const struct lu_env *env, struct cl_io *io,
                         *descr = whole_file;
                         descr->cld_obj   = clob;
                         descr->cld_mode  = CLM_PHANTOM;
                         *descr = whole_file;
                         descr->cld_obj   = clob;
                         descr->cld_mode  = CLM_PHANTOM;

-                        /* The lockreq for glimpse should be mandatory,
-                         * otherwise, osc may decide to use lockless */
-                        io->ci_lockreq = CILR_MANDATORY;

                         cio->cui_glimpse = 1;
                         cio->cui_glimpse = 1;

-                        lock = cl_lock_request(env, io, descr, CEF_ASYNC,
+                        /*
+                         * CEF_ASYNC is used because glimpse sub-locks cannot
+                         * deadlock (because they never conflict with other
+                         * locks) and, hence, can be enqueued out-of-order.
+                         *
+                         * CEF_MUST protects glimpse lock from conversion into
+                         * a lockless mode.
+                         */
+                        lock = cl_lock_request(env, io, descr,
+                                               CEF_ASYNC|CEF_MUST,

                                                "glimpse", cfs_current());
                         cio->cui_glimpse = 0;
                         if (!IS_ERR(lock)) {
                                                "glimpse", cfs_current());
                         cio->cui_glimpse = 0;
                         if (!IS_ERR(lock)) {

diff --git a/lustre/osc/osc_lock.c b/lustre/osc/osc_lock.c
index 0e3521e..247c044 100644 (file)
--- a/lustre/osc/osc_lock.c
+++ b/lustre/osc/osc_lock.c
@@ -414,16 +414,15 @@ static void osc_lock_granted(const struct lu_env *env, struct osc_lock *olck,
                 olck->ols_state = OLS_GRANTED;
                 osc_lock_lvb_update(env, olck, rc);
 
                 olck->ols_state = OLS_GRANTED;
                 osc_lock_lvb_update(env, olck, rc);
 

-                /* release DLM spin-locks to allow cl_lock_modify() to take a
-                 * semaphore on a parent lock. This is safe, because
+                /* release DLM spin-locks to allow cl_lock_{modify,signal}()
+                 * to take a semaphore on a parent lock. This is safe, because

                  * spin-locks are needed to protect consistency of
                  * dlmlock->l_*_mode and LVB, and we have finished processing
                  * them. */
                 unlock_res_and_lock(dlmlock);
                 cl_lock_modify(env, lock, descr);
                  * spin-locks are needed to protect consistency of
                  * dlmlock->l_*_mode and LVB, and we have finished processing
                  * them. */
                 unlock_res_and_lock(dlmlock);
                 cl_lock_modify(env, lock, descr);

-                lock_res_and_lock(dlmlock);
-

                 cl_lock_signal(env, lock);
                 cl_lock_signal(env, lock);

+                lock_res_and_lock(dlmlock);

         }
         EXIT;
 }
         }
         EXIT;
 }


@@ -1061,6 +1060,7 @@ static void osc_lock_to_lockless(const struct lu_env *env,
                         slice->cls_ops = &osc_lock_lockless_ops;
                 }
         }
                         slice->cls_ops = &osc_lock_lockless_ops;
                 }
         }

+        LASSERT(ergo(ols->ols_glimpse, !osc_lock_is_lockless(ols)));

 }
 
 /**
 }
 
 /**


@@ -1296,7 +1296,7 @@ static int osc_lock_enqueue(const struct lu_env *env,
                         ols->ols_state = OLS_GRANTED;
                 }
         }
                         ols->ols_state = OLS_GRANTED;
                 }
         }

-
+        LASSERT(ergo(ols->ols_glimpse, !osc_lock_is_lockless(ols)));

         RETURN(result);
 }
 
         RETURN(result);
 }