More for logging:

- Store recovery status in export, and check that, so we can have a
  non-recovery connection to the obdfilter and not do logging, transnos, etc.
- Split up filter_setup() so that it can be set up by the MDS on an already
  mounted filesystem
- Rename obdfilter "last_rcvd" file to be "filter_data" to avoid clashing on
  an MDS+obdfilter setup
- Pass struct obd_trans_info around, and put the journal handle in that,
  instead of using desc_private (for journal nesting).

Replace "force" and "failover" with "flags" + OBD_OPT_FORCE, OBD_OPT_FAILOVER.

diff --git a/lustre/include/linux/lustre_log.h b/lustre/include/linux/lustre_log.h
index fb24084..1e206d2 100644 (file)
--- a/lustre/include/linux/lustre_log.h
+++ b/lustre/include/linux/lustre_log.h
@@ -46,7 +46,7 @@
 /* Identifier for a single log object */
 struct llog_logid {
         __u64                   lgl_oid;
-        __u64                   lgl_bootcount;
+//      __u64                   lgl_bootcount;
 };
 
 /* On-disk header structure of catalog of available log object (internal) */
@@ -150,6 +150,7 @@ extern int llog_clear_record(struct lustre_handle *conn, __u32 recno);
 extern int llog_delete(struct lustre_handle *conn, struct llog_logid *id);
 
 /* internal api */
+extern int llog_init_catalog(struct lustre_handle *conn);
 extern struct llog_handle *llog_id2handle(struct lustre_handle *conn,
                                           struct llog_logid *logid);
 

diff --git a/lustre/lib/recov_log.c b/lustre/lib/recov_log.c
index de95ee2..c7f5100 100644 (file)
--- a/lustre/lib/recov_log.c
+++ b/lustre/lib/recov_log.c
@@ -25,9 +25,8 @@
 #include <linux/obd.h>
 #include <linux/lustre_log.h>
 
-/* Create a new log or catalog handle */
-static struct log_handle *llog_new_handle(struct lustre_handle *conn,
-                                          struct obd_trans_info *oti)
+/* Allocate a new log or catalog handle */
+struct log_handle *llog_alloc_handle(void)
 {
         struct llog_handle *loghandle;
         int rc;
@@ -46,33 +45,14 @@ static struct log_handle *llog_new_handle(struct lustre_handle *conn,
         if (loghandle->lgh_pga[1].pg == NULL)
                 GOTO(out_pga1, rc = -ENOMEM);
         loghandle->lgh_pga[0].off = LLOG_HEADER_SIZE;
+        LIST_HEAD_INIT(&loghandle->lgh_list);
 
         obdo_alloc(loghandle->lgh_oa);
         if (!loghandle->lgh_oa)
                 GOTO(out_pga2, rc = -ENOMEM);
 
-        rc = obd_create(conn, loghandle->lgh_oa, loghandle->lgh_lsm, oti)
-        if (rc) {
-                CERROR("couldn't create new log object: rc %d\n", rc);
-                GOTO(out_oa, rc);
-        }
-
-        rc = obd_open(conn, loghandle->lgh_oa, loghandle->lgh_lsm, oti, NULL);
-        if (rc) {
-                CERROR("couldn't open new log object "LPX64": rc %d\n",
-                       loghandle->lgh_oa->o_id, rc);
-                GOTO(out_destroy, rc);
-        }
-        LIST_HEAD_INIT(&loghandle->lgh_list);
-        loghandle->lgh_lid.lid_oid = oa->o_id;
-        //loghandle->lgh_lid.lid_bootcount = ????;
-
         RETURN(loghandle);
 
-out_destroy:
-        obd_destroy(conn, loghandle->lgh_oa, loghandle->lgh_lsm, oti);
-out_oa:
-        obd_free(loghandle->lgh_oa);
 out_pga2:
         __free_page(loghandle->lgh_pga[1].pg);
 out_pga1:
@@ -81,6 +61,17 @@ out:
         RETURN(ERR_PTR(rc));
 }
 
+void llog_free_handle(struct llog_handle *loghandle)
+{
+        if (!loghandle)
+                return;
+
+        list_del_init(&loghandle->lgh_list);
+        obdo_free(loghandle->lgh_oa);
+        __free_page(loghandle->lgh_pga[1].pg);
+        __free_page(loghandle->lgh_pga[0].pg);
+}
+
 /* Create a new log handle and add it to the open list.
  * This log handle will be closed when all of the records in it are removed.
  */
@@ -99,6 +90,21 @@ static struct llog_handle *llog_new_log(struct lustre_handle *conn,
         if (IS_ERR(loghandle))
                 RETURN(loghandle);
 
+        rc = obd_create(conn, loghandle->lgh_oa, loghandle->lgh_lsm, oti)
+        if (rc) {
+                CERROR("couldn't create new log object: rc %d\n", rc);
+                GOTO(out_handle, rc);
+        }
+
+        rc = obd_open(conn, loghandle->lgh_oa, loghandle->lgh_lsm, oti, NULL);
+        if (rc) {
+                CERROR("couldn't open new log object "LPX64": rc %d\n",
+                       loghandle->lgh_oa->o_id, rc);
+                GOTO(out_destroy, rc);
+        }
+        loghandle->lgh_lid.lid_oid = oa->o_id;
+        //loghandle->lgh_lid.lid_bootcount = ????;
+
         loh = kmap(loghandle->lgh_pga[0].pg);
         clear_page(loh);
         loh->loh_size = loh->loh_size_end = LLOG_HEADER_SIZE;
@@ -147,12 +153,21 @@ retry:
         rc = obd_brw(OBD_BRW_WRITE, conn, cathandle->lgh_lsm, num_pga,
                      cathandle->lgh_pga, NULL, oti);
         if (rc) {
+                CERROR("error adding log "LPX64" to catalog: rc %d\n",
+                       loghandle->lgh_lid.lid_oid, rc);
+                /* XXX Hmm, what to do?  Under normal ops we would clean this
+                 *     handle up anyways, and at worst we leak some objects.
+                 *     The only danger is if the OST crashes - the log is lost.
+                 */
+        }
         list_add_tail(&loghandle->lgh_list, &cathandle->lgh_list);
 
         RETURN(0);
 
+out_destroy:
+        obd_destroy(conn, loghandle->lgh_oa, loghandle->lgh_lsm, oti);
 out_handle:
-        OBD_FREE(loghandle, sizeof(*loghandle));
+        llog_free(handle);
 
         RETURN(rc);
 }
@@ -184,8 +199,8 @@ int llog_init_catalog(struct lustre_handle *conn, struct obd_trans_info *oti)
 /* We start a new log object here if needed, either because no log has been
  * started, or because the current log cannot fit the new record.
  */
-int llog_get_log(struct lustre_handle *conn, int reclen,
-                 struct obd_trans_info *oti)
+int llog_current_log(struct lustre_handle *conn, int reclen,
+                     struct obd_trans_info *oti)
 {
         struct obd_device *obd = class_conn2obd(conn);
         struct list_head *loglist = &obd->obd_catalog->lgh_list;
@@ -220,7 +235,7 @@ int llog_add_record(struct lustre_handle *conn, struct llog_trans_hdr *rec,
         int rc;
         ENTRY;
 
-        loghandle = llog_get_log(conn, reclen, transhandle);
+        loghandle = llog_current_log(conn, reclen, oti);
 
         offset = loghandle->lgh_pga[1].off;
 
@@ -262,6 +277,46 @@ int llog_add_record(struct lustre_handle *conn, struct llog_trans_hdr *rec,
         RETURN(0);
 }
 
-int llog_clear_records(int count, struct llog_cookie **cookies)
+struct loghandle *llog_id2handle(struct lustre_handle *conn,
+                                 struct llog_logid *lid)
+{
+        struct obd_device *obd = class_conn2obd(conn);
+        struct llog_handle *cathandle = obd->obd_catalog;
+        struct llog_handle *loghandle = NULL, *tmp;
+        ENTRY;
+
+        if (cathandle == NULL)
+                RETURN(NULL);
+
+        list_for_each_entry(tmp, &cathandle->lgh_list, lgh_list) {
+                if (tmp->lgh_lid == lid) {
+                        loghandle = tmp;
+                        break;
+                }
+        }
+
+        if (loghandle == NULL) {
+                obd_open(
+        RETURN(loghandle);
+}
+
+int llog_clear_records(struct lustre_handle *conn, int count,
+                       struct llog_cookie *cookies)
+{
+        int i;
+
+        for (i = 0; i < count; i++) {
+                struct llog_handle *loghandle;
+                loghandle = llog_id2handle(conn, cookies[i].lgc_lid);
+
+}
+
 int llog_clear_record(struct llog_handle *handle, __u32 recno)
+{
+        return 0;
+}
+
 int llog_delete(struct llog_logid *id)
+{
+        return 0;
+}