Branch HEAD

b=20085
i=rread
i=nico

- add missed lock counter increment causing assertion
- introduce class_export_cb_get/put(); class_export_lock_get/put() to make
  counter handling easier

diff --git a/lustre/include/lustre_export.h b/lustre/include/lustre_export.h
index 7ff4284..72a83af 100644 (file)
--- a/lustre/include/lustre_export.h
+++ b/lustre/include/lustre_export.h
@@ -143,9 +143,16 @@ enum obd_option {
 struct obd_export {
         struct portals_handle     exp_handle;
         atomic_t                  exp_refcount;
-        atomic_t                  exp_rpc_count;
-        atomic_t                  exp_cb_count;
-        atomic_t                  exp_locks_count;
+        /**
+         * Set of counters below is to track where export references are
+         * kept. The exp_rpc_count is used for reconnect handling also,
+         * the cb_count and locks_count are for debug purposes only for now.
+         * The sum of them should be less than exp_refcount by 3
+         */
+        atomic_t                  exp_rpc_count; /** RPC references */
+        atomic_t                  exp_cb_count; /** Commit callback references */
+        atomic_t                  exp_locks_count; /** Lock references */
+
         atomic_t                  exp_replay_count;
         struct obd_uuid           exp_client_uuid;
         struct list_head          exp_obd_chain;

diff --git a/lustre/include/obd_class.h b/lustre/include/obd_class.h
index a9e36ba..a62ed99 100644 (file)
--- a/lustre/include/obd_class.h
+++ b/lustre/include/obd_class.h
@@ -182,12 +182,47 @@ void class_del_profiles(void);
 
 #define class_export_rpc_put(exp)                                       \
 ({                                                                      \
+        LASSERT(atomic_read(&exp->exp_rpc_count) > 0);                  \
         atomic_dec(&(exp)->exp_rpc_count);                              \
         CDEBUG(D_INFO, "RPC PUTting export %p : new rpc_count %d\n",    \
                (exp), atomic_read(&(exp)->exp_rpc_count));              \
         class_export_put(exp);                                          \
 })
 
+#define class_export_lock_get(exp)                                      \
+({                                                                      \
+        atomic_inc(&(exp)->exp_locks_count);                            \
+        CDEBUG(D_INFO, "lock GETting export %p : new locks_count %d\n", \
+               (exp), atomic_read(&(exp)->exp_locks_count));            \
+        class_export_get(exp);                                          \
+})
+
+#define class_export_lock_put(exp)                                      \
+({                                                                      \
+        LASSERT(atomic_read(&exp->exp_locks_count) > 0);                \
+        atomic_dec(&(exp)->exp_locks_count);                            \
+        CDEBUG(D_INFO, "lock PUTting export %p : new locks_count %d\n", \
+               (exp), atomic_read(&(exp)->exp_locks_count));            \
+        class_export_put(exp);                                          \
+})
+
+#define class_export_cb_get(exp)                                        \
+({                                                                      \
+        atomic_inc(&(exp)->exp_cb_count);                               \
+        CDEBUG(D_INFO, "callback GETting export %p : new cb_count %d\n",\
+               (exp), atomic_read(&(exp)->exp_cb_count));               \
+        class_export_get(exp);                                          \
+})
+
+#define class_export_cb_put(exp)                                        \
+({                                                                      \
+        LASSERT(atomic_read(&exp->exp_cb_count) > 0);                   \
+        atomic_dec(&(exp)->exp_cb_count);                               \
+        CDEBUG(D_INFO, "callback PUTting export %p : new cb_count %d\n",\
+               (exp), atomic_read(&(exp)->exp_cb_count));               \
+        class_export_put(exp);                                          \
+})
+
 /* genops.c */
 struct obd_export *class_export_get(struct obd_export *exp);
 void class_export_put(struct obd_export *exp);

diff --git a/lustre/ldlm/ldlm_flock.c b/lustre/ldlm/ldlm_flock.c
index f58e176..46ff280 100644 (file)
--- a/lustre/ldlm/ldlm_flock.c
+++ b/lustre/ldlm/ldlm_flock.c
@@ -400,7 +400,7 @@ reprocess:
                         new->l_policy_data.l_flock.end + 1;
                 new2->l_conn_export = lock->l_conn_export;
                 if (lock->l_export != NULL) {
-                        new2->l_export = class_export_get(lock->l_export);
+                        new2->l_export = class_export_lock_get(lock->l_export);
                         if (new2->l_export->exp_lock_hash &&
                             hlist_unhashed(&new2->l_exp_hash))
                                 lustre_hash_add(new2->l_export->exp_lock_hash,

diff --git a/lustre/ldlm/ldlm_lock.c b/lustre/ldlm/ldlm_lock.c
index 3e4f1ac..5d719d4 100644 (file)
--- a/lustre/ldlm/ldlm_lock.c
+++ b/lustre/ldlm/ldlm_lock.c
@@ -164,9 +164,7 @@ void ldlm_lock_put(struct ldlm_lock *lock)
                 ldlm_resource_putref(res);
                 lock->l_resource = NULL;
                 if (lock->l_export) {
-                        LASSERT(atomic_read(&lock->l_export->exp_locks_count) > 0);
-                        atomic_dec(&lock->l_export->exp_locks_count);
-                        class_export_put(lock->l_export);
+                        class_export_lock_put(lock->l_export);
                         lock->l_export = NULL;
                 }
 

diff --git a/lustre/ldlm/ldlm_lockd.c b/lustre/ldlm/ldlm_lockd.c
index 8271bee..5ffd734 100644 (file)
--- a/lustre/ldlm/ldlm_lockd.c
+++ b/lustre/ldlm/ldlm_lockd.c
@@ -208,7 +208,7 @@ static int expired_lock_main(void *arg)
                                 LDLM_LOCK_RELEASE(lock);
                                 continue;
                         }
-                        export = class_export_get(lock->l_export);
+                        export = class_export_lock_get(lock->l_export);
                         spin_unlock_bh(&waiting_locks_spinlock);
 
                         /* release extra ref grabbed by ldlm_add_waiting_lock()
@@ -217,7 +217,7 @@ static int expired_lock_main(void *arg)
 
                         do_dump++;
                         class_fail_export(export);
-                        class_export_put(export);
+                        class_export_lock_put(export);
                         spin_lock_bh(&waiting_locks_spinlock);
                 }
                 spin_unlock_bh(&waiting_locks_spinlock);
@@ -1109,8 +1109,7 @@ int ldlm_handle_enqueue0(struct ldlm_namespace *ns,
                 LDLM_ERROR(lock, "lock on destroyed export %p", req->rq_export);
                 GOTO(out, rc = -ENOTCONN);
         }
-        lock->l_export = class_export_get(req->rq_export);
-        atomic_inc(&lock->l_export->exp_locks_count);
+        lock->l_export = class_export_lock_get(req->rq_export);
         if (lock->l_export->exp_lock_hash)
                 lustre_hash_add(lock->l_export->exp_lock_hash,
                                 &lock->l_remote_handle,

diff --git a/lustre/mdt/mdt_handler.c b/lustre/mdt/mdt_handler.c
index 3b77972..2bb3866 100644 (file)
--- a/lustre/mdt/mdt_handler.c
+++ b/lustre/mdt/mdt_handler.c
@@ -3180,9 +3180,7 @@ int mdt_intent_lock_replace(struct mdt_thread_info *info,
                 new_lock->l_writers--;
         }
 
-        new_lock->l_export = class_export_get(req->rq_export);
-        atomic_inc(&lock->l_export->exp_locks_count);
-
+        new_lock->l_export = class_export_lock_get(req->rq_export);
         new_lock->l_blocking_ast = lock->l_blocking_ast;
         new_lock->l_completion_ast = lock->l_completion_ast;
         new_lock->l_remote_handle = lock->l_remote_handle;

diff --git a/lustre/mdt/mdt_recovery.c b/lustre/mdt/mdt_recovery.c
index f74b694..0ca7baf 100644 (file)
--- a/lustre/mdt/mdt_recovery.c
+++ b/lustre/mdt/mdt_recovery.c
@@ -942,8 +942,7 @@ static int mdt_txn_stop_cb(const struct lu_env *env,
         /* add separate commit callback for transaction handling because we need
          * export as parameter */
         mdt_trans_add_cb(txn, lut_cb_last_committed,
-                         class_export_get(mti->mti_exp));
-        atomic_inc(&mti->mti_exp->exp_cb_count);
+                         class_export_cb_get(mti->mti_exp));
 
         return mdt_last_rcvd_update(mti, txn);
 }

diff --git a/lustre/obdfilter/filter.c b/lustre/obdfilter/filter.c
index e13a2cb..2b59d9a 100644 (file)
--- a/lustre/obdfilter/filter.c
+++ b/lustre/obdfilter/filter.c
@@ -88,8 +88,7 @@ static void filter_commit_cb(struct obd_device *obd, __u64 transno,
         struct obd_export *exp = cb_data;
         LASSERT(exp->exp_obd == obd);
         obd_transno_commit_cb(obd, transno, exp, error);
-        atomic_dec(&exp->exp_cb_count);
-        class_export_put(exp);
+        class_export_cb_put(exp);
 }
 
 int filter_version_get_check(struct obd_export *exp,
@@ -165,8 +164,7 @@ int filter_finish_transno(struct obd_export *exp, struct inode *inode,
                        fed->fed_lr_idx, fed->fed_lr_off);
                 err = -EINVAL;
         } else {
-                class_export_get(exp); /* released when the cb is called */
-                atomic_inc(&exp->exp_cb_count);
+                class_export_cb_get(exp); /* released when the cb is called */
                 if (!force_sync)
                         force_sync = fsfilt_add_journal_cb(exp->exp_obd,
                                                            last_rcvd,

diff --git a/lustre/ptlrpc/target.c b/lustre/ptlrpc/target.c
index ecf1e30..0b6b2fe 100644 (file)
--- a/lustre/ptlrpc/target.c
+++ b/lustre/ptlrpc/target.c
@@ -309,9 +309,7 @@ void lut_cb_last_committed(struct lu_target *lut, __u64 transno,
         } else {
                 spin_unlock(&lut->lut_translock);
         }
-        LASSERT(atomic_read(&exp->exp_cb_count) > 0);
-        atomic_dec(&exp->exp_cb_count);
-        class_export_put(exp);
+        class_export_cb_put(exp);
         if (transno)
                 CDEBUG(D_HA, "%s: transno "LPD64" is committed\n",
                        lut->lut_obd->obd_name, transno);