From 51d9f398eb6931452c86886a076015b48cdeec68 Mon Sep 17 00:00:00 2001
From: adilger <adilger>
Date: Fri, 30 Jul 2004 21:53:14 +0000
Subject: [PATCH] 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)

---
 lustre/include/linux/obd.h | 1 +
 lustre/ldlm/l_lock.c       | 3 ---
 lustre/lov/lov_log.c       | 7 +++++++
 lustre/lov/lov_obd.c       | 1 +
 4 files changed, 9 insertions(+), 3 deletions(-)

diff --git a/lustre/include/linux/obd.h b/lustre/include/linux/obd.h
index ec030ca..0279d97 100644
--- 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
--- 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
--- 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
--- 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);
 }
-- 
1.8.3.1