X-Git-Url: https://git.whamcloud.com/?a=blobdiff_plain;ds=sidebyside;f=lustre%2Ftarget%2Fout_handler.c;h=11dc86427a229e21a3aea0e44d9245aca2c0a549;hb=4f53536d002c13886210b672b657795baa067144;hp=74e8f47c9c6918de742bfca5b6cf25455708e2c9;hpb=0209add4a5099817111c8576afe930d1e2daef03;p=fs%2Flustre-release.git diff --git a/lustre/target/out_handler.c b/lustre/target/out_handler.c index 74e8f47..11dc864 100644 --- a/lustre/target/out_handler.c +++ b/lustre/target/out_handler.c @@ -20,9 +20,9 @@ * GPL HEADER END */ /* - * Copyright (c) 2013, Intel Corporation. + * Copyright (c) 2013, 2014, Intel Corporation. * - * lustre/mdt/out_handler.c + * lustre/target/out_handler.c * * Object update handler between targets. * @@ -36,28 +36,6 @@ #include "tgt_internal.h" #include -struct tx_arg *tx_add_exec(struct thandle_exec_args *ta, tx_exec_func_t func, - tx_exec_func_t undo, char *file, int line) -{ - int i; - - LASSERT(ta); - LASSERT(func); - - LASSERTF(ta->ta_argno + 1 <= TX_MAX_OPS, - "Too many updates(%d) in one trans\n", ta->ta_argno); - i = ta->ta_argno; - - ta->ta_argno++; - - ta->ta_args[i].exec_fn = func; - ta->ta_args[i].undo_fn = undo; - ta->ta_args[i].file = file; - ta->ta_args[i].line = line; - - return &ta->ta_args[i]; -} - static void out_reconstruct(const struct lu_env *env, struct dt_device *dt, struct dt_object *obj, struct object_update_reply *reply, @@ -88,6 +66,22 @@ static inline int out_check_resent(const struct lu_env *env, return 0; if (req_xid_is_last(req)) { + struct lsd_client_data *lcd; + + /* XXX this does not support mulitple transactions yet, i.e. + * only 1 update RPC each time betwee MDTs */ + lcd = req->rq_export->exp_target_data.ted_lcd; + + req->rq_transno = lcd->lcd_last_transno; + req->rq_status = lcd->lcd_last_result; + if (req->rq_status != 0) + req->rq_transno = 0; + lustre_msg_set_transno(req->rq_repmsg, req->rq_transno); + lustre_msg_set_status(req->rq_repmsg, req->rq_status); + + DEBUG_REQ(D_RPCTRACE, req, "restoring transno "LPD64"status %d", + req->rq_transno, req->rq_status); + reconstruct(env, dt, obj, reply, index); return 1; } @@ -96,99 +90,6 @@ static inline int out_check_resent(const struct lu_env *env, return 0; } -static int out_obj_destroy(const struct lu_env *env, struct dt_object *dt_obj, - struct thandle *th) -{ - int rc; - - CDEBUG(D_INFO, "%s: destroy "DFID"\n", dt_obd_name(th->th_dev), - PFID(lu_object_fid(&dt_obj->do_lu))); - - dt_write_lock(env, dt_obj, MOR_TGT_CHILD); - rc = dt_destroy(env, dt_obj, th); - dt_write_unlock(env, dt_obj); - - return rc; -} - -/** - * All of the xxx_undo will be used once execution failed, - * But because all of the required resource has been reserved in - * declare phase, i.e. if declare succeed, it should make sure - * the following executing phase succeed in anyway, so these undo - * should be useless for most of the time in Phase I - */ -int out_tx_create_undo(const struct lu_env *env, struct thandle *th, - struct tx_arg *arg) -{ - int rc; - - rc = out_obj_destroy(env, arg->object, th); - if (rc != 0) - CERROR("%s: undo failure, we are doomed!: rc = %d\n", - dt_obd_name(th->th_dev), rc); - return rc; -} - -int out_tx_create_exec(const struct lu_env *env, struct thandle *th, - struct tx_arg *arg) -{ - struct dt_object *dt_obj = arg->object; - int rc; - - CDEBUG(D_OTHER, "%s: create "DFID": dof %u, mode %o\n", - dt_obd_name(th->th_dev), - PFID(lu_object_fid(&arg->object->do_lu)), - arg->u.create.dof.dof_type, - arg->u.create.attr.la_mode & S_IFMT); - - dt_write_lock(env, dt_obj, MOR_TGT_CHILD); - rc = dt_create(env, dt_obj, &arg->u.create.attr, - &arg->u.create.hint, &arg->u.create.dof, th); - - dt_write_unlock(env, dt_obj); - - CDEBUG(D_INFO, "%s: insert create reply %p index %d: rc = %d\n", - dt_obd_name(th->th_dev), arg->reply, arg->index, rc); - - object_update_result_insert(arg->reply, NULL, 0, arg->index, rc); - - return rc; -} - -static int __out_tx_create(const struct lu_env *env, struct dt_object *obj, - struct lu_attr *attr, struct lu_fid *parent_fid, - struct dt_object_format *dof, - struct thandle_exec_args *ta, - struct object_update_reply *reply, - int index, char *file, int line) -{ - struct tx_arg *arg; - - LASSERT(ta->ta_handle != NULL); - ta->ta_err = dt_declare_create(env, obj, attr, NULL, dof, - ta->ta_handle); - if (ta->ta_err != 0) - return ta->ta_err; - - arg = tx_add_exec(ta, out_tx_create_exec, out_tx_create_undo, file, - line); - LASSERT(arg); - - /* release the object in out_trans_stop */ - lu_object_get(&obj->do_lu); - arg->object = obj; - arg->u.create.attr = *attr; - if (parent_fid != NULL) - arg->u.create.fid = *parent_fid; - memset(&arg->u.create.hint, 0, sizeof(arg->u.create.hint)); - arg->u.create.dof = *dof; - arg->reply = reply; - arg->index = index; - - return 0; -} - static int out_create(struct tgt_session_info *tsi) { struct tgt_thread_info *tti = tgt_th_info(tsi->tsi_env); @@ -199,28 +100,27 @@ static int out_create(struct tgt_session_info *tsi) struct lu_attr *attr = &tti->tti_attr; struct lu_fid *fid = NULL; struct obdo *wobdo; - int size; + size_t size; int rc; ENTRY; wobdo = object_update_param_get(update, 0, &size); - if (wobdo == NULL || size != sizeof(*wobdo)) { + if (wobdo == NULL || IS_ERR(wobdo) || size != sizeof(*wobdo)) { CERROR("%s: obdo is NULL, invalid RPC: rc = %d\n", tgt_name(tsi->tsi_tgt), -EPROTO); RETURN(err_serious(-EPROTO)); } - obdo_le_to_cpu(wobdo, wobdo); + if (ptlrpc_req_need_swab(tsi->tsi_pill->rc_req)) + lustre_swab_obdo(wobdo); lustre_get_wire_obdo(NULL, lobdo, wobdo); la_from_obdo(attr, lobdo, lobdo->o_valid); dof->dof_type = dt_mode_to_dft(attr->la_mode); if (update->ou_params_count > 1) { - int size; - fid = object_update_param_get(update, 1, &size); - if (fid == NULL || size != sizeof(*fid)) { + if (fid == NULL || IS_ERR(fid) || size != sizeof(*fid)) { CERROR("%s: invalid fid: rc = %d\n", tgt_name(tsi->tsi_tgt), -EPROTO); RETURN(err_serious(-EPROTO)); @@ -238,69 +138,13 @@ static int out_create(struct tgt_session_info *tsi) RETURN(-EEXIST); rc = out_tx_create(tsi->tsi_env, obj, attr, fid, dof, - &tti->tti_tea, + &tti->tti_tea, tti->tti_tea.ta_handle, tti->tti_u.update.tti_update_reply, tti->tti_u.update.tti_update_reply_index); RETURN(rc); } -static int out_tx_attr_set_undo(const struct lu_env *env, - struct thandle *th, struct tx_arg *arg) -{ - CERROR("%s: attr set undo "DFID" unimplemented yet!: rc = %d\n", - dt_obd_name(th->th_dev), - PFID(lu_object_fid(&arg->object->do_lu)), -ENOTSUPP); - - return -ENOTSUPP; -} - -static int out_tx_attr_set_exec(const struct lu_env *env, struct thandle *th, - struct tx_arg *arg) -{ - struct dt_object *dt_obj = arg->object; - int rc; - - CDEBUG(D_OTHER, "%s: attr set "DFID"\n", dt_obd_name(th->th_dev), - PFID(lu_object_fid(&dt_obj->do_lu))); - - dt_write_lock(env, dt_obj, MOR_TGT_CHILD); - rc = dt_attr_set(env, dt_obj, &arg->u.attr_set.attr, th, NULL); - dt_write_unlock(env, dt_obj); - - CDEBUG(D_INFO, "%s: insert attr_set reply %p index %d: rc = %d\n", - dt_obd_name(th->th_dev), arg->reply, arg->index, rc); - - object_update_result_insert(arg->reply, NULL, 0, arg->index, rc); - - return rc; -} - -static int __out_tx_attr_set(const struct lu_env *env, - struct dt_object *dt_obj, - const struct lu_attr *attr, - struct thandle_exec_args *th, - struct object_update_reply *reply, - int index, char *file, int line) -{ - struct tx_arg *arg; - - LASSERT(th->ta_handle != NULL); - th->ta_err = dt_declare_attr_set(env, dt_obj, attr, th->ta_handle); - if (th->ta_err != 0) - return th->ta_err; - - arg = tx_add_exec(th, out_tx_attr_set_exec, out_tx_attr_set_undo, - file, line); - LASSERT(arg); - lu_object_get(&dt_obj->do_lu); - arg->object = dt_obj; - arg->u.attr_set.attr = *attr; - arg->reply = reply; - arg->index = index; - return 0; -} - static int out_attr_set(struct tgt_session_info *tsi) { struct tgt_thread_info *tti = tgt_th_info(tsi->tsi_env); @@ -309,13 +153,13 @@ static int out_attr_set(struct tgt_session_info *tsi) struct dt_object *obj = tti->tti_u.update.tti_dt_object; struct obdo *lobdo = &tti->tti_u.update.tti_obdo; struct obdo *wobdo; - int size; + size_t size; int rc; ENTRY; wobdo = object_update_param_get(update, 0, &size); - if (wobdo == NULL || size != sizeof(*wobdo)) { + if (wobdo == NULL || IS_ERR(wobdo) || size != sizeof(*wobdo)) { CERROR("%s: empty obdo in the update: rc = %d\n", tgt_name(tsi->tsi_tgt), -EPROTO); RETURN(err_serious(-EPROTO)); @@ -323,11 +167,14 @@ static int out_attr_set(struct tgt_session_info *tsi) attr->la_valid = 0; attr->la_valid = 0; - obdo_le_to_cpu(wobdo, wobdo); + + if (ptlrpc_req_need_swab(tsi->tsi_pill->rc_req)) + lustre_swab_obdo(wobdo); lustre_get_wire_obdo(NULL, lobdo, wobdo); la_from_obdo(attr, lobdo, lobdo->o_valid); rc = out_tx_attr_set(tsi->tsi_env, obj, attr, &tti->tti_tea, + tti->tti_tea.ta_handle, tti->tti_u.update.tti_update_reply, tti->tti_u.update.tti_update_reply_index); @@ -357,49 +204,12 @@ static int out_attr_get(struct tgt_session_info *tsi) } dt_read_lock(env, obj, MOR_TGT_CHILD); - rc = dt_attr_get(env, obj, la, NULL); + rc = dt_attr_get(env, obj, la); if (rc) GOTO(out_unlock, rc); - /* - * If it is a directory, we will also check whether the - * directory is empty. - * la_flags = 0 : Empty. - * = 1 : Not empty. - */ - la->la_flags = 0; - if (S_ISDIR(la->la_mode)) { - struct dt_it *it; - const struct dt_it_ops *iops; - - if (!dt_try_as_dir(env, obj)) - GOTO(out_unlock, rc = -ENOTDIR); - - iops = &obj->do_index_ops->dio_it; - it = iops->init(env, obj, LUDA_64BITHASH, BYPASS_CAPA); - if (!IS_ERR(it)) { - int result; - result = iops->get(env, it, (const void *)""); - if (result > 0) { - int i; - for (result = 0, i = 0; result == 0 && i < 3; - ++i) - result = iops->next(env, it); - if (result == 0) - la->la_flags = 1; - } else if (result == 0) - /* - * Huh? Index contains no zero key? - */ - rc = -EIO; - - iops->put(env, it); - iops->fini(env, it); - } - } obdo->o_valid = 0; obdo_from_la(obdo, la, la->la_valid); - obdo_cpu_to_le(obdo, obdo); lustre_set_wire_obdo(NULL, obdo, obdo); out_unlock: @@ -430,8 +240,14 @@ static int out_xattr_get(struct tgt_session_info *tsi) ENTRY; + if (!lu_object_exists(&obj->do_lu)) { + set_bit(LU_OBJECT_HEARD_BANSHEE, + &obj->do_lu.lo_header->loh_flags); + RETURN(-ENOENT); + } + name = object_update_param_get(update, 0, NULL); - if (name == NULL) { + if (name == NULL || IS_ERR(name)) { CERROR("%s: empty name for xattr get: rc = %d\n", tgt_name(tsi->tsi_tgt), -EPROTO); RETURN(err_serious(-EPROTO)); @@ -449,16 +265,12 @@ static int out_xattr_get(struct tgt_session_info *tsi) cfs_size_round((unsigned long)update_result->our_data - (unsigned long)update_result); dt_read_lock(env, obj, MOR_TGT_CHILD); - rc = dt_xattr_get(env, obj, lbuf, name, NULL); + rc = dt_xattr_get(env, obj, lbuf, name); dt_read_unlock(env, obj); if (rc < 0) { lbuf->lb_len = 0; GOTO(out, rc); } - if (rc == 0) { - lbuf->lb_len = 0; - GOTO(out, rc = -ENOENT); - } lbuf->lb_len = rc; rc = 0; CDEBUG(D_INFO, "%s: "DFID" get xattr %s len %d\n", @@ -487,7 +299,7 @@ static int out_index_lookup(struct tgt_session_info *tsi) RETURN(-ENOENT); name = object_update_param_get(update, 0, NULL); - if (name == NULL) { + if (name == NULL || IS_ERR(name)) { CERROR("%s: empty name for lookup: rc = %d\n", tgt_name(tsi->tsi_tgt), -EPROTO); RETURN(err_serious(-EPROTO)); @@ -498,7 +310,7 @@ static int out_index_lookup(struct tgt_session_info *tsi) GOTO(out_unlock, rc = -ENOTDIR); rc = dt_lookup(env, obj, (struct dt_rec *)&tti->tti_fid1, - (struct dt_key *)name, NULL); + (struct dt_key *)name); if (rc < 0) GOTO(out_unlock, rc); @@ -506,13 +318,13 @@ static int out_index_lookup(struct tgt_session_info *tsi) if (rc == 0) rc += 1; +out_unlock: + dt_read_unlock(env, obj); + CDEBUG(D_INFO, "lookup "DFID" %s get "DFID" rc %d\n", PFID(lu_object_fid(&obj->do_lu)), name, PFID(&tti->tti_fid1), rc); -out_unlock: - dt_read_unlock(env, obj); - CDEBUG(D_INFO, "%s: insert lookup reply %p index %d: rc = %d\n", tgt_name(tsi->tsi_tgt), tti->tti_u.update.tti_update_reply, 0, rc); @@ -523,68 +335,6 @@ out_unlock: RETURN(rc); } -static int out_tx_xattr_set_exec(const struct lu_env *env, - struct thandle *th, - struct tx_arg *arg) -{ - struct dt_object *dt_obj = arg->object; - int rc; - - CDEBUG(D_INFO, "%s: set xattr buf %p name %s flag %d\n", - dt_obd_name(th->th_dev), arg->u.xattr_set.buf.lb_buf, - arg->u.xattr_set.name, arg->u.xattr_set.flags); - - if (!lu_object_exists(&dt_obj->do_lu)) - GOTO(out, rc = -ENOENT); - - dt_write_lock(env, dt_obj, MOR_TGT_CHILD); - rc = dt_xattr_set(env, dt_obj, &arg->u.xattr_set.buf, - arg->u.xattr_set.name, arg->u.xattr_set.flags, - th, NULL); - dt_write_unlock(env, dt_obj); - /** - * Ignore errors if this is LINK EA - **/ - if (unlikely(rc && !strcmp(arg->u.xattr_set.name, XATTR_NAME_LINK))) - rc = 0; -out: - CDEBUG(D_INFO, "%s: insert xattr set reply %p index %d: rc = %d\n", - dt_obd_name(th->th_dev), arg->reply, arg->index, rc); - - object_update_result_insert(arg->reply, NULL, 0, arg->index, rc); - - return rc; -} - -static int __out_tx_xattr_set(const struct lu_env *env, - struct dt_object *dt_obj, - const struct lu_buf *buf, - const char *name, int flags, - struct thandle_exec_args *ta, - struct object_update_reply *reply, - int index, char *file, int line) -{ - struct tx_arg *arg; - - LASSERT(ta->ta_handle != NULL); - ta->ta_err = dt_declare_xattr_set(env, dt_obj, buf, name, - flags, ta->ta_handle); - if (ta->ta_err != 0) - return ta->ta_err; - - arg = tx_add_exec(ta, out_tx_xattr_set_exec, NULL, file, line); - LASSERT(arg); - lu_object_get(&dt_obj->do_lu); - arg->object = dt_obj; - arg->u.xattr_set.name = name; - arg->u.xattr_set.flags = flags; - arg->u.xattr_set.buf = *buf; - arg->reply = reply; - arg->index = index; - arg->u.xattr_set.csum = 0; - return 0; -} - static int out_xattr_set(struct tgt_session_info *tsi) { struct tgt_thread_info *tti = tgt_th_info(tsi->tsi_env); @@ -593,115 +343,66 @@ static int out_xattr_set(struct tgt_session_info *tsi) struct lu_buf *lbuf = &tti->tti_buf; char *name; char *buf; - char *tmp; - int buf_len = 0; + __u32 *tmp; + size_t buf_len = 0; int flag; + size_t size = 0; int rc; ENTRY; name = object_update_param_get(update, 0, NULL); - if (name == NULL) { + if (name == NULL || IS_ERR(name)) { CERROR("%s: empty name for xattr set: rc = %d\n", tgt_name(tsi->tsi_tgt), -EPROTO); RETURN(err_serious(-EPROTO)); } buf = object_update_param_get(update, 1, &buf_len); - if (buf == NULL || buf_len == 0) { - CERROR("%s: empty buf for xattr set: rc = %d\n", - tgt_name(tsi->tsi_tgt), -EPROTO); + if (IS_ERR(buf)) RETURN(err_serious(-EPROTO)); - } lbuf->lb_buf = buf; lbuf->lb_len = buf_len; - tmp = (char *)object_update_param_get(update, 2, NULL); - if (tmp == NULL) { - CERROR("%s: empty flag for xattr set: rc = %d\n", - tgt_name(tsi->tsi_tgt), -EPROTO); + tmp = object_update_param_get(update, 2, &size); + if (tmp == NULL || size != sizeof(*tmp)) { + CERROR("%s: emptry or wrong size %zu flag: rc = %d\n", + tgt_name(tsi->tsi_tgt), size, -EPROTO); RETURN(err_serious(-EPROTO)); } if (ptlrpc_req_need_swab(tsi->tsi_pill->rc_req)) - __swab32s((__u32 *)tmp); - flag = *(int *)tmp; + __swab32s(tmp); + flag = *tmp; rc = out_tx_xattr_set(tsi->tsi_env, obj, lbuf, name, flag, - &tti->tti_tea, + &tti->tti_tea, tti->tti_tea.ta_handle, tti->tti_u.update.tti_update_reply, tti->tti_u.update.tti_update_reply_index); RETURN(rc); } -static int out_obj_ref_add(const struct lu_env *env, - struct dt_object *dt_obj, - struct thandle *th) +static int out_xattr_del(struct tgt_session_info *tsi) { - int rc; - - dt_write_lock(env, dt_obj, MOR_TGT_CHILD); - rc = dt_ref_add(env, dt_obj, th); - dt_write_unlock(env, dt_obj); - - return rc; -} - -static int out_obj_ref_del(const struct lu_env *env, - struct dt_object *dt_obj, - struct thandle *th) -{ - int rc; - - dt_write_lock(env, dt_obj, MOR_TGT_CHILD); - rc = dt_ref_del(env, dt_obj, th); - dt_write_unlock(env, dt_obj); - - return rc; -} - -static int out_tx_ref_add_exec(const struct lu_env *env, struct thandle *th, - struct tx_arg *arg) -{ - struct dt_object *dt_obj = arg->object; - int rc; - - rc = out_obj_ref_add(env, dt_obj, th); - - CDEBUG(D_INFO, "%s: insert ref_add reply %p index %d: rc = %d\n", - dt_obd_name(th->th_dev), arg->reply, arg->index, rc); - - object_update_result_insert(arg->reply, NULL, 0, arg->index, rc); - return rc; -} + struct tgt_thread_info *tti = tgt_th_info(tsi->tsi_env); + struct object_update *update = tti->tti_u.update.tti_update; + struct dt_object *obj = tti->tti_u.update.tti_dt_object; + char *name; + int rc; + ENTRY; -static int out_tx_ref_add_undo(const struct lu_env *env, struct thandle *th, - struct tx_arg *arg) -{ - return out_obj_ref_del(env, arg->object, th); -} + name = object_update_param_get(update, 0, NULL); + if (name == NULL || IS_ERR(name)) { + CERROR("%s: empty name for xattr set: rc = %d\n", + tgt_name(tsi->tsi_tgt), -EPROTO); + RETURN(err_serious(-EPROTO)); + } -static int __out_tx_ref_add(const struct lu_env *env, - struct dt_object *dt_obj, - struct thandle_exec_args *ta, - struct object_update_reply *reply, - int index, char *file, int line) -{ - struct tx_arg *arg; - - LASSERT(ta->ta_handle != NULL); - ta->ta_err = dt_declare_ref_add(env, dt_obj, ta->ta_handle); - if (ta->ta_err != 0) - return ta->ta_err; - - arg = tx_add_exec(ta, out_tx_ref_add_exec, out_tx_ref_add_undo, file, - line); - LASSERT(arg); - lu_object_get(&dt_obj->do_lu); - arg->object = dt_obj; - arg->reply = reply; - arg->index = index; - return 0; + rc = out_tx_xattr_del(tsi->tsi_env, obj, name, &tti->tti_tea, + tti->tti_tea.ta_handle, + tti->tti_u.update.tti_update_reply, + tti->tti_u.update.tti_update_reply_index); + RETURN(rc); } /** @@ -716,56 +417,12 @@ static int out_ref_add(struct tgt_session_info *tsi) ENTRY; rc = out_tx_ref_add(tsi->tsi_env, obj, &tti->tti_tea, + tti->tti_tea.ta_handle, tti->tti_u.update.tti_update_reply, tti->tti_u.update.tti_update_reply_index); RETURN(rc); } -static int out_tx_ref_del_exec(const struct lu_env *env, struct thandle *th, - struct tx_arg *arg) -{ - struct dt_object *dt_obj = arg->object; - int rc; - - rc = out_obj_ref_del(env, dt_obj, th); - - CDEBUG(D_INFO, "%s: insert ref_del reply %p index %d: rc = %d\n", - dt_obd_name(th->th_dev), arg->reply, arg->index, 0); - - object_update_result_insert(arg->reply, NULL, 0, arg->index, rc); - - return rc; -} - -static int out_tx_ref_del_undo(const struct lu_env *env, struct thandle *th, - struct tx_arg *arg) -{ - return out_obj_ref_add(env, arg->object, th); -} - -static int __out_tx_ref_del(const struct lu_env *env, - struct dt_object *dt_obj, - struct thandle_exec_args *ta, - struct object_update_reply *reply, - int index, char *file, int line) -{ - struct tx_arg *arg; - - LASSERT(ta->ta_handle != NULL); - ta->ta_err = dt_declare_ref_del(env, dt_obj, ta->ta_handle); - if (ta->ta_err != 0) - return ta->ta_err; - - arg = tx_add_exec(ta, out_tx_ref_del_exec, out_tx_ref_del_undo, file, - line); - LASSERT(arg); - lu_object_get(&dt_obj->do_lu); - arg->object = dt_obj; - arg->reply = reply; - arg->index = index; - return 0; -} - static int out_ref_del(struct tgt_session_info *tsi) { struct tgt_thread_info *tti = tgt_th_info(tsi->tsi_env); @@ -778,139 +435,37 @@ static int out_ref_del(struct tgt_session_info *tsi) RETURN(-ENOENT); rc = out_tx_ref_del(tsi->tsi_env, obj, &tti->tti_tea, + tti->tti_tea.ta_handle, tti->tti_u.update.tti_update_reply, tti->tti_u.update.tti_update_reply_index); RETURN(rc); } -static int out_obj_index_insert(const struct lu_env *env, - struct dt_object *dt_obj, - const struct dt_rec *rec, - const struct dt_key *key, - struct thandle *th) -{ - int rc; - - CDEBUG(D_INFO, "%s: index insert "DFID" name: %s fid "DFID"\n", - dt_obd_name(th->th_dev), PFID(lu_object_fid(&dt_obj->do_lu)), - (char *)key, PFID((struct lu_fid *)rec)); - - if (dt_try_as_dir(env, dt_obj) == 0) - return -ENOTDIR; - - dt_write_lock(env, dt_obj, MOR_TGT_CHILD); - rc = dt_insert(env, dt_obj, rec, key, th, NULL, 0); - dt_write_unlock(env, dt_obj); - - return rc; -} - -static int out_obj_index_delete(const struct lu_env *env, - struct dt_object *dt_obj, - const struct dt_key *key, - struct thandle *th) -{ - int rc; - - CDEBUG(D_INFO, "%s: index delete "DFID" name: %s\n", - dt_obd_name(th->th_dev), PFID(lu_object_fid(&dt_obj->do_lu)), - (char *)key); - - if (dt_try_as_dir(env, dt_obj) == 0) - return -ENOTDIR; - - dt_write_lock(env, dt_obj, MOR_TGT_CHILD); - rc = dt_delete(env, dt_obj, key, th, NULL); - dt_write_unlock(env, dt_obj); - - return rc; -} - -static int out_tx_index_insert_exec(const struct lu_env *env, - struct thandle *th, struct tx_arg *arg) -{ - struct dt_object *dt_obj = arg->object; - int rc; - - rc = out_obj_index_insert(env, dt_obj, arg->u.insert.rec, - arg->u.insert.key, th); - - CDEBUG(D_INFO, "%s: insert idx insert reply %p index %d: rc = %d\n", - dt_obd_name(th->th_dev), arg->reply, arg->index, rc); - - object_update_result_insert(arg->reply, NULL, 0, arg->index, rc); - - return rc; -} - -static int out_tx_index_insert_undo(const struct lu_env *env, - struct thandle *th, struct tx_arg *arg) -{ - return out_obj_index_delete(env, arg->object, arg->u.insert.key, th); -} - -static int __out_tx_index_insert(const struct lu_env *env, - struct dt_object *dt_obj, - char *name, struct lu_fid *fid, - struct thandle_exec_args *ta, - struct object_update_reply *reply, - int index, char *file, int line) -{ - struct tx_arg *arg; - - LASSERT(ta->ta_handle != NULL); - - if (dt_try_as_dir(env, dt_obj) == 0) { - ta->ta_err = -ENOTDIR; - return ta->ta_err; - } - - ta->ta_err = dt_declare_insert(env, dt_obj, - (struct dt_rec *)fid, - (struct dt_key *)name, - ta->ta_handle); - - if (ta->ta_err != 0) - return ta->ta_err; - - arg = tx_add_exec(ta, out_tx_index_insert_exec, - out_tx_index_insert_undo, file, - line); - LASSERT(arg); - lu_object_get(&dt_obj->do_lu); - arg->object = dt_obj; - arg->reply = reply; - arg->index = index; - arg->u.insert.rec = (struct dt_rec *)fid; - arg->u.insert.key = (struct dt_key *)name; - - return 0; -} - static int out_index_insert(struct tgt_session_info *tsi) { - struct tgt_thread_info *tti = tgt_th_info(tsi->tsi_env); - struct object_update *update = tti->tti_u.update.tti_update; - struct dt_object *obj = tti->tti_u.update.tti_dt_object; - struct lu_fid *fid; - char *name; - int rc = 0; - int size; - + struct tgt_thread_info *tti = tgt_th_info(tsi->tsi_env); + struct object_update *update = tti->tti_u.update.tti_update; + struct dt_object *obj = tti->tti_u.update.tti_dt_object; + struct dt_insert_rec *rec = &tti->tti_rec; + struct lu_fid *fid; + char *name; + __u32 *ptype; + int rc = 0; + size_t size; ENTRY; name = object_update_param_get(update, 0, NULL); - if (name == NULL) { + if (name == NULL || IS_ERR(name)) { CERROR("%s: empty name for index insert: rc = %d\n", tgt_name(tsi->tsi_tgt), -EPROTO); RETURN(err_serious(-EPROTO)); } fid = object_update_param_get(update, 1, &size); - if (fid == NULL || size != sizeof(*fid)) { + if (fid == NULL || IS_ERR(fid) || size != sizeof(*fid)) { CERROR("%s: invalid fid: rc = %d\n", tgt_name(tsi->tsi_tgt), -EPROTO); - RETURN(err_serious(-EPROTO)); + RETURN(err_serious(-EPROTO)); } if (ptlrpc_req_need_swab(tsi->tsi_pill->rc_req)) @@ -922,68 +477,25 @@ static int out_index_insert(struct tgt_session_info *tsi) RETURN(err_serious(-EPROTO)); } - rc = out_tx_index_insert(tsi->tsi_env, obj, name, fid, - &tti->tti_tea, - tti->tti_u.update.tti_update_reply, - tti->tti_u.update.tti_update_reply_index); - RETURN(rc); -} - -static int out_tx_index_delete_exec(const struct lu_env *env, - struct thandle *th, - struct tx_arg *arg) -{ - int rc; - - rc = out_obj_index_delete(env, arg->object, arg->u.insert.key, th); - - CDEBUG(D_INFO, "%s: insert idx insert reply %p index %d: rc = %d\n", - dt_obd_name(th->th_dev), arg->reply, arg->index, rc); - - object_update_result_insert(arg->reply, NULL, 0, arg->index, rc); - - return rc; -} - -static int out_tx_index_delete_undo(const struct lu_env *env, - struct thandle *th, - struct tx_arg *arg) -{ - CERROR("%s: Oops, can not rollback index_delete yet: rc = %d\n", - dt_obd_name(th->th_dev), -ENOTSUPP); - return -ENOTSUPP; -} + ptype = object_update_param_get(update, 2, &size); + if (ptype == NULL || IS_ERR(ptype) || size != sizeof(*ptype)) { + CERROR("%s: invalid type for index insert: rc = %d\n", + tgt_name(tsi->tsi_tgt), -EPROTO); + RETURN(err_serious(-EPROTO)); + } -static int __out_tx_index_delete(const struct lu_env *env, - struct dt_object *dt_obj, char *name, - struct thandle_exec_args *ta, - struct object_update_reply *reply, - int index, char *file, int line) -{ - struct tx_arg *arg; + if (ptlrpc_req_need_swab(tsi->tsi_pill->rc_req)) + __swab32s(ptype); - if (dt_try_as_dir(env, dt_obj) == 0) { - ta->ta_err = -ENOTDIR; - return ta->ta_err; - } + rec->rec_fid = fid; + rec->rec_type = *ptype; - LASSERT(ta->ta_handle != NULL); - ta->ta_err = dt_declare_delete(env, dt_obj, - (struct dt_key *)name, - ta->ta_handle); - if (ta->ta_err != 0) - return ta->ta_err; - - arg = tx_add_exec(ta, out_tx_index_delete_exec, - out_tx_index_delete_undo, file, - line); - LASSERT(arg); - lu_object_get(&dt_obj->do_lu); - arg->object = dt_obj; - arg->reply = reply; - arg->index = index; - arg->u.insert.key = (struct dt_key *)name; - return 0; + rc = out_tx_index_insert(tsi->tsi_env, obj, (const struct dt_rec *)rec, + (const struct dt_key *)name, &tti->tti_tea, + tti->tti_tea.ta_handle, + tti->tti_u.update.tti_update_reply, + tti->tti_u.update.tti_update_reply_index); + RETURN(rc); } static int out_index_delete(struct tgt_session_info *tsi) @@ -998,64 +510,19 @@ static int out_index_delete(struct tgt_session_info *tsi) RETURN(-ENOENT); name = object_update_param_get(update, 0, NULL); - if (name == NULL) { + if (name == NULL || IS_ERR(name)) { CERROR("%s: empty name for index delete: rc = %d\n", tgt_name(tsi->tsi_tgt), -EPROTO); RETURN(err_serious(-EPROTO)); } - rc = out_tx_index_delete(tsi->tsi_env, obj, name, &tti->tti_tea, + rc = out_tx_index_delete(tsi->tsi_env, obj, (const struct dt_key *)name, + &tti->tti_tea, tti->tti_tea.ta_handle, tti->tti_u.update.tti_update_reply, tti->tti_u.update.tti_update_reply_index); RETURN(rc); } -static int out_tx_destroy_exec(const struct lu_env *env, struct thandle *th, - struct tx_arg *arg) -{ - struct dt_object *dt_obj = arg->object; - int rc; - - rc = out_obj_destroy(env, dt_obj, th); - - CDEBUG(D_INFO, "%s: insert destroy reply %p index %d: rc = %d\n", - dt_obd_name(th->th_dev), arg->reply, arg->index, rc); - - object_update_result_insert(arg->reply, NULL, 0, arg->index, rc); - - RETURN(rc); -} - -static int out_tx_destroy_undo(const struct lu_env *env, struct thandle *th, - struct tx_arg *arg) -{ - CERROR("%s: not support destroy undo yet!: rc = %d\n", - dt_obd_name(th->th_dev), -ENOTSUPP); - return -ENOTSUPP; -} - -static int __out_tx_destroy(const struct lu_env *env, struct dt_object *dt_obj, - struct thandle_exec_args *ta, - struct object_update_reply *reply, - int index, char *file, int line) -{ - struct tx_arg *arg; - - LASSERT(ta->ta_handle != NULL); - ta->ta_err = dt_declare_destroy(env, dt_obj, ta->ta_handle); - if (ta->ta_err) - return ta->ta_err; - - arg = tx_add_exec(ta, out_tx_destroy_exec, out_tx_destroy_undo, - file, line); - LASSERT(arg); - lu_object_get(&dt_obj->do_lu); - arg->object = dt_obj; - arg->reply = reply; - arg->index = index; - return 0; -} - static int out_destroy(struct tgt_session_info *tsi) { struct tgt_thread_info *tti = tgt_th_info(tsi->tsi_env); @@ -1076,57 +543,13 @@ static int out_destroy(struct tgt_session_info *tsi) RETURN(-ENOENT); rc = out_tx_destroy(tsi->tsi_env, obj, &tti->tti_tea, + tti->tti_tea.ta_handle, tti->tti_u.update.tti_update_reply, tti->tti_u.update.tti_update_reply_index); RETURN(rc); } -static int out_tx_write_exec(const struct lu_env *env, struct thandle *th, - struct tx_arg *arg) -{ - struct dt_object *dt_obj = arg->object; - int rc; - - dt_write_lock(env, dt_obj, MOR_TGT_CHILD); - rc = dt_record_write(env, dt_obj, &arg->u.write.buf, - &arg->u.write.pos, th); - dt_write_unlock(env, dt_obj); - - if (rc == 0) - rc = arg->u.write.buf.lb_len; - - object_update_result_insert(arg->reply, NULL, 0, arg->index, rc); - - return rc > 0 ? 0 : rc; -} - -static int __out_tx_write(const struct lu_env *env, - struct dt_object *dt_obj, - const struct lu_buf *buf, - loff_t pos, struct thandle_exec_args *ta, - struct object_update_reply *reply, - int index, char *file, int line) -{ - struct tx_arg *arg; - - LASSERT(ta->ta_handle != NULL); - ta->ta_err = dt_declare_record_write(env, dt_obj, buf, pos, - ta->ta_handle); - if (ta->ta_err != 0) - return ta->ta_err; - - arg = tx_add_exec(ta, out_tx_write_exec, NULL, file, line); - LASSERT(arg); - lu_object_get(&dt_obj->do_lu); - arg->object = dt_obj; - arg->u.write.buf = *buf; - arg->u.write.pos = pos; - arg->reply = reply; - arg->index = index; - return 0; -} - static int out_write(struct tgt_session_info *tsi) { struct tgt_thread_info *tti = tgt_th_info(tsi->tsi_env); @@ -1134,14 +557,15 @@ static int out_write(struct tgt_session_info *tsi) struct dt_object *obj = tti->tti_u.update.tti_dt_object; struct lu_buf *lbuf = &tti->tti_buf; char *buf; - char *tmp; - int buf_len = 0; + __u64 *tmp; + size_t size = 0; + size_t buf_len = 0; loff_t pos; int rc; ENTRY; buf = object_update_param_get(update, 0, &buf_len); - if (buf == NULL || buf_len == 0) { + if (buf == NULL || IS_ERR(buf) || buf_len == 0) { CERROR("%s: empty buf for xattr set: rc = %d\n", tgt_name(tsi->tsi_tgt), -EPROTO); RETURN(err_serious(-EPROTO)); @@ -1149,24 +573,95 @@ static int out_write(struct tgt_session_info *tsi) lbuf->lb_buf = buf; lbuf->lb_len = buf_len; - tmp = (char *)object_update_param_get(update, 1, NULL); - if (tmp == NULL) { - CERROR("%s: empty flag for xattr set: rc = %d\n", - tgt_name(tsi->tsi_tgt), -EPROTO); + tmp = object_update_param_get(update, 1, &size); + if (tmp == NULL || IS_ERR(tmp) || size != sizeof(*tmp)) { + CERROR("%s: empty or wrong size %zu pos: rc = %d\n", + tgt_name(tsi->tsi_tgt), size, -EPROTO); RETURN(err_serious(-EPROTO)); } if (ptlrpc_req_need_swab(tsi->tsi_pill->rc_req)) - __swab64s((__u64 *)tmp); - pos = *(loff_t *)tmp; + __swab64s(tmp); + pos = *tmp; rc = out_tx_write(tsi->tsi_env, obj, lbuf, pos, - &tti->tti_tea, + &tti->tti_tea, tti->tti_tea.ta_handle, tti->tti_u.update.tti_update_reply, tti->tti_u.update.tti_update_reply_index); RETURN(rc); } +static int out_read(struct tgt_session_info *tsi) +{ + const struct lu_env *env = tsi->tsi_env; + struct tgt_thread_info *tti = tgt_th_info(env); + struct object_update *update = tti->tti_u.update.tti_update; + struct dt_object *obj = tti->tti_u.update.tti_dt_object; + struct object_update_reply *reply = tti->tti_u.update.tti_update_reply; + int index = tti->tti_u.update.tti_update_reply_index; + struct object_update_result *update_result; + struct lu_buf *lbuf = &tti->tti_buf; + struct out_read_reply *orr; + void *tmp; + size_t size; + __u64 pos; + int rc; + ENTRY; + + update_result = object_update_result_get(reply, index, NULL); + LASSERT(update_result != NULL); + update_result->our_datalen = sizeof(*orr); + + if (!lu_object_exists(&obj->do_lu)) + GOTO(out, rc = -ENOENT); + + tmp = object_update_param_get(update, 0, NULL); + if (tmp == NULL) { + CERROR("%s: empty size for read: rc = %d\n", + tgt_name(tsi->tsi_tgt), -EPROTO); + GOTO(out, rc = err_serious(-EPROTO)); + } + size = le64_to_cpu(*(size_t *)(tmp)); + + tmp = object_update_param_get(update, 1, NULL); + if (tmp == NULL) { + CERROR("%s: empty pos for read: rc = %d\n", + tgt_name(tsi->tsi_tgt), -EPROTO); + GOTO(out, rc = err_serious(-EPROTO)); + } + pos = le64_to_cpu(*(__u64 *)(tmp)); + + if (size > OUT_UPDATE_REPLY_SIZE - + cfs_size_round((unsigned long)update_result->our_data - + (unsigned long)update_result) - sizeof(pos)) { + CERROR("%s: get %zu the biggest read size is %d: rc = %d\n", + tgt_name(tsi->tsi_tgt), size, OUT_UPDATE_REPLY_SIZE, + -EPROTO); + GOTO(out, rc = err_serious(-EPROTO)); + } + + /* Put the offset into the begining of the buffer in reply */ + orr = (struct out_read_reply *)update_result->our_data; + + lbuf->lb_buf = orr->orr_data; + lbuf->lb_len = size; + + dt_read_lock(env, obj, MOR_TGT_CHILD); + rc = dt_read(env, obj, lbuf, &pos); + dt_read_unlock(env, obj); + orr->orr_size = rc < 0 ? 0 : rc; + orr->orr_offset = pos; + + orr_cpu_to_le(orr, orr); + update_result->our_datalen += orr->orr_size; +out: + /* Insert read buffer */ + update_result->our_rc = ptlrpc_status_hton(rc); + reply->ourp_lens[index] = cfs_size_round(update_result->our_datalen + + sizeof(*update_result)); + RETURN(rc); +} + #define DEF_OUT_HNDL(opc, name, flags, fn) \ [opc - OUT_CREATE] = { \ .th_name = name, \ @@ -1178,7 +673,6 @@ static int out_write(struct tgt_session_info *tsi) .th_version = 0, \ } -#define out_handler mdt_handler static struct tgt_handler out_update_ops[] = { DEF_OUT_HNDL(OUT_CREATE, "out_create", MUTABOR | HABEO_REFERO, out_create), @@ -1194,6 +688,8 @@ static struct tgt_handler out_update_ops[] = { out_attr_get), DEF_OUT_HNDL(OUT_XATTR_SET, "out_xattr_set", MUTABOR | HABEO_REFERO, out_xattr_set), + DEF_OUT_HNDL(OUT_XATTR_DEL, "out_xattr_del", MUTABOR | HABEO_REFERO, + out_xattr_del), DEF_OUT_HNDL(OUT_XATTR_GET, "out_xattr_get", HABEO_REFERO, out_xattr_get), DEF_OUT_HNDL(OUT_INDEX_LOOKUP, "out_index_lookup", HABEO_REFERO, @@ -1203,9 +699,11 @@ static struct tgt_handler out_update_ops[] = { DEF_OUT_HNDL(OUT_INDEX_DELETE, "out_index_delete", MUTABOR | HABEO_REFERO, out_index_delete), DEF_OUT_HNDL(OUT_WRITE, "out_write", MUTABOR | HABEO_REFERO, out_write), + DEF_OUT_HNDL(OUT_READ, "out_read", HABEO_REFERO, out_read), + }; -struct tgt_handler *out_handler_find(__u32 opc) +static struct tgt_handler *out_handler_find(__u32 opc) { struct tgt_handler *h; @@ -1223,18 +721,17 @@ struct tgt_handler *out_handler_find(__u32 opc) static int out_tx_start(const struct lu_env *env, struct dt_device *dt, struct thandle_exec_args *ta, struct obd_export *exp) { - memset(ta, 0, sizeof(*ta)); + ta->ta_argno = 0; ta->ta_handle = dt_trans_create(env, dt); if (IS_ERR(ta->ta_handle)) { int rc; rc = PTR_ERR(ta->ta_handle); ta->ta_handle = NULL; - CERROR("%s: start handle error: rc = %d\n", - dt_obd_name(dt), rc); + CERROR("%s: start handle error: rc = %d\n", dt_obd_name(dt), + rc); return rc; } - ta->ta_dev = dt; if (exp->exp_need_sync) ta->ta_handle->th_sync = 1; @@ -1244,7 +741,7 @@ static int out_tx_start(const struct lu_env *env, struct dt_device *dt, static int out_trans_start(const struct lu_env *env, struct thandle_exec_args *ta) { - return dt_trans_start(env, ta->ta_dev, ta->ta_handle); + return dt_trans_start(env, ta->ta_handle->th_dev, ta->ta_handle); } static int out_trans_stop(const struct lu_env *env, @@ -1254,73 +751,87 @@ static int out_trans_stop(const struct lu_env *env, int rc; ta->ta_handle->th_result = err; - rc = dt_trans_stop(env, ta->ta_dev, ta->ta_handle); + rc = dt_trans_stop(env, ta->ta_handle->th_dev, ta->ta_handle); for (i = 0; i < ta->ta_argno; i++) { - if (ta->ta_args[i].object != NULL) { - struct dt_object *obj = ta->ta_args[i].object; + if (ta->ta_args[i]->object != NULL) { + struct dt_object *obj = ta->ta_args[i]->object; /* If the object is being created during this * transaction, we need to remove them from the * cache immediately, because a few layers are * missing in OUT handler, i.e. the object might * not be initialized in all layers */ - if (ta->ta_args[i].exec_fn == out_tx_create_exec) + if (ta->ta_args[i]->exec_fn == out_tx_create_exec) set_bit(LU_OBJECT_HEARD_BANSHEE, &obj->do_lu.lo_header->loh_flags); - lu_object_put(env, &ta->ta_args[i].object->do_lu); - ta->ta_args[i].object = NULL; + lu_object_put(env, &ta->ta_args[i]->object->do_lu); + ta->ta_args[i]->object = NULL; } } + ta->ta_handle = NULL; + ta->ta_argno = 0; return rc; } -int out_tx_end(const struct lu_env *env, struct thandle_exec_args *ta) +static int out_tx_end(const struct lu_env *env, struct thandle_exec_args *ta, + int declare_ret) { - struct tgt_session_info *tsi = tgt_ses_info(env); - int i = 0, rc; + struct tgt_session_info *tsi = tgt_ses_info(env); + int i; + int rc; + int rc1; + ENTRY; - LASSERT(ta->ta_dev); - LASSERT(ta->ta_handle); + if (ta->ta_handle == NULL) + RETURN(0); - if (ta->ta_err != 0 || ta->ta_argno == 0) - GOTO(stop, rc = ta->ta_err); + if (declare_ret != 0 || ta->ta_argno == 0) + GOTO(stop, rc = declare_ret); + LASSERT(ta->ta_handle->th_dev != NULL); rc = out_trans_start(env, ta); - if (unlikely(rc)) + if (unlikely(rc != 0)) GOTO(stop, rc); for (i = 0; i < ta->ta_argno; i++) { - rc = ta->ta_args[i].exec_fn(env, ta->ta_handle, - &ta->ta_args[i]); + rc = ta->ta_args[i]->exec_fn(env, ta->ta_handle, + ta->ta_args[i]); if (unlikely(rc != 0)) { CDEBUG(D_INFO, "error during execution of #%u from" - " %s:%d: rc = %d\n", i, ta->ta_args[i].file, - ta->ta_args[i].line, rc); + " %s:%d: rc = %d\n", i, ta->ta_args[i]->file, + ta->ta_args[i]->line, rc); while (--i >= 0) { - if (ta->ta_args[i].undo_fn != NULL) - ta->ta_args[i].undo_fn(env, + if (ta->ta_args[i]->undo_fn != NULL) + ta->ta_args[i]->undo_fn(env, ta->ta_handle, - &ta->ta_args[i]); + ta->ta_args[i]); else CERROR("%s: undo for %s:%d: rc = %d\n", - dt_obd_name(ta->ta_dev), - ta->ta_args[i].file, - ta->ta_args[i].line, -ENOTSUPP); + dt_obd_name(ta->ta_handle->th_dev), + ta->ta_args[i]->file, + ta->ta_args[i]->line, -ENOTSUPP); } break; } + CDEBUG(D_INFO, "%s: executed %u/%u: rc = %d\n", + dt_obd_name(ta->ta_handle->th_dev), i, ta->ta_argno, rc); } - /* Only fail for real update */ - tsi->tsi_reply_fail_id = OBD_FAIL_OUT_UPDATE_NET_REP; + /* Only fail for real updates, XXX right now llog updates will be + * ignore, whose updates count is usually 1, so failover test + * case will spot this FAIL_UPDATE_NET_REP precisely, and it will + * be removed after async update patch is landed. */ + if (ta->ta_argno > 1) + tsi->tsi_reply_fail_id = OBD_FAIL_OUT_UPDATE_NET_REP; + stop: - CDEBUG(D_INFO, "%s: executed %u/%u: rc = %d\n", - dt_obd_name(ta->ta_dev), i, ta->ta_argno, rc); - out_trans_stop(env, ta, rc); + rc1 = out_trans_stop(env, ta, rc); + if (rc == 0) + rc = rc1; + ta->ta_handle = NULL; ta->ta_argno = 0; - ta->ta_err = 0; RETURN(rc); } @@ -1348,7 +859,7 @@ int out_handle(struct tgt_session_info *tsi) struct object_update_reply *reply; int bufsize; int count; - int old_batchid = -1; + int current_batchid = -1; int i; int rc = 0; int rc1 = 0; @@ -1399,11 +910,6 @@ int out_handle(struct tgt_session_info *tsi) RETURN(err_serious(-EPROTO)); object_update_reply_init(reply, count); tti->tti_u.update.tti_update_reply = reply; - - rc = out_tx_start(env, dt, ta, tsi->tsi_exp); - if (rc != 0) - RETURN(rc); - tti->tti_mult_trans = !req_is_replay(tgt_ses_req(tsi)); /* Walk through updates in the request to execute them synchronously */ @@ -1418,21 +924,6 @@ int out_handle(struct tgt_session_info *tsi) if (ptlrpc_req_need_swab(pill->rc_req)) lustre_swab_object_update(update); - if (old_batchid == -1) { - old_batchid = update->ou_batchid; - } else if (old_batchid != update->ou_batchid) { - /* Stop the current update transaction, - * create a new one */ - rc = out_tx_end(env, ta); - if (rc < 0) - RETURN(rc); - - rc = out_tx_start(env, dt, ta, tsi->tsi_exp); - if (rc != 0) - RETURN(rc); - old_batchid = update->ou_batchid; - } - if (!fid_is_sane(&update->ou_fid)) { CERROR("%s: invalid FID "DFID": rc = %d\n", tgt_name(tsi->tsi_tgt), PFID(&update->ou_fid), @@ -1446,8 +937,10 @@ int out_handle(struct tgt_session_info *tsi) if (dt->dd_record_fid_accessed) { lfsck_pack_rfa(&tti->tti_lr, - lu_object_fid(&dt_obj->do_lu)); - tgt_lfsck_in_notify(env, dt, &tti->tti_lr); + lu_object_fid(&dt_obj->do_lu), + LE_FID_ACCESSED, + LFSCK_TYPE_LAYOUT); + tgt_lfsck_in_notify(env, dt, &tti->tti_lr, NULL); } tti->tti_u.update.tti_dt_object = dt_obj; @@ -1455,33 +948,67 @@ int out_handle(struct tgt_session_info *tsi) tti->tti_u.update.tti_update_reply_index = i; h = out_handler_find(update->ou_type); - if (likely(h != NULL)) { - /* For real modification RPC, check if the update - * has been executed */ - if (h->th_flags & MUTABOR) { - struct ptlrpc_request *req = tgt_ses_req(tsi); + if (unlikely(h == NULL)) { + CERROR("%s: unsupported opc: 0x%x\n", + tgt_name(tsi->tsi_tgt), update->ou_type); + GOTO(next, rc = -ENOTSUPP); + } + + /* Check resend case only for modifying RPC */ + if (h->th_flags & MUTABOR) { + struct ptlrpc_request *req = tgt_ses_req(tsi); + + if (out_check_resent(env, dt, dt_obj, req, + out_reconstruct, reply, i)) + GOTO(next, rc = 0); + } + + /* start transaction for modification RPC only */ + if (h->th_flags & MUTABOR && current_batchid == -1) { + current_batchid = update->ou_batchid; + rc = out_tx_start(env, dt, ta, tsi->tsi_exp); + if (rc != 0) + GOTO(next, rc); + + if (update->ou_flags & UPDATE_FL_SYNC) + ta->ta_handle->th_sync = 1; + } + + /* Stop the current update transaction, if the update has + * different batchid, or read-only update */ + if (((current_batchid != update->ou_batchid) || + !(h->th_flags & MUTABOR)) && ta->ta_handle != NULL) { + rc = out_tx_end(env, ta, rc); + current_batchid = -1; + if (rc != 0) + GOTO(next, rc); - if (out_check_resent(env, dt, dt_obj, req, - out_reconstruct, reply, i)) + /* start a new transaction if needed */ + if (h->th_flags & MUTABOR) { + rc = out_tx_start(env, dt, ta, tsi->tsi_exp); + if (rc != 0) GOTO(next, rc); - } - rc = h->th_act(tsi); - } else { - CERROR("%s: The unsupported opc: 0x%x\n", - tgt_name(tsi->tsi_tgt), update->ou_type); - lu_object_put(env, &dt_obj->do_lu); - GOTO(out, rc = -ENOTSUPP); + if (update->ou_flags & UPDATE_FL_SYNC) + ta->ta_handle->th_sync = 1; + + current_batchid = update->ou_batchid; + } } + + rc = h->th_act(tsi); next: lu_object_put(env, &dt_obj->do_lu); if (rc < 0) GOTO(out, rc); } out: - rc1 = out_tx_end(env, ta); - if (rc == 0) - rc = rc1; + if (current_batchid != -1) { + rc1 = out_tx_end(env, ta, rc); + if (rc == 0) + rc = rc1; + } + RETURN(rc); }