b=21571 stacksize and locking fixes for loadgen patch from umka

[fs/lustre-release.git] / lustre / obdfilter / filter_log.c

diff --git a/lustre/obdfilter/filter_log.c b/lustre/obdfilter/filter_log.c
index 26c8399..a8caad3 100644 (file)
--- a/lustre/obdfilter/filter_log.c
+++ b/lustre/obdfilter/filter_log.c
@@ -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",
@@ -126,11 +126,11 @@ void filter_cancel_cookies_cb(struct obd_device *obd, __u64 transno,
         }
 
         olg = filter_find_olg(obd, cookie->lgc_lgl.lgl_ogr);
-        if (!olg) { 
+        if (!olg) {
                 CDEBUG(D_HA, "unknown group "LPU64"!\n", cookie->lgc_lgl.lgl_ogr);
                 GOTO(out, rc = 0);
         }
-        
+
         ctxt = llog_group_get_ctxt(olg, cookie->lgc_subsys + 1);
         if (!ctxt) {
                 CERROR("no valid context for group "LPU64"\n",
@@ -154,35 +154,41 @@ 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_gr = lur->lur_ogr;
         oa->o_valid = OBD_MD_FLID | OBD_MD_FLGROUP;
         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;
+
+        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++;
         }
-
-        if (rc == 0)
-                CDEBUG(D_RPCTRACE, "object "LPU64" is destroyed\n", oid);
+        OBDO_FREE(oa);
 
         RETURN(rc);
 }
@@ -195,24 +201,34 @@ 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_gr = lsr->lsr_ogr;
+                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_gr = lsr->lsr_ogr;
+                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;
         oinfo.oi_oa->o_lcookie = *cookie;
         oid = oinfo.oi_oa->o_id;
 
@@ -232,7 +248,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 +258,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) {
+                spin_lock_bh(&ctxt->loc_obd->obd_processing_task_lock);
+                ctxt->loc_obd->u.filter.fo_mds_ost_sync = 0;
+                spin_unlock_bh(&ctxt->loc_obd->obd_processing_task_lock);
+                RETURN(0);
+        }
+
         if (!(llh->lgh_hdr->llh_flags & LLOG_F_IS_PLAIN)) {
                 CERROR("log is not plain\n");
                 RETURN(-EINVAL);
@@ -257,6 +280,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: {