fix paralel init for llog.

make llog more smp safe.
don't write big array into journal.
fix return codes in llog_put_cat_list/llog_get_cat_list.

Branch b1_8_gate
b=16679
i=umka
i=adilger

diff --git a/lustre/obdclass/llog_obd.c b/lustre/obdclass/llog_obd.c
index 638f1fd..d03b637 100644 (file)
--- a/lustre/obdclass/llog_obd.c
+++ b/lustre/obdclass/llog_obd.c
@@ -57,10 +57,10 @@ static struct llog_ctxt* llog_new_ctxt(struct obd_device *obd)
         OBD_ALLOC(ctxt, sizeof(*ctxt));
         if (!ctxt)
                 return NULL;
-        
+
         ctxt->loc_obd = obd;
         atomic_set(&ctxt->loc_refcount, 1);
-        
+
         return ctxt;
 }
 
@@ -144,18 +144,22 @@ int llog_setup(struct obd_device *obd, int index, struct obd_device *disk_obd,
         if (index < 0 || index >= LLOG_MAX_CTXTS)
                 RETURN(-EFAULT);
 
-        ctxt = llog_get_context(obd, index); 
+        /* someone can call lov_llog_init with NULL uuid - this can produce
+         * parallel enter to this function */
+        mutex_down(&obd->obd_llog_alloc);
+        ctxt = llog_get_context(obd, index);
         if (ctxt) {
                 /* mds_lov_update_mds might call here multiple times. So if the
                    llog is already set up then don't to do it again. */
-                CDEBUG(D_CONFIG, "obd %s ctxt %d already set up\n", 
+                CDEBUG(D_CONFIG, "obd %s ctxt %d already set up\n",
                        obd->obd_name, index);
                 LASSERT(ctxt->loc_obd == obd);
                 LASSERT(ctxt->loc_exp == disk_obd->obd_self_export);
                 LASSERT(ctxt->loc_logops == op);
-                llog_ctxt_put(ctxt); 
+                llog_ctxt_put(ctxt);
                 GOTO(out, rc = 0);
         }
+
         ctxt = llog_new_ctxt(obd);
         if (!ctxt)
                 GOTO(out, rc = -ENOMEM);
@@ -178,6 +182,7 @@ int llog_setup(struct obd_device *obd, int index, struct obd_device *disk_obd,
                 obd->obd_llog_ctxt[index] = NULL;
         }
 out:
+        mutex_up(&obd->obd_llog_alloc);
         RETURN(rc);
 }
 EXPORT_SYMBOL(llog_setup);
@@ -396,42 +401,33 @@ int llog_obd_origin_add(struct llog_ctxt *ctxt,
 }
 EXPORT_SYMBOL(llog_obd_origin_add);
 
-int llog_cat_initialize(struct obd_device *obd, int count,
+int llog_cat_initialize(struct obd_device *obd, int idx,
                         struct obd_uuid *uuid)
 {
+        struct llog_catid idarray;
         char name[32] = CATLIST;
-        struct llog_catid *idarray = NULL;
-        int size = sizeof(*idarray) * count;
         int rc;
         ENTRY;
 
-        if (count) {
-                OBD_VMALLOC(idarray, size);
-                if (!idarray)
-                        RETURN(-ENOMEM);
-        }
-
-        rc = llog_get_cat_list(obd, obd, name, count, idarray);
+        rc = llog_get_cat_list(obd, obd, name, idx, 1, &idarray);
         if (rc) {
                 CERROR("rc: %d\n", rc);
                 GOTO(out, rc);
         }
 
-        rc = obd_llog_init(obd, obd, count, idarray, uuid);
+        rc = obd_llog_init(obd, obd, 1, &idarray, uuid);
         if (rc) {
                 CERROR("rc: %d\n", rc);
                 GOTO(out, rc);
         }
 
-        rc = llog_put_cat_list(obd, obd, name, count, idarray);
+        rc = llog_put_cat_list(obd, obd, name, idx, 1, &idarray);
         if (rc) {
                 CERROR("rc: %d\n", rc);
                 GOTO(out, rc);
         }
 
  out:
-        if (idarray)
-                OBD_VFREE(idarray, size);
         RETURN(rc);
 }
 EXPORT_SYMBOL(llog_cat_initialize);