LU-3469 osp: osp_sync() to flush pending changes synchronously

[fs/lustre-release.git] / lustre / osp / osp_sync.c

diff --git a/lustre/osp/osp_sync.c b/lustre/osp/osp_sync.c
index 388c80e..34a0e2c 100644 (file)
--- a/lustre/osp/osp_sync.c
+++ b/lustre/osp/osp_sync.c
@@ -151,6 +151,8 @@ static inline int osp_sync_can_process_new(struct osp_device *d,
 {
        LASSERT(d);
 
+       if (unlikely(atomic_read(&d->opd_syn_barrier) > 0))
+               return 0;
        if (!osp_sync_low_in_progress(d))
                return 0;
        if (!osp_sync_low_in_flight(d))
@@ -235,6 +237,7 @@ static int osp_sync_add_rec(const struct lu_env *env, struct osp_device *d,
                LASSERT(attr);
                osi->osi_setattr.lsr_uid = attr->la_uid;
                osi->osi_setattr.lsr_gid = attr->la_gid;
+               osi->osi_setattr.lsr_valid = attr->la_valid;
                break;
        default:
                LBUG();
@@ -398,6 +401,8 @@ static int osp_sync_interpret(const struct lu_env *env,
        spin_lock(&d->opd_syn_lock);
        d->opd_syn_rpc_in_flight--;
        spin_unlock(&d->opd_syn_lock);
+       if (unlikely(atomic_read(&d->opd_syn_barrier) > 0))
+               wake_up(&d->opd_syn_barrier_waitq);
        CDEBUG(D_OTHER, "%s: %d in flight, %d in progress\n",
               d->opd_obd->obd_name, d->opd_syn_rpc_in_flight,
               d->opd_syn_rpc_in_progress);
@@ -478,6 +483,13 @@ static int osp_sync_new_setattr_job(struct osp_device *d,
        ENTRY;
        LASSERT(h->lrh_type == MDS_SETATTR64_REC);
 
+       /* lsr_valid can only be 0 or LA_UID/GID set */
+       if (!rec->lsr_valid && !(rec->lsr_valid & ~(LA_UID | LA_GID))) {
+               CERROR("%s: invalid setattr record, lsr_valid:"LPU64"\n",
+                      d->opd_obd->obd_name, rec->lsr_valid);
+               RETURN(-EINVAL);
+       }
+
        req = osp_sync_new_job(d, llh, h, OST_SETATTR, &RQF_OST_SETATTR);
        if (IS_ERR(req))
                RETURN(PTR_ERR(req));
@@ -487,8 +499,17 @@ static int osp_sync_new_setattr_job(struct osp_device *d,
        body->oa.o_oi = rec->lsr_oi;
        body->oa.o_uid = rec->lsr_uid;
        body->oa.o_gid = rec->lsr_gid;
-       body->oa.o_valid = OBD_MD_FLGROUP | OBD_MD_FLID |
-                          OBD_MD_FLUID | OBD_MD_FLGID;
+       body->oa.o_valid = OBD_MD_FLGROUP | OBD_MD_FLID;
+       /* old setattr record (prior 2.6.0) doesn't have 'valid' stored,
+        * we assume that both UID and GID are valid in that case. */
+       if (rec->lsr_valid == 0) {
+               body->oa.o_valid |= (OBD_MD_FLUID | OBD_MD_FLGID);
+       } else {
+               if (rec->lsr_valid & LA_UID)
+                       body->oa.o_valid |= OBD_MD_FLUID;
+               if (rec->lsr_valid & LA_GID)
+                       body->oa.o_valid |= OBD_MD_FLGID;
+       }
 
        osp_sync_send_new_rpc(d, req);
        RETURN(0);
@@ -692,8 +713,10 @@ static int osp_sync_process_record(const struct lu_env *env,
                         * NOTE: it's possible to meet same id if
                         * OST stores few stripes of same file
                         */
-                       if (rec->lrh_id > d->opd_syn_last_processed_id)
+                       if (rec->lrh_id > d->opd_syn_last_processed_id) {
                                d->opd_syn_last_processed_id = rec->lrh_id;
+                               wake_up(&d->opd_syn_barrier_waitq);
+                       }
 
                        d->opd_syn_changes--;
                }
@@ -1131,6 +1154,7 @@ int osp_sync_init(const struct lu_env *env, struct osp_device *d)
        d->opd_syn_max_rpc_in_progress = OSP_MAX_IN_PROGRESS;
        spin_lock_init(&d->opd_syn_lock);
        init_waitqueue_head(&d->opd_syn_waitq);
+       init_waitqueue_head(&d->opd_syn_barrier_waitq);
        init_waitqueue_head(&d->opd_syn_thread.t_ctl_waitq);
        INIT_LIST_HEAD(&d->opd_syn_committed_there);