X-Git-Url: https://git.whamcloud.com/?a=blobdiff_plain;f=lustre%2Fobdfilter%2Ffilter_log.c;h=7ae5f8866c62645d92191f9f8fcccad513116db1;hb=bd12b1a4fc839c789630eea92fb816b453f2c503;hp=b2c42d1144d8f1ff86d41306cfdba231c97c8c32;hpb=6869932b552ac705f411de3362f01bd50c1f6f7d;p=fs%2Flustre-release.git

diff --git a/lustre/obdfilter/filter_log.c b/lustre/obdfilter/filter_log.c
index b2c42d1..7ae5f88 100644
--- a/lustre/obdfilter/filter_log.c
+++ b/lustre/obdfilter/filter_log.c
@@ -26,7 +26,7 @@
  * GPL HEADER END
  */
 /*
- * Copyright  2008 Sun Microsystems, Inc. All rights reserved
+ * Copyright (c) 2003, 2010, Oracle and/or its affiliates. All rights reserved.
  * Use is subject to license terms.
  */
 /*
@@ -66,7 +66,7 @@ int filter_log_sz_change(struct llog_handle *cathandle,
         ENTRY;
 
         LOCK_INODE_MUTEX(inode);
-        ofd = inode->i_filterdata;
+        ofd = INODE_PRIVATE_DATA(inode);
 
         if (ofd && ofd->ofd_epoch >= ioepoch) {
                 if (ofd->ofd_epoch > ioepoch)
@@ -83,7 +83,7 @@ int filter_log_sz_change(struct llog_handle *cathandle,
                 if (!ofd)
                         GOTO(out, rc = -ENOMEM);
                 igrab(inode);
-                inode->i_filterdata = ofd;
+                INODE_PRIVATE_DATA(inode) = ofd;
                 ofd->ofd_epoch = ioepoch;
         }
         /* the decision to write a record is now made, unlock */
@@ -117,7 +117,7 @@ void filter_cancel_cookies_cb(struct obd_device *obd, __u64 transno,
         struct obd_llog_group *olg;
         struct llog_ctxt *ctxt;
         int rc;
-        
+
         /* we have to find context for right group */
         if (error != 0 || obd->obd_stopping) {
                 CDEBUG(D_INODE, "not cancel logcookie err %d stopping %d \n",
@@ -125,16 +125,16 @@ void filter_cancel_cookies_cb(struct obd_device *obd, __u64 transno,
                 GOTO (out, rc = 0);
         }
 
-        olg = filter_find_olg(obd, cookie->lgc_lgl.lgl_ogr);
-        if (!olg) { 
-                CDEBUG(D_HA, "unknown group "LPU64"!\n", cookie->lgc_lgl.lgl_ogr);
+        olg = filter_find_olg(obd, cookie->lgc_lgl.lgl_oseq);
+        if (!olg) {
+                CDEBUG(D_HA, "unknown group "LPU64"!\n", cookie->lgc_lgl.lgl_oseq);
                 GOTO(out, rc = 0);
         }
-        
+
         ctxt = llog_group_get_ctxt(olg, cookie->lgc_subsys + 1);
         if (!ctxt) {
                 CERROR("no valid context for group "LPU64"\n",
-                        cookie->lgc_lgl.lgl_ogr);
+                        cookie->lgc_lgl.lgl_oseq);
                 GOTO(out, rc = 0);
         }
 
@@ -148,41 +148,55 @@ out:
         OBD_FREE(cookie, sizeof(*cookie));
 }
 
-/* Callback for processing the unlink log record received from MDS by 
+/* Callback for processing the unlink log record received from MDS by
  * llog_client_api. */
 static int filter_recov_log_unlink_cb(struct llog_ctxt *ctxt,
                                       struct llog_rec_hdr *rec,
                                       struct llog_cookie *cookie)
 {
-        struct obd_device *obd = ctxt->loc_obd;
-        struct obd_export *exp = obd->obd_self_export;
+        struct obd_export *exp = ctxt->loc_obd->obd_self_export;
         struct llog_unlink_rec *lur;
         struct obdo *oa;
         obd_id oid;
+        obd_count count;
         int rc = 0;
         ENTRY;
 
         lur = (struct llog_unlink_rec *)rec;
         OBDO_ALLOC(oa);
-        if (oa == NULL) 
+        if (oa == NULL)
                 RETURN(-ENOMEM);
         oa->o_valid |= OBD_MD_FLCOOKIE;
         oa->o_id = lur->lur_oid;
-        oa->o_gr = lur->lur_ogen;
+        oa->o_seq = lur->lur_oseq;
         oa->o_valid = OBD_MD_FLID | OBD_MD_FLGROUP;
-        memcpy(obdo_logcookie(oa), cookie, sizeof(*cookie));
+        oa->o_lcookie = *cookie;
         oid = oa->o_id;
-
-        rc = filter_destroy(exp, oa, NULL, NULL, NULL);
-        OBDO_FREE(oa);
-        if (rc == -ENOENT) {
-                CDEBUG(D_RPCTRACE, "object already removed, send cookie\n");
-                llog_cancel(ctxt, NULL, 1, cookie, 0);
-                RETURN(0);
+        /* objid gap may require to destroy several objects in row */
+        count = lur->lur_count + 1;
+
+        /* This check is only valid before FID-on-OST and it should
+         * be removed after FID-on-OST is implemented */
+        if (oa->o_seq > FID_SEQ_OST_MAX) {
+                CERROR("%s: invalid group number "LPU64" > MAX_CMD_GROUP %u\n",
+                        exp->exp_obd->obd_name, oa->o_seq, FID_SEQ_OST_MAX);
+                RETURN(-EINVAL);
         }
 
-        if (rc == 0)
-                CDEBUG(D_RPCTRACE, "object "LPU64" is destroyed\n", oid);
+        while (count > 0) {
+                rc = filter_destroy(exp, oa, NULL, NULL, NULL, NULL);
+                if (rc == 0)
+                        CDEBUG(D_RPCTRACE, "object "LPU64" is destroyed\n",
+                               oid);
+                else if (rc != -ENOENT)
+                        CEMERG("error destroying object "LPU64": %d\n",
+                               oid, rc);
+                else
+                        rc = 0;
+                count--;
+                oid++;
+        }
+        OBDO_FREE(oa);
 
         RETURN(rc);
 }
@@ -195,25 +209,35 @@ static int filter_recov_log_setattr_cb(struct llog_ctxt *ctxt,
 {
         struct obd_device *obd = ctxt->loc_obd;
         struct obd_export *exp = obd->obd_self_export;
-        struct llog_setattr_rec *lsr;
         struct obd_info oinfo = { { { 0 } } };
         obd_id oid;
         int rc = 0;
         ENTRY;
 
-        lsr = (struct llog_setattr_rec *)rec;
         OBDO_ALLOC(oinfo.oi_oa);
         if (oinfo.oi_oa == NULL)
                 RETURN(-ENOMEM);
 
+        if (rec->lrh_type == MDS_SETATTR_REC) {
+                struct llog_setattr_rec *lsr = (struct llog_setattr_rec *)rec;
+
+                oinfo.oi_oa->o_id = lsr->lsr_oid;
+                oinfo.oi_oa->o_seq = lsr->lsr_oseq;
+                oinfo.oi_oa->o_uid = lsr->lsr_uid;
+                oinfo.oi_oa->o_gid = lsr->lsr_gid;
+        } else {
+                struct llog_setattr64_rec *lsr = (struct llog_setattr64_rec *)rec;
+
+                oinfo.oi_oa->o_id = lsr->lsr_oid;
+                oinfo.oi_oa->o_seq = lsr->lsr_oseq;
+                oinfo.oi_oa->o_uid = lsr->lsr_uid;
+                oinfo.oi_oa->o_gid = lsr->lsr_gid;
+        }
+
         oinfo.oi_oa->o_valid |= (OBD_MD_FLID | OBD_MD_FLUID | OBD_MD_FLGID |
                                  OBD_MD_FLCOOKIE);
-        oinfo.oi_oa->o_id = lsr->lsr_oid;
-        oinfo.oi_oa->o_gr = lsr->lsr_ogen;
         oinfo.oi_oa->o_valid = OBD_MD_FLID | OBD_MD_FLGROUP;
-        oinfo.oi_oa->o_uid = lsr->lsr_uid;
-        oinfo.oi_oa->o_gid = lsr->lsr_gid;
-        memcpy(obdo_logcookie(oinfo.oi_oa), cookie, sizeof(*cookie));
+        oinfo.oi_oa->o_lcookie = *cookie;
         oid = oinfo.oi_oa->o_id;
 
         rc = filter_setattr(exp, &oinfo, NULL);
@@ -232,7 +256,7 @@ static int filter_recov_log_setattr_cb(struct llog_ctxt *ctxt,
 }
 
 int filter_recov_log_mds_ost_cb(struct llog_handle *llh,
-                               struct llog_rec_hdr *rec, void *data)
+                                struct llog_rec_hdr *rec, void *data)
 {
         struct llog_ctxt *ctxt = llh->lgh_ctxt;
         struct llog_cookie cookie;
@@ -242,6 +266,13 @@ int filter_recov_log_mds_ost_cb(struct llog_handle *llh,
         if (ctxt->loc_obd->obd_stopping)
                 RETURN(LLOG_PROC_BREAK);
 
+        if (rec == NULL) {
+                cfs_spin_lock(&ctxt->loc_obd->u.filter.fo_flags_lock);
+                ctxt->loc_obd->u.filter.fo_mds_ost_sync = 0;
+                cfs_spin_unlock(&ctxt->loc_obd->u.filter.fo_flags_lock);
+                RETURN(0);
+        }
+
         if (!(llh->lgh_hdr->llh_flags & LLOG_F_IS_PLAIN)) {
                 CERROR("log is not plain\n");
                 RETURN(-EINVAL);
@@ -257,6 +288,7 @@ int filter_recov_log_mds_ost_cb(struct llog_handle *llh,
                 rc = filter_recov_log_unlink_cb(ctxt, rec, &cookie);
                 break;
         case MDS_SETATTR_REC:
+        case MDS_SETATTR64_REC:
                 rc = filter_recov_log_setattr_cb(ctxt, rec, &cookie);
                 break;
         case LLOG_GEN_REC: {