LU-11089 obdclass: make key_set_version an atomic_t

As a first step to simplifying the locking in lu_object.c,
change key_set_version to an atomic_t.  This will mean
we don't need to hold a spinlock when incrementing.

It is clear that keys_fill() (and so lu_context_refill())
cannot race with itself as it holds no locks for the main
part of the work.  So updates to lc_version, and testing of
that value cannot need locking either.
So remove the locking from keys_fill() and lu_context_refill()
as there is no longer anything to protect.  The locking around
lu_keys_initing_cnt is preserved for now.

Also, don't increment when deregistering or quiescing a key.
key_set_version is *only* use to avoid filling new key values
if there have been no changes to the set of key.
Deregistering a key does not mean that we need to try filling
any new value, so the increment is pointless.

Finally, remove the refill loop in keys_fill().  If a key
is registered or revived while keys_fill() is running it must be safe
to ignore it just as we would if it was registered immediately
after keys_fill() ran.  The important thing is that the
keys_set_version stored in ctx->lc_version must be sampled
*before* those unseen keys were added.
So sample keys_set_version early.

Linux-commit : 0fbfbc5ad0f892cf4c5e087a4e7e67102b2289af

Change-Id: Ic6907561d0bb864b10e1c53fb3e5469d0c60f888
Signed-off-by: NeilBrown <neilb@suse.com>
Signed-off-by: James Simmons <uja.ornl@yahoo.com>
Reviewed-on: https://review.whamcloud.com/32711
Reviewed-by: Gu Zheng <gzheng@ddn.com>
Tested-by: Jenkins
Tested-by: Maloo <maloo@whamcloud.com>
Reviewed-by: Alexey Lyashkov <c17817@cray.com>
Reviewed-by: Oleg Drokin <green@whamcloud.com>

diff --git a/lustre/obdclass/lu_object.c b/lustre/obdclass/lu_object.c
index f17a8ee..16bb8bb 100644 (file)
--- a/lustre/obdclass/lu_object.c
+++ b/lustre/obdclass/lu_object.c
@@ -1348,7 +1348,7 @@ static atomic_t lu_key_initing_cnt = ATOMIC_INIT(0);
  * lu_context_refill(). No locking is provided, as initialization and shutdown
  * are supposed to be externally serialized.
  */
-static unsigned key_set_version = 0;
+static atomic_t key_set_version = ATOMIC_INIT(0);
 
 /**
  * Register new key.
@@ -1372,7 +1372,7 @@ int lu_context_key_register(struct lu_context_key *key)
                         lu_keys[i] = key;
                         lu_ref_init(&key->lct_reference);
                         result = 0;
-                        ++key_set_version;
+                       atomic_inc(&key_set_version);
                         break;
                 }
         }
@@ -1415,7 +1415,6 @@ void lu_context_key_degister(struct lu_context_key *key)
        lu_context_key_quiesce(key);
 
        write_lock(&lu_keys_guard);
-       ++key_set_version;
        key_fini(&lu_shrink_env.le_ctx, key->lct_index);
 
        /**
@@ -1576,7 +1575,6 @@ void lu_context_key_quiesce(struct lu_context_key *key)
                                    lc_remember)
                        key_fini(ctx, key->lct_index);
 
-               ++key_set_version;
                write_unlock(&lu_keys_guard);
        }
 }
@@ -1585,7 +1583,7 @@ void lu_context_key_revive(struct lu_context_key *key)
 {
        write_lock(&lu_keys_guard);
        key->lct_tags &= ~LCT_QUIESCENT;
-       ++key_set_version;
+       atomic_inc(&key_set_version);
        write_unlock(&lu_keys_guard);
 }
 
@@ -1606,7 +1604,6 @@ static void keys_fini(struct lu_context *ctx)
 static int keys_fill(struct lu_context *ctx)
 {
        unsigned int i;
-       unsigned pre_version;
 
        /*
         * A serialisation with lu_context_key_quiesce() is needed, but some
@@ -1619,16 +1616,15 @@ static int keys_fill(struct lu_context *ctx)
         */
        read_lock(&lu_keys_guard);
        atomic_inc(&lu_key_initing_cnt);
-       pre_version = key_set_version;
        read_unlock(&lu_keys_guard);
+       ctx->lc_version = atomic_read(&key_set_version);
 
-refill:
-       LINVRNT(ctx->lc_value != NULL);
+       LINVRNT(ctx->lc_value);
        for (i = 0; i < ARRAY_SIZE(lu_keys); ++i) {
                struct lu_context_key *key;
 
                key = lu_keys[i];
-               if (ctx->lc_value[i] == NULL && key != NULL &&
+               if (!ctx->lc_value[i] && key &&
                    (key->lct_tags & ctx->lc_tags) &&
                    /*
                     * Don't create values for a LCT_QUIESCENT key, as this
@@ -1666,17 +1662,7 @@ refill:
                }
        }
 
-       read_lock(&lu_keys_guard);
-       if (pre_version != key_set_version) {
-               pre_version = key_set_version;
-               read_unlock(&lu_keys_guard);
-               goto refill;
-       }
-
-       ctx->lc_version = key_set_version;
-
        atomic_dec(&lu_key_initing_cnt);
-       read_unlock(&lu_keys_guard);
        return 0;
 }
 
@@ -1785,13 +1771,9 @@ EXPORT_SYMBOL(lu_context_exit);
  */
 int lu_context_refill(struct lu_context *ctx)
 {
-       read_lock(&lu_keys_guard);
-       if (likely(ctx->lc_version == key_set_version)) {
-               read_unlock(&lu_keys_guard);
+       if (likely(ctx->lc_version == atomic_read(&key_set_version)))
                return 0;
-       }
 
-       read_unlock(&lu_keys_guard);
        return keys_fill(ctx);
 }
 
@@ -1802,14 +1784,15 @@ int lu_context_refill(struct lu_context *ctx)
  * predefined when the lu_device type are registered, during the module probe
  * phase.
  */
-__u32 lu_context_tags_default = 0;
-__u32 lu_session_tags_default = 0;
+u32 lu_context_tags_default;
+u32 lu_session_tags_default;
 
+#ifdef HAVE_SERVER_SUPPORT
 void lu_context_tags_update(__u32 tags)
 {
        write_lock(&lu_keys_guard);
        lu_context_tags_default |= tags;
-       key_set_version++;
+       atomic_inc(&key_set_version);
        write_unlock(&lu_keys_guard);
 }
 EXPORT_SYMBOL(lu_context_tags_update);
@@ -1818,7 +1801,7 @@ void lu_context_tags_clear(__u32 tags)
 {
        write_lock(&lu_keys_guard);
        lu_context_tags_default &= ~tags;
-       key_set_version++;
+       atomic_inc(&key_set_version);
        write_unlock(&lu_keys_guard);
 }
 EXPORT_SYMBOL(lu_context_tags_clear);
@@ -1827,7 +1810,7 @@ void lu_session_tags_update(__u32 tags)
 {
        write_lock(&lu_keys_guard);
        lu_session_tags_default |= tags;
-       key_set_version++;
+       atomic_inc(&key_set_version);
        write_unlock(&lu_keys_guard);
 }
 EXPORT_SYMBOL(lu_session_tags_update);
@@ -1836,10 +1819,11 @@ void lu_session_tags_clear(__u32 tags)
 {
        write_lock(&lu_keys_guard);
        lu_session_tags_default &= ~tags;
-       key_set_version++;
+       atomic_inc(&key_set_version);
        write_unlock(&lu_keys_guard);
 }
 EXPORT_SYMBOL(lu_session_tags_clear);
+#endif /* HAVE_SERVER_SUPPORT */
 
 int lu_env_init(struct lu_env *env, __u32 tags)
 {