LU-8346 obdclass: guarantee all keys filled

[fs/lustre-release.git] / lustre / obdclass / lu_object.c

diff --git a/lustre/obdclass/lu_object.c b/lustre/obdclass/lu_object.c
index b633d26..cc89029 100644 (file)
--- a/lustre/obdclass/lu_object.c
+++ b/lustre/obdclass/lu_object.c
@@ -831,35 +831,20 @@ struct lu_object *lu_object_find_slice(const struct lu_env *env,
 }
 EXPORT_SYMBOL(lu_object_find_slice);
 
-/**
- * Global list of all device types.
- */
-static struct list_head lu_device_types;
-
 int lu_device_type_init(struct lu_device_type *ldt)
 {
        int result = 0;
 
        atomic_set(&ldt->ldt_device_nr, 0);
-       INIT_LIST_HEAD(&ldt->ldt_linkage);
        if (ldt->ldt_ops->ldto_init)
                result = ldt->ldt_ops->ldto_init(ldt);
 
-       if (result == 0) {
-               spin_lock(&obd_types_lock);
-               list_add(&ldt->ldt_linkage, &lu_device_types);
-               spin_unlock(&obd_types_lock);
-       }
-
        return result;
 }
 EXPORT_SYMBOL(lu_device_type_init);
 
 void lu_device_type_fini(struct lu_device_type *ldt)
 {
-       spin_lock(&obd_types_lock);
-       list_del_init(&ldt->ldt_linkage);
-       spin_unlock(&obd_types_lock);
        if (ldt->ldt_ops->ldto_fini)
                ldt->ldt_ops->ldto_fini(ldt);
 }
@@ -868,8 +853,8 @@ EXPORT_SYMBOL(lu_device_type_fini);
 /**
  * Global list of all sites on this node
  */
-static struct list_head lu_sites;
-static struct rw_semaphore lu_sites_guard;
+static LIST_HEAD(lu_sites);
+static DECLARE_RWSEM(lu_sites_guard);
 
 /**
  * Global environment used by site shrinker.
@@ -1594,7 +1579,7 @@ EXPORT_SYMBOL(lu_context_key_get);
 /**
  * List of remembered contexts. XXX document me.
  */
-static struct list_head lu_context_remembered;
+static LIST_HEAD(lu_context_remembered);
 
 /**
  * Destroy \a key in all remembered contexts. This is used to destroy key
@@ -1630,15 +1615,18 @@ void lu_context_key_quiesce(struct lu_context_key *key)
                list_for_each_entry(ctx, &lu_context_remembered,
                                    lc_remember)
                        key_fini(ctx, key->lct_index);
-               write_unlock(&lu_keys_guard);
+
                ++key_set_version;
+               write_unlock(&lu_keys_guard);
        }
 }
 
 void lu_context_key_revive(struct lu_context_key *key)
 {
-        key->lct_tags &= ~LCT_QUIESCENT;
-        ++key_set_version;
+       write_lock(&lu_keys_guard);
+       key->lct_tags &= ~LCT_QUIESCENT;
+       ++key_set_version;
+       write_unlock(&lu_keys_guard);
 }
 
 static void keys_fini(struct lu_context *ctx)
@@ -1658,6 +1646,7 @@ 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
@@ -1670,24 +1659,26 @@ 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);
 
-        LINVRNT(ctx->lc_value != NULL);
-        for (i = 0; i < ARRAY_SIZE(lu_keys); ++i) {
-                struct lu_context_key *key;
+refill:
+       LINVRNT(ctx->lc_value != NULL);
+       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 &&
-                    (key->lct_tags & ctx->lc_tags) &&
-                    /*
-                     * Don't create values for a LCT_QUIESCENT key, as this
-                     * will pin module owning a key.
-                     */
-                    !(key->lct_tags & LCT_QUIESCENT)) {
-                        void *value;
-
-                        LINVRNT(key->lct_init != NULL);
-                        LINVRNT(key->lct_index == i);
+               key = lu_keys[i];
+               if (ctx->lc_value[i] == NULL && key != NULL &&
+                   (key->lct_tags & ctx->lc_tags) &&
+                   /*
+                    * Don't create values for a LCT_QUIESCENT key, as this
+                    * will pin module owning a key.
+                    */
+                   !(key->lct_tags & LCT_QUIESCENT)) {
+                       void *value;
+
+                       LINVRNT(key->lct_init != NULL);
+                       LINVRNT(key->lct_index == i);
 
                        LASSERT(key->lct_owner != NULL);
                        if (!(ctx->lc_tags & LCT_NOREF) &&
@@ -1704,19 +1695,27 @@ static int keys_fill(struct lu_context *ctx)
 
                        lu_ref_add_atomic(&key->lct_reference, "ctx", ctx);
                        atomic_inc(&key->lct_used);
-                        /*
-                         * This is the only place in the code, where an
-                         * element of ctx->lc_value[] array is set to non-NULL
-                         * value.
-                         */
-                        ctx->lc_value[i] = value;
-                        if (key->lct_exit != NULL)
-                                ctx->lc_tags |= LCT_HAS_EXIT;
-                }
-                ctx->lc_version = key_set_version;
-        }
+                       /*
+                        * This is the only place in the code, where an
+                        * element of ctx->lc_value[] array is set to non-NULL
+                        * value.
+                        */
+                       ctx->lc_value[i] = value;
+                       if (key->lct_exit != NULL)
+                               ctx->lc_tags |= LCT_HAS_EXIT;
+               }
+       }
+
+       read_lock(&lu_keys_guard);
+       if (pre_version != key_set_version) {
+               pre_version = key_set_version;
+               read_unlock(&lu_keys_guard);
+               goto refill;
+       }
+
        atomic_dec(&lu_key_initing_cnt);
-        return 0;
+       read_unlock(&lu_keys_guard);
+       return 0;
 }
 
 static int keys_init(struct lu_context *ctx)
@@ -2148,11 +2147,6 @@ int lu_global_init(void)
 
         CDEBUG(D_INFO, "Lustre LU module (%p).\n", &lu_keys);
 
-       INIT_LIST_HEAD(&lu_device_types);
-       INIT_LIST_HEAD(&lu_context_remembered);
-       INIT_LIST_HEAD(&lu_sites);
-       init_rwsem(&lu_sites_guard);
-
         result = lu_ref_global_init();
         if (result != 0)
                 return result;
@@ -2412,3 +2406,4 @@ int lu_buf_check_and_grow(struct lu_buf *buf, size_t len)
        buf->lb_len = len;
        return 0;
 }
+EXPORT_SYMBOL(lu_buf_check_and_grow);