- if (lustre_msg_get_transno(req->rq_reqmsg) != 0) {
- pre_versions = lustre_msg_get_versions(req->rq_reqmsg);
- LASSERT(index < PTLRPC_NUM_VERSIONS);
- /** Sanity check for malformed buffers */
- if (pre_versions == NULL) {
- CERROR("No versions in request buffer\n");
- spin_lock(&req->rq_export->exp_lock);
- req->rq_export->exp_vbr_failed = 1;
- spin_unlock(&req->rq_export->exp_lock);
- RETURN(-EOVERFLOW);
- } else if (pre_versions[index] != curr_version) {
- CDEBUG(D_INODE, "Version mismatch "LPX64" != "LPX64"\n",
- pre_versions[index], curr_version);
- spin_lock(&req->rq_export->exp_lock);
- req->rq_export->exp_vbr_failed = 1;
- spin_unlock(&req->rq_export->exp_lock);
- RETURN(-EOVERFLOW);
+ LASSERT(idx < PTLRPC_NUM_VERSIONS);
+ /** Sanity check for malformed buffers */
+ if (pre_ver == NULL) {
+ CERROR("No versions in request buffer\n");
+ spin_lock(&req->rq_export->exp_lock);
+ req->rq_export->exp_vbr_failed = 1;
+ spin_unlock(&req->rq_export->exp_lock);
+ RETURN(-EOVERFLOW);
+ } else if (pre_ver[idx] != version) {
+ CDEBUG(D_INODE, "Version mismatch "LPX64" != "LPX64"\n",
+ pre_ver[idx], version);
+ spin_lock(&req->rq_export->exp_lock);
+ req->rq_export->exp_vbr_failed = 1;
+ spin_unlock(&req->rq_export->exp_lock);
+ RETURN(-EOVERFLOW);
+ }
+ RETURN(0);
+}
+
+/**
+ * Save pre-versions in reply.
+ */
+static void mdt_version_save(struct ptlrpc_request *req, __u64 version,
+ int idx)
+{
+ __u64 *reply_ver;
+
+ if (!exp_connect_vbr(req->rq_export))
+ return;
+
+ LASSERT(!req_is_replay(req));
+ LASSERT(req->rq_repmsg != NULL);
+ reply_ver = lustre_msg_get_versions(req->rq_repmsg);
+ if (reply_ver)
+ reply_ver[idx] = version;
+}
+
+/**
+ * Save enoent version, it is needed when it is obvious that object doesn't
+ * exist, e.g. child during create.
+ */
+static void mdt_enoent_version_save(struct mdt_thread_info *info, int idx)
+{
+ /* save version of file name for replay, it must be ENOENT here */
+ if (!req_is_replay(mdt_info_req(info))) {
+ info->mti_ver[idx] = ENOENT_VERSION;
+ mdt_version_save(mdt_info_req(info), info->mti_ver[idx], idx);
+ }
+}
+
+/**
+ * Get version from disk and save in reply buffer.
+ *
+ * Versions are saved in reply only during normal operations not replays.
+ */
+void mdt_version_get_save(struct mdt_thread_info *info,
+ struct mdt_object *mto, int idx)
+{
+ /* don't save versions during replay */
+ if (!req_is_replay(mdt_info_req(info))) {
+ mdt_obj_version_get(info, mto, &info->mti_ver[idx]);
+ mdt_version_save(mdt_info_req(info), info->mti_ver[idx], idx);
+ }
+}
+
+/**
+ * Get version from disk and check it, no save in reply.
+ */
+int mdt_version_get_check(struct mdt_thread_info *info,
+ struct mdt_object *mto, int idx)
+{
+ /* only check versions during replay */
+ if (!req_is_replay(mdt_info_req(info)))
+ return 0;
+
+ mdt_obj_version_get(info, mto, &info->mti_ver[idx]);
+ return mdt_version_check(mdt_info_req(info), info->mti_ver[idx], idx);
+}
+
+/**
+ * Get version from disk and check if recovery or just save.
+ */
+int mdt_version_get_check_save(struct mdt_thread_info *info,
+ struct mdt_object *mto, int idx)
+{
+ int rc = 0;
+
+ mdt_obj_version_get(info, mto, &info->mti_ver[idx]);
+ if (req_is_replay(mdt_info_req(info)))
+ rc = mdt_version_check(mdt_info_req(info), info->mti_ver[idx],
+ idx);
+ else
+ mdt_version_save(mdt_info_req(info), info->mti_ver[idx], idx);
+ return rc;
+}
+
+/**
+ * Lookup with version checking.
+ *
+ * This checks version of 'name'. Many reint functions uses 'name' for child not
+ * FID, therefore we need to get object by name and check its version.
+ */
+int mdt_lookup_version_check(struct mdt_thread_info *info,
+ struct mdt_object *p, struct lu_name *lname,
+ struct lu_fid *fid, int idx)
+{
+ int rc, vbrc;
+
+ rc = mdo_lookup(info->mti_env, mdt_object_child(p), lname, fid,
+ &info->mti_spec);
+ /* Check version only during replay */
+ if (!req_is_replay(mdt_info_req(info)))
+ return rc;
+
+ info->mti_ver[idx] = ENOENT_VERSION;
+ if (rc == 0) {
+ struct mdt_object *child;
+ child = mdt_object_find(info->mti_env, info->mti_mdt, fid);
+ if (likely(!IS_ERR(child))) {
+ mdt_obj_version_get(info, child, &info->mti_ver[idx]);
+ mdt_object_put(info->mti_env, child);