LU-1302 llog: structures changes, llog_thread_info

[fs/lustre-release.git] / lustre / obdclass / llog_cat.c

diff --git a/lustre/obdclass/llog_cat.c b/lustre/obdclass/llog_cat.c
index a757e68..54e911f 100644 (file)
--- a/lustre/obdclass/llog_cat.c
+++ b/lustre/obdclass/llog_cat.c
@@ -1,40 +1,49 @@
-/* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
- * vim:expandtab:shiftwidth=8:tabstop=8:
+/*
+ * GPL HEADER START

  *
  *

- *  Copyright (C) 2001-2003 Cluster File Systems, Inc.
- *   Author: Andreas Dilger <adilger@clusterfs.com>
+ * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.

  *
  *

- *   This file is part of the Lustre file system, http://www.lustre.org
- *   Lustre is a trademark of Cluster File Systems, Inc.
+ * This program is free software; you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License version 2 only,
+ * as published by the Free Software Foundation.

  *
  *

- *   You may have signed or agreed to another license before downloading
- *   this software.  If so, you are bound by the terms and conditions
- *   of that agreement, and the following does not apply to you.  See the
- *   LICENSE file included with this distribution for more information.
+ * This program is distributed in the hope that it will be useful, but
+ * WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+ * General Public License version 2 for more details (a copy is included
+ * in the LICENSE file that accompanied this code).

  *
  *

- *   If you did not agree to a different license, then this copy of Lustre
- *   is open source software; you can redistribute it and/or modify it
- *   under the terms of version 2 of the GNU General Public License as
- *   published by the Free Software Foundation.
+ * You should have received a copy of the GNU General Public License
+ * version 2 along with this program; If not, see
+ * http://www.sun.com/software/products/lustre/docs/GPLv2.pdf

  *
  *

- *   In either case, Lustre is distributed in the hope that it will be
- *   useful, but WITHOUT ANY WARRANTY; without even the implied warranty
- *   of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
- *   license text for more details.
+ * Please contact Sun Microsystems, Inc., 4150 Network Circle, Santa Clara,
+ * CA 95054 USA or visit www.sun.com if you need additional information or
+ * have any questions.
+ *
+ * GPL HEADER END
+ */
+/*
+ * Copyright (c) 2003, 2010, Oracle and/or its affiliates. All rights reserved.
+ * Use is subject to license terms.
+ */
+/*
+ * This file is part of Lustre, http://www.lustre.org/
+ * Lustre is a trademark of Sun Microsystems, Inc.
+ *
+ * lustre/obdclass/llog_cat.c

  *
  * OST<->MDS recovery logging infrastructure.
  *
  * Invariants in implementation:
  * - we do not share logs among different OST<->MDS connections, so that
  *   if an OST or MDS fails it need only look at log(s) relevant to itself
  *
  * OST<->MDS recovery logging infrastructure.
  *
  * Invariants in implementation:
  * - we do not share logs among different OST<->MDS connections, so that
  *   if an OST or MDS fails it need only look at log(s) relevant to itself

+ *
+ * Author: Andreas Dilger <adilger@clusterfs.com>

  */
 
 #define DEBUG_SUBSYSTEM S_LOG
 
  */
 
 #define DEBUG_SUBSYSTEM S_LOG
 

-#ifndef EXPORT_SYMTAB
-#define EXPORT_SYMTAB
-#endif
-

 #ifndef __KERNEL__
 #include <liblustre.h>
 #endif
 #ifndef __KERNEL__
 #include <liblustre.h>
 #endif


@@ -67,13 +76,13 @@ static struct llog_handle *llog_cat_new_log(struct llog_handle *cathandle)
                 RETURN(ERR_PTR(-ENOSPC));
         }
 
                 RETURN(ERR_PTR(-ENOSPC));
         }
 

-        if (OBD_FAIL_CHECK_ONCE(OBD_FAIL_MDS_LLOG_CREATE_FAILED))
+        if (OBD_FAIL_CHECK(OBD_FAIL_MDS_LLOG_CREATE_FAILED))

                 RETURN(ERR_PTR(-ENOSPC));
                 RETURN(ERR_PTR(-ENOSPC));

- 
+

         rc = llog_create(cathandle->lgh_ctxt, &loghandle, NULL, NULL);
         rc = llog_create(cathandle->lgh_ctxt, &loghandle, NULL, NULL);

-        if (rc) 
+        if (rc)

                 RETURN(ERR_PTR(rc));
                 RETURN(ERR_PTR(rc));

-        
+

         rc = llog_init_handle(loghandle,
                               LLOG_F_IS_PLAIN | LLOG_F_ZAP_WHEN_EMPTY,
                               &cathandle->lgh_hdr->llh_tgtuuid);
         rc = llog_init_handle(loghandle,
                               LLOG_F_IS_PLAIN | LLOG_F_ZAP_WHEN_EMPTY,
                               &cathandle->lgh_hdr->llh_tgtuuid);


@@ -82,16 +91,21 @@ static struct llog_handle *llog_cat_new_log(struct llog_handle *cathandle)
 
         if (index == 0)
                 index = 1;
 
         if (index == 0)
                 index = 1;

+
+       cfs_spin_lock(&loghandle->lgh_hdr_lock);
+       llh->llh_count++;

         if (ext2_set_bit(index, llh->llh_bitmap)) {
                 CERROR("argh, index %u already set in log bitmap?\n",
                        index);
         if (ext2_set_bit(index, llh->llh_bitmap)) {
                 CERROR("argh, index %u already set in log bitmap?\n",
                        index);

+               cfs_spin_unlock(&loghandle->lgh_hdr_lock);

                 LBUG(); /* should never happen */
         }
                 LBUG(); /* should never happen */
         }

+       cfs_spin_unlock(&loghandle->lgh_hdr_lock);
+

         cathandle->lgh_last_idx = index;
         cathandle->lgh_last_idx = index;

-        llh->llh_count++;

         llh->llh_tail.lrt_index = index;
 
         llh->llh_tail.lrt_index = index;
 

-        CDEBUG(D_HA, "new recovery log "LPX64":%x for index %u of catalog "
+        CDEBUG(D_RPCTRACE,"new recovery log "LPX64":%x for index %u of catalog "

                LPX64"\n", loghandle->lgh_id.lgl_oid, loghandle->lgh_id.lgl_ogen,
                index, cathandle->lgh_id.lgl_oid);
         /* build the record for this log in the catalog */
                LPX64"\n", loghandle->lgh_id.lgl_oid, loghandle->lgh_id.lgl_ogen,
                index, cathandle->lgh_id.lgl_oid);
         /* build the record for this log in the catalog */


@@ -111,16 +125,16 @@ static struct llog_handle *llog_cat_new_log(struct llog_handle *cathandle)
 
         loghandle->lgh_hdr->llh_cat_idx = index;
         cathandle->u.chd.chd_current_log = loghandle;
 
         loghandle->lgh_hdr->llh_cat_idx = index;
         cathandle->u.chd.chd_current_log = loghandle;

-        LASSERT(list_empty(&loghandle->u.phd.phd_entry));
-        list_add_tail(&loghandle->u.phd.phd_entry, &cathandle->u.chd.chd_head);
+        LASSERT(cfs_list_empty(&loghandle->u.phd.phd_entry));
+        cfs_list_add_tail(&loghandle->u.phd.phd_entry,
+                          &cathandle->u.chd.chd_head);

 
 

- out_destroy:
+out_destroy:

         if (rc < 0)
                 llog_destroy(loghandle);
 
         RETURN(loghandle);
 }
         if (rc < 0)
                 llog_destroy(loghandle);
 
         RETURN(loghandle);
 }

-EXPORT_SYMBOL(llog_cat_new_log);

 
 /* Open an existent log handle and add it to the open list.
  * This log handle will be closed when all of the records in it are removed.
 
 /* Open an existent log handle and add it to the open list.
  * This log handle will be closed when all of the records in it are removed.


@@ -138,8 +152,8 @@ int llog_cat_id2handle(struct llog_handle *cathandle, struct llog_handle **res,
         if (cathandle == NULL)
                 RETURN(-EBADF);
 
         if (cathandle == NULL)
                 RETURN(-EBADF);
 

-        list_for_each_entry(loghandle, &cathandle->u.chd.chd_head,
-                            u.phd.phd_entry) {
+        cfs_list_for_each_entry(loghandle, &cathandle->u.chd.chd_head,
+                                u.phd.phd_entry) {

                 struct llog_logid *cgl = &loghandle->lgh_id;
                 if (cgl->lgl_oid == logid->lgl_oid) {
                         if (cgl->lgl_ogen != logid->lgl_ogen) {
                 struct llog_logid *cgl = &loghandle->lgh_id;
                 if (cgl->lgl_oid == logid->lgl_oid) {
                         if (cgl->lgl_ogen != logid->lgl_ogen) {


@@ -160,14 +174,14 @@ int llog_cat_id2handle(struct llog_handle *cathandle, struct llog_handle **res,
         } else {
                 rc = llog_init_handle(loghandle, LLOG_F_IS_PLAIN, NULL);
                 if (!rc) {
         } else {
                 rc = llog_init_handle(loghandle, LLOG_F_IS_PLAIN, NULL);
                 if (!rc) {

-                        list_add(&loghandle->u.phd.phd_entry,
-                                 &cathandle->u.chd.chd_head);
+                        cfs_list_add(&loghandle->u.phd.phd_entry,
+                                     &cathandle->u.chd.chd_head);

                 }
         }
         if (!rc) {
                 loghandle->u.phd.phd_cat_handle = cathandle;
                 loghandle->u.phd.phd_cookie.lgc_lgl = cathandle->lgh_id;
                 }
         }
         if (!rc) {
                 loghandle->u.phd.phd_cat_handle = cathandle;
                 loghandle->u.phd.phd_cookie.lgc_lgl = cathandle->lgh_id;

-                loghandle->u.phd.phd_cookie.lgc_index = 
+                loghandle->u.phd.phd_cookie.lgc_index =

                         loghandle->lgh_hdr->llh_cat_idx;
         }
 
                         loghandle->lgh_hdr->llh_cat_idx;
         }
 


@@ -182,8 +196,8 @@ int llog_cat_put(struct llog_handle *cathandle)
         int rc;
         ENTRY;
 
         int rc;
         ENTRY;
 

-        list_for_each_entry_safe(loghandle, n, &cathandle->u.chd.chd_head,
-                                 u.phd.phd_entry) {
+        cfs_list_for_each_entry_safe(loghandle, n, &cathandle->u.chd.chd_head,
+                                     u.phd.phd_entry) {

                 int err = llog_close(loghandle);
                 if (err)
                         CERROR("error closing loghandle\n");
                 int err = llog_close(loghandle);
                 if (err)
                         CERROR("error closing loghandle\n");


@@ -193,7 +207,15 @@ int llog_cat_put(struct llog_handle *cathandle)
 }
 EXPORT_SYMBOL(llog_cat_put);
 
 }
 EXPORT_SYMBOL(llog_cat_put);
 

-/* Return the currently active log handle.  If the current log handle doesn't
+/**
+ * lockdep markers for nested struct llog_handle::lgh_lock locking.
+ */
+enum {
+        LLOGH_CAT,
+        LLOGH_LOG
+};
+
+/** Return the currently active log handle.  If the current log handle doesn't

  * have enough space left for the current record, start a new one.
  *
  * If reclen is 0, we only want to know what the currently active log is,
  * have enough space left for the current record, start a new one.
  *
  * If reclen is 0, we only want to know what the currently active log is,


@@ -209,47 +231,47 @@ static struct llog_handle *llog_cat_current_log(struct llog_handle *cathandle,
         struct llog_handle *loghandle = NULL;
         ENTRY;
 
         struct llog_handle *loghandle = NULL;
         ENTRY;
 

-        down_read(&cathandle->lgh_lock);
+        cfs_down_read_nested(&cathandle->lgh_lock, LLOGH_CAT);

         loghandle = cathandle->u.chd.chd_current_log;
         if (loghandle) {
                 struct llog_log_hdr *llh = loghandle->lgh_hdr;
         loghandle = cathandle->u.chd.chd_current_log;
         if (loghandle) {
                 struct llog_log_hdr *llh = loghandle->lgh_hdr;

-                down_write(&loghandle->lgh_lock);
+                cfs_down_write_nested(&loghandle->lgh_lock, LLOGH_LOG);

                 if (loghandle->lgh_last_idx < LLOG_BITMAP_SIZE(llh) - 1) {
                 if (loghandle->lgh_last_idx < LLOG_BITMAP_SIZE(llh) - 1) {

-                        up_read(&cathandle->lgh_lock);
+                        cfs_up_read(&cathandle->lgh_lock);

                         RETURN(loghandle);
                 } else {
                         RETURN(loghandle);
                 } else {

-                        up_write(&loghandle->lgh_lock);
+                        cfs_up_write(&loghandle->lgh_lock);

                 }
         }
         if (!create) {
                 if (loghandle)
                 }
         }
         if (!create) {
                 if (loghandle)

-                        down_write(&loghandle->lgh_lock);
-                up_read(&cathandle->lgh_lock);
+                        cfs_down_write(&loghandle->lgh_lock);
+                cfs_up_read(&cathandle->lgh_lock);

                 RETURN(loghandle);
         }
                 RETURN(loghandle);
         }

-        up_read(&cathandle->lgh_lock);
+        cfs_up_read(&cathandle->lgh_lock);

 
         /* time to create new log */
 
         /* first, we have to make sure the state hasn't changed */
 
         /* time to create new log */
 
         /* first, we have to make sure the state hasn't changed */

-        down_write(&cathandle->lgh_lock);
+        cfs_down_write_nested(&cathandle->lgh_lock, LLOGH_CAT);

         loghandle = cathandle->u.chd.chd_current_log;
         if (loghandle) {
                 struct llog_log_hdr *llh = loghandle->lgh_hdr;
         loghandle = cathandle->u.chd.chd_current_log;
         if (loghandle) {
                 struct llog_log_hdr *llh = loghandle->lgh_hdr;

-                down_write(&loghandle->lgh_lock);
+                cfs_down_write_nested(&loghandle->lgh_lock, LLOGH_LOG);

                 if (loghandle->lgh_last_idx < LLOG_BITMAP_SIZE(llh) - 1) {
                 if (loghandle->lgh_last_idx < LLOG_BITMAP_SIZE(llh) - 1) {

-                        up_write(&cathandle->lgh_lock);
+                        cfs_up_write(&cathandle->lgh_lock);

                         RETURN(loghandle);
                 } else {
                         RETURN(loghandle);
                 } else {

-                        up_write(&loghandle->lgh_lock);
+                        cfs_up_write(&loghandle->lgh_lock);

                 }
         }
 
         CDEBUG(D_INODE, "creating new log\n");
         loghandle = llog_cat_new_log(cathandle);
         if (!IS_ERR(loghandle))
                 }
         }
 
         CDEBUG(D_INODE, "creating new log\n");
         loghandle = llog_cat_new_log(cathandle);
         if (!IS_ERR(loghandle))

-                down_write(&loghandle->lgh_lock);
-        up_write(&cathandle->lgh_lock);
+                cfs_down_write_nested(&loghandle->lgh_lock, LLOGH_LOG);
+        cfs_up_write(&cathandle->lgh_lock);

         RETURN(loghandle);
 }
 
         RETURN(loghandle);
 }
 


@@ -271,14 +293,16 @@ int llog_cat_add_rec(struct llog_handle *cathandle, struct llog_rec_hdr *rec,
                 RETURN(PTR_ERR(loghandle));
         /* loghandle is already locked by llog_cat_current_log() for us */
         rc = llog_write_rec(loghandle, rec, reccookie, 1, buf, -1);
                 RETURN(PTR_ERR(loghandle));
         /* loghandle is already locked by llog_cat_current_log() for us */
         rc = llog_write_rec(loghandle, rec, reccookie, 1, buf, -1);

-        up_write(&loghandle->lgh_lock);
+        if (rc < 0)
+                CERROR("llog_write_rec %d: lh=%p\n", rc, loghandle);
+        cfs_up_write(&loghandle->lgh_lock);

         if (rc == -ENOSPC) {
                 /* to create a new plain log */
                 loghandle = llog_cat_current_log(cathandle, 1);
                 if (IS_ERR(loghandle))
                         RETURN(PTR_ERR(loghandle));
                 rc = llog_write_rec(loghandle, rec, reccookie, 1, buf, -1);
         if (rc == -ENOSPC) {
                 /* to create a new plain log */
                 loghandle = llog_cat_current_log(cathandle, 1);
                 if (IS_ERR(loghandle))
                         RETURN(PTR_ERR(loghandle));
                 rc = llog_write_rec(loghandle, rec, reccookie, 1, buf, -1);

-                up_write(&loghandle->lgh_lock);
+                cfs_up_write(&loghandle->lgh_lock);

         }
 
         RETURN(rc);
         }
 
         RETURN(rc);


@@ -295,12 +319,12 @@ EXPORT_SYMBOL(llog_cat_add_rec);
  * Assumes caller has already pushed us into the kernel context.
  */
 int llog_cat_cancel_records(struct llog_handle *cathandle, int count,
  * Assumes caller has already pushed us into the kernel context.
  */
 int llog_cat_cancel_records(struct llog_handle *cathandle, int count,

-                        struct llog_cookie *cookies)
+                            struct llog_cookie *cookies)

 {
         int i, index, rc = 0;
         ENTRY;
 
 {
         int i, index, rc = 0;
         ENTRY;
 

-        down_write(&cathandle->lgh_lock);
+        cfs_down_write_nested(&cathandle->lgh_lock, LLOGH_CAT);

         for (i = 0; i < count; i++, cookies++) {
                 struct llog_handle *loghandle;
                 struct llog_logid *lgl = &cookies->lgc_lgl;
         for (i = 0; i < count; i++, cookies++) {
                 struct llog_handle *loghandle;
                 struct llog_logid *lgl = &cookies->lgc_lgl;


@@ -311,9 +335,9 @@ int llog_cat_cancel_records(struct llog_handle *cathandle, int count,
                         break;
                 }
 
                         break;
                 }
 

-                down_write(&loghandle->lgh_lock);
+                cfs_down_write_nested(&loghandle->lgh_lock, LLOGH_LOG);

                 rc = llog_cancel_rec(loghandle, cookies->lgc_index);
                 rc = llog_cancel_rec(loghandle, cookies->lgc_index);

-                up_write(&loghandle->lgh_lock);
+                cfs_up_write(&loghandle->lgh_lock);

 
                 if (rc == 1) {          /* log has been destroyed */
                         index = loghandle->u.phd.phd_cookie.lgc_index;
 
                 if (rc == 1) {          /* log has been destroyed */
                         index = loghandle->u.phd.phd_cookie.lgc_index;


@@ -325,12 +349,12 @@ int llog_cat_cancel_records(struct llog_handle *cathandle, int count,
                         llog_cat_set_first_idx(cathandle, index);
                         rc = llog_cancel_rec(cathandle, index);
                         if (rc == 0)
                         llog_cat_set_first_idx(cathandle, index);
                         rc = llog_cancel_rec(cathandle, index);
                         if (rc == 0)

-                                CDEBUG(D_HA, "cancel plain log at index %u "
-                                       "of catalog "LPX64"\n",
+                                CDEBUG(D_RPCTRACE,"cancel plain log at index %u"
+                                       " of catalog "LPX64"\n",

                                        index, cathandle->lgh_id.lgl_oid);
                 }
         }
                                        index, cathandle->lgh_id.lgl_oid);
                 }
         }

-        up_write(&cathandle->lgh_lock);
+        cfs_up_write(&cathandle->lgh_lock);

 
         RETURN(rc);
 }
 
         RETURN(rc);
 }


@@ -360,14 +384,31 @@ int llog_cat_process_cb(struct llog_handle *cat_llh, struct llog_rec_hdr *rec,
                 RETURN(rc);
         }
 
                 RETURN(rc);
         }
 

-        rc = llog_process(llh, d->lpd_cb, d->lpd_data, NULL);
+        if (rec->lrh_index < d->lpd_startcat)
+                /* Skip processing of the logs until startcat */
+                RETURN(0);
+
+        if (d->lpd_startidx > 0) {
+                struct llog_process_cat_data cd;
+
+                cd.lpcd_first_idx = d->lpd_startidx;
+                cd.lpcd_last_idx = 0;
+                rc = llog_process_flags(llh, d->lpd_cb, d->lpd_data, &cd,
+                                        d->lpd_flags);
+                /* Continue processing the next log from idx 0 */
+                d->lpd_startidx = 0;
+        } else {
+                rc = llog_process_flags(llh, d->lpd_cb, d->lpd_data, NULL,
+                                        d->lpd_flags);
+        }
+

         RETURN(rc);
 }
 
         RETURN(rc);
 }
 

-int llog_cat_process(struct llog_handle *cat_llh, llog_cb_t cb, void *data)
+int llog_cat_process_flags(struct llog_handle *cat_llh, llog_cb_t cb,
+                           void *data, int flags, int startcat, int startidx)

 {
         struct llog_process_data d;
 {
         struct llog_process_data d;

-        struct llog_process_cat_data cd;

         struct llog_log_hdr *llh = cat_llh->lgh_hdr;
         int rc;
         ENTRY;
         struct llog_log_hdr *llh = cat_llh->lgh_hdr;
         int rc;
         ENTRY;


@@ -375,28 +416,94 @@ int llog_cat_process(struct llog_handle *cat_llh, llog_cb_t cb, void *data)
         LASSERT(llh->llh_flags & LLOG_F_IS_CAT);
         d.lpd_data = data;
         d.lpd_cb = cb;
         LASSERT(llh->llh_flags & LLOG_F_IS_CAT);
         d.lpd_data = data;
         d.lpd_cb = cb;

+        d.lpd_startcat = startcat;
+        d.lpd_startidx = startidx;
+        d.lpd_flags = flags;

 
         if (llh->llh_cat_idx > cat_llh->lgh_last_idx) {
 
         if (llh->llh_cat_idx > cat_llh->lgh_last_idx) {

+                struct llog_process_cat_data cd;
+

                 CWARN("catlog "LPX64" crosses index zero\n",
                       cat_llh->lgh_id.lgl_oid);
 
                 CWARN("catlog "LPX64" crosses index zero\n",
                       cat_llh->lgh_id.lgl_oid);
 

-                cd.first_idx = llh->llh_cat_idx;
-                cd.last_idx = 0;
-                rc = llog_process(cat_llh, llog_cat_process_cb, &d, &cd);
+                cd.lpcd_first_idx = llh->llh_cat_idx;
+                cd.lpcd_last_idx = 0;
+                rc = llog_process_flags(cat_llh, llog_cat_process_cb, &d, &cd,
+                                        flags);

                 if (rc != 0)
                         RETURN(rc);
 
                 if (rc != 0)
                         RETURN(rc);
 

-                cd.first_idx = 0;
-                cd.last_idx = cat_llh->lgh_last_idx;
-                rc = llog_process(cat_llh, llog_cat_process_cb, &d, &cd);
+                cd.lpcd_first_idx = 0;
+                cd.lpcd_last_idx = cat_llh->lgh_last_idx;
+                rc = llog_process_flags(cat_llh, llog_cat_process_cb, &d, &cd,
+                                        flags);

         } else {
         } else {

-                rc = llog_process(cat_llh, llog_cat_process_cb, &d, NULL);
+                rc = llog_process_flags(cat_llh, llog_cat_process_cb, &d, NULL,
+                                        flags);

         }
 
         RETURN(rc);
 }
         }
 
         RETURN(rc);
 }

+EXPORT_SYMBOL(llog_cat_process_flags);
+
+int llog_cat_process(struct llog_handle *cat_llh, llog_cb_t cb, void *data,
+                     int startcat, int startidx)
+{
+        return llog_cat_process_flags(cat_llh, cb, data, 0, startcat, startidx);
+}

 EXPORT_SYMBOL(llog_cat_process);
 
 EXPORT_SYMBOL(llog_cat_process);
 

+#ifdef __KERNEL__
+int llog_cat_process_thread(void *data)
+{
+        struct llog_process_cat_args *args = data;
+        struct llog_ctxt *ctxt = args->lpca_ctxt;
+        struct llog_handle *llh = NULL;
+        llog_cb_t cb = args->lpca_cb;
+        struct llog_logid logid;
+        int rc;
+        ENTRY;
+
+        cfs_daemonize_ctxt("ll_log_process");
+
+        logid = *(struct llog_logid *)(args->lpca_arg);
+        rc = llog_create(ctxt, &llh, &logid, NULL);
+        if (rc) {
+                CERROR("llog_create() failed %d\n", rc);
+                GOTO(out, rc);
+        }
+        rc = llog_init_handle(llh, LLOG_F_IS_CAT, NULL);
+        if (rc) {
+                CERROR("llog_init_handle failed %d\n", rc);
+                GOTO(release_llh, rc);
+        }
+
+        if (cb) {
+                rc = llog_cat_process(llh, cb, NULL, 0, 0);
+                if (rc != LLOG_PROC_BREAK && rc != 0)
+                        CERROR("llog_cat_process() failed %d\n", rc);
+                cb(llh, NULL, NULL);
+        } else {
+                CWARN("No callback function for recovery\n");
+        }
+
+        /*
+         * Make sure that all cached data is sent.
+         */
+       llog_sync(ctxt, NULL, 0);
+        GOTO(release_llh, rc);
+release_llh:
+        rc = llog_cat_put(llh);
+        if (rc)
+                CERROR("llog_cat_put() failed %d\n", rc);
+out:
+        llog_ctxt_put(ctxt);
+        OBD_FREE_PTR(args);
+        return rc;
+}
+EXPORT_SYMBOL(llog_cat_process_thread);
+#endif
+

 static int llog_cat_reverse_process_cb(struct llog_handle *cat_llh,
                                        struct llog_rec_hdr *rec, void *data)
 {
 static int llog_cat_reverse_process_cb(struct llog_handle *cat_llh,
                                        struct llog_rec_hdr *rec, void *data)
 {


@@ -441,15 +548,15 @@ int llog_cat_reverse_process(struct llog_handle *cat_llh,
                 CWARN("catalog "LPX64" crosses index zero\n",
                       cat_llh->lgh_id.lgl_oid);
 
                 CWARN("catalog "LPX64" crosses index zero\n",
                       cat_llh->lgh_id.lgl_oid);
 

-                cd.first_idx = 0;
-                cd.last_idx = cat_llh->lgh_last_idx;
+                cd.lpcd_first_idx = 0;
+                cd.lpcd_last_idx = cat_llh->lgh_last_idx;

                 rc = llog_reverse_process(cat_llh, llog_cat_reverse_process_cb,
                                           &d, &cd);
                 if (rc != 0)
                         RETURN(rc);
 
                 rc = llog_reverse_process(cat_llh, llog_cat_reverse_process_cb,
                                           &d, &cd);
                 if (rc != 0)
                         RETURN(rc);
 

-                cd.first_idx = le32_to_cpu(llh->llh_cat_idx);
-                cd.last_idx = 0;
+                cd.lpcd_first_idx = le32_to_cpu(llh->llh_cat_idx);
+                cd.lpcd_last_idx = 0;

                 rc = llog_reverse_process(cat_llh, llog_cat_reverse_process_cb,
                                           &d, &cd);
         } else {
                 rc = llog_reverse_process(cat_llh, llog_cat_reverse_process_cb,
                                           &d, &cd);
         } else {


@@ -486,55 +593,9 @@ int llog_cat_set_first_idx(struct llog_handle *cathandle, int index)
                         }
                 }
 out:
                         }
                 }
 out:

-                CDEBUG(D_HA, "set catlog "LPX64" first idx %u\n",
+                CDEBUG(D_RPCTRACE, "set catlog "LPX64" first idx %u\n",

                        cathandle->lgh_id.lgl_oid, llh->llh_cat_idx);
         }
 
         RETURN(0);
 }
                        cathandle->lgh_id.lgl_oid, llh->llh_cat_idx);
         }
 
         RETURN(0);
 }

-
-#if 0
-/* Assumes caller has already pushed us into the kernel context. */
-int llog_cat_init(struct llog_handle *cathandle, struct obd_uuid *tgtuuid)
-{
-        struct llog_log_hdr *llh;
-        loff_t offset = 0;
-        int rc = 0;
-        ENTRY;
-
-        LASSERT(sizeof(*llh) == LLOG_CHUNK_SIZE);
-
-        down(&cathandle->lgh_lock);
-        llh = cathandle->lgh_hdr;
-
-        if (cathandle->lgh_file->f_dentry->d_inode->i_size == 0) {
-                llog_write_rec(cathandle, &llh->llh_hdr, NULL, 0, NULL, 0);
-
-write_hdr:
-                rc = lustre_fwrite(cathandle->lgh_file, llh, LLOG_CHUNK_SIZE,
-                                   &offset);
-                if (rc != LLOG_CHUNK_SIZE) {
-                        CERROR("error writing catalog header: rc %d\n", rc);
-                        OBD_FREE(llh, sizeof(*llh));
-                        if (rc >= 0)
-                                rc = -ENOSPC;
-                } else
-                        rc = 0;
-        } else {
-                rc = lustre_fread(cathandle->lgh_file, llh, LLOG_CHUNK_SIZE,
-                                  &offset);
-                if (rc != LLOG_CHUNK_SIZE) {
-                        CERROR("error reading catalog header: rc %d\n", rc);
-                        /* Can we do much else if the header is bad? */
-                        goto write_hdr;
-                } else
-                        rc = 0;
-        }
-
-        cathandle->lgh_tgtuuid = &llh->llh_tgtuuid;
-        up(&cathandle->lgh_lock);
-        RETURN(rc);
-}
-EXPORT_SYMBOL(llog_cat_init);
-
-#endif