Fix hash_lock / keystore.sk_dk_lock lock inversion

The keystore.sk_dk_lock should not be held while performing I/O.
Drop the lock when reading from disk and update the code so
they the first successful caller adds the key.

Improve error handling in spa_keystore_create_mapping_impl().

Reviewed by: Thomas Caputi <tcaputi@datto.com>
Reviewed-by: RageLtMan <rageltman@sempervictus>
Signed-off-by: Brian Behlendorf <behlendorf1@llnl.gov>
Closes #7112
Closes #7115

diff --git a/module/zfs/dsl_crypt.c b/module/zfs/dsl_crypt.c
index 6a63d54cadb30a0cc138fe78380e6f0c4ac37b4f..98a6f8cc89a25239671215cb76ea896fd5ea71e3 100644 (file)
--- a/module/zfs/dsl_crypt.c
+++ b/module/zfs/dsl_crypt.c
@@ -655,55 +655,56 @@ spa_keystore_dsl_key_hold_dd(spa_t *spa, dsl_dir_t *dd, void *tag,
 {
        int ret;
        avl_index_t where;
-       dsl_crypto_key_t *dck = NULL;
+       dsl_crypto_key_t *dck_io = NULL, *dck_ks = NULL;
        dsl_wrapping_key_t *wkey = NULL;
        uint64_t dckobj = dd->dd_crypto_obj;
 
-       rw_enter(&spa->spa_keystore.sk_dk_lock, RW_WRITER);
-
-       /* lookup the key in the tree of currently loaded keys */
-       ret = spa_keystore_dsl_key_hold_impl(spa, dckobj, tag, &dck);
-       if (!ret) {
-               rw_exit(&spa->spa_keystore.sk_dk_lock);
-               *dck_out = dck;
+       /* Lookup the key in the tree of currently loaded keys */
+       rw_enter(&spa->spa_keystore.sk_dk_lock, RW_READER);
+       ret = spa_keystore_dsl_key_hold_impl(spa, dckobj, tag, &dck_ks);
+       rw_exit(&spa->spa_keystore.sk_dk_lock);
+       if (ret == 0) {
+               *dck_out = dck_ks;
                return (0);
        }
 
-       /* lookup the wrapping key from the keystore */
+       /* Lookup the wrapping key from the keystore */
        ret = spa_keystore_wkey_hold_dd(spa, dd, FTAG, &wkey);
        if (ret != 0) {
-               ret = SET_ERROR(EACCES);
-               goto error_unlock;
+               *dck_out = NULL;
+               return (SET_ERROR(EACCES));
        }
 
-       /* read the key from disk */
+       /* Read the key from disk */
        ret = dsl_crypto_key_open(spa->spa_meta_objset, wkey, dckobj,
-           tag, &dck);
-       if (ret != 0)
-               goto error_unlock;
+           tag, &dck_io);
+       if (ret != 0) {
+               dsl_wrapping_key_rele(wkey, FTAG);
+               *dck_out = NULL;
+               return (ret);
+       }
 
        /*
-        * add the key to the keystore (this should always succeed
-        * since we made sure it didn't exist before)
+        * Add the key to the keystore.  It may already exist if it was
+        * added while performing the read from disk.  In this case discard
+        * it and return the key from the keystore.
         */
-       avl_find(&spa->spa_keystore.sk_dsl_keys, dck, &where);
-       avl_insert(&spa->spa_keystore.sk_dsl_keys, dck, where);
+       rw_enter(&spa->spa_keystore.sk_dk_lock, RW_WRITER);
+       ret = spa_keystore_dsl_key_hold_impl(spa, dckobj, tag, &dck_ks);
+       if (ret != 0) {
+               avl_find(&spa->spa_keystore.sk_dsl_keys, dck_io, &where);
+               avl_insert(&spa->spa_keystore.sk_dsl_keys, dck_io, where);
+               *dck_out = dck_io;
+       } else {
+               dsl_crypto_key_free(dck_io);
+               *dck_out = dck_ks;
+       }
 
-       /* release the wrapping key (the dsl key now has a reference to it) */
+       /* Release the wrapping key (the dsl key now has a reference to it) */
        dsl_wrapping_key_rele(wkey, FTAG);
-
        rw_exit(&spa->spa_keystore.sk_dk_lock);
 
-       *dck_out = dck;
        return (0);
-
-error_unlock:
-       rw_exit(&spa->spa_keystore.sk_dk_lock);
-       if (wkey != NULL)
-               dsl_wrapping_key_rele(wkey, FTAG);
-
-       *dck_out = NULL;
-       return (ret);
 }
 
 void
@@ -934,20 +935,19 @@ spa_keystore_create_mapping_impl(spa_t *spa, uint64_t dsobj,
 {
        int ret;
        avl_index_t where;
-       dsl_key_mapping_t *km = NULL, *found_km;
+       dsl_key_mapping_t *km, *found_km;
        boolean_t should_free = B_FALSE;
 
-       /* allocate the mapping */
-       km = kmem_alloc(sizeof (dsl_key_mapping_t), KM_SLEEP);
-       if (!km)
-               return (SET_ERROR(ENOMEM));
-
-       /* initialize the mapping */
+       /* Allocate and initialize the mapping */
+       km = kmem_zalloc(sizeof (dsl_key_mapping_t), KM_SLEEP);
        refcount_create(&km->km_refcnt);
 
        ret = spa_keystore_dsl_key_hold_dd(spa, dd, km, &km->km_key);
-       if (ret != 0)
-               goto error;
+       if (ret != 0) {
+               refcount_destroy(&km->km_refcnt);
+               kmem_free(km, sizeof (dsl_key_mapping_t));
+               return (ret);
+       }
 
        km->km_dsobj = dsobj;
 
@@ -979,15 +979,6 @@ spa_keystore_create_mapping_impl(spa_t *spa, uint64_t dsobj,
        }
 
        return (0);
-
-error:
-       if (km->km_key)
-               spa_keystore_dsl_key_rele(spa, km->km_key, km);
-
-       refcount_destroy(&km->km_refcnt);
-       kmem_free(km, sizeof (dsl_key_mapping_t));
-
-       return (ret);
 }
 
 int