Whamcloud - gitweb
LU-3534 osp: move RPC pack from declare to execution phase
[fs/lustre-release.git] / lustre / osp / osp_sync.c
index 4d266fc..2f9b78a 100644 (file)
@@ -27,7 +27,7 @@
  * Copyright (c) 2007, 2010, Oracle and/or its affiliates. All rights reserved.
  * Use is subject to license terms.
  *
- * Copyright (c) 2012, 2013, Intel Corporation.
+ * Copyright (c) 2012, 2014, Intel Corporation.
  */
 /*
  * This file is part of Lustre, http://www.lustre.org/
@@ -94,14 +94,13 @@ static void osp_sync_remove_from_tracker(struct osp_device *d);
 
 #define OSP_JOB_MAGIC          0x26112005
 
-/**
- * Return status: whether OSP thread should keep running
- *
- * \param[in] d                OSP device
- *
- * \retval 1           should keep running
- * \retval 0           should stop
- */
+struct osp_job_req_args {
+       /** bytes reserved for ptlrpc_replay_req() */
+       struct ptlrpc_replay_async_args jra_raa;
+       struct list_head                jra_link;
+       __u32                           jra_magic;
+};
+
 static inline int osp_sync_running(struct osp_device *d)
 {
        return !!(d->opd_syn_thread.t_flags & SVC_RUNNING);
@@ -244,13 +243,17 @@ int osp_sync_declare_add(const struct lu_env *env, struct osp_object *o,
        struct osp_thread_info  *osi = osp_env_info(env);
        struct osp_device       *d = lu2osp_dev(o->opo_obj.do_lu.lo_dev);
        struct llog_ctxt        *ctxt;
+       struct thandle          *storage_th;
        int                      rc;
 
        ENTRY;
 
        /* it's a layering violation, to access internals of th,
         * but we can do this as a sanity check, for a while */
-       LASSERT(th->th_dev == d->opd_storage);
+       LASSERT(th->th_top != NULL);
+       storage_th = thandle_get_sub_by_dt(env, th->th_top, d->opd_storage);
+       if (IS_ERR(storage_th))
+               RETURN(PTR_ERR(storage_th));
 
        switch (type) {
        case MDS_UNLINK64_REC:
@@ -264,12 +267,13 @@ int osp_sync_declare_add(const struct lu_env *env, struct osp_object *o,
        }
 
        /* we want ->dt_trans_start() to allocate per-thandle structure */
-       th->th_tags |= LCT_OSP_THREAD;
+       storage_th->th_tags |= LCT_OSP_THREAD;
 
        ctxt = llog_get_context(d->opd_obd, LLOG_MDS_OST_ORIG_CTXT);
        LASSERT(ctxt);
 
-       rc = llog_declare_add(env, ctxt->loc_handle, &osi->osi_hdr, th);
+       rc = llog_declare_add(env, ctxt->loc_handle, &osi->osi_hdr,
+                             storage_th);
        llog_ctxt_put(ctxt);
 
        RETURN(rc);
@@ -304,13 +308,17 @@ static int osp_sync_add_rec(const struct lu_env *env, struct osp_device *d,
        struct osp_thread_info  *osi = osp_env_info(env);
        struct llog_ctxt        *ctxt;
        struct osp_txn_info     *txn;
+       struct thandle          *storage_th;
        int                      rc;
 
        ENTRY;
 
        /* it's a layering violation, to access internals of th,
         * but we can do this as a sanity check, for a while */
-       LASSERT(th->th_dev == d->opd_storage);
+       LASSERT(th->th_top != NULL);
+       storage_th = thandle_get_sub_by_dt(env, th->th_top, d->opd_storage);
+       if (IS_ERR(storage_th))
+               RETURN(PTR_ERR(storage_th));
 
        switch (type) {
        case MDS_UNLINK64_REC:
@@ -336,7 +344,7 @@ static int osp_sync_add_rec(const struct lu_env *env, struct osp_device *d,
                LBUG();
        }
 
-       txn = osp_txn_info(&th->th_ctx);
+       txn = osp_txn_info(&storage_th->th_ctx);
        LASSERT(txn);
 
        txn->oti_current_id = osp_sync_id_get(d, txn->oti_current_id);
@@ -345,25 +353,23 @@ static int osp_sync_add_rec(const struct lu_env *env, struct osp_device *d,
        ctxt = llog_get_context(d->opd_obd, LLOG_MDS_OST_ORIG_CTXT);
        if (ctxt == NULL)
                RETURN(-ENOMEM);
+
        rc = llog_add(env, ctxt->loc_handle, &osi->osi_hdr, &osi->osi_cookie,
-                     th);
+                     storage_th);
        llog_ctxt_put(ctxt);
 
-       CDEBUG(D_OTHER, "%s: new record "DOSTID":%lu/%lu: %d\n",
-              d->opd_obd->obd_name, POSTID(&osi->osi_cookie.lgc_lgl.lgl_oi),
-              (unsigned long) osi->osi_cookie.lgc_lgl.lgl_ogen,
-              (unsigned long) osi->osi_cookie.lgc_index, rc);
-
-       if (rc > 0)
-               rc = 0;
-
-       if (likely(rc == 0)) {
+       if (likely(rc >= 0)) {
+               CDEBUG(D_OTHER, "%s: new record "DOSTID":%lu/%lu: %d\n",
+                      d->opd_obd->obd_name,
+                      POSTID(&osi->osi_cookie.lgc_lgl.lgl_oi),
+                      (unsigned long)osi->osi_cookie.lgc_lgl.lgl_ogen,
+                      (unsigned long)osi->osi_cookie.lgc_index, rc);
                spin_lock(&d->opd_syn_lock);
                d->opd_syn_changes++;
                spin_unlock(&d->opd_syn_lock);
        }
-
-       RETURN(rc);
+       /* return 0 always here, error case just cause no llog record */
+       RETURN(0);
 }
 
 int osp_sync_add(const struct lu_env *env, struct osp_object *o,
@@ -376,7 +382,7 @@ int osp_sync_add(const struct lu_env *env, struct osp_object *o,
 }
 
 int osp_sync_gap(const struct lu_env *env, struct osp_device *d,
-                struct lu_fid *fid, int lost, struct thandle *th)
+                       struct lu_fid *fid, int lost, struct thandle *th)
 {
        return osp_sync_add_rec(env, d, fid, MDS_UNLINK64_REC, lost, th, NULL);
 }
@@ -410,6 +416,7 @@ int osp_sync_gap(const struct lu_env *env, struct osp_device *d,
 static void osp_sync_request_commit_cb(struct ptlrpc_request *req)
 {
        struct osp_device *d = req->rq_cb_data;
+       struct osp_job_req_args *jra;
 
        CDEBUG(D_HA, "commit req %p, transno "LPU64"\n", req, req->rq_transno);
 
@@ -418,15 +425,16 @@ static void osp_sync_request_commit_cb(struct ptlrpc_request *req)
 
        /* do not do any opd_dyn_rpc_* accounting here
         * it's done in osp_sync_interpret sooner or later */
-
        LASSERT(d);
-       LASSERT(req->rq_svc_thread == (void *) OSP_JOB_MAGIC);
-       LASSERT(list_empty(&req->rq_exp_list));
+
+       jra = ptlrpc_req_async_args(req);
+       LASSERT(jra->jra_magic == OSP_JOB_MAGIC);
+       LASSERT(list_empty(&jra->jra_link));
 
        ptlrpc_request_addref(req);
 
        spin_lock(&d->opd_syn_lock);
-       list_add(&req->rq_exp_list, &d->opd_syn_committed_there);
+       list_add(&jra->jra_link, &d->opd_syn_committed_there);
        spin_unlock(&d->opd_syn_lock);
 
        /* XXX: some batching wouldn't hurt */
@@ -454,10 +462,12 @@ static int osp_sync_interpret(const struct lu_env *env,
                              struct ptlrpc_request *req, void *aa, int rc)
 {
        struct osp_device *d = req->rq_cb_data;
+       struct osp_job_req_args *jra = aa;
 
-       if (req->rq_svc_thread != (void *) OSP_JOB_MAGIC)
-               DEBUG_REQ(D_ERROR, req, "bad magic %p\n", req->rq_svc_thread);
-       LASSERT(req->rq_svc_thread == (void *) OSP_JOB_MAGIC);
+       if (jra->jra_magic != OSP_JOB_MAGIC) {
+               DEBUG_REQ(D_ERROR, req, "bad magic %u\n", jra->jra_magic);
+               LBUG();
+       }
        LASSERT(d);
 
        CDEBUG(D_HA, "reply req %p/%d, rc %d, transno %u\n", req,
@@ -471,12 +481,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(list_empty(&req->rq_exp_list));
+               LASSERT(list_empty(&jra->jra_link));
 
                ptlrpc_request_addref(req);
 
                spin_lock(&d->opd_syn_lock);
-               list_add(&req->rq_exp_list, &d->opd_syn_committed_there);
+               list_add(&jra->jra_link, &d->opd_syn_committed_there);
                spin_unlock(&d->opd_syn_lock);
 
                wake_up(&d->opd_syn_waitq);
@@ -537,8 +547,13 @@ static int osp_sync_interpret(const struct lu_env *env,
 static void osp_sync_send_new_rpc(struct osp_device *d,
                                  struct ptlrpc_request *req)
 {
+       struct osp_job_req_args *jra;
+
        LASSERT(d->opd_syn_rpc_in_flight <= d->opd_syn_max_rpc_in_flight);
-       LASSERT(req->rq_svc_thread == (void *) OSP_JOB_MAGIC);
+
+       jra = ptlrpc_req_async_args(req);
+       jra->jra_magic = OSP_JOB_MAGIC;
+       INIT_LIST_HEAD(&jra->jra_link);
 
        ptlrpcd_add_req(req, PDL_POLICY_ROUND, -1);
 }
@@ -596,8 +611,6 @@ 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;
-       INIT_LIST_HEAD(&req->rq_exp_list);
-       req->rq_svc_thread = (void *) OSP_JOB_MAGIC;
 
        req->rq_interpret_reply = osp_sync_interpret;
        req->rq_commit_cb = osp_sync_request_commit_cb;
@@ -707,85 +720,6 @@ static int osp_sync_new_unlink_job(struct osp_device *d,
 }
 
 /**
- * Prepare OUT-based object destroy RPC.
- *
- * The function allocates a new RPC with OUT format. Then initializes the RPC
- * to contain OUT_DESTROY update against the object specified in the llog
- * record provided by the caller.
- *
- * \param[in] env      LU environment provided by the caller
- * \param[in] osp      OSP device
- * \param[in] llh      llog handle where the record is stored
- * \param[in] h                llog record
- * \param[out] reqp    request prepared
- *
- * \retval 0           on success
- * \retval negative    negated errno on error
- */
-static int osp_prep_unlink_update_req(const struct lu_env *env,
-                                     struct osp_device *osp,
-                                     struct llog_handle *llh,
-                                     struct llog_rec_hdr *h,
-                                     struct ptlrpc_request **reqp)
-{
-       struct llog_unlink64_rec        *rec = (struct llog_unlink64_rec *)h;
-       struct dt_update_request        *update = NULL;
-       struct ptlrpc_request           *req;
-       struct llog_cookie              lcookie;
-       const void                      *buf;
-       __u16                           size;
-       int                             rc;
-       ENTRY;
-
-       update = dt_update_request_create(&osp->opd_dt_dev);
-       if (IS_ERR(update))
-               RETURN(PTR_ERR(update));
-
-       /* This can only happens for unlink slave directory, so decrease
-        * ref for ".." and "." */
-       rc = out_update_pack(env, &update->dur_buf, OUT_REF_DEL, &rec->lur_fid,
-                            0, NULL, NULL, 0);
-       if (rc != 0)
-               GOTO(out, rc);
-
-       rc = out_update_pack(env, &update->dur_buf, OUT_REF_DEL, &rec->lur_fid,
-                            0, NULL, NULL, 0);
-       if (rc != 0)
-               GOTO(out, rc);
-
-       lcookie.lgc_lgl = llh->lgh_id;
-       lcookie.lgc_subsys = LLOG_MDS_OST_ORIG_CTXT;
-       lcookie.lgc_index = h->lrh_index;
-       size = sizeof(lcookie);
-       buf = &lcookie;
-
-       rc = out_update_pack(env, &update->dur_buf, OUT_DESTROY, &rec->lur_fid,
-                            1, &size, &buf, 0);
-       if (rc != 0)
-               GOTO(out, rc);
-
-       rc = out_prep_update_req(env, osp->opd_obd->u.cli.cl_import,
-                                update->dur_buf.ub_req, &req);
-       if (rc != 0)
-               GOTO(out, rc);
-
-       INIT_LIST_HEAD(&req->rq_exp_list);
-       req->rq_svc_thread = (void *)OSP_JOB_MAGIC;
-
-       req->rq_interpret_reply = osp_sync_interpret;
-       req->rq_commit_cb = osp_sync_request_commit_cb;
-       req->rq_cb_data = osp;
-
-       ptlrpc_request_set_replen(req);
-       *reqp = req;
-out:
-       if (update != NULL)
-               dt_update_request_destroy(update);
-
-       RETURN(rc);
-}
-
-/**
  * Generate a request for unlink change.
  *
  * The function prepares a new RPC, initializes it with unlink(destroy)
@@ -816,27 +750,20 @@ static int osp_sync_new_unlink64_job(const struct lu_env *env,
 
        ENTRY;
        LASSERT(h->lrh_type == MDS_UNLINK64_REC);
+       req = osp_sync_new_job(d, llh, h, OST_DESTROY,
+                              &RQF_OST_DESTROY);
+       if (IS_ERR(req))
+               RETURN(PTR_ERR(req));
 
-       if (d->opd_connect_mdt) {
-               rc = osp_prep_unlink_update_req(env, d, llh, h, &req);
-               if (rc != 0)
-                       RETURN(rc);
-       } else {
-               req = osp_sync_new_job(d, llh, h, OST_DESTROY,
-                                      &RQF_OST_DESTROY);
-               if (IS_ERR(req))
-                       RETURN(PTR_ERR(req));
-
-               body = req_capsule_client_get(&req->rq_pill, &RMF_OST_BODY);
-               if (body == NULL)
-                       RETURN(-EFAULT);
-               rc = fid_to_ostid(&rec->lur_fid, &body->oa.o_oi);
-               if (rc < 0)
-                       RETURN(rc);
-               body->oa.o_misc = rec->lur_count;
-               body->oa.o_valid = OBD_MD_FLGROUP | OBD_MD_FLID |
-                                  OBD_MD_FLOBJCOUNT;
-       }
+       body = req_capsule_client_get(&req->rq_pill, &RMF_OST_BODY);
+       if (body == NULL)
+               RETURN(-EFAULT);
+       rc = fid_to_ostid(&rec->lur_fid, &body->oa.o_oi);
+       if (rc < 0)
+               RETURN(rc);
+       body->oa.o_misc = rec->lur_count;
+       body->oa.o_valid = OBD_MD_FLGROUP | OBD_MD_FLID |
+                          OBD_MD_FLOBJCOUNT;
        osp_sync_send_new_rpc(d, req);
        RETURN(1);
 }
@@ -969,7 +896,7 @@ static void osp_sync_process_committed(const struct lu_env *env,
        struct obd_device       *obd = d->opd_obd;
        struct obd_import       *imp = obd->u.cli.cl_import;
        struct ost_body         *body;
-       struct ptlrpc_request   *req, *tmp;
+       struct ptlrpc_request   *req;
        struct llog_ctxt        *ctxt;
        struct llog_handle      *llh;
        struct list_head         list;
@@ -1008,33 +935,20 @@ static void osp_sync_process_committed(const struct lu_env *env,
        INIT_LIST_HEAD(&d->opd_syn_committed_there);
        spin_unlock(&d->opd_syn_lock);
 
-       list_for_each_entry_safe(req, tmp, &list, rq_exp_list) {
+       while (!list_empty(&list)) {
                struct llog_cookie *lcookie = NULL;
-
-               LASSERT(req->rq_svc_thread == (void *) OSP_JOB_MAGIC);
-               list_del_init(&req->rq_exp_list);
-
-               if (d->opd_connect_mdt) {
-                       struct object_update_request *ureq;
-                       struct object_update *update;
-                       ureq = req_capsule_client_get(&req->rq_pill,
-                                                     &RMF_OUT_UPDATE);
-                       LASSERT(ureq != NULL &&
-                               ureq->ourq_magic == UPDATE_REQUEST_MAGIC);
-
-                       /* 1st/2nd is for decref . and .., 3rd one is for
-                        * destroy, where the log cookie is stored.
-                        * See osp_prep_unlink_update_req */
-                       update = object_update_request_get(ureq, 2, NULL);
-                       LASSERT(update != NULL);
-                       lcookie = object_update_param_get(update, 0, NULL);
-                       LASSERT(lcookie != NULL);
-               } else {
-                       body = req_capsule_client_get(&req->rq_pill,
-                                                     &RMF_OST_BODY);
-                       LASSERT(body);
-                       lcookie = &body->oa.o_lcookie;
-               }
+               struct osp_job_req_args *jra;
+
+               jra = list_entry(list.next, struct osp_job_req_args, jra_link);
+               LASSERT(jra->jra_magic == OSP_JOB_MAGIC);
+               list_del_init(&jra->jra_link);
+
+               req = container_of((void *)jra, struct ptlrpc_request,
+                                  rq_async_args);
+               body = req_capsule_client_get(&req->rq_pill,
+                                             &RMF_OST_BODY);
+               LASSERT(body);
+               lcookie = &body->oa.o_lcookie;
                /* import can be closing, thus all commit cb's are
                 * called we can check committness directly */
                if (req->rq_transno <= imp->imp_peer_committed_transno) {
@@ -1294,17 +1208,23 @@ static int osp_sync_llog_init(const struct lu_env *env, struct osp_device *d)
        OBD_SET_CTXT_MAGIC(&obd->obd_lvfs_ctxt);
        obd->obd_lvfs_ctxt.dt = d->opd_storage;
 
-       if (d->opd_connect_mdt)
-               lu_local_obj_fid(fid, SLAVE_LLOG_CATALOGS_OID);
-       else
-               lu_local_obj_fid(fid, LLOG_CATALOGS_OID);
+       lu_local_obj_fid(fid, LLOG_CATALOGS_OID);
 
        rc = llog_osd_get_cat_list(env, d->opd_storage, d->opd_index, 1,
                                   &osi->osi_cid, fid);
-       if (rc) {
-               CERROR("%s: can't get id from catalogs: rc = %d\n",
-                      obd->obd_name, rc);
-               RETURN(rc);
+       if (rc < 0) {
+               if (rc != -EFAULT) {
+                       CERROR("%s: can't get id from catalogs: rc = %d\n",
+                              obd->obd_name, rc);
+                       RETURN(rc);
+               }
+
+               /* After sparse OST indices is supported, the CATALOG file
+                * may become a sparse file that results in failure on
+                * reading. Skip this error as the llog will be created
+                * later */
+               memset(&osi->osi_cid, 0, sizeof(osi->osi_cid));
+               rc = 0;
        }
 
        CDEBUG(D_INFO, "%s: Init llog for %d - catid "DOSTID":%x\n",