LU-9706 dt: remove dt_txn_hook_commit()

[fs/lustre-release.git] / lustre / target / tgt_main.c

diff --git a/lustre/target/tgt_main.c b/lustre/target/tgt_main.c
index 391f295..b767d1d 100644 (file)
--- a/lustre/target/tgt_main.c
+++ b/lustre/target/tgt_main.c
@@ -21,7 +21,7 @@
  * GPL HEADER END
  */
 /*
  * GPL HEADER END
  */
 /*

- * Copyright (c) 2012, 2015, Intel Corporation.
+ * Copyright (c) 2012, 2017, Intel Corporation.

  */
 /*
  * lustre/target/tgt_main.c
  */
 /*
  * lustre/target/tgt_main.c


@@ -37,23 +37,22 @@
 #include "tgt_internal.h"
 #include "../ptlrpc/ptlrpc_internal.h"
 
 #include "tgt_internal.h"
 #include "../ptlrpc/ptlrpc_internal.h"
 

-static spinlock_t uncommitted_slc_locks_guard;
-static struct list_head uncommitted_slc_locks;
-

 /*
 /*

- * Save cross-MDT lock in uncommitted_slc_locks.
+ * Save cross-MDT lock in lut_slc_locks.

  *
  * Lock R/W count is not saved, but released in unlock (not canceled remotely),
  * instead only a refcount is taken, so that the remote MDT where the object
  * resides can detect conflict with this lock there.
  *
  *
  * Lock R/W count is not saved, but released in unlock (not canceled remotely),
  * instead only a refcount is taken, so that the remote MDT where the object
  * resides can detect conflict with this lock there.
  *

+ * \param lut target

  * \param lock cross-MDT lock to save
  * \param transno when the transaction with this transno is committed, this lock
  *               can be canceled.
  */
  * \param lock cross-MDT lock to save
  * \param transno when the transaction with this transno is committed, this lock
  *               can be canceled.
  */

-void tgt_save_slc_lock(struct ldlm_lock *lock, __u64 transno)
+void tgt_save_slc_lock(struct lu_target *lut, struct ldlm_lock *lock,
+                      __u64 transno)

 {
 {

-       spin_lock(&uncommitted_slc_locks_guard);
+       spin_lock(&lut->lut_slc_locks_guard);

        lock_res_and_lock(lock);
        if (ldlm_is_cbpending(lock)) {
                /* if it was canceld by server, don't save, because remote MDT
        lock_res_and_lock(lock);
        if (ldlm_is_cbpending(lock)) {
                /* if it was canceld by server, don't save, because remote MDT


@@ -65,27 +64,27 @@ void tgt_save_slc_lock(struct ldlm_lock *lock, __u64 transno)
                 * both use this lock, and save it after use, so for the second
                 * one, just put the refcount. */
                if (list_empty(&lock->l_slc_link))
                 * both use this lock, and save it after use, so for the second
                 * one, just put the refcount. */
                if (list_empty(&lock->l_slc_link))

-                       list_add_tail(&lock->l_slc_link,
-                                     &uncommitted_slc_locks);
+                       list_add_tail(&lock->l_slc_link, &lut->lut_slc_locks);

                else
                        LDLM_LOCK_PUT(lock);
        }
        unlock_res_and_lock(lock);
                else
                        LDLM_LOCK_PUT(lock);
        }
        unlock_res_and_lock(lock);

-       spin_unlock(&uncommitted_slc_locks_guard);
+       spin_unlock(&lut->lut_slc_locks_guard);

 }
 EXPORT_SYMBOL(tgt_save_slc_lock);
 
 /*
 }
 EXPORT_SYMBOL(tgt_save_slc_lock);
 
 /*

- * Discard cross-MDT lock from uncommitted_slc_locks.
+ * Discard cross-MDT lock from lut_slc_locks.

  *
  *

- * This is called upon BAST, just remove lock from uncommitted_slc_locks and put
- * lock refcount. The BAST will cancel this lock.
+ * This is called upon BAST, just remove lock from lut_slc_locks and put lock
+ * refcount. The BAST will cancel this lock.

  *
  *

+ * \param lut target

  * \param lock cross-MDT lock to discard
  */
  * \param lock cross-MDT lock to discard
  */

-void tgt_discard_slc_lock(struct ldlm_lock *lock)
+void tgt_discard_slc_lock(struct lu_target *lut, struct ldlm_lock *lock)

 {
 {

-       spin_lock(&uncommitted_slc_locks_guard);
+       spin_lock(&lut->lut_slc_locks_guard);

        lock_res_and_lock(lock);
        /* may race with tgt_cancel_slc_locks() */
        if (lock->l_transno != 0) {
        lock_res_and_lock(lock);
        /* may race with tgt_cancel_slc_locks() */
        if (lock->l_transno != 0) {


@@ -96,26 +95,26 @@ void tgt_discard_slc_lock(struct ldlm_lock *lock)
                LDLM_LOCK_PUT(lock);
        }
        unlock_res_and_lock(lock);
                LDLM_LOCK_PUT(lock);
        }
        unlock_res_and_lock(lock);

-       spin_unlock(&uncommitted_slc_locks_guard);
+       spin_unlock(&lut->lut_slc_locks_guard);

 }
 EXPORT_SYMBOL(tgt_discard_slc_lock);
 
 /*
  * Cancel cross-MDT locks upon transaction commit.
  *
 }
 EXPORT_SYMBOL(tgt_discard_slc_lock);
 
 /*
  * Cancel cross-MDT locks upon transaction commit.
  *

- * Remove cross-MDT locks from uncommitted_slc_locks, cancel them and put lock
- * refcount.
+ * Remove cross-MDT locks from lut_slc_locks, cancel them and put lock refcount.

  *
  *

+ * \param lut target

  * \param transno transaction with this number was committed.
  */
  * \param transno transaction with this number was committed.
  */

-void tgt_cancel_slc_locks(__u64 transno)
+void tgt_cancel_slc_locks(struct lu_target *lut, __u64 transno)

 {
        struct ldlm_lock *lock, *next;
        LIST_HEAD(list);
        struct lustre_handle lockh;
 
 {
        struct ldlm_lock *lock, *next;
        LIST_HEAD(list);
        struct lustre_handle lockh;
 

-       spin_lock(&uncommitted_slc_locks_guard);
-       list_for_each_entry_safe(lock, next, &uncommitted_slc_locks,
+       spin_lock(&lut->lut_slc_locks_guard);
+       list_for_each_entry_safe(lock, next, &lut->lut_slc_locks,

                                 l_slc_link) {
                lock_res_and_lock(lock);
                LASSERT(lock->l_transno != 0);
                                 l_slc_link) {
                lock_res_and_lock(lock);
                LASSERT(lock->l_transno != 0);


@@ -134,7 +133,7 @@ void tgt_cancel_slc_locks(__u64 transno)
                list_move(&lock->l_slc_link, &list);
                unlock_res_and_lock(lock);
        }
                list_move(&lock->l_slc_link, &list);
                unlock_res_and_lock(lock);
        }

-       spin_unlock(&uncommitted_slc_locks_guard);
+       spin_unlock(&lut->lut_slc_locks_guard);

 
        list_for_each_entry_safe(lock, next, &list, l_slc_link) {
                list_del_init(&lock->l_slc_link);
 
        list_for_each_entry_safe(lock, next, &list, l_slc_link) {
                list_del_init(&lock->l_slc_link);


@@ -153,7 +152,9 @@ int tgt_init(const struct lu_env *env, struct lu_target *lut,
        struct lu_attr           attr;
        struct lu_fid            fid;
        struct dt_object        *o;
        struct lu_attr           attr;
        struct lu_fid            fid;
        struct dt_object        *o;

-       int                      rc = 0;
+       struct tg_grants_data   *tgd = &lut->lut_tgd;
+       struct obd_statfs       *osfs;
+       int i, rc = 0;

 
        ENTRY;
 
 
        ENTRY;
 


@@ -182,10 +183,45 @@ int tgt_init(const struct lu_env *env, struct lu_target *lut,
        spin_lock_init(&lut->lut_flags_lock);
        lut->lut_sync_lock_cancel = NEVER_SYNC_ON_CANCEL;
 
        spin_lock_init(&lut->lut_flags_lock);
        lut->lut_sync_lock_cancel = NEVER_SYNC_ON_CANCEL;
 

+       spin_lock_init(&lut->lut_slc_locks_guard);
+       INIT_LIST_HEAD(&lut->lut_slc_locks);
+

        /* last_rcvd initialization is needed by replayable targets only */
        if (!obd->obd_replayable)
                RETURN(0);
 
        /* last_rcvd initialization is needed by replayable targets only */
        if (!obd->obd_replayable)
                RETURN(0);
 

+       /* initialize grant and statfs data in target */
+       dt_conf_get(env, lut->lut_bottom, &lut->lut_dt_conf);
+
+       /* statfs data */
+       spin_lock_init(&tgd->tgd_osfs_lock);
+       tgd->tgd_osfs_age = ktime_get_seconds() - 1000;
+       tgd->tgd_osfs_unstable = 0;
+       tgd->tgd_statfs_inflight = 0;
+       tgd->tgd_osfs_inflight = 0;
+
+       /* grant data */
+       spin_lock_init(&tgd->tgd_grant_lock);
+       tgd->tgd_tot_dirty = 0;
+       tgd->tgd_tot_granted = 0;
+       tgd->tgd_tot_pending = 0;
+       tgd->tgd_grant_compat_disable = 0;
+
+       /* populate cached statfs data */
+       osfs = &tgt_th_info(env)->tti_u.osfs;
+       rc = tgt_statfs_internal(env, lut, osfs, 0, NULL);
+       if (rc != 0) {
+               CERROR("%s: can't get statfs data, rc %d\n", tgt_name(lut),
+                       rc);
+               GOTO(out, rc);
+       }
+       if (!is_power_of_2(osfs->os_bsize)) {
+               CERROR("%s: blocksize (%d) is not a power of 2\n",
+                       tgt_name(lut), osfs->os_bsize);
+               GOTO(out, rc = -EPROTO);
+       }
+       tgd->tgd_blockbits = fls(osfs->os_bsize) - 1;
+

        spin_lock_init(&lut->lut_translock);
        spin_lock_init(&lut->lut_client_bitmap_lock);
 
        spin_lock_init(&lut->lut_translock);
        spin_lock_init(&lut->lut_client_bitmap_lock);
 


@@ -216,7 +252,6 @@ int tgt_init(const struct lu_env *env, struct lu_target *lut,
        /* prepare transactions callbacks */
        lut->lut_txn_cb.dtc_txn_start = tgt_txn_start_cb;
        lut->lut_txn_cb.dtc_txn_stop = tgt_txn_stop_cb;
        /* prepare transactions callbacks */
        lut->lut_txn_cb.dtc_txn_start = tgt_txn_start_cb;
        lut->lut_txn_cb.dtc_txn_stop = tgt_txn_stop_cb;

-       lut->lut_txn_cb.dtc_txn_commit = NULL;

        lut->lut_txn_cb.dtc_cookie = lut;
        lut->lut_txn_cb.dtc_tag = LCT_DT_THREAD | LCT_MD_THREAD;
        INIT_LIST_HEAD(&lut->lut_txn_cb.dtc_linkage);
        lut->lut_txn_cb.dtc_cookie = lut;
        lut->lut_txn_cb.dtc_tag = LCT_DT_THREAD | LCT_MD_THREAD;
        INIT_LIST_HEAD(&lut->lut_txn_cb.dtc_linkage);


@@ -231,7 +266,7 @@ int tgt_init(const struct lu_env *env, struct lu_target *lut,
        OBD_ALLOC(lut->lut_reply_bitmap,
                  LUT_REPLY_SLOTS_MAX_CHUNKS * sizeof(unsigned long *));
        if (lut->lut_reply_bitmap == NULL)
        OBD_ALLOC(lut->lut_reply_bitmap,
                  LUT_REPLY_SLOTS_MAX_CHUNKS * sizeof(unsigned long *));
        if (lut->lut_reply_bitmap == NULL)

-               GOTO(out, rc);
+               GOTO(out, rc = -ENOMEM);

 
        memset(&attr, 0, sizeof(attr));
        attr.la_valid = LA_MODE;
 
        memset(&attr, 0, sizeof(attr));
        attr.la_valid = LA_MODE;


@@ -263,18 +298,26 @@ out_put:
        obd->u.obt.obt_magic = 0;
        obd->u.obt.obt_lut = NULL;
        if (lut->lut_last_rcvd != NULL) {
        obd->u.obt.obt_magic = 0;
        obd->u.obt.obt_lut = NULL;
        if (lut->lut_last_rcvd != NULL) {

-               lu_object_put(env, &lut->lut_last_rcvd->do_lu);
+               dt_object_put(env, lut->lut_last_rcvd);

                lut->lut_last_rcvd = NULL;
        }
        if (lut->lut_client_bitmap != NULL)
                OBD_FREE(lut->lut_client_bitmap, LR_MAX_CLIENTS >> 3);
        lut->lut_client_bitmap = NULL;
        if (lut->lut_reply_data != NULL)
                lut->lut_last_rcvd = NULL;
        }
        if (lut->lut_client_bitmap != NULL)
                OBD_FREE(lut->lut_client_bitmap, LR_MAX_CLIENTS >> 3);
        lut->lut_client_bitmap = NULL;
        if (lut->lut_reply_data != NULL)

-               lu_object_put(env, &lut->lut_reply_data->do_lu);
+               dt_object_put(env, lut->lut_reply_data);

        lut->lut_reply_data = NULL;
        lut->lut_reply_data = NULL;

-       if (lut->lut_reply_bitmap != NULL)
+       if (lut->lut_reply_bitmap != NULL) {
+               for (i = 0; i < LUT_REPLY_SLOTS_MAX_CHUNKS; i++) {
+                       if (lut->lut_reply_bitmap[i] != NULL)
+                               OBD_FREE_LARGE(lut->lut_reply_bitmap[i],
+                                   BITS_TO_LONGS(LUT_REPLY_SLOTS_PER_CHUNK) *
+                                   sizeof(long));
+                       lut->lut_reply_bitmap[i] = NULL;
+               }

                OBD_FREE(lut->lut_reply_bitmap,
                         LUT_REPLY_SLOTS_MAX_CHUNKS * sizeof(unsigned long *));
                OBD_FREE(lut->lut_reply_bitmap,
                         LUT_REPLY_SLOTS_MAX_CHUNKS * sizeof(unsigned long *));

+       }

        lut->lut_reply_bitmap = NULL;
        return rc;
 }
        lut->lut_reply_bitmap = NULL;
        return rc;
 }


@@ -301,12 +344,12 @@ void tgt_fini(const struct lu_env *env, struct lu_target *lut)
        sptlrpc_rule_set_free(&lut->lut_sptlrpc_rset);
 
        if (lut->lut_reply_data != NULL)
        sptlrpc_rule_set_free(&lut->lut_sptlrpc_rset);
 
        if (lut->lut_reply_data != NULL)

-               lu_object_put(env, &lut->lut_reply_data->do_lu);
+               dt_object_put(env, lut->lut_reply_data);

        lut->lut_reply_data = NULL;
        if (lut->lut_reply_bitmap != NULL) {
                for (i = 0; i < LUT_REPLY_SLOTS_MAX_CHUNKS; i++) {
                        if (lut->lut_reply_bitmap[i] != NULL)
        lut->lut_reply_data = NULL;
        if (lut->lut_reply_bitmap != NULL) {
                for (i = 0; i < LUT_REPLY_SLOTS_MAX_CHUNKS; i++) {
                        if (lut->lut_reply_bitmap[i] != NULL)

-                               OBD_FREE(lut->lut_reply_bitmap[i],
+                               OBD_FREE_LARGE(lut->lut_reply_bitmap[i],

                                    BITS_TO_LONGS(LUT_REPLY_SLOTS_PER_CHUNK) *
                                    sizeof(long));
                        lut->lut_reply_bitmap[i] = NULL;
                                    BITS_TO_LONGS(LUT_REPLY_SLOTS_PER_CHUNK) *
                                    sizeof(long));
                        lut->lut_reply_bitmap[i] = NULL;


@@ -321,15 +364,44 @@ void tgt_fini(const struct lu_env *env, struct lu_target *lut)
        }
        if (lut->lut_last_rcvd) {
                dt_txn_callback_del(lut->lut_bottom, &lut->lut_txn_cb);
        }
        if (lut->lut_last_rcvd) {
                dt_txn_callback_del(lut->lut_bottom, &lut->lut_txn_cb);

-               lu_object_put(env, &lut->lut_last_rcvd->do_lu);
+               dt_object_put(env, lut->lut_last_rcvd);

                lut->lut_last_rcvd = NULL;
        }
        EXIT;
 }
 EXPORT_SYMBOL(tgt_fini);
 
                lut->lut_last_rcvd = NULL;
        }
        EXIT;
 }
 EXPORT_SYMBOL(tgt_fini);
 

+static struct kmem_cache *tgt_thread_kmem;
+static struct kmem_cache *tgt_session_kmem;
+static struct lu_kmem_descr tgt_caches[] = {
+       {
+               .ckd_cache = &tgt_thread_kmem,
+               .ckd_name  = "tgt_thread_kmem",
+               .ckd_size  = sizeof(struct tgt_thread_info),
+       },
+       {
+               .ckd_cache = &tgt_session_kmem,
+               .ckd_name  = "tgt_session_kmem",
+               .ckd_size  = sizeof(struct tgt_session_info)
+       },
+       {
+               .ckd_cache = NULL
+       }
+};
+
+

 /* context key constructor/destructor: tg_key_init, tg_key_fini */
 /* context key constructor/destructor: tg_key_init, tg_key_fini */

-LU_KEY_INIT(tgt, struct tgt_thread_info);
+static void *tgt_key_init(const struct lu_context *ctx,
+                                 struct lu_context_key *key)
+{
+       struct tgt_thread_info *thread;
+
+       OBD_SLAB_ALLOC_PTR_GFP(thread, tgt_thread_kmem, GFP_NOFS);
+       if (thread == NULL)
+               return ERR_PTR(-ENOMEM);
+
+       return thread;
+}

 
 static void tgt_key_fini(const struct lu_context *ctx,
                         struct lu_context_key *key, void *data)
 
 static void tgt_key_fini(const struct lu_context *ctx,
                         struct lu_context_key *key, void *data)


@@ -346,7 +418,7 @@ static void tgt_key_fini(const struct lu_context *ctx,
        if (args->ta_args != NULL)
                OBD_FREE(args->ta_args, sizeof(args->ta_args[0]) *
                                        args->ta_alloc_args);
        if (args->ta_args != NULL)
                OBD_FREE(args->ta_args, sizeof(args->ta_args[0]) *
                                        args->ta_alloc_args);

-       OBD_FREE_PTR(info);
+       OBD_SLAB_FREE_PTR(info, tgt_thread_kmem);

 }
 
 static void tgt_key_exit(const struct lu_context *ctx,
 }
 
 static void tgt_key_exit(const struct lu_context *ctx,


@@ -368,8 +440,25 @@ struct lu_context_key tgt_thread_key = {
 
 LU_KEY_INIT_GENERIC(tgt);
 
 
 LU_KEY_INIT_GENERIC(tgt);
 

-/* context key constructor/destructor: tgt_ses_key_init, tgt_ses_key_fini */
-LU_KEY_INIT_FINI(tgt_ses, struct tgt_session_info);
+static void *tgt_ses_key_init(const struct lu_context *ctx,
+                             struct lu_context_key *key)
+{
+       struct tgt_session_info *session;
+
+       OBD_SLAB_ALLOC_PTR_GFP(session, tgt_session_kmem, GFP_NOFS);
+       if (session == NULL)
+               return ERR_PTR(-ENOMEM);
+
+       return session;
+}
+
+static void tgt_ses_key_fini(const struct lu_context *ctx,
+                            struct lu_context_key *key, void *data)
+{
+       struct tgt_session_info *session = data;
+
+       OBD_SLAB_FREE_PTR(session, tgt_session_kmem);
+}

 
 /* context key: tgt_session_key */
 struct lu_context_key tgt_session_key = {
 
 /* context key: tgt_session_key */
 struct lu_context_key tgt_session_key = {


@@ -392,31 +481,37 @@ struct page *tgt_page_to_corrupt;
 
 int tgt_mod_init(void)
 {
 
 int tgt_mod_init(void)
 {

+       int     result;

        ENTRY;
 
        ENTRY;
 

-       tgt_page_to_corrupt = alloc_page(GFP_IOFS);
+       result = lu_kmem_init(tgt_caches);
+       if (result != 0)
+               RETURN(result);
+
+       tgt_page_to_corrupt = alloc_page(GFP_KERNEL);

 
        tgt_key_init_generic(&tgt_thread_key, NULL);
        lu_context_key_register_many(&tgt_thread_key, NULL);
 
        tgt_ses_key_init_generic(&tgt_session_key, NULL);
        lu_context_key_register_many(&tgt_session_key, NULL);
 
        tgt_key_init_generic(&tgt_thread_key, NULL);
        lu_context_key_register_many(&tgt_thread_key, NULL);
 
        tgt_ses_key_init_generic(&tgt_session_key, NULL);
        lu_context_key_register_many(&tgt_session_key, NULL);

+       barrier_init();

 
        update_info_init();
 
 
        update_info_init();
 

-       spin_lock_init(&uncommitted_slc_locks_guard);
-       INIT_LIST_HEAD(&uncommitted_slc_locks);
-

        RETURN(0);
 }
 
 void tgt_mod_exit(void)
 {
        RETURN(0);
 }
 
 void tgt_mod_exit(void)
 {

+       barrier_fini();

        if (tgt_page_to_corrupt != NULL)
        if (tgt_page_to_corrupt != NULL)

-               page_cache_release(tgt_page_to_corrupt);
+               put_page(tgt_page_to_corrupt);

 
        lu_context_key_degister(&tgt_thread_key);
        lu_context_key_degister(&tgt_session_key);
        update_info_fini();
 
        lu_context_key_degister(&tgt_thread_key);
        lu_context_key_degister(&tgt_session_key);
        update_info_fini();

+
+       lu_kmem_fini(tgt_caches);

 }
 
 }