Whamcloud - gitweb
LU-3469 osp: osp_sync() to flush pending changes synchronously
[fs/lustre-release.git] / lustre / osp / osp_sync.c
index b216ece..34a0e2c 100644 (file)
@@ -128,7 +128,7 @@ static inline int osp_sync_has_work(struct osp_device *d)
                return 1;
 
        /* has remotely committed? */
-       if (!cfs_list_empty(&d->opd_syn_committed_there))
+       if (!list_empty(&d->opd_syn_committed_there))
                return 1;
 
        return 0;
@@ -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();
@@ -250,7 +253,7 @@ static int osp_sync_add_rec(const struct lu_env *env, struct osp_device *d,
        if (ctxt == NULL)
                RETURN(-ENOMEM);
        rc = llog_add(env, ctxt->loc_handle, &osi->osi_hdr, &osi->osi_cookie,
-                     NULL, th);
+                     th);
        llog_ctxt_put(ctxt);
 
        CDEBUG(D_OTHER, "%s: new record "DOSTID":%lu/%lu: %d\n",
@@ -321,12 +324,12 @@ static void osp_sync_request_commit_cb(struct ptlrpc_request *req)
 
        LASSERT(d);
        LASSERT(req->rq_svc_thread == (void *) OSP_JOB_MAGIC);
-       LASSERT(cfs_list_empty(&req->rq_exp_list));
+       LASSERT(list_empty(&req->rq_exp_list));
 
        ptlrpc_request_addref(req);
 
        spin_lock(&d->opd_syn_lock);
-       cfs_list_add(&req->rq_exp_list, &d->opd_syn_committed_there);
+       list_add(&req->rq_exp_list, &d->opd_syn_committed_there);
        spin_unlock(&d->opd_syn_lock);
 
        /* XXX: some batching wouldn't hurt */
@@ -344,7 +347,7 @@ static int osp_sync_interpret(const struct lu_env *env,
        LASSERT(d);
 
        CDEBUG(D_HA, "reply req %p/%d, rc %d, transno %u\n", req,
-              cfs_atomic_read(&req->rq_refcount),
+              atomic_read(&req->rq_refcount),
               rc, (unsigned) req->rq_transno);
        LASSERT(rc || req->rq_transno);
 
@@ -354,12 +357,12 @@ static int osp_sync_interpret(const struct lu_env *env,
                 * but object doesn't exist anymore - cancell llog record
                 */
                LASSERT(req->rq_transno == 0);
-               LASSERT(cfs_list_empty(&req->rq_exp_list));
+               LASSERT(list_empty(&req->rq_exp_list));
 
                ptlrpc_request_addref(req);
 
                spin_lock(&d->opd_syn_lock);
-               cfs_list_add(&req->rq_exp_list, &d->opd_syn_committed_there);
+               list_add(&req->rq_exp_list, &d->opd_syn_committed_there);
                spin_unlock(&d->opd_syn_lock);
 
                wake_up(&d->opd_syn_waitq);
@@ -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);
@@ -455,7 +460,7 @@ static struct ptlrpc_request *osp_sync_new_job(struct osp_device *d,
        body->oa.o_lcookie.lgc_lgl = llh->lgh_id;
        body->oa.o_lcookie.lgc_subsys = LLOG_MDS_OST_ORIG_CTXT;
        body->oa.o_lcookie.lgc_index = h->lrh_index;
-       CFS_INIT_LIST_HEAD(&req->rq_exp_list);
+       INIT_LIST_HEAD(&req->rq_exp_list);
        req->rq_svc_thread = (void *) OSP_JOB_MAGIC;
 
        req->rq_interpret_reply = osp_sync_interpret;
@@ -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--;
                }
@@ -722,12 +745,12 @@ static void osp_sync_process_committed(const struct lu_env *env,
        struct ptlrpc_request   *req, *tmp;
        struct llog_ctxt        *ctxt;
        struct llog_handle      *llh;
-       cfs_list_t               list;
+       struct list_head         list;
        int                      rc, done = 0;
 
        ENTRY;
 
-       if (cfs_list_empty(&d->opd_syn_committed_there))
+       if (list_empty(&d->opd_syn_committed_there))
                return;
 
        /*
@@ -752,17 +775,17 @@ static void osp_sync_process_committed(const struct lu_env *env,
        llh = ctxt->loc_handle;
        LASSERT(llh);
 
-       CFS_INIT_LIST_HEAD(&list);
+       INIT_LIST_HEAD(&list);
        spin_lock(&d->opd_syn_lock);
-       cfs_list_splice(&d->opd_syn_committed_there, &list);
-       CFS_INIT_LIST_HEAD(&d->opd_syn_committed_there);
+       list_splice(&d->opd_syn_committed_there, &list);
+       INIT_LIST_HEAD(&d->opd_syn_committed_there);
        spin_unlock(&d->opd_syn_lock);
 
-       cfs_list_for_each_entry_safe(req, tmp, &list, rq_exp_list) {
+       list_for_each_entry_safe(req, tmp, &list, rq_exp_list) {
                struct llog_cookie *lcookie = NULL;
 
                LASSERT(req->rq_svc_thread == (void *) OSP_JOB_MAGIC);
-               cfs_list_del_init(&req->rq_exp_list);
+               list_del_init(&req->rq_exp_list);
 
                if (d->opd_connect_mdt) {
                        struct object_update_request *ureq;
@@ -884,7 +907,7 @@ static int osp_sync_process_queues(const struct lu_env *env,
                l_wait_event(d->opd_syn_waitq,
                             !osp_sync_running(d) ||
                             osp_sync_can_process_new(d, rec) ||
-                            !cfs_list_empty(&d->opd_syn_committed_there),
+                            !list_empty(&d->opd_syn_committed_there),
                             &lwi);
        } while (1);
 }
@@ -966,7 +989,7 @@ static int osp_sync_thread(void *_arg)
                LASSERTF(count < 10, "%s: %d %d %sempty\n",
                         d->opd_obd->obd_name, d->opd_syn_rpc_in_progress,
                         d->opd_syn_rpc_in_flight,
-                        cfs_list_empty(&d->opd_syn_committed_there) ? "" :"!");
+                        list_empty(&d->opd_syn_committed_there) ? "" : "!");
 
        }
 
@@ -979,7 +1002,7 @@ out:
                 "%s: %d %d %sempty\n",
                 d->opd_obd->obd_name, d->opd_syn_rpc_in_progress,
                 d->opd_syn_rpc_in_flight,
-                cfs_list_empty(&d->opd_syn_committed_there) ? "" : "!");
+                list_empty(&d->opd_syn_committed_there) ? "" : "!");
 
        thread->t_flags = SVC_STOPPED;
 
@@ -1077,8 +1100,7 @@ static int osp_sync_llog_init(const struct lu_env *env, struct osp_device *d)
        memcpy(&osi->osi_gen.lgr_gen, &d->opd_syn_generation,
               sizeof(osi->osi_gen.lgr_gen));
 
-       rc = llog_cat_add(env, lgh, &osi->osi_gen.lgr_hdr, &osi->osi_cookie,
-                         NULL);
+       rc = llog_cat_add(env, lgh, &osi->osi_gen.lgr_hdr, &osi->osi_cookie);
        if (rc < 0)
                GOTO(out_close, rc);
        llog_ctxt_put(ctxt);
@@ -1106,6 +1128,7 @@ static void osp_sync_llog_fini(const struct lu_env *env, struct osp_device *d)
 int osp_sync_init(const struct lu_env *env, struct osp_device *d)
 {
        struct l_wait_info       lwi = { 0 };
+       struct task_struct      *task;
        int                      rc;
 
        ENTRY;
@@ -1131,13 +1154,15 @@ 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);
-       CFS_INIT_LIST_HEAD(&d->opd_syn_committed_there);
+       INIT_LIST_HEAD(&d->opd_syn_committed_there);
 
-       rc = PTR_ERR(kthread_run(osp_sync_thread, d,
-                                "osp-syn-%u-%u", d->opd_index, d->opd_group));
-       if (IS_ERR_VALUE(rc)) {
-               CERROR("%s: can't start sync thread: rc = %d\n",
+       task = kthread_run(osp_sync_thread, d, "osp-syn-%u-%u",
+                          d->opd_index, d->opd_group);
+       if (IS_ERR(task)) {
+               rc = PTR_ERR(task);
+               CERROR("%s: cannot start sync thread: rc = %d\n",
                       d->opd_obd->obd_name, rc);
                GOTO(err_llog, rc);
        }
@@ -1173,7 +1198,8 @@ int osp_sync_fini(struct osp_device *d)
 }
 
 static DEFINE_MUTEX(osp_id_tracker_sem);
-static CFS_LIST_HEAD(osp_id_tracker_list);
+static struct list_head osp_id_tracker_list =
+               LIST_HEAD_INIT(osp_id_tracker_list);
 
 static void osp_sync_tracker_commit_cb(struct thandle *th, void *cookie)
 {
@@ -1193,8 +1219,8 @@ static void osp_sync_tracker_commit_cb(struct thandle *th, void *cookie)
                       tr->otr_committed_id, txn->oti_current_id);
                tr->otr_committed_id = txn->oti_current_id;
 
-               cfs_list_for_each_entry(d, &tr->otr_wakeup_list,
-                                       opd_syn_ontrack) {
+               list_for_each_entry(d, &tr->otr_wakeup_list,
+                                   opd_syn_ontrack) {
                        d->opd_syn_last_committed_id = tr->otr_committed_id;
                        wake_up(&d->opd_syn_waitq);
                }
@@ -1210,13 +1236,13 @@ static int osp_sync_id_traction_init(struct osp_device *d)
        LASSERT(d);
        LASSERT(d->opd_storage);
        LASSERT(d->opd_syn_tracker == NULL);
-       CFS_INIT_LIST_HEAD(&d->opd_syn_ontrack);
+       INIT_LIST_HEAD(&d->opd_syn_ontrack);
 
        mutex_lock(&osp_id_tracker_sem);
-       cfs_list_for_each_entry(tr, &osp_id_tracker_list, otr_list) {
+       list_for_each_entry(tr, &osp_id_tracker_list, otr_list) {
                if (tr->otr_dev == d->opd_storage) {
-                       LASSERT(cfs_atomic_read(&tr->otr_refcount));
-                       cfs_atomic_inc(&tr->otr_refcount);
+                       LASSERT(atomic_read(&tr->otr_refcount));
+                       atomic_inc(&tr->otr_refcount);
                        d->opd_syn_tracker = tr;
                        found = tr;
                        break;
@@ -1232,9 +1258,9 @@ static int osp_sync_id_traction_init(struct osp_device *d)
                        tr->otr_dev = d->opd_storage;
                        tr->otr_next_id = 1;
                        tr->otr_committed_id = 0;
-                       cfs_atomic_set(&tr->otr_refcount, 1);
-                       CFS_INIT_LIST_HEAD(&tr->otr_wakeup_list);
-                       cfs_list_add(&tr->otr_list, &osp_id_tracker_list);
+                       atomic_set(&tr->otr_refcount, 1);
+                       INIT_LIST_HEAD(&tr->otr_wakeup_list);
+                       list_add(&tr->otr_list, &osp_id_tracker_list);
                        tr->otr_tx_cb.dtc_txn_commit =
                                                osp_sync_tracker_commit_cb;
                        tr->otr_tx_cb.dtc_cookie = tr;
@@ -1264,10 +1290,10 @@ static void osp_sync_id_traction_fini(struct osp_device *d)
        osp_sync_remove_from_tracker(d);
 
        mutex_lock(&osp_id_tracker_sem);
-       if (cfs_atomic_dec_and_test(&tr->otr_refcount)) {
+       if (atomic_dec_and_test(&tr->otr_refcount)) {
                dt_txn_callback_del(d->opd_storage, &tr->otr_tx_cb);
-               LASSERT(cfs_list_empty(&tr->otr_wakeup_list));
-               cfs_list_del(&tr->otr_list);
+               LASSERT(list_empty(&tr->otr_wakeup_list));
+               list_del(&tr->otr_list);
                OBD_FREE_PTR(tr);
                d->opd_syn_tracker = NULL;
        }
@@ -1300,8 +1326,8 @@ static __u32 osp_sync_id_get(struct osp_device *d, __u32 id)
                id = tr->otr_next_id++;
        if (id > d->opd_syn_last_used_id)
                d->opd_syn_last_used_id = id;
-       if (cfs_list_empty(&d->opd_syn_ontrack))
-               cfs_list_add(&d->opd_syn_ontrack, &tr->otr_wakeup_list);
+       if (list_empty(&d->opd_syn_ontrack))
+               list_add(&d->opd_syn_ontrack, &tr->otr_wakeup_list);
        spin_unlock(&tr->otr_lock);
        CDEBUG(D_OTHER, "new id %u\n", (unsigned) id);
 
@@ -1315,11 +1341,11 @@ static void osp_sync_remove_from_tracker(struct osp_device *d)
        tr = d->opd_syn_tracker;
        LASSERT(tr);
 
-       if (cfs_list_empty(&d->opd_syn_ontrack))
+       if (list_empty(&d->opd_syn_ontrack))
                return;
 
        spin_lock(&tr->otr_lock);
-       cfs_list_del_init(&d->opd_syn_ontrack);
+       list_del_init(&d->opd_syn_ontrack);
        spin_unlock(&tr->otr_lock);
 }