Grab a lov llog lock so we ensure the llog records for a single transno

are allocated in the same order they will be replayed in.  Concurrent
unlinks might otherwise have misordered llog records and not be replayable.
b=4033
r=andreas (for Tian)

diff --git a/lustre/include/linux/obd.h b/lustre/include/linux/obd.h
index ec030ca..0279d97 100644 (file)
--- a/lustre/include/linux/obd.h
+++ b/lustre/include/linux/obd.h
@@ -420,6 +420,7 @@ struct lov_tgt_desc {
 struct lov_obd {
         spinlock_t lov_lock;
         struct lov_desc desc;
+        struct semaphore lov_llog_sem;
         int bufsize;
         int refcount;
         int lo_catalog_loaded:1;

diff --git a/lustre/ldlm/l_lock.c b/lustre/ldlm/l_lock.c
index 69d3f14..024ec26 100644 (file)
--- a/lustre/ldlm/l_lock.c
+++ b/lustre/ldlm/l_lock.c
@@ -123,9 +123,6 @@ void l_check_no_ns_lock(struct ldlm_namespace *ns)
         if (l_has_lock(&ns->ns_lock) && time_after(jiffies, next_msg)) {
                 CERROR("namespace %s lock held illegally; tell phil\n",
                        ns->ns_name);
-#if (LUSTRE_KERNEL_VERSION >= 30)
-                CERROR(portals_debug_dumpstack());
-#endif
                 next_msg = jiffies + 60 * HZ;
         }
 }

diff --git a/lustre/lov/lov_log.c b/lustre/lov/lov_log.c
index c6174d7..9a483bb 100644 (file)
--- a/lustre/lov/lov_log.c
+++ b/lustre/lov/lov_log.c
@@ -77,6 +77,12 @@ static int lov_llog_origin_add(struct llog_ctxt *ctxt, struct llog_rec_hdr *rec,
 
         LASSERT(logcookies && numcookies >= lsm->lsm_stripe_count);
 
+        /* We need this to serialize llog records between parallel unlinks so
+         * we can replay llog records in strict transno and llog order.  If
+         * and when we want to make this more scalable we need to lock and
+         * write records in strictly ost_idx order not lsm order.  Consider
+         * file 1 on ost_idx [1, 2, 3, 4] and file 2 on ost_idx [3, 4, 1, 2] */
+        down(&lov->lov_llog_sem);
         for (i = 0,loi = lsm->lsm_oinfo; i < lsm->lsm_stripe_count; i++,loi++) {
                 struct obd_device *child =
                         lov->tgts[loi->loi_ost_idx].ltd_exp->exp_obd; 
@@ -90,6 +96,7 @@ static int lov_llog_origin_add(struct llog_ctxt *ctxt, struct llog_rec_hdr *rec,
                                numcookies - rc, NULL,
                                lock != NULL ? lock + rc : NULL, lock_count);
         }
+        up(&lov->lov_llog_sem);
         OBD_FREE(lur, sizeof(*lur));
 
         RETURN(rc);

diff --git a/lustre/lov/lov_obd.c b/lustre/lov/lov_obd.c
index 6029ed1..3b1e561 100644 (file)
--- a/lustre/lov/lov_obd.c
+++ b/lustre/lov/lov_obd.c
@@ -500,6 +500,7 @@ static int lov_setup(struct obd_device *obd, obd_count len, void *buf)
         CDEBUG(D_CONFIG, "tgts: %p bufsize: 0x%x\n", lov->tgts, lov->bufsize);
         lov->desc = *desc;
         spin_lock_init(&lov->lov_lock);
+        sema_init(&lov->lov_llog_sem, 1);
 
         RETURN(0);
 }