lustre/mds/mds_reint.c

   1 /* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
   2  * vim:expandtab:shiftwidth=8:tabstop=8:
   3  *
   4  * GPL HEADER START
   5  *
   6  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
   7  *
   8  * This program is free software; you can redistribute it and/or modify
   9  * it under the terms of the GNU General Public License version 2 only,
  10  * as published by the Free Software Foundation.
  11  *
  12  * This program is distributed in the hope that it will be useful, but
  13  * WITHOUT ANY WARRANTY; without even the implied warranty of
  14  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  15  * General Public License version 2 for more details (a copy is included
  16  * in the LICENSE file that accompanied this code).
  17  *
  18  * You should have received a copy of the GNU General Public License
  19  * version 2 along with this program; If not, see
  20  * http://www.sun.com/software/products/lustre/docs/GPLv2.pdf
  21  *
  22  * Please contact Sun Microsystems, Inc., 4150 Network Circle, Santa Clara,
  23  * CA 95054 USA or visit www.sun.com if you need additional information or
  24  * have any questions.
  25  *
  26  * GPL HEADER END
  27  */
  28 /*
  29  * Copyright (c) 2002, 2010, Oracle and/or its affiliates. All rights reserved.
  30  * Use is subject to license terms.
  31  */
  32 /*
  33  * This file is part of Lustre, http://www.lustre.org/
  34  * Lustre is a trademark of Sun Microsystems, Inc.
  35  *
  36  * lustre/mds/mds_reint.c
  37  *
  38  * Lustre Metadata Server (mds) reintegration routines
  39  *
  40  * Author: Peter Braam <braam@clusterfs.com>
  41  * Author: Andreas Dilger <adilger@clusterfs.com>
  42  * Author: Phil Schwan <phil@clusterfs.com>
  43  */
  44
  45 #ifndef EXPORT_SYMTAB
  46 # define EXPORT_SYMTAB
  47 #endif
  48 #define DEBUG_SUBSYSTEM S_MDS
  49
  50 #include <linux/fs.h>
  51 #include <obd_support.h>
  52 #include <obd_class.h>
  53 #include <obd.h>
  54 #include <lustre_lib.h>
  55 #include <lustre/lustre_idl.h>
  56 #include <lustre_mds.h>
  57 #include <lustre_dlm.h>
  58 #include <lustre_fsfilt.h>
  59 #include <lustre_ucache.h>
  60 #include <lustre_net.h>
  61
  62 #include "mds_internal.h"
  63
  64 void mds_commit_cb(struct obd_device *obd, __u64 transno, void *data,
  65                    int error)
  66 {
  67         struct obd_export *exp = data;
  68         LASSERTF(exp->exp_obd == obd,
  69                  "%s: bad export (%p), obd (%p) != exp->exp_obd (%p)\n",
  70                  obd->obd_name, exp, obd, exp->exp_obd);
  71         obd_transno_commit_cb(obd, transno, exp, error);
  72         class_export_put(exp);
  73 }
  74
  75 struct mds_logcancel_data {
  76         struct lov_mds_md      *mlcd_lmm;
  77         int                     mlcd_size;
  78         int                     mlcd_cookielen;
  79         int                     mlcd_eadatalen;
  80         struct llog_cookie      mlcd_cookies[0];
  81 };
  82
  83 /** lookup child dentry in parent dentry according to the name.
  84  *  if dentry is found, delete "lustre_mdt_attrs" EA (with name "lma")
  85  *  if it exists by checking OBD_INCOMPAT_FID.
  86  */
  87 struct dentry *mds_lookup(struct obd_device *obd, const char *fid_name,
  88                           struct dentry *dparent, int fid_namelen)
  89 {
  90         struct dentry *dchild;
  91         struct lr_server_data *lsd = obd->u.mds.mds_server_data;
  92         ENTRY;
  93
  94         dchild = ll_lookup_one_len(fid_name, dparent, fid_namelen);
  95         if (!IS_ERR(dchild) &&
  96             unlikely((lsd->lsd_feature_incompat & OBD_INCOMPAT_FID) ||
  97                       OBD_FAIL_CHECK(OBD_FAIL_MDS_REMOVE_COMMON_EA))) {
  98                 struct inode *inode = dchild->d_inode;
  99                 void         *handle;
 100                 if (inode != NULL) {
 101                         LOCK_INODE_MUTEX(inode);
 102                         handle = fsfilt_start(obd, inode, FSFILT_OP_SETATTR,
 103                                               NULL);
 104                         if (!IS_ERR(handle)) {
 105                                 fsfilt_set_md(obd, inode, handle, NULL, 0,
 106                                               "lma");
 107                                 /* result is ignored. */
 108                                 UNLOCK_INODE_MUTEX(inode);
 109                                 fsfilt_commit(obd, inode, handle, 0);
 110                         } else {
 111                                 UNLOCK_INODE_MUTEX(inode);
 112                         }
 113                 }
 114         }
 115         RETURN(dchild);
 116 }
 117
 118 static void mds_cancel_cookies_cb(struct obd_device *obd, __u64 transno,
 119                                   void *cb_data, int error)
 120 {
 121         struct mds_logcancel_data *mlcd = cb_data;
 122         struct lov_stripe_md *lsm = NULL;
 123         struct llog_ctxt *ctxt;
 124         int rc;
 125
 126         obd_transno_commit_cb(obd, transno, NULL, error);
 127
 128         CDEBUG(D_RPCTRACE, "cancelling %d cookies\n",
 129                (int)(mlcd->mlcd_cookielen / sizeof(*mlcd->mlcd_cookies)));
 130
 131         rc = obd_unpackmd(obd->u.mds.mds_lov_exp, &lsm, mlcd->mlcd_lmm,
 132                           mlcd->mlcd_eadatalen);
 133         if (rc < 0) {
 134                 CERROR("bad LSM cancelling %d log cookies: rc %d\n",
 135                        (int)(mlcd->mlcd_cookielen/sizeof(*mlcd->mlcd_cookies)),
 136                        rc);
 137         } else {
 138                 rc = obd_checkmd(obd->u.mds.mds_lov_exp, obd->obd_self_export,
 139                                  lsm);
 140                 if (rc)
 141                         CERROR("Can not revalidate lsm %p \n", lsm);
 142
 143                 ctxt = llog_get_context(obd,mlcd->mlcd_cookies[0].lgc_subsys+1);
 144                 /* XXX 0 normally, SENDNOW for debug */
 145                 rc = llog_cancel(ctxt, lsm, mlcd->mlcd_cookielen /
 146                                                 sizeof(*mlcd->mlcd_cookies),
 147                                  mlcd->mlcd_cookies, OBD_LLOG_FL_SENDNOW);
 148                 llog_ctxt_put(ctxt);
 149
 150                 if (rc)
 151                         CERROR("error cancelling %d log cookies: rc %d\n",
 152                                (int)(mlcd->mlcd_cookielen /
 153                                      sizeof(*mlcd->mlcd_cookies)), rc);
 154         }
 155
 156         OBD_FREE(mlcd, mlcd->mlcd_size);
 157 }
 158
 159 /* fsfilt_set_version return old version. use that here */
 160 static void mds_versions_set(struct obd_device *obd,
 161                              struct inode **inodes, __u64 version)
 162 {
 163         int i;
 164
 165         if (inodes == NULL)
 166                 return;
 167
 168         for (i = 0; i < PTLRPC_NUM_VERSIONS; i++)
 169                 if (inodes[i] != NULL)
 170                         fsfilt_set_version(obd, inodes[i], version);
 171 }
 172
 173 int mds_version_get_check(struct ptlrpc_request *req, struct inode *inode,
 174                           int index)
 175 {
 176         /* version recovery */
 177         struct obd_device *obd = req->rq_export->exp_obd;
 178         __u64 curr_version, *pre_versions;
 179         ENTRY;
 180
 181         if (inode == NULL || !exp_connect_vbr(req->rq_export))
 182                 RETURN(0);
 183
 184         curr_version = fsfilt_get_version(obd, inode);
 185         if ((__s64)curr_version == -EOPNOTSUPP)
 186                 RETURN(0);
 187         /* VBR: version is checked always because costs nothing */
 188         if (lustre_msg_get_transno(req->rq_reqmsg) != 0) {
 189                 pre_versions = lustre_msg_get_versions(req->rq_reqmsg);
 190                 LASSERT(index < PTLRPC_NUM_VERSIONS);
 191                 /* Sanity check for malformed buffers */
 192                 if (pre_versions == NULL) {
 193                         CERROR("No versions in request buffer\n");
 194                         spin_lock(&req->rq_export->exp_lock);
 195                         req->rq_export->exp_vbr_failed = 1;
 196                         spin_unlock(&req->rq_export->exp_lock);
 197                         RETURN(-EOVERFLOW);
 198                 } else if (pre_versions[index] != curr_version) {
 199                         CDEBUG(D_INODE, "Version mismatch "LPX64" != "LPX64"\n",
 200                                pre_versions[index], curr_version);
 201                         spin_lock(&req->rq_export->exp_lock);
 202                         req->rq_export->exp_vbr_failed = 1;
 203                         spin_unlock(&req->rq_export->exp_lock);
 204                         RETURN(-EOVERFLOW);
 205                 }
 206         }
 207         /* save pre-versions in reply */
 208         LASSERT(req->rq_repmsg != NULL);
 209         pre_versions = lustre_msg_get_versions(req->rq_repmsg);
 210         if (pre_versions)
 211                 pre_versions[index] = curr_version;
 212         RETURN(0);
 213 }
 214
 215 /* Assumes caller has already pushed us into the kernel context. */
 216 int mds_finish_transno(struct mds_obd *mds, struct inode **inodes, void *handle,
 217                        struct ptlrpc_request *req, int rc, __u32 op_data,
 218                        int force_sync)
 219 {
 220         struct mds_export_data *med = &req->rq_export->exp_mds_data;
 221         struct lsd_client_data *lcd = med->med_lcd;
 222         struct obd_device *obd = req->rq_export->exp_obd;
 223         __u64 transno, prev_transno;
 224         int err;
 225         loff_t off;
 226         int log_pri = D_RPCTRACE;
 227         struct inode *inode = inodes ? inodes[0] : NULL;
 228         int version_set = handle ? 1 : 0;
 229         ENTRY;
 230
 231         if (IS_ERR(handle)) {
 232                 LASSERT(rc != 0);
 233                 RETURN(rc);
 234         }
 235
 236         /* if the export has already been failed, we have no last_rcvd slot */
 237         if (req->rq_export->exp_failed || obd->obd_fail) {
 238                 CWARN("commit transaction for disconnected client %s: rc %d\n",
 239                       req->rq_export->exp_client_uuid.uuid, rc);
 240                 if (rc == 0)
 241                         rc = -ENOTCONN;
 242                 if (handle)
 243                         GOTO(commit, rc);
 244                 RETURN(rc);
 245         }
 246
 247         if (handle == NULL) {
 248                 /* if we're starting our own xaction, use our own inode */
 249                 inode = mds->mds_rcvd_filp->f_dentry->d_inode;
 250                 handle = fsfilt_start(obd, inode, FSFILT_OP_SETATTR, NULL);
 251                 if (IS_ERR(handle)) {
 252                         CERROR("fsfilt_start: %ld\n", PTR_ERR(handle));
 253                         RETURN(PTR_ERR(handle));
 254                 }
 255         }
 256
 257         off = med->med_lr_off;
 258
 259         transno = lustre_msg_get_transno(req->rq_reqmsg);
 260         if (rc != 0) {
 261                 if (transno != 0) {
 262                         CERROR("%s: replay %s transno "LPU64" failed: rc %d\n",
 263                                obd->obd_name,
 264                                libcfs_nid2str(req->rq_export->exp_connection->c_peer.nid),
 265                                transno, rc);
 266                 }
 267         } else if (transno == 0) {
 268                 spin_lock(&mds->mds_transno_lock);
 269                 transno = ++mds->mds_last_transno;
 270                 spin_unlock(&mds->mds_transno_lock);
 271                 /* VBR: set versions */
 272                 if (inodes && version_set)
 273                         mds_versions_set(obd, inodes, transno);
 274         } else {
 275                 spin_lock(&mds->mds_transno_lock);
 276                 if (transno > mds->mds_last_transno)
 277                         mds->mds_last_transno = transno;
 278                 spin_unlock(&mds->mds_transno_lock);
 279
 280                 /* VBR: replay case. Copy version from replay req and
 281                  * set new versions */
 282                 mds_versions_set(obd, inodes, transno);
 283         }
 284
 285         req->rq_transno = transno;
 286         lustre_msg_set_transno(req->rq_repmsg, transno);
 287
 288         if (transno == 0)
 289                 LASSERT(rc != 0);
 290         if (lustre_msg_get_opc(req->rq_reqmsg) == MDS_CLOSE) {
 291                 if (transno != 0)
 292                         lcd->lcd_last_close_transno = cpu_to_le64(transno);
 293                 lcd->lcd_last_close_xid = cpu_to_le64(req->rq_xid);
 294                 lcd->lcd_last_close_result = cpu_to_le32(rc);
 295                 lcd->lcd_last_close_data = cpu_to_le32(op_data);
 296         } else {
 297                 prev_transno = le64_to_cpu(lcd->lcd_last_transno);
 298                 if (((lustre_msg_get_flags(req->rq_reqmsg) &
 299                       (MSG_RESENT | MSG_REPLAY)) == 0) ||
 300                     (transno > prev_transno)) {
 301                         /* VBR: save versions in last_rcvd for reconstruct. */
 302                         __u64 *pre_versions = lustre_msg_get_versions(req->rq_repmsg);
 303                         if (pre_versions) {
 304                                 lcd->lcd_pre_versions[0] = cpu_to_le64(pre_versions[0]);
 305                                 lcd->lcd_pre_versions[1] = cpu_to_le64(pre_versions[1]);
 306                                 lcd->lcd_pre_versions[2] = cpu_to_le64(pre_versions[2]);
 307                                 lcd->lcd_pre_versions[3] = cpu_to_le64(pre_versions[3]);
 308                         }
 309                         if (transno != 0)
 310                                 lcd->lcd_last_transno = cpu_to_le64(transno);
 311                         lcd->lcd_last_xid     = cpu_to_le64(req->rq_xid);
 312                         lcd->lcd_last_result  = cpu_to_le32(rc);
 313                         lcd->lcd_last_data    = cpu_to_le32(op_data);
 314                 }
 315         }
 316         /** update trans table */
 317         target_trans_table_update(req->rq_export, transno);
 318
 319         if (off <= 0) {
 320                 CERROR("client idx %d has offset %lld\n", med->med_lr_idx, off);
 321                 err = -EINVAL;
 322         } else {
 323                 struct obd_export *exp = req->rq_export;
 324
 325                 class_export_get(exp); /* released once the cb is called */
 326                 if (!force_sync)
 327                         force_sync = fsfilt_add_journal_cb(obd, transno,
 328                                                            handle, mds_commit_cb,
 329                                                            exp);
 330
 331                 err = fsfilt_write_record(obd, mds->mds_rcvd_filp, lcd,
 332                                           sizeof(*lcd), &off,
 333                                           force_sync | exp->exp_need_sync);
 334                 if (force_sync)
 335                         mds_commit_cb(obd, transno, exp, err);
 336         }
 337
 338         if (err) {
 339                 log_pri = D_ERROR;
 340                 if (rc == 0)
 341                         rc = err;
 342         }
 343
 344         DEBUG_REQ(log_pri, req,
 345                   "wrote trans #"LPU64" rc %d client %s at idx %u: err = %d",
 346                   transno, rc, lcd->lcd_uuid, med->med_lr_idx, err);
 347
 348         err = mds_lov_write_objids(obd);
 349         if (err) {
 350                 log_pri = D_ERROR;
 351                 if (rc == 0)
 352                         rc = err;
 353         }
 354         CDEBUG(log_pri, "wrote objids: err = %d\n", err);
 355
 356 commit:
 357         err = fsfilt_commit(obd, inode, handle, 0);
 358         if (err) {
 359                 CERROR("error committing transaction: %d\n", err);
 360                 if (!rc)
 361                         rc = err;
 362         }
 363
 364         RETURN(rc);
 365 }
 366
 367 /* this gives the same functionality as the code between
 368  * sys_chmod and inode_setattr
 369  * chown_common and inode_setattr
 370  * utimes and inode_setattr
 371  */
 372 int mds_fix_attr(struct inode *inode, struct mds_update_record *rec)
 373 {
 374         time_t now = CURRENT_SECONDS;
 375         struct iattr *attr = &rec->ur_iattr;
 376         unsigned int ia_valid = attr->ia_valid;
 377         int error;
 378         ENTRY;
 379
 380         if (ia_valid & ATTR_RAW)
 381                 attr->ia_valid &= ~ATTR_RAW;
 382
 383         if (!(ia_valid & ATTR_CTIME_SET))
 384                 LTIME_S(attr->ia_ctime) = now;
 385         else
 386                 attr->ia_valid &= ~ATTR_CTIME_SET;
 387         if (!(ia_valid & ATTR_ATIME_SET))
 388                 LTIME_S(attr->ia_atime) = now;
 389         if (!(ia_valid & ATTR_MTIME_SET))
 390                 LTIME_S(attr->ia_mtime) = now;
 391
 392         if (IS_IMMUTABLE(inode) || IS_APPEND(inode))
 393                 RETURN((attr->ia_valid & ~ATTR_ATTR_FLAG) ? -EPERM : 0);
 394
 395         /* times */
 396         if ((ia_valid & (ATTR_MTIME|ATTR_ATIME)) == (ATTR_MTIME|ATTR_ATIME)) {
 397                 if (current_fsuid() != inode->i_uid &&
 398                     (error = ll_permission(inode, MAY_WRITE, NULL)) != 0)
 399                         RETURN(error);
 400         }
 401
 402         if (ia_valid & ATTR_SIZE &&
 403             /* NFSD hack for open(O_CREAT|O_TRUNC)=mknod+truncate (bug 5781) */
 404             !(rec->ur_uc.luc_fsuid == inode->i_uid &&
 405               ia_valid & MDS_OPEN_OWNEROVERRIDE)) {
 406                 if ((error = ll_permission(inode, MAY_WRITE, NULL)) != 0)
 407                         RETURN(error);
 408         }
 409
 410         if (ia_valid & (ATTR_UID | ATTR_GID)) {
 411                 /* chown */
 412                 error = -EPERM;
 413                 if (IS_IMMUTABLE(inode) || IS_APPEND(inode))
 414                         RETURN(-EPERM);
 415                 if (attr->ia_uid == (uid_t) -1)
 416                         attr->ia_uid = inode->i_uid;
 417                 if (attr->ia_gid == (gid_t) -1)
 418                         attr->ia_gid = inode->i_gid;
 419                 if (!(ia_valid & ATTR_MODE))
 420                         attr->ia_mode = inode->i_mode;
 421                 /*
 422                  * If the user or group of a non-directory has been
 423                  * changed by a non-root user, remove the setuid bit.
 424                  * 19981026 David C Niemi <niemi@tux.org>
 425                  *
 426                  * Changed this to apply to all users, including root,
 427                  * to avoid some races. This is the behavior we had in
 428                  * 2.0. The check for non-root was definitely wrong
 429                  * for 2.2 anyway, as it should have been using
 430                  * CAP_FSETID rather than fsuid -- 19990830 SD.
 431                  */
 432                 if ((inode->i_mode & S_ISUID) == S_ISUID &&
 433                     !S_ISDIR(inode->i_mode)) {
 434                         attr->ia_mode &= ~S_ISUID;
 435                         attr->ia_valid |= ATTR_MODE;
 436                 }
 437                 /*
 438                  * Likewise, if the user or group of a non-directory
 439                  * has been changed by a non-root user, remove the
 440                  * setgid bit UNLESS there is no group execute bit
 441                  * (this would be a file marked for mandatory
 442                  * locking).  19981026 David C Niemi <niemi@tux.org>
 443                  *
 444                  * Removed the fsuid check (see the comment above) --
 445                  * 19990830 SD.
 446                  */
 447                 if (((inode->i_mode & (S_ISGID | S_IXGRP)) ==
 448                      (S_ISGID | S_IXGRP)) && !S_ISDIR(inode->i_mode)) {
 449                         attr->ia_mode &= ~S_ISGID;
 450                         attr->ia_valid |= ATTR_MODE;
 451                 }
 452         } else if (ia_valid & ATTR_MODE) {
 453                 int mode;
 454                 if (!(attr->ia_valid & ATTR_FORCE)) {
 455                         mode = inode->i_mode;
 456                         if (((mode & S_ISUID) && (!(attr->ia_mode & S_ISUID))) ||
 457                             ((mode & S_ISGID) && (mode & S_IXGRP) &&
 458                             (!(attr->ia_mode & S_ISGID))))
 459                                 attr->ia_valid |= ATTR_FORCE;
 460                 }
 461                 mode = attr->ia_mode;
 462                 /* chmod */
 463                 if (attr->ia_mode == (umode_t)-1)
 464                         mode = inode->i_mode;
 465                 attr->ia_mode =
 466                         (mode & S_IALLUGO) | (inode->i_mode & ~S_IALLUGO);
 467         }
 468         RETURN(0);
 469 }
 470
 471 void mds_steal_ack_locks(struct ptlrpc_request *req)
 472 {
 473         struct obd_export         *exp = req->rq_export;
 474         struct list_head          *tmp;
 475         struct ptlrpc_reply_state *oldrep;
 476         struct ptlrpc_service     *svc;
 477         int                        i;
 478
 479         /* CAVEAT EMPTOR: spinlock order */
 480         spin_lock(&exp->exp_lock);
 481         list_for_each (tmp, &exp->exp_outstanding_replies) {
 482                 oldrep = list_entry(tmp, struct ptlrpc_reply_state,rs_exp_list);
 483
 484                 if (oldrep->rs_xid != req->rq_xid)
 485                         continue;
 486
 487                 if (lustre_msg_get_opc(oldrep->rs_msg) !=
 488                     lustre_msg_get_opc(req->rq_reqmsg))
 489                         CERROR ("Resent req xid "LPU64" has mismatched opc: "
 490                                 "new %d old %d\n", req->rq_xid,
 491                                 lustre_msg_get_opc(req->rq_reqmsg),
 492                                 lustre_msg_get_opc(oldrep->rs_msg));
 493
 494                 svc = oldrep->rs_service;
 495                 spin_lock (&svc->srv_lock);
 496
 497                 list_del_init (&oldrep->rs_exp_list);
 498
 499                 CWARN("Stealing %d locks from rs %p x"LPD64".t"LPD64
 500                       " o%d NID %s\n",
 501                       oldrep->rs_nlocks, oldrep,
 502                       oldrep->rs_xid, oldrep->rs_transno,
 503                       lustre_msg_get_opc(oldrep->rs_msg),
 504                       libcfs_nid2str(exp->exp_connection->c_peer.nid));
 505
 506                 for (i = 0; i < oldrep->rs_nlocks; i++)
 507                         ptlrpc_save_lock(req,
 508                                          &oldrep->rs_locks[i],
 509                                          oldrep->rs_modes[i]);
 510                 oldrep->rs_nlocks = 0;
 511
 512                 DEBUG_REQ(D_DLMTRACE, req, "stole locks for");
 513                 ptlrpc_schedule_difficult_reply (oldrep);
 514
 515                 spin_unlock (&svc->srv_lock);
 516                 break;
 517         }
 518         spin_unlock(&exp->exp_lock);
 519 }
 520
 521 /**
 522  * VBR: restore versions
 523  */
 524 void mds_vbr_reconstruct(struct ptlrpc_request *req,
 525                          struct lsd_client_data *lcd)
 526 {
 527         __u64 pre_versions[4] = {0};
 528         pre_versions[0] = le64_to_cpu(lcd->lcd_pre_versions[0]);
 529         pre_versions[1] = le64_to_cpu(lcd->lcd_pre_versions[1]);
 530         pre_versions[2] = le64_to_cpu(lcd->lcd_pre_versions[2]);
 531         pre_versions[3] = le64_to_cpu(lcd->lcd_pre_versions[3]);
 532         lustre_msg_set_versions(req->rq_repmsg, pre_versions);
 533 }
 534
 535 void mds_req_from_lcd(struct ptlrpc_request *req, struct lsd_client_data *lcd)
 536 {
 537         if (lustre_msg_get_opc(req->rq_reqmsg) == MDS_CLOSE) {
 538                 req->rq_transno = le64_to_cpu(lcd->lcd_last_close_transno);
 539                 req->rq_status = le32_to_cpu(lcd->lcd_last_close_result);
 540         } else {
 541                 req->rq_transno = le64_to_cpu(lcd->lcd_last_transno);
 542                 req->rq_status = le32_to_cpu(lcd->lcd_last_result);
 543                 mds_vbr_reconstruct(req, lcd);
 544         }
 545         if (req->rq_status != 0)
 546                 req->rq_transno = 0;
 547         lustre_msg_set_transno(req->rq_repmsg, req->rq_transno);
 548         lustre_msg_set_status(req->rq_repmsg, req->rq_status);
 549         DEBUG_REQ(D_RPCTRACE, req, "restoring transno "LPD64"/status %d",
 550                   req->rq_transno, req->rq_status);
 551
 552         mds_steal_ack_locks(req);
 553 }
 554
 555 static void reconstruct_reint_setattr(struct mds_update_record *rec,
 556                                       int offset, struct ptlrpc_request *req)
 557 {
 558         struct obd_export *exp = req->rq_export;
 559         struct mds_export_data *med = &exp->exp_mds_data;
 560         struct mds_obd *obd = &exp->exp_obd->u.mds;
 561         struct dentry *de;
 562         struct mds_body *body;
 563
 564         mds_req_from_lcd(req, med->med_lcd);
 565
 566         de = mds_fid2dentry(obd, rec->ur_fid1, NULL);
 567         if (IS_ERR(de)) {
 568                 int rc;
 569                 rc = PTR_ERR(de);
 570                 LCONSOLE_WARN("FID "LPU64"/%u lookup error %d."
 571                               " Evicting client %s with export %s.\n",
 572                               rec->ur_fid1->id, rec->ur_fid1->generation, rc,
 573                               obd_uuid2str(&exp->exp_client_uuid),
 574                               obd_export_nid2str(exp));
 575                 mds_export_evict(exp);
 576                 EXIT;
 577                 return;
 578         }
 579
 580         body = lustre_msg_buf(req->rq_repmsg, offset, sizeof(*body));
 581         mds_pack_inode2body(body, de->d_inode);
 582
 583         /* Don't return OST-specific attributes if we didn't just set them */
 584         if (rec->ur_iattr.ia_valid & ATTR_SIZE)
 585                 body->valid |= OBD_MD_FLSIZE | OBD_MD_FLBLOCKS;
 586         if (rec->ur_iattr.ia_valid & (ATTR_MTIME | ATTR_MTIME_SET))
 587                 body->valid |= OBD_MD_FLMTIME;
 588         if (rec->ur_iattr.ia_valid & (ATTR_ATIME | ATTR_ATIME_SET))
 589                 body->valid |= OBD_MD_FLATIME;
 590
 591         l_dput(de);
 592 }
 593
 594 int mds_osc_setattr_async(struct obd_device *obd, struct inode *inode,
 595                           struct lov_mds_md *lmm, int lmm_size,
 596                           struct llog_cookie *logcookies, struct ll_fid *fid)
 597 {
 598         struct mds_obd *mds = &obd->u.mds;
 599         struct obd_trans_info oti = { 0 };
 600         struct obd_info oinfo = { { { 0 } } };
 601         int rc;
 602         ENTRY;
 603
 604         if (OBD_FAIL_CHECK(OBD_FAIL_MDS_OST_SETATTR))
 605                 RETURN(0);
 606
 607         /* first get memory EA */
 608         OBDO_ALLOC(oinfo.oi_oa);
 609         if (!oinfo.oi_oa)
 610                 RETURN(-ENOMEM);
 611
 612         LASSERT(lmm);
 613
 614         rc = obd_unpackmd(mds->mds_lov_exp, &oinfo.oi_md, lmm, lmm_size);
 615         if (rc < 0) {
 616                 CERROR("Error unpack md %p for inode %lu\n", lmm, inode->i_ino);
 617                 GOTO(out, rc);
 618         }
 619
 620         rc = obd_checkmd(mds->mds_lov_exp, obd->obd_self_export, oinfo.oi_md);
 621         if (rc) {
 622                 CERROR("Error revalidate lsm %p \n", oinfo.oi_md);
 623                 GOTO(out, rc);
 624         }
 625
 626         /* then fill oa */
 627         obdo_from_inode(oinfo.oi_oa, inode, OBD_MD_FLUID | OBD_MD_FLGID);
 628         oinfo.oi_oa->o_valid |= OBD_MD_FLID | OBD_MD_FLGROUP;
 629         oinfo.oi_oa->o_id = oinfo.oi_md->lsm_object_id;
 630         oinfo.oi_oa->o_gr = 0;
 631         if (logcookies) {
 632                 oinfo.oi_oa->o_valid |= OBD_MD_FLCOOKIE;
 633                 oti.oti_logcookies = logcookies;
 634         }
 635
 636         LASSERT(fid != NULL);
 637         oinfo.oi_oa->o_fid = fid->id;
 638         oinfo.oi_oa->o_generation = fid->generation;
 639         oinfo.oi_oa->o_valid |= OBD_MD_FLFID | OBD_MD_FLGENER;
 640
 641         /* do async setattr from mds to ost not waiting for responses. */
 642         rc = obd_setattr_async(mds->mds_lov_exp, &oinfo, &oti, NULL);
 643         if (rc)
 644                 CDEBUG(D_INODE, "mds to ost setattr objid 0x"LPX64
 645                        " on ost error %d\n", oinfo.oi_md->lsm_object_id, rc);
 646 out:
 647         if (oinfo.oi_md)
 648                 obd_free_memmd(mds->mds_lov_exp, &oinfo.oi_md);
 649         OBDO_FREE(oinfo.oi_oa);
 650         RETURN(rc);
 651 }
 652
 653 /* In the raw-setattr case, we lock the child inode.
 654  * In the write-back case or if being called from open, the client holds a lock
 655  * already.
 656  *
 657  * We use the ATTR_FROM_OPEN flag to tell these cases apart. */
 658 static int mds_reint_setattr(struct mds_update_record *rec, int offset,
 659                              struct ptlrpc_request *req,
 660                              struct lustre_handle *lh)
 661 {
 662         unsigned int ia_valid = rec->ur_iattr.ia_valid;
 663         struct mds_obd *mds = mds_req2mds(req);
 664         struct obd_device *obd = req->rq_export->exp_obd;
 665         struct mds_body *body;
 666         struct dentry *de;
 667         struct inode *inodes[PTLRPC_NUM_VERSIONS] = { NULL };
 668         struct inode *inode = NULL;
 669         struct lustre_handle lockh;
 670         void *handle = NULL;
 671         struct mds_logcancel_data *mlcd = NULL;
 672         struct lov_mds_md *lmm = NULL;
 673         struct llog_cookie *logcookies = NULL;
 674         int lmm_size = 0, need_lock = 1, cookie_size = 0;
 675         int rc = 0, cleanup_phase = 0, err = 0, locked = 0, sync = 0;
 676         int do_vbr = rec->ur_iattr.ia_valid &
 677                      (ATTR_MODE|ATTR_UID|ATTR_GID|
 678                       ATTR_FROM_OPEN|ATTR_RAW|ATTR_ATTR_FLAG);
 679         unsigned int qcids[MAXQUOTAS] = { 0, 0 };
 680         unsigned int qpids[MAXQUOTAS] = { rec->ur_iattr.ia_uid,
 681                                           rec->ur_iattr.ia_gid };
 682         ENTRY;
 683
 684         LASSERT(offset == REQ_REC_OFF);
 685         offset = REPLY_REC_OFF;
 686
 687         DEBUG_REQ(D_INODE, req, "setattr "LPU64"/%u %x", rec->ur_fid1->id,
 688                   rec->ur_fid1->generation, rec->ur_iattr.ia_valid);
 689         OBD_COUNTER_INCREMENT(obd, setattr);
 690
 691         MDS_CHECK_RESENT(req, reconstruct_reint_setattr(rec, offset, req));
 692
 693         if (rec->ur_dlm)
 694                 ldlm_request_cancel(req, rec->ur_dlm, 0);
 695
 696         if (rec->ur_iattr.ia_valid & ATTR_FROM_OPEN ||
 697             (req->rq_export->exp_connect_flags & OBD_CONNECT_RDONLY)) {
 698                 de = mds_fid2dentry(mds, rec->ur_fid1, NULL);
 699                 if (IS_ERR(de))
 700                         GOTO(cleanup, rc = PTR_ERR(de));
 701                 if (req->rq_export->exp_connect_flags & OBD_CONNECT_RDONLY)
 702                         GOTO(cleanup, rc = -EROFS);
 703         } else {
 704                 __u64 lockpart = MDS_INODELOCK_UPDATE;
 705                 if (rec->ur_iattr.ia_valid & (ATTR_MODE|ATTR_UID|ATTR_GID))
 706                         lockpart |= MDS_INODELOCK_LOOKUP;
 707
 708                 de = mds_fid2locked_dentry(obd, rec->ur_fid1, NULL, LCK_EX,
 709                                            &lockh, NULL, 0, lockpart);
 710                 if (IS_ERR(de))
 711                         GOTO(cleanup, rc = PTR_ERR(de));
 712                 locked = 1;
 713         }
 714
 715         cleanup_phase = 1;
 716         inode = de->d_inode;
 717         LASSERT(inode);
 718
 719         if ((rec->ur_iattr.ia_valid & ATTR_FROM_OPEN) ||
 720             (rec->ur_iattr.ia_valid & ATTR_SIZE)) {
 721                 /* Check write access for the O_TRUNC case */
 722                 if (mds_query_write_access(inode) < 0)
 723                         GOTO(cleanup, rc = -ETXTBSY);
 724         }
 725
 726         /* save uid/gid for quota acq/rel */
 727         qcids[USRQUOTA] = inode->i_uid;
 728         qcids[GRPQUOTA] = inode->i_gid;
 729
 730         if ((S_ISREG(inode->i_mode) || S_ISDIR(inode->i_mode)) &&
 731             rec->ur_eadata != NULL) {
 732                 LOCK_INODE_MUTEX(inode);
 733                 need_lock = 0;
 734         }
 735
 736         OBD_FAIL_WRITE(obd, OBD_FAIL_MDS_REINT_SETATTR_WRITE, inode->i_sb);
 737
 738         /* VBR: update version if attr changed are important for recovery */
 739         if (do_vbr) {
 740                 rc = mds_version_get_check(req, inode, 0);
 741                 if (rc)
 742                         GOTO(cleanup_no_trans, rc);
 743         }
 744         /* start a log jounal handle if needed */
 745         if (S_ISREG(inode->i_mode) &&
 746             rec->ur_iattr.ia_valid & (ATTR_UID | ATTR_GID)) {
 747                 lmm_size = mds->mds_max_mdsize;
 748                 OBD_ALLOC(lmm, lmm_size);
 749                 if (lmm == NULL)
 750                         GOTO(cleanup, rc = -ENOMEM);
 751
 752                 cleanup_phase = 2;
 753                 rc = mds_get_md(obd, inode, lmm, &lmm_size, need_lock, 0,
 754                                 req->rq_export->exp_connect_flags);
 755                 if (rc < 0)
 756                         GOTO(cleanup, rc);
 757                 rc = 0;
 758
 759                 handle = fsfilt_start_log(obd, inode, FSFILT_OP_SETATTR, NULL,
 760                                           le32_to_cpu(lmm->lmm_stripe_count));
 761         } else {
 762                 handle = fsfilt_start(obd, inode, FSFILT_OP_SETATTR, NULL);
 763         }
 764         if (IS_ERR(handle))
 765                 GOTO(cleanup, rc = PTR_ERR(handle));
 766
 767         if (rec->ur_iattr.ia_valid & (ATTR_MTIME | ATTR_CTIME))
 768                 CDEBUG(D_INODE, "setting mtime %lu, ctime %lu\n",
 769                        LTIME_S(rec->ur_iattr.ia_mtime),
 770                        LTIME_S(rec->ur_iattr.ia_ctime));
 771         rc = mds_fix_attr(inode, rec);
 772         if (rc)
 773                 GOTO(cleanup, rc);
 774
 775         if (rec->ur_iattr.ia_valid & ATTR_ATTR_FLAG) {  /* ioctl */
 776                 rc = fsfilt_iocontrol(obd, de, FSFILT_IOC_SETFLAGS,
 777                                       (long)&rec->ur_flags);
 778         } else if (rec->ur_iattr.ia_valid) {            /* setattr */
 779                 rc = fsfilt_setattr(obd, de, handle, &rec->ur_iattr, 0);
 780                 /* journal chown/chgrp in llog, just like unlink */
 781                 if (rc == 0 && lmm_size){
 782                         cookie_size = mds_get_cookie_size(obd, lmm);
 783                         OBD_ALLOC(logcookies, cookie_size);
 784                         if (logcookies == NULL)
 785                                 GOTO(cleanup, rc = -ENOMEM);
 786
 787                         if (mds_log_op_setattr(obd, inode, lmm, lmm_size,
 788                                                logcookies, cookie_size) <= 0) {
 789                                 OBD_FREE(logcookies, cookie_size);
 790                                 logcookies = NULL;
 791                         }
 792                 }
 793         }
 794
 795         if (rc == 0 && (S_ISREG(inode->i_mode) || S_ISDIR(inode->i_mode)) &&
 796             rec->ur_eadata != NULL) {
 797                 struct lov_stripe_md *lsm = NULL;
 798                 struct lov_user_md *lum = NULL;
 799
 800                 rc = ll_permission(inode, MAY_WRITE, NULL);
 801                 if (rc < 0)
 802                         GOTO(cleanup, rc);
 803
 804                 lum = rec->ur_eadata;
 805                 /* if { size, offset, count } = { 0, -1, 0 } and no pool
 806                  * (i.e. all default values specified) then delete default
 807                  * striping from dir. */
 808                 if (S_ISDIR(inode->i_mode) &&
 809                     LOVEA_DELETE_VALUES(lum->lmm_stripe_size,
 810                                         lum->lmm_stripe_count,
 811                                         lum->lmm_stripe_offset) &&
 812                     (le32_to_cpu(lum->lmm_magic) != LOV_USER_MAGIC_V3)) {
 813                         rc = fsfilt_set_md(obd, inode, handle, NULL, 0, "lov");
 814                         if (rc)
 815                                 GOTO(cleanup, rc);
 816                 } else {
 817                         rc = obd_iocontrol(OBD_IOC_LOV_SETSTRIPE,
 818                                            mds->mds_lov_exp, 0,
 819                                            &lsm, rec->ur_eadata);
 820                         if (rc)
 821                                 GOTO(cleanup, rc);
 822
 823                         obd_free_memmd(mds->mds_lov_exp, &lsm);
 824
 825                         rc = fsfilt_set_md(obd, inode, handle, rec->ur_eadata,
 826                                            rec->ur_eadatalen, "lov");
 827                         if (rc)
 828                                 GOTO(cleanup, rc);
 829                 }
 830         }
 831
 832         body = lustre_msg_buf(req->rq_repmsg, offset, sizeof(*body));
 833         mds_pack_inode2body(body, inode);
 834
 835         /* don't return OST-specific attributes if we didn't just set them. */
 836         if (ia_valid & ATTR_SIZE)
 837                 body->valid |= OBD_MD_FLSIZE | OBD_MD_FLBLOCKS;
 838         if (ia_valid & (ATTR_MTIME | ATTR_MTIME_SET))
 839                 body->valid |= OBD_MD_FLMTIME;
 840         if (ia_valid & (ATTR_ATIME | ATTR_ATIME_SET))
 841                 body->valid |= OBD_MD_FLATIME;
 842
 843         if (rc == 0 && rec->ur_cookielen && !IS_ERR(mds->mds_lov_obd)) {
 844                 OBD_ALLOC(mlcd, sizeof(*mlcd) + rec->ur_cookielen +
 845                           rec->ur_eadatalen);
 846                 if (mlcd) {
 847                         mlcd->mlcd_size = sizeof(*mlcd) + rec->ur_cookielen +
 848                                 rec->ur_eadatalen;
 849                         mlcd->mlcd_eadatalen = rec->ur_eadatalen;
 850                         mlcd->mlcd_cookielen = rec->ur_cookielen;
 851                         mlcd->mlcd_lmm = (void *)&mlcd->mlcd_cookies +
 852                                          mlcd->mlcd_cookielen;
 853                         memcpy(&mlcd->mlcd_cookies, rec->ur_logcookies,
 854                                mlcd->mlcd_cookielen);
 855                         memcpy(mlcd->mlcd_lmm, rec->ur_eadata,
 856                                mlcd->mlcd_eadatalen);
 857                 } else {
 858                         CERROR("unable to allocate log cancel data\n");
 859                 }
 860         }
 861         EXIT;
 862  cleanup:
 863         if (mlcd != NULL)
 864                 sync = fsfilt_add_journal_cb(req->rq_export->exp_obd, 0, handle,
 865                                              mds_cancel_cookies_cb, mlcd);
 866
 867         /* permission changes may require sync operation */
 868         if (rc == 0 && ia_valid & (ATTR_MODE|ATTR_UID|ATTR_GID))
 869                 sync |= mds->mds_sync_permission;
 870         inodes[0] = inode;
 871         err = mds_finish_transno(mds, do_vbr ? inodes : NULL, handle, req,
 872                                  rc, 0, sync);
 873
 874  cleanup_no_trans:
 875         /* do mds to ost setattr if needed */
 876         if (!rc && !err && lmm_size)
 877                 mds_osc_setattr_async(obd, inode, lmm, lmm_size,
 878                                       logcookies, rec->ur_fid1);
 879
 880         switch (cleanup_phase) {
 881         case 2:
 882                 OBD_FREE(lmm, mds->mds_max_mdsize);
 883                 if (logcookies)
 884                         OBD_FREE(logcookies, cookie_size);
 885         case 1:
 886                 if ((S_ISREG(inode->i_mode) || S_ISDIR(inode->i_mode)) &&
 887                     rec->ur_eadata != NULL)
 888                         UNLOCK_INODE_MUTEX(inode);
 889                 l_dput(de);
 890                 if (locked) {
 891                         if (rc) {
 892                                 ldlm_lock_decref(&lockh, LCK_EX);
 893                         } else {
 894                                 ptlrpc_save_lock (req, &lockh, LCK_EX);
 895                         }
 896                 }
 897         case 0:
 898                 break;
 899         default:
 900                 LBUG();
 901         }
 902         if (err && !rc)
 903                 rc = err;
 904
 905         req->rq_status = rc;
 906
 907         /* trigger dqrel/dqacq for original owner and new owner */
 908         if (ia_valid & (ATTR_UID | ATTR_GID))
 909                 lquota_adjust(mds_quota_interface_ref, obd, qcids, qpids, rc,
 910                               FSFILT_OP_SETATTR);
 911
 912         return 0;
 913 }
 914
 915 static void reconstruct_reint_create(struct mds_update_record *rec, int offset,
 916                                      struct ptlrpc_request *req)
 917 {
 918         struct obd_export *exp = req->rq_export;
 919         struct mds_export_data *med = &exp->exp_mds_data;
 920         struct mds_obd *obd = &exp->exp_obd->u.mds;
 921         struct dentry *parent, *child;
 922         struct mds_body *body;
 923         int rc;
 924
 925         mds_req_from_lcd(req, med->med_lcd);
 926
 927         if (req->rq_status)
 928                 return;
 929
 930         parent = mds_fid2dentry(obd, rec->ur_fid1, NULL);
 931         if (IS_ERR(parent)) {
 932                 rc = PTR_ERR(parent);
 933                 LCONSOLE_WARN("Parent "LPU64"/%u lookup error %d."
 934                               " Evicting client %s with export %s.\n",
 935                               rec->ur_fid1->id, rec->ur_fid1->generation, rc,
 936                               obd_uuid2str(&exp->exp_client_uuid),
 937                               obd_export_nid2str(exp));
 938                 mds_export_evict(exp);
 939                 EXIT;
 940                 return;
 941         }
 942         child = mds_lookup(exp->exp_obd, rec->ur_name, parent,
 943                            rec->ur_namelen - 1);
 944         if (IS_ERR(child)) {
 945                 rc = PTR_ERR(child);
 946                 LCONSOLE_WARN("Child "LPU64"/%u lookup error %d."
 947                               " Evicting client %s with export %s.\n",
 948                               rec->ur_fid1->id, rec->ur_fid1->generation, rc,
 949                               obd_uuid2str(&exp->exp_client_uuid),
 950                               obd_export_nid2str(exp));
 951                 mds_export_evict(exp);
 952                 l_dput(parent);
 953                 EXIT;
 954                 return;
 955         }
 956
 957         body = lustre_msg_buf(req->rq_repmsg, offset, sizeof(*body));
 958         mds_pack_inode2body(body, child->d_inode);
 959
 960         l_dput(parent);
 961         l_dput(child);
 962 }
 963
 964 static int mds_reint_create(struct mds_update_record *rec, int offset,
 965                             struct ptlrpc_request *req,
 966                             struct lustre_handle *lh)
 967 {
 968         struct dentry *dparent = NULL;
 969         struct mds_obd *mds = mds_req2mds(req);
 970         struct obd_device *obd = req->rq_export->exp_obd;
 971         struct dentry *dchild = NULL;
 972         struct inode *inodes[PTLRPC_NUM_VERSIONS] = { NULL };
 973         struct inode *dir = NULL;
 974         void *handle = NULL;
 975         struct lustre_handle lockh;
 976         int rc = 0, err = 0, type = rec->ur_mode & S_IFMT, cleanup_phase = 0;
 977         int created = 0;
 978         unsigned int qcids[MAXQUOTAS] = { current_fsuid(), current_fsgid() };
 979         unsigned int qpids[MAXQUOTAS] = { 0, 0 };
 980         unsigned int ids[MAXQUOTAS] = { 0, 0 };
 981         struct lvfs_dentry_params dp = LVFS_DENTRY_PARAMS_INIT;
 982         int quota_pending[2] = {0, 0};
 983         unsigned int gid = current_fsgid();
 984         ENTRY;
 985
 986         LASSERT(offset == REQ_REC_OFF);
 987         offset = REPLY_REC_OFF;
 988
 989         LASSERT(!strcmp(req->rq_export->exp_obd->obd_type->typ_name,
 990                         LUSTRE_MDS_NAME));
 991
 992         DEBUG_REQ(D_INODE, req, "parent "LPU64"/%u name %s mode %o",
 993                   rec->ur_fid1->id, rec->ur_fid1->generation,
 994                   rec->ur_name, rec->ur_mode);
 995
 996         MDS_CHECK_RESENT(req, reconstruct_reint_create(rec, offset, req));
 997
 998         if (OBD_FAIL_CHECK(OBD_FAIL_MDS_REINT_CREATE))
 999                 GOTO(cleanup, rc = -ESTALE);
1000
1001         if (rec->ur_dlm)
1002                 ldlm_request_cancel(req, rec->ur_dlm, 0);
1003
1004         dparent = mds_fid2locked_dentry(obd, rec->ur_fid1, NULL, LCK_EX, &lockh,
1005                                         rec->ur_name, rec->ur_namelen - 1,
1006                                         MDS_INODELOCK_UPDATE);
1007         if (IS_ERR(dparent)) {
1008                 rc = PTR_ERR(dparent);
1009                 CDEBUG(D_DENTRY, "parent "LPU64"/%u lookup error %d\n",
1010                                rec->ur_fid1->id, rec->ur_fid1->generation, rc);
1011                 GOTO(cleanup, rc);
1012         }
1013         cleanup_phase = 1; /* locked parent dentry */
1014         dir = dparent->d_inode;
1015         LASSERT(dir);
1016
1017         ldlm_lock_dump_handle(D_OTHER, &lockh);
1018
1019         dchild = mds_lookup(obd, rec->ur_name, dparent, rec->ur_namelen - 1);
1020         if (IS_ERR(dchild)) {
1021                 rc = PTR_ERR(dchild);
1022                 dchild = NULL;
1023                 CDEBUG(D_DENTRY, "child lookup error %d\n", rc);
1024                 GOTO(cleanup, rc);
1025         }
1026
1027         cleanup_phase = 2; /* child dentry */
1028
1029         OBD_FAIL_WRITE(obd, OBD_FAIL_MDS_REINT_CREATE_WRITE, dir->i_sb);
1030
1031         if (req->rq_export->exp_connect_flags & OBD_CONNECT_RDONLY) {
1032                 if (dchild->d_inode)
1033                         GOTO(cleanup, rc = -EEXIST);
1034                 GOTO(cleanup, rc = -EROFS);
1035         }
1036
1037         /** check there is no stale orphan with same inode number */
1038         rc = mds_check_stale_orphan(obd, rec->ur_fid2);
1039         if (rc)
1040                 GOTO(cleanup, rc);
1041
1042         /* version recovery check */
1043         rc = mds_version_get_check(req, dir, 0);
1044         if (rc)
1045                 GOTO(cleanup_no_trans, rc);
1046
1047         if (dir->i_mode & S_ISGID && S_ISDIR(rec->ur_mode))
1048                 rec->ur_mode |= S_ISGID;
1049
1050         dchild->d_fsdata = (void *)&dp;
1051         dp.ldp_inum = (unsigned long)rec->ur_fid2->id;
1052         dp.ldp_ptr = req;
1053
1054         if (dir->i_mode & S_ISGID)
1055                 gid = dir->i_gid;
1056         else
1057                 gid = current_fsgid();
1058
1059         /* we try to get enough quota to write here, and let ldiskfs
1060          * decide if it is out of quota or not b=14783
1061          * FIXME: after CMD is used, pointer to obd_trans_info* couldn't
1062          * be NULL, b=14840 */
1063         ids[0] = current_fsuid();
1064         ids[1] = gid;
1065         lquota_chkquota(mds_quota_interface_ref, req->rq_export, ids[0], ids[1],
1066                         1, quota_pending, NULL, NULL, 0);
1067
1068         switch (type) {
1069         case S_IFREG:{
1070                 handle = fsfilt_start(obd, dir, FSFILT_OP_CREATE, NULL);
1071                 if (IS_ERR(handle))
1072                         GOTO(cleanup, rc = PTR_ERR(handle));
1073                 LOCK_INODE_MUTEX(dir);
1074                 rc = ll_vfs_create(dir, dchild, rec->ur_mode, NULL);
1075                 UNLOCK_INODE_MUTEX(dir);
1076                 mds_counter_incr(req->rq_export, LPROC_MDS_MKNOD);
1077                 EXIT;
1078                 break;
1079         }
1080         case S_IFDIR:{
1081                 handle = fsfilt_start(obd, dir, FSFILT_OP_MKDIR, NULL);
1082                 if (IS_ERR(handle))
1083                         GOTO(cleanup, rc = PTR_ERR(handle));
1084                 LOCK_INODE_MUTEX(dir);
1085                 rc = ll_vfs_mkdir(dir, dchild, mds->mds_vfsmnt, rec->ur_mode);
1086                 UNLOCK_INODE_MUTEX(dir);
1087                 mds_counter_incr(req->rq_export, LPROC_MDS_MKDIR);
1088                 EXIT;
1089                 break;
1090         }
1091         case S_IFLNK:{
1092                 handle = fsfilt_start(obd, dir, FSFILT_OP_SYMLINK, NULL);
1093                 if (IS_ERR(handle))
1094                         GOTO(cleanup, rc = PTR_ERR(handle));
1095                 LOCK_INODE_MUTEX(dir);
1096                 if (rec->ur_tgt == NULL)        /* no target supplied */
1097                         rc = -EINVAL;           /* -EPROTO? */
1098                 else
1099                         rc = ll_vfs_symlink(dir, dchild, mds->mds_vfsmnt,
1100                                             rec->ur_tgt, S_IALLUGO);
1101                 UNLOCK_INODE_MUTEX(dir);
1102                 mds_counter_incr(req->rq_export, LPROC_MDS_MKNOD);
1103                 EXIT;
1104                 break;
1105         }
1106         case S_IFCHR:
1107         case S_IFBLK:
1108         case S_IFIFO:
1109         case S_IFSOCK:{
1110                 int rdev = rec->ur_rdev;
1111                 handle = fsfilt_start(obd, dir, FSFILT_OP_MKNOD, NULL);
1112                 if (IS_ERR(handle))
1113                         GOTO(cleanup, rc = PTR_ERR(handle));
1114                 LOCK_INODE_MUTEX(dir);
1115                 rc = ll_vfs_mknod(dir, dchild, mds->mds_vfsmnt, rec->ur_mode,
1116                                   rdev);
1117                 UNLOCK_INODE_MUTEX(dir);
1118                 mds_counter_incr(req->rq_export, LPROC_MDS_MKNOD);
1119                 EXIT;
1120                 break;
1121         }
1122         default:
1123                 CERROR("bad file type %o creating %s\n", type, rec->ur_name);
1124                 dchild->d_fsdata = NULL;
1125                 GOTO(cleanup, rc = -EINVAL);
1126         }
1127
1128         /* In case we stored the desired inum in here, we want to clean up. */
1129         if (dchild->d_fsdata == (void *)(unsigned long)rec->ur_fid2->id)
1130                 dchild->d_fsdata = NULL;
1131
1132         if (rc) {
1133                 CDEBUG(D_INODE, "error during create: %d\n", rc);
1134                 GOTO(cleanup, rc);
1135         } else {
1136                 struct iattr iattr;
1137                 struct inode *inode = dchild->d_inode;
1138                 struct mds_body *body;
1139
1140                 created = 1;
1141                 LTIME_S(iattr.ia_atime) = rec->ur_time;
1142                 LTIME_S(iattr.ia_ctime) = rec->ur_time;
1143                 LTIME_S(iattr.ia_mtime) = rec->ur_time;
1144                 iattr.ia_uid = current_fsuid();  /* set by push_ctxt already */
1145                 iattr.ia_gid = gid;
1146                 iattr.ia_valid = ATTR_UID | ATTR_GID | ATTR_ATIME |
1147                         ATTR_MTIME | ATTR_CTIME;
1148
1149                 if (rec->ur_fid2->id) {
1150                         if (rec->ur_fid2->id != inode->i_ino) {
1151                                 if (req->rq_export->exp_delayed)
1152                                         rc = -EOVERFLOW;
1153                                 else
1154                                         rc = -EFAULT;
1155                                 GOTO(cleanup, rc);
1156                         }
1157                         inode->i_generation = rec->ur_fid2->generation;
1158                         /* Dirtied and committed by the upcoming setattr. */
1159                         CDEBUG(D_INODE, "recreated ino %lu with gen %u\n",
1160                                inode->i_ino, inode->i_generation);
1161                 } else {
1162                         CDEBUG(D_INODE, "created ino %lu with gen %x\n",
1163                                inode->i_ino, inode->i_generation);
1164                 }
1165
1166                 rc = fsfilt_setattr(obd, dchild, handle, &iattr, 0);
1167                 if (rc)
1168                         CERROR("error on child setattr: rc = %d\n", rc);
1169
1170                 iattr.ia_valid = ATTR_MTIME | ATTR_CTIME;
1171                 rc = fsfilt_setattr(obd, dparent, handle, &iattr, 0);
1172                 if (rc)
1173                         CERROR("error on parent setattr: rc = %d\n", rc);
1174
1175                 if (S_ISDIR(inode->i_mode)) {
1176                         struct lov_mds_md_v3 lmm;
1177                         int lmm_size = sizeof(lmm);
1178                         rc = mds_get_md(obd, dir, &lmm, &lmm_size, 1, 0,
1179                                         req->rq_export->exp_connect_flags);
1180                         if (rc > 0) {
1181                                 LOCK_INODE_MUTEX(inode);
1182                                 rc = fsfilt_set_md(obd, inode, handle,
1183                                                    &lmm, lmm_size, "lov");
1184                                 UNLOCK_INODE_MUTEX(inode);
1185                         }
1186                         if (rc)
1187                                 CERROR("error on copy stripe info: rc = %d\n",
1188                                         rc);
1189                 }
1190
1191                 body = lustre_msg_buf(req->rq_repmsg, offset, sizeof(*body));
1192                 mds_pack_inode2body(body, inode);
1193         }
1194         EXIT;
1195
1196 cleanup:
1197         inodes[0] = dir;
1198         inodes[1] = dchild ? dchild->d_inode : NULL;
1199         err = mds_finish_transno(mds, inodes, handle, req, rc, 0, 0);
1200
1201 cleanup_no_trans:
1202         if (quota_pending[0] || quota_pending[1])
1203                 lquota_pending_commit(mds_quota_interface_ref, obd,
1204                                       ids[0], ids[1], quota_pending);
1205
1206         if (rc && created) {
1207                 /* Destroy the file we just created.  This should not need
1208                  * extra journal credits, as we have already modified all of
1209                  * the blocks needed in order to create the file in the first
1210                  * place.
1211                  */
1212                 switch (type) {
1213                 case S_IFDIR:
1214                         LOCK_INODE_MUTEX(dir);
1215                         err = ll_vfs_rmdir(dir, dchild, mds->mds_vfsmnt);
1216                         UNLOCK_INODE_MUTEX(dir);
1217                         if (err)
1218                                 CERROR("rmdir in error path: %d\n", err);
1219                         break;
1220                 default:
1221                         LOCK_INODE_MUTEX(dir);
1222                         err = ll_vfs_unlink(dir, dchild, mds->mds_vfsmnt);
1223                         UNLOCK_INODE_MUTEX(dir);
1224                         if (err)
1225                                 CERROR("unlink in error path: %d\n", err);
1226                         break;
1227                 }
1228         } else if (created) {
1229                 /* The inode we were allocated may have just been freed
1230                  * by an unlink operation.  We take this lock to
1231                  * synchronize against the matching reply-ack-lock taken
1232                  * in unlink, to avoid replay problems if this reply
1233                  * makes it out to the client but the unlink's does not.
1234                  * See bug 2029 for more detail.*/
1235                 mds_lock_new_child(obd, dchild->d_inode, NULL);
1236                 /* save uid/gid of create inode and parent */
1237                 qpids[USRQUOTA] = dir->i_uid;
1238                 qpids[GRPQUOTA] = dir->i_gid;
1239         } else {
1240                 rc = err;
1241         }
1242
1243         switch (cleanup_phase) {
1244         case 2: /* child dentry */
1245                 l_dput(dchild);
1246         case 1: /* locked parent dentry */
1247                 if (rc) {
1248                         ldlm_lock_decref(&lockh, LCK_EX);
1249                 } else {
1250                         ptlrpc_save_lock (req, &lockh, LCK_EX);
1251                 }
1252                 l_dput(dparent);
1253         case 0:
1254                 break;
1255         default:
1256                 CERROR("invalid cleanup_phase %d\n", cleanup_phase);
1257                 LBUG();
1258         }
1259         req->rq_status = rc;
1260
1261         /* trigger dqacq on the owner of child and parent */
1262         lquota_adjust(mds_quota_interface_ref, obd, qcids, qpids, rc,
1263                       FSFILT_OP_CREATE);
1264         return 0;
1265 }
1266
1267 int res_gt(struct ldlm_res_id *res1, struct ldlm_res_id *res2,
1268            ldlm_policy_data_t *p1, ldlm_policy_data_t *p2)
1269 {
1270         int i;
1271
1272         for (i = 0; i < RES_NAME_SIZE; i++) {
1273                 /* return 1 here, because enqueue_ordered will skip resources
1274                  * of all zeroes if they're sorted to the end of the list. */
1275                 if (res1->name[i] == 0 && res2->name[i] != 0)
1276                         return 1;
1277                 if (res2->name[i] == 0 && res1->name[i] != 0)
1278                         return 0;
1279
1280                 if (res1->name[i] > res2->name[i])
1281                         return 1;
1282                 if (res1->name[i] < res2->name[i])
1283                         return 0;
1284         }
1285         if (!p1 || !p2)
1286                 return 0;
1287         if (memcmp(p1, p2, sizeof(*p1)) < 0)
1288                 return 1;
1289         return 0;
1290 }
1291
1292 /* This function doesn't use ldlm_match_or_enqueue because we're always called
1293  * with EX or PW locks, and the MDS is no longer allowed to match write locks,
1294  * because they take the place of local semaphores.
1295  *
1296  * One or two locks are taken in numerical order.  A res_id->name[0] of 0 means
1297  * no lock is taken for that res_id.  Must be at least one non-zero res_id. */
1298 int enqueue_ordered_locks(struct obd_device *obd, struct ldlm_res_id *p1_res_id,
1299                           struct lustre_handle *p1_lockh, int p1_lock_mode,
1300                           ldlm_policy_data_t *p1_policy,
1301                           struct ldlm_res_id *p2_res_id,
1302                           struct lustre_handle *p2_lockh, int p2_lock_mode,
1303                           ldlm_policy_data_t *p2_policy)
1304 {
1305         struct ldlm_res_id *res_id[2] = { p1_res_id, p2_res_id };
1306         struct lustre_handle *handles[2] = { p1_lockh, p2_lockh };
1307         int lock_modes[2] = { p1_lock_mode, p2_lock_mode };
1308         ldlm_policy_data_t *policies[2] = {p1_policy, p2_policy};
1309         int rc, flags;
1310         ENTRY;
1311
1312         LASSERT(p1_res_id != NULL && p2_res_id != NULL);
1313
1314         CDEBUG(D_INFO, "locks before: "LPU64"/"LPU64"\n",
1315                res_id[0]->name[0], res_id[1]->name[0]);
1316
1317         if (res_gt(p1_res_id, p2_res_id, p1_policy, p2_policy)) {
1318                 handles[1] = p1_lockh;
1319                 handles[0] = p2_lockh;
1320                 res_id[1] = p1_res_id;
1321                 res_id[0] = p2_res_id;
1322                 lock_modes[1] = p1_lock_mode;
1323                 lock_modes[0] = p2_lock_mode;
1324                 policies[1] = p1_policy;
1325                 policies[0] = p2_policy;
1326         }
1327
1328         CDEBUG(D_DLMTRACE, "lock order: "LPU64"/"LPU64"\n",
1329                res_id[0]->name[0], res_id[1]->name[0]);
1330
1331         flags = LDLM_FL_LOCAL_ONLY | LDLM_FL_ATOMIC_CB;
1332         rc = ldlm_cli_enqueue_local(obd->obd_namespace, res_id[0],
1333                                     LDLM_IBITS, policies[0], lock_modes[0],
1334                                     &flags, ldlm_blocking_ast,
1335                                     ldlm_completion_ast, NULL, NULL, 0,
1336                                     NULL, handles[0]);
1337         if (rc != ELDLM_OK)
1338                 RETURN(-EIO);
1339         ldlm_lock_dump_handle(D_OTHER, handles[0]);
1340
1341         if (memcmp(res_id[0], res_id[1], sizeof(*res_id[0])) == 0 &&
1342             (policies[0]->l_inodebits.bits & policies[1]->l_inodebits.bits)) {
1343                 memcpy(handles[1], handles[0], sizeof(*(handles[1])));
1344                 ldlm_lock_addref(handles[1], lock_modes[1]);
1345         } else if (res_id[1]->name[0] != 0) {
1346                 flags = LDLM_FL_LOCAL_ONLY | LDLM_FL_ATOMIC_CB;
1347                 rc = ldlm_cli_enqueue_local(obd->obd_namespace, res_id[1],
1348                                             LDLM_IBITS, policies[1],
1349                                             lock_modes[1], &flags,
1350                                             ldlm_blocking_ast,
1351                                             ldlm_completion_ast, NULL, NULL,
1352                                             0, NULL, handles[1]);
1353                 if (rc != ELDLM_OK) {
1354                         ldlm_lock_decref(handles[0], lock_modes[0]);
1355                         RETURN(-EIO);
1356                 }
1357                 ldlm_lock_dump_handle(D_OTHER, handles[1]);
1358         }
1359
1360         RETURN(0);
1361 }
1362
1363 static inline int res_eq(struct ldlm_res_id *res1, struct ldlm_res_id *res2)
1364 {
1365         return !memcmp(res1, res2, sizeof(*res1));
1366 }
1367
1368 static inline void
1369 try_to_aggregate_locks(struct ldlm_res_id *res1, ldlm_policy_data_t *p1,
1370                         struct ldlm_res_id *res2, ldlm_policy_data_t *p2)
1371 {
1372         if (!res_eq(res1, res2))
1373                 return;
1374         /* XXX: any additional inodebits (to current LOOKUP and UPDATE)
1375          * should be taken with great care here */
1376         p1->l_inodebits.bits |= p2->l_inodebits.bits;
1377 }
1378
1379 int enqueue_4ordered_locks(struct obd_device *obd,struct ldlm_res_id *p1_res_id,
1380                            struct lustre_handle *p1_lockh, int p1_lock_mode,
1381                            ldlm_policy_data_t *p1_policy,
1382                            struct ldlm_res_id *p2_res_id,
1383                            struct lustre_handle *p2_lockh, int p2_lock_mode,
1384                            ldlm_policy_data_t *p2_policy,
1385                            struct ldlm_res_id *c1_res_id,
1386                            struct lustre_handle *c1_lockh, int c1_lock_mode,
1387                            ldlm_policy_data_t *c1_policy,
1388                            struct ldlm_res_id *c2_res_id,
1389                            struct lustre_handle *c2_lockh, int c2_lock_mode,
1390                            ldlm_policy_data_t *c2_policy)
1391 {
1392         struct ldlm_res_id *res_id[5] = { p1_res_id, p2_res_id,
1393                                           c1_res_id, c2_res_id };
1394         struct lustre_handle *dlm_handles[5] = { p1_lockh, p2_lockh,
1395                                                  c1_lockh, c2_lockh };
1396         int lock_modes[5] = { p1_lock_mode, p2_lock_mode,
1397                               c1_lock_mode, c2_lock_mode };
1398         ldlm_policy_data_t *policies[5] = {p1_policy, p2_policy,
1399                                            c1_policy, c2_policy};
1400         int rc, i, j, sorted, flags;
1401         ENTRY;
1402
1403         CDEBUG(D_DLMTRACE, "locks before: "LPU64"/"LPU64"/"LPU64"/"LPU64"\n",
1404                res_id[0]->name[0], res_id[1]->name[0], res_id[2]->name[0],
1405                res_id[3]->name[0]);
1406
1407         /* simple insertion sort - we have at most 4 elements */
1408         for (i = 1; i < 4; i++) {
1409                 j = i - 1;
1410                 dlm_handles[4] = dlm_handles[i];
1411                 res_id[4] = res_id[i];
1412                 lock_modes[4] = lock_modes[i];
1413                 policies[4] = policies[i];
1414
1415                 sorted = 0;
1416                 do {
1417                         if (res_gt(res_id[j], res_id[4], policies[j],
1418                                    policies[4])) {
1419                                 dlm_handles[j + 1] = dlm_handles[j];
1420                                 res_id[j + 1] = res_id[j];
1421                                 lock_modes[j + 1] = lock_modes[j];
1422                                 policies[j + 1] = policies[j];
1423                                 j--;
1424                         } else {
1425                                 sorted = 1;
1426                         }
1427                 } while (j >= 0 && !sorted);
1428
1429                 dlm_handles[j + 1] = dlm_handles[4];
1430                 res_id[j + 1] = res_id[4];
1431                 lock_modes[j + 1] = lock_modes[4];
1432                 policies[j + 1] = policies[4];
1433         }
1434
1435         CDEBUG(D_DLMTRACE, "lock order: "LPU64"/"LPU64"/"LPU64"/"LPU64"\n",
1436                res_id[0]->name[0], res_id[1]->name[0], res_id[2]->name[0],
1437                res_id[3]->name[0]);
1438
1439         /* XXX we could send ASTs on all these locks first before blocking? */
1440         for (i = 0; i < 4; i++) {
1441                 flags = LDLM_FL_ATOMIC_CB;
1442                 if (res_id[i]->name[0] == 0)
1443                         break;
1444                 if (i && res_eq(res_id[i], res_id[i-1])) {
1445                         memcpy(dlm_handles[i], dlm_handles[i-1],
1446                                sizeof(*(dlm_handles[i])));
1447                         ldlm_lock_addref(dlm_handles[i], lock_modes[i]);
1448                 } else {
1449                         /* we need to enqueue locks with different inodebits
1450                          * at once, because otherwise concurrent thread can
1451                          * hit the windown between these two locks and we'll
1452                          * get to deadlock. see bug 10360. note also, that it
1453                          * is impossible to have >2 equal res. */
1454                         if (i < 3)
1455                                 try_to_aggregate_locks(res_id[i], policies[i],
1456                                                        res_id[i+1], policies[i+1]);
1457                         rc = ldlm_cli_enqueue_local(obd->obd_namespace,
1458                                                     res_id[i], LDLM_IBITS,
1459                                                     policies[i], lock_modes[i],
1460                                                     &flags, ldlm_blocking_ast,
1461                                                     ldlm_completion_ast, NULL,
1462                                                     NULL, 0, NULL,
1463                                                     dlm_handles[i]);
1464                         if (rc != ELDLM_OK)
1465                                 GOTO(out_err, rc = -EIO);
1466                         ldlm_lock_dump_handle(D_OTHER, dlm_handles[i]);
1467                 }
1468         }
1469
1470         RETURN(0);
1471 out_err:
1472         while (i-- > 0)
1473                 ldlm_lock_decref(dlm_handles[i], lock_modes[i]);
1474
1475         return rc;
1476 }
1477
1478 /* In the unlikely case that the child changed while we were waiting
1479  * on the lock, we need to drop the lock on the old child and either:
1480  * - if the child has a lower resource name, then we have to also
1481  *   drop the parent lock and regain the locks in the right order
1482  * - in the rename case, if the child has a lower resource name than one of
1483  *   the other parent/child resources (maxres) we also need to reget the locks
1484  * - if the child has a higher resource name (this is the common case)
1485  *   we can just get the lock on the new child (still in lock order)
1486  *
1487  * Returns 0 if the child did not change or if it changed but could be locked.
1488  * Returns 1 if the child changed and we need to re-lock (no locks held).
1489  * Returns -ve error with a valid dchild (no locks held). */
1490 static int mds_verify_child(struct obd_device *obd,
1491                             struct ldlm_res_id *parent_res_id,
1492                             struct lustre_handle *parent_lockh,
1493                             struct dentry *dparent, int parent_mode,
1494                             struct ldlm_res_id *child_res_id,
1495                             struct lustre_handle *child_lockh,
1496                             struct dentry **dchildp, int child_mode,
1497                             ldlm_policy_data_t *child_policy,
1498                             const char *name, int namelen,
1499                             struct ldlm_res_id *maxres)
1500 {
1501         struct dentry *vchild, *dchild = *dchildp;
1502         int rc = 0, cleanup_phase = 2; /* parent, child locks */
1503         ENTRY;
1504
1505         /* not want child - not check it */
1506         if (name == NULL)
1507                 RETURN(0);
1508
1509         vchild = ll_lookup_one_len(name, dparent, namelen - 1);
1510         if (IS_ERR(vchild))
1511                 GOTO(cleanup, rc = PTR_ERR(vchild));
1512
1513         if (likely((vchild->d_inode == NULL && child_res_id->name[0] == 0) ||
1514                    (vchild->d_inode != NULL &&
1515                     child_res_id->name[0] == vchild->d_inode->i_ino &&
1516                     child_res_id->name[1] == vchild->d_inode->i_generation))) {
1517                 if (dchild != NULL)
1518                         l_dput(dchild);
1519                 *dchildp = vchild;
1520
1521                 RETURN(0);
1522         }
1523         /* resouce is changed, but not want child lock, return new child */
1524         if (child_lockh == NULL) {
1525                 dput(dchild);
1526                 *dchildp = vchild;
1527                 GOTO(cleanup, rc = 0);
1528         }
1529
1530         CDEBUG(D_DLMTRACE, "child inode changed: %p != %p (%lu != "LPU64")\n",
1531                vchild->d_inode, dchild ? dchild->d_inode : 0,
1532                vchild->d_inode ? vchild->d_inode->i_ino : 0,
1533                child_res_id->name[0]);
1534         if (child_res_id->name[0] != 0)
1535                 ldlm_lock_decref(child_lockh, child_mode);
1536         if (dchild)
1537                 l_dput(dchild);
1538
1539         cleanup_phase = 1; /* parent lock only */
1540         *dchildp = dchild = vchild;
1541
1542         if (dchild->d_inode) {
1543                 int flags = LDLM_FL_ATOMIC_CB;
1544                 child_res_id->name[0] = dchild->d_inode->i_ino;
1545                 child_res_id->name[1] = dchild->d_inode->i_generation;
1546
1547                 /* Make sure that we don't try to re-enqueue a lock on the
1548                  * same resource if it happens that the source is renamed to
1549                  * the target by another thread (bug 9974, thanks racer :-) */
1550                 if (!res_gt(child_res_id, parent_res_id, NULL, NULL) ||
1551                     !res_gt(child_res_id, maxres, NULL, NULL)) {
1552                         CDEBUG(D_DLMTRACE, "relock "LPU64"<("LPU64"|"LPU64")\n",
1553                                child_res_id->name[0], parent_res_id->name[0],
1554                                maxres->name[0]);
1555                         GOTO(cleanup, rc = 1);
1556                 }
1557
1558                 rc = ldlm_cli_enqueue_local(obd->obd_namespace, child_res_id,
1559                                             LDLM_IBITS, child_policy,
1560                                             child_mode, &flags,
1561                                             ldlm_blocking_ast,
1562                                             ldlm_completion_ast, NULL,
1563                                             NULL, 0, NULL, child_lockh);
1564                 if (rc != ELDLM_OK)
1565                         GOTO(cleanup, rc = -EIO);
1566         } else {
1567                 memset(child_res_id, 0, sizeof(*child_res_id));
1568                 memset(child_lockh, 0, sizeof(*child_lockh));
1569         }
1570
1571         EXIT;
1572 cleanup:
1573         if (rc) {
1574                 switch(cleanup_phase) {
1575                 case 2:
1576                         if (child_res_id->name[0] != 0)
1577                                 ldlm_lock_decref(child_lockh, child_mode);
1578                 case 1:
1579                         ldlm_lock_decref(parent_lockh, parent_mode);
1580                 }
1581         }
1582         return rc;
1583 }
1584
1585 #define INODE_CTIME_AGE (10)
1586 #define INODE_CTIME_OLD(inode) (LTIME_S(inode->i_ctime) +               \
1587                                 INODE_CTIME_AGE < CURRENT_SECONDS)
1588
1589 int mds_get_parent_child_locked(struct obd_device *obd, struct mds_obd *mds,
1590                                 struct ll_fid *fid,
1591                                 struct lustre_handle *parent_lockh,
1592                                 struct dentry **dparentp, int parent_mode,
1593                                 __u64 parent_lockpart,
1594                                 char *name, int namelen,
1595                                 struct lustre_handle *child_lockh,
1596                                 struct dentry **dchildp, int child_mode,
1597                                 __u64 child_lockpart)
1598 {
1599         struct ldlm_res_id child_res_id = { .name = {0} };
1600         struct ldlm_res_id parent_res_id = { .name = {0} };
1601         ldlm_policy_data_t parent_policy = {.l_inodebits = { parent_lockpart }};
1602         ldlm_policy_data_t child_policy = {.l_inodebits = { child_lockpart }};
1603         static struct ldlm_res_id child_res_id_nolock = { .name = {0} };
1604         struct inode *inode;
1605         int rc = 0, cleanup_phase = 0;
1606         ENTRY;
1607
1608         /* Step 1: Lookup parent */
1609         *dparentp = mds_fid2dentry(mds, fid, NULL);
1610         if (IS_ERR(*dparentp)) {
1611                 rc = PTR_ERR(*dparentp);
1612                 *dparentp = NULL;
1613                 RETURN(rc);
1614         }
1615
1616         CDEBUG(D_INODE, "parent ino %lu, name %s\n",
1617                (*dparentp)->d_inode->i_ino, name);
1618
1619         parent_res_id.name[0] = (*dparentp)->d_inode->i_ino;
1620         parent_res_id.name[1] = (*dparentp)->d_inode->i_generation;
1621
1622         cleanup_phase = 1; /* parent dentry */
1623
1624         if (name == NULL)
1625                 GOTO(retry_locks, rc);
1626         /* Step 2: Lookup child (without DLM lock, to get resource name) */
1627         *dchildp = mds_lookup(obd, name, *dparentp, namelen - 1);
1628         if (IS_ERR(*dchildp)) {
1629                 rc = PTR_ERR(*dchildp);
1630                 *dchildp = NULL;
1631                 CDEBUG(D_INODE, "child lookup error %d\n", rc);
1632                 GOTO(cleanup, rc);
1633         }
1634
1635         cleanup_phase = 2; /* child dentry */
1636
1637         inode = (*dchildp)->d_inode;
1638         if (inode != NULL) {
1639                 if (is_bad_inode(inode)) {
1640                         CERROR("bad inode returned %lu/%u\n",
1641                                inode->i_ino, inode->i_generation);
1642                         GOTO(cleanup, rc = -ENOENT);
1643                 }
1644                 inode = igrab(inode);
1645         }
1646         if (inode == NULL)
1647                 goto retry_locks;
1648
1649         child_res_id.name[0] = inode->i_ino;
1650         child_res_id.name[1] = inode->i_generation;
1651
1652         /* If we want a LCK_CR for a directory, and this directory has not been
1653            changed for some time, we return not only a LOOKUP lock, but also an
1654            UPDATE lock to have negative dentry starts working for this dir.
1655            Also we apply same logic to non-directories. If the file is rarely
1656            changed - we return both locks and this might save us RPC on
1657            later STAT. */
1658         if ((child_mode & (LCK_CR|LCK_PR|LCK_CW)) && INODE_CTIME_OLD(inode))
1659                 child_policy.l_inodebits.bits |= MDS_INODELOCK_UPDATE;
1660
1661         iput(inode);
1662
1663 retry_locks:
1664         cleanup_phase = 2; /* child dentry */
1665
1666         /* Step 3: Lock parent and child in resource order.  If child doesn't
1667          *         exist, we still have to lock the parent and re-lookup. */
1668         rc = enqueue_ordered_locks(obd,&parent_res_id,parent_lockh,parent_mode,
1669                                    &parent_policy,
1670                                    child_lockh ? &child_res_id :
1671                                                  &child_res_id_nolock,
1672                                    child_lockh, child_mode,
1673                                    &child_policy);
1674         if (rc)
1675                 GOTO(cleanup, rc);
1676
1677         if (IS_DEADDIR((*dparentp)->d_inode))
1678                 GOTO(cleanup, -ENOENT);
1679
1680         /* Step 4: Re-lookup child to verify it hasn't changed since locking */
1681         rc = mds_verify_child(obd, &parent_res_id, parent_lockh, *dparentp,
1682                               parent_mode, &child_res_id, child_lockh, dchildp,
1683                               child_mode,&child_policy, name, namelen, &parent_res_id);
1684         if (rc > 0)
1685                 goto retry_locks;
1686         if (rc < 0) {
1687                 GOTO(cleanup, rc);
1688         }
1689
1690 cleanup:
1691         if (rc) {
1692                 switch (cleanup_phase) {
1693                 case 2:
1694                         l_dput(*dchildp);
1695                 case 1:
1696                         l_dput(*dparentp);
1697                 default: ;
1698                 }
1699         }
1700         return rc;
1701 }
1702
1703 void mds_reconstruct_generic(struct ptlrpc_request *req)
1704 {
1705         struct mds_export_data *med = &req->rq_export->exp_mds_data;
1706
1707         mds_req_from_lcd(req, med->med_lcd);
1708 }
1709
1710 /* If we are unlinking an open file/dir (i.e. creating an orphan) then
1711  * we instead link the inode into the PENDING directory until it is
1712  * finally released.  We can't simply call mds_reint_rename() or some
1713  * part thereof, because we don't have the inode to check for link
1714  * count/open status until after it is locked.
1715  *
1716  * returns 1 on success
1717  * returns 0 if we lost a race and didn't make a new link
1718  * returns negative on error
1719  */
1720 static int mds_orphan_add_link(struct mds_update_record *rec,
1721                                struct obd_device *obd, struct dentry *dentry)
1722 {
1723         struct mds_obd *mds = &obd->u.mds;
1724         struct inode *pending_dir = mds->mds_pending_dir->d_inode;
1725         struct inode *inode = dentry->d_inode;
1726         struct dentry *pending_child;
1727         char fidname[LL_FID_NAMELEN];
1728         int fidlen = 0, rc, mode;
1729         int ignoring_quota;
1730         ENTRY;
1731
1732         LASSERT(inode != NULL);
1733         LASSERT(!mds_inode_is_orphan(inode));
1734         LASSERT(TRYLOCK_INODE_MUTEX(pending_dir) == 0);
1735
1736         fidlen = ll_fid2str(fidname, inode->i_ino, inode->i_generation);
1737
1738         CDEBUG(D_INODE, "pending destroy of %dx open %d linked %s %s = %s\n",
1739                mds_orphan_open_count(inode), inode->i_nlink,
1740                S_ISDIR(inode->i_mode) ? "dir" :
1741                 S_ISREG(inode->i_mode) ? "file" : "other",rec->ur_name,fidname);
1742
1743         if (!mds_orphan_needed(obd, inode) || inode->i_nlink != 0)
1744                 RETURN(0);
1745
1746         pending_child = lookup_one_len(fidname, mds->mds_pending_dir, fidlen);
1747         if (IS_ERR(pending_child))
1748                 RETURN(PTR_ERR(pending_child));
1749
1750         if (pending_child->d_inode != NULL) {
1751                 CERROR("re-destroying orphan file %s?\n", rec->ur_name);
1752                 LASSERT(pending_child->d_inode == inode);
1753                 GOTO(out_dput, rc = 0);
1754         }
1755
1756         /* link() is semanticaly-wrong for S_IFDIR, so we set S_IFREG
1757          * for linking and return real mode back then -bzzz */
1758         mode = inode->i_mode;
1759         inode->i_mode = S_IFREG;
1760         /* avoid vfs_link upon 0 nlink inode, inc by 2 instead of 1 because
1761          * ext3_inc_count() can reset i_nlink for indexed directory */
1762         inode->i_nlink += 2;
1763
1764         /* Temporarily raise the resource capability as we do not want to
1765          * get -EDQUOT from VFS during this unlink operation */
1766         ignoring_quota = lquota_enforce(mds_quota_interface_ref, obd, 1);
1767
1768         rc = ll_vfs_link(dentry, mds->mds_vfsmnt, pending_dir, pending_child,
1769                          mds->mds_vfsmnt);
1770         if (rc)
1771                 CERROR("error linking orphan %s %s to PENDING: rc = %d\n",
1772                        S_ISDIR(mode) ? "dir" : S_ISREG(mode) ? "file" : "other",
1773                        rec->ur_name, rc);
1774         else
1775                 mds_inode_set_orphan(inode);
1776
1777         /* return mode and correct i_nlink if inode is directory */
1778         inode->i_mode = mode;
1779         LASSERTF(rc || inode->i_nlink == 3, "%s nlink == %d\n",
1780                  S_ISDIR(mode) ? "dir" : S_ISREG(mode) ? "file" : "other",
1781                  inode->i_nlink);
1782
1783         if (S_ISDIR(mode)) {
1784                 pending_dir->i_nlink++;
1785                 if (pending_dir->i_sb->s_op->dirty_inode)
1786                         pending_dir->i_sb->s_op->dirty_inode(pending_dir);
1787         }
1788
1789         inode->i_nlink -= 2;
1790         if (inode->i_sb->s_op->dirty_inode)
1791                 inode->i_sb->s_op->dirty_inode(inode);
1792
1793         if (!ignoring_quota)
1794                 lquota_enforce(mds_quota_interface_ref, obd, 0);
1795
1796         if (rc)
1797                 GOTO(out_dput, rc);
1798
1799         GOTO(out_dput, rc = 1);
1800 out_dput:
1801         l_dput(pending_child);
1802         RETURN(rc);
1803 }
1804
1805 int mds_get_cookie_size(struct obd_device *obd, struct lov_mds_md *lmm)
1806 {
1807         int count = le32_to_cpu(lmm->lmm_stripe_count);
1808         int real_csize = count * sizeof(struct llog_cookie);
1809         return real_csize;
1810 }
1811
1812 static void mds_shrink_reply(struct ptlrpc_request *req,
1813                            int reply_mdoff, int have_md, int have_acl)
1814 {
1815         struct obd_device *obd = req->rq_export->exp_obd;
1816         struct mds_body *reply_body;
1817         int cookie_size = 0, md_size = 0;
1818         ENTRY;
1819
1820         /* LSM and cookie is always placed after mds_body */
1821         reply_body =  lustre_msg_buf(req->rq_repmsg, reply_mdoff,
1822                                      sizeof(*reply_body));
1823         if (reply_body == NULL)
1824                 /* if there is no mds_body, no point in shrinking the reply */
1825                  return;
1826
1827         reply_mdoff++;
1828
1829         if (have_md || have_acl) {
1830                 if (reply_body->valid & (OBD_MD_FLEASIZE | OBD_MD_FLDIREA)) {
1831                         md_size = reply_body->eadatasize;
1832                 } else if (reply_body->valid & OBD_MD_LINKNAME)
1833                         md_size = reply_body->eadatasize;
1834
1835                 if (reply_body->valid & OBD_MD_FLCOOKIE) {
1836                         LASSERT(reply_body->valid & OBD_MD_FLEASIZE);
1837                         cookie_size = mds_get_cookie_size(obd, lustre_msg_buf(
1838                                                           req->rq_repmsg,
1839                                                           reply_mdoff, 0));
1840                 } else if (reply_body->valid & OBD_MD_FLACL) {
1841                         cookie_size = reply_body->aclsize;
1842                 }
1843         }
1844         CDEBUG(D_INFO, "Shrink %d/%d to md_size %d cookie_size %d \n",
1845                have_md, have_acl, md_size, cookie_size);
1846
1847         if (likely(have_md))
1848                 lustre_shrink_reply(req, reply_mdoff, md_size, 1);
1849
1850         if (likely(have_acl))
1851                 lustre_shrink_reply(req, reply_mdoff + (md_size > 0),
1852                                     cookie_size, 1);
1853 }
1854
1855 void mds_shrink_body_reply(struct ptlrpc_request *req,
1856                            int req_mdoff,
1857                            int reply_mdoff)
1858 {
1859         struct mds_body *rq_body;
1860         const long long have_acl = OBD_MD_FLCOOKIE | OBD_MD_FLACL;
1861         const long have_md = OBD_MD_FLEASIZE | OBD_MD_FLDIREA;
1862         ENTRY;
1863
1864         /* LSM and cookie is always placed after mds_body */
1865         rq_body =  lustre_msg_buf(req->rq_reqmsg, req_mdoff,
1866                                   sizeof(*rq_body));
1867         LASSERT(rq_body);
1868
1869         /* this check is need for avoid hit asset in case
1870          * OBD_MDS_FLFLAGS */
1871         mds_shrink_reply(req, reply_mdoff,
1872                          rq_body->valid & have_md,
1873                          rq_body->valid & have_acl);
1874 }
1875
1876 void mds_shrink_intent_reply(struct ptlrpc_request *req,
1877                              int opc, int reply_mdoff)
1878 {
1879         switch (opc) {
1880                 case REINT_UNLINK:
1881                 case REINT_RENAME:
1882                         mds_shrink_reply(req, reply_mdoff, 1, 1);
1883                         break;
1884                 case REINT_OPEN:
1885                         mds_shrink_reply(req, reply_mdoff, 1, 0);
1886                         break;
1887                 default:
1888                         break;
1889         }
1890 }
1891
1892 static int mds_reint_unlink(struct mds_update_record *rec, int offset,
1893                             struct ptlrpc_request *req,
1894                             struct lustre_handle *lh)
1895 {
1896         struct dentry *dparent = NULL, *dchild;
1897         struct mds_obd *mds = mds_req2mds(req);
1898         struct obd_device *obd = req->rq_export->exp_obd;
1899         struct mds_body *body = NULL;
1900         struct inode *inodes[PTLRPC_NUM_VERSIONS] = { NULL };
1901         struct inode *child_inode = NULL;
1902         struct lustre_handle parent_lockh, child_lockh, child_reuse_lockh;
1903         void *handle = NULL;
1904         int rc = 0, cleanup_phase = 0;
1905         unsigned int qcids[MAXQUOTAS] = { 0, 0 };
1906         unsigned int qpids[MAXQUOTAS] = { 0, 0 };
1907         ENTRY;
1908
1909         LASSERT(offset == REQ_REC_OFF); /*  || offset == DLM_INTENT_REC_OFF); */
1910         offset = REPLY_REC_OFF;
1911
1912         DEBUG_REQ(D_INODE, req, "parent ino "LPU64"/%u, child %s",
1913                   rec->ur_fid1->id, rec->ur_fid1->generation, rec->ur_name);
1914
1915         MDS_CHECK_RESENT(req, mds_reconstruct_generic(req));
1916
1917         if (OBD_FAIL_CHECK(OBD_FAIL_MDS_REINT_UNLINK))
1918                 GOTO(cleanup, rc = -ENOENT);
1919
1920         if (rec->ur_dlm)
1921                 ldlm_request_cancel(req, rec->ur_dlm, 0);
1922
1923         rc = mds_get_parent_child_locked(obd, mds, rec->ur_fid1,
1924                                          &parent_lockh, &dparent, LCK_EX,
1925                                          MDS_INODELOCK_UPDATE,
1926                                          rec->ur_name, rec->ur_namelen,
1927                                          &child_lockh, &dchild, LCK_EX,
1928                                          MDS_INODELOCK_FULL);
1929         if (rc)
1930                 GOTO(cleanup, rc);
1931
1932         cleanup_phase = 1; /* dchild, dparent, locks */
1933
1934         dget(dchild);
1935         child_inode = dchild->d_inode;
1936         if (child_inode == NULL) {
1937                 CDEBUG(D_INODE, "child doesn't exist (dir %lu, name %s)\n",
1938                        dparent->d_inode->i_ino, rec->ur_name);
1939                 GOTO(cleanup, rc = -ENOENT);
1940         }
1941
1942         /* save uid/gid for quota acquire/release */
1943         qcids[USRQUOTA] = child_inode->i_uid;
1944         qcids[GRPQUOTA] = child_inode->i_gid;
1945         qpids[USRQUOTA] = dparent->d_inode->i_uid;
1946         qpids[GRPQUOTA] = dparent->d_inode->i_gid;
1947
1948         cleanup_phase = 2; /* dchild has a lock */
1949
1950         /* VBR: version recovery check for parent */
1951         rc = mds_version_get_check(req, dparent->d_inode, 0);
1952         if (rc)
1953                 GOTO(cleanup_no_trans, rc);
1954
1955         /* version recovery check */
1956         rc = mds_version_get_check(req, child_inode, 1);
1957         if (rc)
1958                 GOTO(cleanup_no_trans, rc);
1959
1960         /* We have to do these checks ourselves, in case we are making an
1961          * orphan.  The client tells us whether rmdir() or unlink() was called,
1962          * so we need to return appropriate errors (bug 72). */
1963         if ((rec->ur_mode & S_IFMT) == S_IFDIR) {
1964                 if (!S_ISDIR(child_inode->i_mode))
1965                         GOTO(cleanup, rc = -ENOTDIR);
1966         } else {
1967                 if (S_ISDIR(child_inode->i_mode))
1968                         GOTO(cleanup, rc = -EISDIR);
1969         }
1970
1971         /* Check for EROFS after we check ENODENT, ENOTDIR, and EISDIR */
1972         if (req->rq_export->exp_connect_flags & OBD_CONNECT_RDONLY)
1973                 GOTO(cleanup, rc = -EROFS);
1974
1975         /* Step 3: Get a lock on the ino to sync with creation WRT inode
1976          * reuse (see bug 2029). */
1977         rc = mds_lock_new_child(obd, child_inode, &child_reuse_lockh);
1978         if (rc != ELDLM_OK)
1979                 GOTO(cleanup, rc);
1980
1981         cleanup_phase = 3; /* child inum lock */
1982
1983         OBD_FAIL_WRITE(obd, OBD_FAIL_MDS_REINT_UNLINK_WRITE, dparent->d_inode->i_sb);
1984
1985         /* ldlm_reply in buf[0] if called via intent */
1986         if (offset == DLM_INTENT_REC_OFF)
1987                 offset = DLM_REPLY_REC_OFF;
1988
1989         body = lustre_msg_buf(req->rq_repmsg, offset, sizeof(*body));
1990         LASSERT(body != NULL);
1991
1992         /* child orphan sem protects orphan_dec_test && is_orphan race */
1993         MDS_DOWN_READ_ORPHAN_SEM(child_inode);
1994         cleanup_phase = 4; /* MDS_UP_READ_ORPHAN_SEM(new_inode) when finished */
1995
1996         /* If this is potentially the last reference to this inode, get the
1997          * OBD EA data first so the client can destroy OST objects.  We
1998          * only do the object removal later if no open files/links remain. */
1999         if ((S_ISDIR(child_inode->i_mode) && child_inode->i_nlink == 2) ||
2000             child_inode->i_nlink == 1) {
2001                 if (mds_orphan_needed(obd, child_inode)) {
2002                         /* need to lock pending_dir before transaction */
2003                         LOCK_INODE_MUTEX(mds->mds_pending_dir->d_inode);
2004                         cleanup_phase = 5; /* UNLOCK_INODE_MUTEX(mds->mds_pending_dir->d_inode); */
2005                 } else if (S_ISREG(child_inode->i_mode)) {
2006                         mds_pack_inode2body(body, child_inode);
2007                         mds_pack_md(obd, req->rq_repmsg, offset + 1, body,
2008                                     child_inode, MDS_PACK_MD_LOCK, 0,
2009                                     req->rq_export->exp_connect_flags);
2010                 }
2011         }
2012
2013         /* Step 4: Do the unlink: we already verified ur_mode above (bug 72) */
2014         switch (child_inode->i_mode & S_IFMT) {
2015         case S_IFDIR:
2016                 /* Drop any lingering child directories before we start our
2017                  * transaction, to avoid doing multiple inode dirty/delete
2018                  * in our compound transaction (bug 1321). */
2019                 shrink_dcache_parent(dchild);
2020                 handle = fsfilt_start(obd, dparent->d_inode, FSFILT_OP_RMDIR,
2021                                       NULL);
2022                 if (IS_ERR(handle))
2023                         GOTO(cleanup, rc = PTR_ERR(handle));
2024                 LOCK_INODE_MUTEX(dparent->d_inode);
2025                 rc = ll_vfs_rmdir(dparent->d_inode, dchild, mds->mds_vfsmnt);
2026                 UNLOCK_INODE_MUTEX(dparent->d_inode);
2027                 mds_counter_incr(req->rq_export, LPROC_MDS_RMDIR);
2028                 break;
2029         case S_IFREG: {
2030                 struct lov_mds_md *lmm = lustre_msg_buf(req->rq_repmsg,
2031                                                         offset + 1, 0);
2032                 int sz = lustre_msg_buflen(req->rq_repmsg, offset + 1) > 0 ?
2033                          le32_to_cpu(lmm->lmm_stripe_count) : 0;
2034
2035                 handle = fsfilt_start_log(obd, dparent->d_inode,
2036                                           FSFILT_OP_UNLINK, NULL, sz);
2037                 if (IS_ERR(handle))
2038                         GOTO(cleanup, rc = PTR_ERR(handle));
2039                 LOCK_INODE_MUTEX(dparent->d_inode);
2040                 rc = ll_vfs_unlink(dparent->d_inode, dchild, mds->mds_vfsmnt);
2041                 UNLOCK_INODE_MUTEX(dparent->d_inode);
2042                 mds_counter_incr(req->rq_export, LPROC_MDS_UNLINK);
2043                 break;
2044         }
2045         case S_IFLNK:
2046         case S_IFCHR:
2047         case S_IFBLK:
2048         case S_IFIFO:
2049         case S_IFSOCK:
2050                 handle = fsfilt_start(obd, dparent->d_inode, FSFILT_OP_UNLINK,
2051                                       NULL);
2052                 if (IS_ERR(handle))
2053                         GOTO(cleanup, rc = PTR_ERR(handle));
2054                 LOCK_INODE_MUTEX(dparent->d_inode);
2055                 rc = ll_vfs_unlink(dparent->d_inode, dchild, mds->mds_vfsmnt);
2056                 UNLOCK_INODE_MUTEX(dparent->d_inode);
2057                 mds_counter_incr(req->rq_export, LPROC_MDS_UNLINK);
2058                 break;
2059         default:
2060                 CERROR("bad file type %o unlinking %s\n", rec->ur_mode,
2061                        rec->ur_name);
2062                 LBUG();
2063                 GOTO(cleanup, rc = -EINVAL);
2064         }
2065
2066         if (rc == 0 && child_inode->i_nlink == 0) {
2067                 if (mds_orphan_needed(obd, child_inode)) {
2068                         rc = mds_orphan_add_link(rec, obd, dchild);
2069                         if (rc == 1)
2070                                 /* child inode was successfully linked
2071                                  * to PENDING */
2072                                 GOTO(cleanup, rc = 0);
2073                         else
2074                                 /* we failed to move the file to PENDING,
2075                                  * really unlink the file as if there were
2076                                  * no more openers */
2077                                 rc = 0;
2078                 }
2079
2080                 if (!S_ISREG(child_inode->i_mode))
2081                         GOTO(cleanup, rc);
2082
2083                 if (!(body->valid & OBD_MD_FLEASIZE)) {
2084                         body->valid |=(OBD_MD_FLSIZE | OBD_MD_FLBLOCKS |
2085                                        OBD_MD_FLATIME | OBD_MD_FLMTIME);
2086                 } else if (mds_log_op_unlink(obd,
2087                                 lustre_msg_buf(req->rq_repmsg, offset + 1, 0),
2088                                 lustre_msg_buflen(req->rq_repmsg, offset + 1),
2089                                 lustre_msg_buf(req->rq_repmsg, offset + 2, 0),
2090                                 lustre_msg_buflen(req->rq_repmsg, offset+2)) >
2091                            0) {
2092                         body->valid |= OBD_MD_FLCOOKIE;
2093                 }
2094         }
2095
2096         GOTO(cleanup, rc);
2097 cleanup:
2098         if (rc == 0) {
2099                 struct iattr iattr;
2100                 int err;
2101
2102                 if (child_inode->i_nlink > 0 ||
2103                     mds_orphan_open_count(child_inode) > 0) {
2104                         /* update ctime of unlinked file only if it is still
2105                          * opened or a link still exists */
2106                         iattr.ia_valid = ATTR_CTIME;
2107                         LTIME_S(iattr.ia_ctime) = rec->ur_time;
2108                         err = fsfilt_setattr(obd, dchild, handle, &iattr, 0);
2109                         if (err)
2110                                 CERROR("error on unlinked inode time update: "
2111                                        "rc = %d\n", err);
2112                 }
2113
2114                 /* update mtime and ctime of parent directory*/
2115                 iattr.ia_valid = ATTR_MTIME | ATTR_CTIME;
2116                 LTIME_S(iattr.ia_mtime) = rec->ur_time;
2117                 LTIME_S(iattr.ia_ctime) = rec->ur_time;
2118                 err = fsfilt_setattr(obd, dparent, handle, &iattr, 0);
2119                 if (err)
2120                         CERROR("error on parent setattr: rc = %d\n", err);
2121         }
2122         inodes[0] = dparent ? dparent->d_inode : NULL;
2123         inodes[1] = child_inode;
2124         rc = mds_finish_transno(mds, inodes, handle, req, rc, 0, 0);
2125
2126 cleanup_no_trans:
2127         switch(cleanup_phase) {
2128         case 5: /* pending_dir semaphore */
2129                 UNLOCK_INODE_MUTEX(mds->mds_pending_dir->d_inode);
2130         case 4: /* child inode semaphore */
2131                 MDS_UP_READ_ORPHAN_SEM(child_inode);
2132         case 3: /* child ino-reuse lock */
2133                 if (rc && body != NULL) {
2134                         // Don't unlink the OST objects if the MDS unlink failed
2135                         body->valid = 0;
2136                 }
2137                 if (rc)
2138                         ldlm_lock_decref(&child_reuse_lockh, LCK_EX);
2139                 else
2140                         ptlrpc_save_lock(req, &child_reuse_lockh, LCK_EX);
2141         case 2: /* child lock */
2142                 ldlm_lock_decref(&child_lockh, LCK_EX);
2143         case 1: /* child and parent dentry, parent lock */
2144                 if (rc)
2145                         ldlm_lock_decref(&parent_lockh, LCK_EX);
2146                 else
2147                         ptlrpc_save_lock(req, &parent_lockh, LCK_EX);
2148                 l_dput(dchild);
2149                 l_dput(dchild);
2150                 l_dput(dparent);
2151         case 0:
2152                 break;
2153         default:
2154                 CERROR("invalid cleanup_phase %d\n", cleanup_phase);
2155                 LBUG();
2156         }
2157         req->rq_status = rc;
2158
2159         /* trigger dqrel on the owner of child and parent */
2160         lquota_adjust(mds_quota_interface_ref, obd, qcids, qpids, rc,
2161                       FSFILT_OP_UNLINK);
2162         return 0;
2163 }
2164
2165 static int mds_reint_link(struct mds_update_record *rec, int offset,
2166                           struct ptlrpc_request *req,
2167                           struct lustre_handle *lh)
2168 {
2169         struct obd_device *obd = req->rq_export->exp_obd;
2170         struct dentry *de_src = NULL;
2171         struct dentry *de_tgt_dir = NULL;
2172         struct dentry *dchild = NULL;
2173         struct inode *inodes[PTLRPC_NUM_VERSIONS] = { NULL };
2174         struct mds_obd *mds = mds_req2mds(req);
2175         struct lustre_handle *handle = NULL, tgt_dir_lockh, src_lockh;
2176         struct ldlm_res_id src_res_id = { .name = {0} };
2177         struct ldlm_res_id tgt_dir_res_id = { .name = {0} };
2178         ldlm_policy_data_t src_policy ={.l_inodebits = {MDS_INODELOCK_UPDATE}};
2179         ldlm_policy_data_t tgt_dir_policy =
2180                                        {.l_inodebits = {MDS_INODELOCK_UPDATE}};
2181         int rc = 0, cleanup_phase = 0;
2182         ENTRY;
2183
2184         LASSERT(offset == REQ_REC_OFF);
2185
2186         DEBUG_REQ(D_INODE, req, "original "LPU64"/%u to "LPU64"/%u %s",
2187                   rec->ur_fid1->id, rec->ur_fid1->generation,
2188                   rec->ur_fid2->id, rec->ur_fid2->generation, rec->ur_name);
2189
2190         mds_counter_incr(req->rq_export, LPROC_MDS_LINK);
2191
2192         MDS_CHECK_RESENT(req, mds_reconstruct_generic(req));
2193
2194         if (OBD_FAIL_CHECK(OBD_FAIL_MDS_REINT_LINK))
2195                 GOTO(cleanup, rc = -ENOENT);
2196
2197         if (rec->ur_dlm)
2198                 ldlm_request_cancel(req, rec->ur_dlm, 0);
2199
2200         /* Step 1: Lookup the source inode and target directory by FID */
2201         de_src = mds_fid2dentry(mds, rec->ur_fid1, NULL);
2202         if (IS_ERR(de_src))
2203                 GOTO(cleanup, rc = PTR_ERR(de_src));
2204
2205         cleanup_phase = 1; /* source dentry */
2206
2207         de_tgt_dir = mds_fid2dentry(mds, rec->ur_fid2, NULL);
2208         if (IS_ERR(de_tgt_dir)) {
2209                 rc = PTR_ERR(de_tgt_dir);
2210                 de_tgt_dir = NULL;
2211                 GOTO(cleanup, rc);
2212         }
2213
2214         cleanup_phase = 2; /* target directory dentry */
2215
2216         CDEBUG(D_INODE, "linking %.*s/%s to inode %lu\n",
2217                de_tgt_dir->d_name.len, de_tgt_dir->d_name.name, rec->ur_name,
2218                de_src->d_inode->i_ino);
2219
2220         /* Step 2: Take the two locks */
2221         src_res_id.name[0] = de_src->d_inode->i_ino;
2222         src_res_id.name[1] = de_src->d_inode->i_generation;
2223         tgt_dir_res_id.name[0] = de_tgt_dir->d_inode->i_ino;
2224         tgt_dir_res_id.name[1] = de_tgt_dir->d_inode->i_generation;
2225
2226         rc = enqueue_ordered_locks(obd, &src_res_id, &src_lockh, LCK_EX,
2227                                    &src_policy,
2228                                    &tgt_dir_res_id, &tgt_dir_lockh, LCK_EX,
2229                                    &tgt_dir_policy);
2230         if (rc)
2231                 GOTO(cleanup, rc);
2232
2233         cleanup_phase = 3; /* locks */
2234
2235         /* version recovery check */
2236         /* directory check */
2237         rc = mds_version_get_check(req, de_tgt_dir->d_inode, 0);
2238         if (rc)
2239                 GOTO(cleanup_no_trans, rc);
2240         /* inode version check */
2241         rc = mds_version_get_check(req, de_src->d_inode, 1);
2242         if (rc)
2243                 GOTO(cleanup_no_trans, rc);
2244
2245         if (mds_inode_is_orphan(de_src->d_inode)) {
2246                 CDEBUG(D_INODE, "an attempt to link an orphan inode %lu/%u\n",
2247                        de_src->d_inode->i_ino,
2248                        de_src->d_inode->i_generation);
2249                 GOTO(cleanup, rc = -ENOENT);
2250         }
2251
2252         /* Step 3: Lookup the child */
2253         dchild = mds_lookup(obd, rec->ur_name, de_tgt_dir, rec->ur_namelen-1);
2254         if (IS_ERR(dchild)) {
2255                 rc = PTR_ERR(dchild);
2256                 dchild = NULL;
2257                 if (rc != -EPERM && rc != -EACCES && rc != -ENAMETOOLONG)
2258                         CERROR("child lookup error %d\n", rc);
2259                 GOTO(cleanup, rc);
2260         }
2261
2262         cleanup_phase = 4; /* child dentry */
2263
2264         if (dchild->d_inode) {
2265                 CDEBUG(D_INODE, "child exists (dir %lu, name %s)\n",
2266                        de_tgt_dir->d_inode->i_ino, rec->ur_name);
2267                 rc = -EEXIST;
2268                 GOTO(cleanup, rc);
2269         }
2270
2271         /* Step 4: Do it. */
2272         OBD_FAIL_WRITE(obd, OBD_FAIL_MDS_REINT_LINK_WRITE, de_src->d_inode->i_sb);
2273
2274         if (req->rq_export->exp_connect_flags & OBD_CONNECT_RDONLY)
2275                 GOTO(cleanup, rc = -EROFS);
2276
2277         handle = fsfilt_start(obd, de_tgt_dir->d_inode, FSFILT_OP_LINK, NULL);
2278         if (IS_ERR(handle))
2279                 GOTO(cleanup, rc = PTR_ERR(handle));
2280
2281         LOCK_INODE_MUTEX(de_tgt_dir->d_inode);
2282         rc = ll_vfs_link(de_src, mds->mds_vfsmnt, de_tgt_dir->d_inode, dchild,
2283                          mds->mds_vfsmnt);
2284         UNLOCK_INODE_MUTEX(de_tgt_dir->d_inode);
2285         if (rc && rc != -EPERM && rc != -EACCES)
2286                 CERROR("vfs_link error %d\n", rc);
2287         if (rc == 0) {
2288                 struct iattr iattr;
2289                 int err;
2290
2291                 /* update ctime of old file */
2292                 iattr.ia_valid = ATTR_CTIME;
2293                 LTIME_S(iattr.ia_ctime) = rec->ur_time;
2294                 err = fsfilt_setattr(obd, de_src, handle, &iattr, 0);
2295                 if (err)
2296                         CERROR("error on old inode time update: "
2297                                "rc = %d\n", err);
2298
2299                 /* update mtime and ctime of target directory */
2300                 iattr.ia_valid = ATTR_MTIME | ATTR_CTIME;
2301                 LTIME_S(iattr.ia_mtime) = rec->ur_time;
2302                 LTIME_S(iattr.ia_ctime) = rec->ur_time;
2303                 err = fsfilt_setattr(obd, de_tgt_dir, handle, &iattr, 0);
2304                 if (err)
2305                         CERROR("error on target dir inode time update: "
2306                                "rc = %d\n", err);
2307         }
2308 cleanup:
2309         inodes[0] = de_tgt_dir ? de_tgt_dir->d_inode : NULL;
2310         inodes[1] = (dchild && !IS_ERR(dchild)) ? dchild->d_inode : NULL;
2311         rc = mds_finish_transno(mds, inodes, handle, req, rc, 0, 0);
2312         EXIT;
2313
2314 cleanup_no_trans:
2315         switch (cleanup_phase) {
2316         case 4: /* child dentry */
2317                 l_dput(dchild);
2318         case 3: /* locks */
2319                 if (rc) {
2320                         ldlm_lock_decref(&src_lockh, LCK_EX);
2321                         ldlm_lock_decref(&tgt_dir_lockh, LCK_EX);
2322                 } else {
2323                         ptlrpc_save_lock(req, &src_lockh, LCK_EX);
2324                         ptlrpc_save_lock(req, &tgt_dir_lockh, LCK_EX);
2325                 }
2326         case 2: /* target dentry */
2327                 l_dput(de_tgt_dir);
2328         case 1: /* source dentry */
2329                 l_dput(de_src);
2330         case 0:
2331                 break;
2332         default:
2333                 CERROR("invalid cleanup_phase %d\n", cleanup_phase);
2334                 LBUG();
2335         }
2336         req->rq_status = rc;
2337         return 0;
2338 }
2339
2340 /* The idea here is that we need to get four locks in the end:
2341  * one on each parent directory, one on each child.  We need to take
2342  * these locks in some kind of order (to avoid deadlocks), and the order
2343  * I selected is "increasing resource number" order.  We need to look up
2344  * the children, however, before we know what the resource number(s) are.
2345  * Thus the following plan:
2346  *
2347  * 1,2. Look up the parents
2348  * 3,4. Look up the children
2349  * 5. Take locks on the parents and children, in order
2350  * 6. Verify that the children haven't changed since they were looked up
2351  *
2352  * If there was a race and the children changed since they were first looked
2353  * up, it is possible that mds_verify_child() will be able to just grab the
2354  * lock on the new child resource (if it has a higher resource than any other)
2355  * but we need to compare against not only its parent, but also against the
2356  * parent and child of the "other half" of the rename, hence maxres_{src,tgt}.
2357  *
2358  * We need the fancy igrab() on the child inodes because we aren't holding a
2359  * lock on the parent after the lookup is done, so dentry->d_inode may change
2360  * at any time, and igrab() itself doesn't like getting passed a NULL argument.
2361  */
2362 int mds_get_parents_children_locked(struct obd_device *obd,
2363                                     struct mds_obd *mds,
2364                                     struct ll_fid *p1_fid,
2365                                     struct dentry **de_srcdirp,
2366                                     struct ll_fid *p2_fid,
2367                                     struct dentry **de_tgtdirp,
2368                                     int parent_mode,
2369                                     const char *old_name, int old_len,
2370                                     struct dentry **de_oldp,
2371                                     const char *new_name, int new_len,
2372                                     struct dentry **de_newp,
2373                                     struct lustre_handle *dlm_handles,
2374                                     int child_mode)
2375 {
2376         struct ldlm_res_id p1_res_id = { .name = {0} };
2377         struct ldlm_res_id p2_res_id = { .name = {0} };
2378         struct ldlm_res_id c1_res_id = { .name = {0} };
2379         struct ldlm_res_id c2_res_id = { .name = {0} };
2380         ldlm_policy_data_t p_policy = {.l_inodebits = {MDS_INODELOCK_UPDATE}};
2381         /* Only dentry should disappear, but the inode itself would be
2382            intact otherwise. */
2383         ldlm_policy_data_t c1_policy = {.l_inodebits = {MDS_INODELOCK_LOOKUP}};
2384         /* If something is going to be replaced, both dentry and inode locks are
2385          * needed */
2386         ldlm_policy_data_t c2_policy = {.l_inodebits = {MDS_INODELOCK_FULL}};
2387         struct ldlm_res_id *maxres_src, *maxres_tgt;
2388         struct inode *inode;
2389         int rc = 0, cleanup_phase = 0;
2390         ENTRY;
2391
2392         /* Step 1: Lookup the source directory */
2393         *de_srcdirp = mds_fid2dentry(mds, p1_fid, NULL);
2394         if (IS_ERR(*de_srcdirp))
2395                 GOTO(cleanup, rc = PTR_ERR(*de_srcdirp));
2396
2397         cleanup_phase = 1; /* source directory dentry */
2398
2399         p1_res_id.name[0] = (*de_srcdirp)->d_inode->i_ino;
2400         p1_res_id.name[1] = (*de_srcdirp)->d_inode->i_generation;
2401
2402         /* Step 2: Lookup the target directory */
2403         if (memcmp(p1_fid, p2_fid, sizeof(*p1_fid)) == 0) {
2404                 *de_tgtdirp = dget(*de_srcdirp);
2405         } else {
2406                 *de_tgtdirp = mds_fid2dentry(mds, p2_fid, NULL);
2407                 if (IS_ERR(*de_tgtdirp)) {
2408                         rc = PTR_ERR(*de_tgtdirp);
2409                         *de_tgtdirp = NULL;
2410                         GOTO(cleanup, rc);
2411                 }
2412         }
2413
2414         cleanup_phase = 2; /* target directory dentry */
2415
2416         p2_res_id.name[0] = (*de_tgtdirp)->d_inode->i_ino;
2417         p2_res_id.name[1] = (*de_tgtdirp)->d_inode->i_generation;
2418
2419         /* Step 3: Lookup the source child entry */
2420         *de_oldp = mds_lookup(obd, old_name, *de_srcdirp, old_len - 1);
2421         if (IS_ERR(*de_oldp)) {
2422                 rc = PTR_ERR(*de_oldp);
2423                 CDEBUG(D_INODE, "old child lookup error (%.*s): rc %d\n",
2424                        old_len - 1, old_name, rc);
2425                 GOTO(cleanup, rc);
2426         }
2427
2428         cleanup_phase = 3; /* original name dentry */
2429
2430         inode = (*de_oldp)->d_inode;
2431         if (inode != NULL)
2432                 inode = igrab(inode);
2433         if (inode == NULL)
2434                 GOTO(cleanup, rc = -ENOENT);
2435
2436         c1_res_id.name[0] = inode->i_ino;
2437         c1_res_id.name[1] = inode->i_generation;
2438
2439         iput(inode);
2440
2441         /* Step 4: Lookup the target child entry */
2442         if (!new_name)
2443                 GOTO(retry_locks, rc);
2444         *de_newp = mds_lookup(obd, new_name, *de_tgtdirp, new_len - 1);
2445         if (IS_ERR(*de_newp)) {
2446                 rc = PTR_ERR(*de_newp);
2447                 CDEBUG(D_DENTRY, "new child lookup error (%.*s): rc %d\n",
2448                        old_len - 1, old_name, rc);
2449                 GOTO(cleanup, rc);
2450         }
2451
2452         cleanup_phase = 4; /* target dentry */
2453
2454         inode = (*de_newp)->d_inode;
2455         if (inode != NULL) {
2456                 if (is_bad_inode(inode)) {
2457                         CERROR("bad inode returned %lu/%u\n",
2458                                inode->i_ino, inode->i_generation);
2459                         GOTO(cleanup, rc = -ENOENT);
2460                 }
2461                 inode = igrab(inode);
2462         }
2463         if (inode == NULL)
2464                 goto retry_locks;
2465
2466         c2_res_id.name[0] = inode->i_ino;
2467         c2_res_id.name[1] = inode->i_generation;
2468         iput(inode);
2469
2470 retry_locks:
2471         /* Step 5: Take locks on the parents and child(ren) */
2472         maxres_src = &p1_res_id;
2473         maxres_tgt = &p2_res_id;
2474         cleanup_phase = 4; /* target dentry */
2475
2476         if (c2_res_id.name[0] != 0 && res_gt(&c2_res_id, &p2_res_id,NULL,NULL))
2477                 maxres_tgt = &c2_res_id;
2478
2479         rc = enqueue_4ordered_locks(obd, &p1_res_id,&dlm_handles[0],parent_mode,
2480                                     &p_policy,
2481                                     &p2_res_id, &dlm_handles[1], parent_mode,
2482                                     &p_policy,
2483                                     &c1_res_id, &dlm_handles[2], child_mode,
2484                                     &c1_policy,
2485                                     &c2_res_id, &dlm_handles[3], child_mode,
2486                                     &c2_policy);
2487         if (rc)
2488                 GOTO(cleanup, rc);
2489
2490         cleanup_phase = 6; /* parent and child(ren) locks */
2491
2492         /* Step 6a: Re-lookup source child to verify it hasn't changed */
2493         rc = mds_verify_child(obd, &p1_res_id, &dlm_handles[0], *de_srcdirp,
2494                               parent_mode, &c1_res_id, &dlm_handles[2], de_oldp,
2495                               child_mode, &c1_policy, old_name, old_len,
2496                               maxres_tgt);
2497         if (rc) {
2498                 if (c2_res_id.name[0] != 0)
2499                         ldlm_lock_decref(&dlm_handles[3], child_mode);
2500                 ldlm_lock_decref(&dlm_handles[1], parent_mode);
2501                 cleanup_phase = 4;
2502                 if (rc > 0)
2503                         goto retry_locks;
2504                 GOTO(cleanup, rc);
2505         }
2506
2507         if ((*de_oldp)->d_inode == NULL)
2508                 GOTO(cleanup, rc = -ENOENT);
2509
2510         if (!new_name)
2511                 GOTO(cleanup, rc);
2512
2513         /* Safe to skip check for child res being all zero */
2514         if (res_gt(&c1_res_id, maxres_src, NULL, NULL))
2515                 maxres_src = &c1_res_id;
2516
2517         /* Step 6b: Re-lookup target child to verify it hasn't changed */
2518         rc = mds_verify_child(obd, &p2_res_id, &dlm_handles[1], *de_tgtdirp,
2519                               parent_mode, &c2_res_id, &dlm_handles[3], de_newp,
2520                               child_mode, &c2_policy, new_name, new_len,
2521                               maxres_src);
2522         if (rc) {
2523                 ldlm_lock_decref(&dlm_handles[2], child_mode);
2524                 ldlm_lock_decref(&dlm_handles[0], parent_mode);
2525                 cleanup_phase = 4;
2526                 if (rc > 0)
2527                         goto retry_locks;
2528                 GOTO(cleanup, rc);
2529         }
2530
2531         EXIT;
2532 cleanup:
2533         if (rc) {
2534                 switch (cleanup_phase) {
2535                 case 6: /* child lock(s) */
2536                         if (c2_res_id.name[0] != 0)
2537                                 ldlm_lock_decref(&dlm_handles[3], child_mode);
2538                         if (c1_res_id.name[0] != 0)
2539                                 ldlm_lock_decref(&dlm_handles[2], child_mode);
2540                 case 5: /* parent locks */
2541                         ldlm_lock_decref(&dlm_handles[1], parent_mode);
2542                         ldlm_lock_decref(&dlm_handles[0], parent_mode);
2543                 case 4: /* target dentry */
2544                         l_dput(*de_newp);
2545                 case 3: /* source dentry */
2546                         l_dput(*de_oldp);
2547                 case 2: /* target directory dentry */
2548                         l_dput(*de_tgtdirp);
2549                 case 1: /* source directry dentry */
2550                         l_dput(*de_srcdirp);
2551                 }
2552         }
2553
2554         return rc;
2555 }
2556
2557 static int mds_reint_rename(struct mds_update_record *rec, int offset,
2558                             struct ptlrpc_request *req,
2559                             struct lustre_handle *lockh)
2560 {
2561         struct obd_device *obd = req->rq_export->exp_obd;
2562         struct dentry *de_srcdir = NULL;
2563         struct dentry *de_tgtdir = NULL;
2564         struct dentry *de_old = NULL;
2565         struct dentry *de_new = NULL;
2566         struct dentry *trap;
2567         struct inode *old_inode = NULL, *new_inode = NULL;
2568         struct inode *inodes[PTLRPC_NUM_VERSIONS] = { NULL };
2569         struct mds_obd *mds = mds_req2mds(req);
2570         struct lustre_handle dlm_handles[4];
2571         struct mds_body *body = NULL;
2572         struct lov_mds_md *lmm = NULL;
2573         int rc = 0, lock_count = 3, cleanup_phase = 0, sz;
2574         void *handle = NULL;
2575         unsigned int qcids[MAXQUOTAS] = { 0, 0 };
2576         unsigned int qpids[4] = { 0, 0, 0, 0 };
2577         ENTRY;
2578
2579         LASSERT(offset == REQ_REC_OFF);
2580         offset = REPLY_REC_OFF;
2581
2582         DEBUG_REQ(D_INODE, req, "parent "LPU64"/%u %s to "LPU64"/%u %s",
2583                   rec->ur_fid1->id, rec->ur_fid1->generation, rec->ur_name,
2584                   rec->ur_fid2->id, rec->ur_fid2->generation, rec->ur_tgt);
2585
2586         mds_counter_incr(req->rq_export, LPROC_MDS_RENAME);
2587
2588         MDS_CHECK_RESENT(req, mds_reconstruct_generic(req));
2589
2590         if (rec->ur_dlm)
2591                 ldlm_request_cancel(req, rec->ur_dlm, 0);
2592
2593         rc = mds_get_parents_children_locked(obd, mds, rec->ur_fid1, &de_srcdir,
2594                                              rec->ur_fid2, &de_tgtdir, LCK_EX,
2595                                              rec->ur_name, rec->ur_namelen,
2596                                              &de_old, rec->ur_tgt,
2597                                              rec->ur_tgtlen, &de_new,
2598                                              dlm_handles, LCK_EX);
2599         if (rc)
2600                 GOTO(cleanup, rc);
2601
2602         cleanup_phase = 1; /* parent(s), children, locks */
2603
2604         old_inode = de_old->d_inode;
2605         new_inode = de_new->d_inode;
2606
2607         if (new_inode != NULL)
2608                 lock_count = 4;
2609
2610         /* version recovery check */
2611         rc = mds_version_get_check(req, de_srcdir->d_inode, 0);
2612         if (rc)
2613                 GOTO(cleanup_no_trans, rc);
2614         rc = mds_version_get_check(req, old_inode, 1);
2615         if (rc)
2616                 GOTO(cleanup_no_trans, rc);
2617         rc = mds_version_get_check(req, de_tgtdir->d_inode, 2);
2618         if (rc)
2619                 GOTO(cleanup_no_trans, rc);
2620         rc = mds_version_get_check(req, new_inode, 3);
2621         if (rc)
2622                 GOTO(cleanup_no_trans, rc);
2623
2624         /* sanity check for src inode */
2625         if (old_inode->i_ino == de_srcdir->d_inode->i_ino ||
2626             old_inode->i_ino == de_tgtdir->d_inode->i_ino)
2627                 GOTO(cleanup, rc = -EINVAL);
2628
2629         if (req->rq_export->exp_connect_flags & OBD_CONNECT_RDONLY)
2630                 GOTO(cleanup, rc = -EROFS);
2631
2632         if (new_inode == NULL)
2633                 goto no_unlink;
2634
2635         igrab(new_inode);
2636         cleanup_phase = 2; /* iput(new_inode) when finished */
2637
2638         /* sanity check for dest inode */
2639         if (new_inode->i_ino == de_srcdir->d_inode->i_ino ||
2640             new_inode->i_ino == de_tgtdir->d_inode->i_ino)
2641                 GOTO(cleanup, rc = -EINVAL);
2642
2643         if (old_inode == new_inode)
2644                 GOTO(cleanup, rc = 0);
2645
2646         /* save uids/gids for qunit acquire/release */
2647         qcids[USRQUOTA] = old_inode->i_uid;
2648         qcids[GRPQUOTA] = old_inode->i_gid;
2649         qpids[USRQUOTA] = de_tgtdir->d_inode->i_uid;
2650         qpids[GRPQUOTA] = de_tgtdir->d_inode->i_gid;
2651         qpids[2] = de_srcdir->d_inode->i_uid;
2652         qpids[3] = de_srcdir->d_inode->i_gid;
2653
2654         /* if we are about to remove the target at first, pass the EA of
2655          * that inode to client to perform and cleanup on OST */
2656         body = lustre_msg_buf(req->rq_repmsg, offset, sizeof(*body));
2657         LASSERT(body != NULL);
2658
2659         /* child orphan sem protects orphan_dec_test && is_orphan race */
2660         MDS_DOWN_READ_ORPHAN_SEM(new_inode);
2661         cleanup_phase = 3; /* MDS_UP_READ_ORPHAN_SEM(new_inode) when finished */
2662
2663         if ((S_ISDIR(new_inode->i_mode) && new_inode->i_nlink == 2) ||
2664             new_inode->i_nlink == 1) {
2665                 if (mds_orphan_needed(obd, new_inode)) {
2666                         /* need to lock pending_dir before transaction */
2667                         LOCK_INODE_MUTEX(mds->mds_pending_dir->d_inode);
2668                         cleanup_phase = 4; /* UNLOCK_INODE_MUTEX(mds->mds_pending_dir->d_inode); */
2669                 } else if (S_ISREG(new_inode->i_mode)) {
2670                         mds_pack_inode2body(body, new_inode);
2671                         mds_pack_md(obd, req->rq_repmsg, offset + 1, body,
2672                                     new_inode, MDS_PACK_MD_LOCK, 0,
2673                                     req->rq_export->exp_connect_flags);
2674                 }
2675         }
2676
2677 no_unlink:
2678         OBD_FAIL_WRITE(obd, OBD_FAIL_MDS_REINT_RENAME_WRITE,
2679                        de_srcdir->d_inode->i_sb);
2680
2681         lmm = lustre_msg_buf(req->rq_repmsg, offset + 1, 0);
2682         /* check that lmm size is not 0 */
2683         sz = lustre_msg_buflen(req->rq_repmsg, offset + 1) > 0 ?
2684              le32_to_cpu(lmm->lmm_stripe_count) : 0;
2685
2686         handle = fsfilt_start_log(obd, de_tgtdir->d_inode, FSFILT_OP_RENAME,
2687                                   NULL, sz);
2688         if (IS_ERR(handle))
2689                 GOTO(cleanup, rc = PTR_ERR(handle));
2690
2691         trap = lock_rename(de_tgtdir, de_srcdir);
2692         /* source should not be ancestor of target */
2693         if (de_old == trap) {
2694                 unlock_rename(de_tgtdir, de_srcdir);
2695                 GOTO(cleanup, rc = -EINVAL);
2696         }
2697         /* target should not be an ancestor of source */
2698         if (de_new == trap) {
2699                 unlock_rename(de_tgtdir, de_srcdir);
2700                 GOTO(cleanup, rc = -ENOTEMPTY);
2701         }
2702
2703         de_old->d_fsdata = req;
2704         de_new->d_fsdata = req;
2705
2706         rc = ll_vfs_rename(de_srcdir->d_inode, de_old, mds->mds_vfsmnt,
2707                            de_tgtdir->d_inode, de_new, mds->mds_vfsmnt);
2708
2709         unlock_rename(de_tgtdir, de_srcdir);
2710
2711         if (rc == 0) {
2712                 struct iattr iattr;
2713                 int err;
2714
2715                 /* update ctime of renamed file */
2716                 iattr.ia_valid = ATTR_CTIME;
2717                 LTIME_S(iattr.ia_ctime) = rec->ur_time;
2718                 if (S_ISDIR(de_old->d_inode->i_mode) &&
2719                     de_srcdir->d_inode != de_tgtdir->d_inode) {
2720                         /* cross directory rename of a directory, ".."
2721                            changed, update mtime also */
2722                         iattr.ia_valid |= ATTR_MTIME;
2723                         LTIME_S(iattr.ia_mtime) = rec->ur_time;
2724                 }
2725                 err = fsfilt_setattr(obd, de_old, handle, &iattr, 0);
2726                 if (err)
2727                         CERROR("error on old inode time update: "
2728                                "rc = %d\n", err);
2729
2730                 if (de_new->d_inode) {
2731                         /* target file exists, update its ctime as it
2732                            gets unlinked */
2733                         iattr.ia_valid = ATTR_CTIME;
2734                         LTIME_S(iattr.ia_ctime) = rec->ur_time;
2735                         err = fsfilt_setattr(obd, de_new, handle, &iattr, 0);
2736                         if (err)
2737                                 CERROR("error on target inode time update: "
2738                                        "rc = %d\n", err);
2739                 }
2740
2741                 /* update mtime and ctime of old directory */
2742                 iattr.ia_valid = ATTR_MTIME | ATTR_CTIME;
2743                 LTIME_S(iattr.ia_mtime) = rec->ur_time;
2744                 LTIME_S(iattr.ia_ctime) = rec->ur_time;
2745                 err = fsfilt_setattr(obd, de_srcdir, handle, &iattr, 0);
2746                 if (err)
2747                         CERROR("error on old dir inode update: "
2748                                "rc = %d\n", err);
2749
2750                 if (de_srcdir->d_inode != de_tgtdir->d_inode) {
2751                         /* cross directory rename, update
2752                            mtime and ctime of new directory */
2753                         iattr.ia_valid = ATTR_MTIME | ATTR_CTIME;
2754                         LTIME_S(iattr.ia_mtime) = rec->ur_time;
2755                         LTIME_S(iattr.ia_ctime) = rec->ur_time;
2756                         err = fsfilt_setattr(obd, de_tgtdir, handle, &iattr, 0);
2757                         if (err)
2758                                 CERROR("error on new dir inode time update: "
2759                                        "rc = %d\n", err);
2760                 }
2761         }
2762
2763         if (rc == 0 && new_inode != NULL && new_inode->i_nlink == 0) {
2764                 if (mds_orphan_needed(obd, new_inode)) {
2765                         rc = mds_orphan_add_link(rec, obd, de_new);
2766
2767                         if (rc == 1)
2768                                 /* inode successfully linked to PENDING */
2769                                 GOTO(cleanup, rc = 0);
2770                         else
2771                                 /* we failed to move the file to PENDING,
2772                                  * really unlink the file as if there were
2773                                  * no more openers */
2774                                 rc = 0;
2775                 }
2776
2777                 if (!S_ISREG(new_inode->i_mode))
2778                         GOTO(cleanup, rc);
2779
2780                 if (!(body->valid & OBD_MD_FLEASIZE)) {
2781                         body->valid |= (OBD_MD_FLSIZE | OBD_MD_FLBLOCKS |
2782                                         OBD_MD_FLATIME | OBD_MD_FLMTIME);
2783                 } else if (mds_log_op_unlink(obd,
2784                                              lustre_msg_buf(req->rq_repmsg,
2785                                                             offset + 1, 0),
2786                                              lustre_msg_buflen(req->rq_repmsg,
2787                                                                offset + 1),
2788                                              lustre_msg_buf(req->rq_repmsg,
2789                                                             offset + 2, 0),
2790                                              lustre_msg_buflen(req->rq_repmsg,
2791                                                                offset + 2))
2792                            > 0) {
2793                         body->valid |= OBD_MD_FLCOOKIE;
2794                 }
2795         }
2796
2797         GOTO(cleanup, rc);
2798 cleanup:
2799         inodes[0] = de_srcdir && !IS_ERR(de_srcdir) ? de_srcdir->d_inode : NULL;
2800         inodes[1] = old_inode;
2801         inodes[2] = de_tgtdir && !IS_ERR(de_tgtdir) ? de_tgtdir->d_inode : NULL;
2802         inodes[3] = new_inode;
2803         rc = mds_finish_transno(mds, inodes, handle, req, rc, 0, 0);
2804
2805 cleanup_no_trans:
2806         switch (cleanup_phase) {
2807         case 4:
2808                 UNLOCK_INODE_MUTEX(mds->mds_pending_dir->d_inode);
2809         case 3:
2810                 MDS_UP_READ_ORPHAN_SEM(new_inode);
2811         case 2:
2812                 iput(new_inode);
2813         case 1:
2814                 if (rc) {
2815                         if (lock_count == 4)
2816                                 ldlm_lock_decref(&(dlm_handles[3]), LCK_EX);
2817                         ldlm_lock_decref(&(dlm_handles[2]), LCK_EX);
2818                         ldlm_lock_decref(&(dlm_handles[1]), LCK_EX);
2819                         ldlm_lock_decref(&(dlm_handles[0]), LCK_EX);
2820                 } else {
2821                         if (lock_count == 4)
2822                                 ptlrpc_save_lock(req,&(dlm_handles[3]), LCK_EX);
2823                         ptlrpc_save_lock(req, &(dlm_handles[2]), LCK_EX);
2824                         ptlrpc_save_lock(req, &(dlm_handles[1]), LCK_EX);
2825                         ptlrpc_save_lock(req, &(dlm_handles[0]), LCK_EX);
2826                 }
2827                 l_dput(de_new);
2828                 l_dput(de_old);
2829                 l_dput(de_tgtdir);
2830                 l_dput(de_srcdir);
2831         case 0:
2832                 break;
2833         default:
2834                 CERROR("invalid cleanup_phase %d\n", cleanup_phase);
2835                 LBUG();
2836         }
2837         req->rq_status = rc;
2838
2839         /* acquire/release qunit */
2840         lquota_adjust(mds_quota_interface_ref, obd, qcids, qpids, rc,
2841                       FSFILT_OP_RENAME);
2842         return 0;
2843 }
2844
2845 typedef int (*mds_reinter)(struct mds_update_record *, int offset,
2846                            struct ptlrpc_request *, struct lustre_handle *);
2847
2848 static mds_reinter reinters[REINT_MAX] = {
2849         [REINT_SETATTR] mds_reint_setattr,
2850         [REINT_CREATE] mds_reint_create,
2851         [REINT_LINK] mds_reint_link,
2852         [REINT_UNLINK] mds_reint_unlink,
2853         [REINT_RENAME] mds_reint_rename,
2854         [REINT_OPEN] mds_open
2855 };
2856
2857 int mds_reint_rec(struct mds_update_record *rec, int offset,
2858                   struct ptlrpc_request *req, struct lustre_handle *lockh)
2859 {
2860         struct obd_device *obd = req->rq_export->exp_obd;
2861         struct mds_obd *mds = &obd->u.mds;
2862         struct lvfs_run_ctxt saved;
2863         int rc;
2864 #ifdef CRAY_XT3
2865         gid_t fsgid = rec->ur_uc.luc_fsgid;
2866 #endif
2867         ENTRY;
2868
2869 #ifdef CRAY_XT3
2870         if (req->rq_uid != LNET_UID_ANY) {
2871                 /* non-root local cluster client
2872                  * NB root's creds are believed... */
2873                 LASSERT (req->rq_uid != 0);
2874                 rec->ur_uc.luc_fsuid = req->rq_uid;
2875                 cfs_kernel_cap_unpack(&rec->ur_uc.luc_cap, 0);
2876         }
2877 #endif
2878
2879         /* get group info of this user */
2880         rec->ur_uc.luc_uce = upcall_cache_get_entry(mds->mds_group_hash,
2881                                                     rec->ur_uc.luc_fsuid,
2882                                                     rec->ur_uc.luc_fsgid, 2,
2883                                                     &rec->ur_uc.luc_suppgid1);
2884
2885         if (IS_ERR(rec->ur_uc.luc_uce)) {
2886                 rc = PTR_ERR(rec->ur_uc.luc_uce);
2887                 rec->ur_uc.luc_uce = NULL;
2888                 RETURN(rc);
2889         }
2890
2891         /* checked by unpacker */
2892         LASSERT(rec->ur_opcode < REINT_MAX && reinters[rec->ur_opcode] != NULL);
2893
2894 #ifdef CRAY_XT3
2895         if (rec->ur_uc.luc_uce)
2896                 rec->ur_uc.luc_fsgid = rec->ur_uc.luc_uce->ue_primary;
2897 #endif
2898
2899         push_ctxt(&saved, &obd->obd_lvfs_ctxt, &rec->ur_uc);
2900
2901 #ifdef CRAY_XT3
2902         if (rec->ur_uc.luc_uce && fsgid != rec->ur_uc.luc_fsgid &&
2903             in_group_p(fsgid)) {
2904                 struct cred *cred;
2905                 rec->ur_uc.luc_fsgid = saved.luc.luc_fsgid = fsgid;
2906                 if ((cred = prepare_creds())) {
2907                         cred->fsgid = fsgid;
2908                         commit_creds(cred);
2909                 }
2910         }
2911 #endif
2912
2913         rc = reinters[rec->ur_opcode] (rec, offset, req, lockh);
2914         pop_ctxt(&saved, &obd->obd_lvfs_ctxt, &rec->ur_uc);
2915
2916         upcall_cache_put_entry(mds->mds_group_hash, rec->ur_uc.luc_uce);
2917         RETURN(rc);
2918 }