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  2008 Sun Microsystems, Inc. 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         ENTRY;
1730
1731         LASSERT(inode != NULL);
1732         LASSERT(!mds_inode_is_orphan(inode));
1733         LASSERT(TRYLOCK_INODE_MUTEX(pending_dir) == 0);
1734
1735         fidlen = ll_fid2str(fidname, inode->i_ino, inode->i_generation);
1736
1737         CDEBUG(D_INODE, "pending destroy of %dx open %d linked %s %s = %s\n",
1738                mds_orphan_open_count(inode), inode->i_nlink,
1739                S_ISDIR(inode->i_mode) ? "dir" :
1740                 S_ISREG(inode->i_mode) ? "file" : "other",rec->ur_name,fidname);
1741
1742         if (!mds_orphan_needed(obd, inode) || inode->i_nlink != 0)
1743                 RETURN(0);
1744
1745         pending_child = lookup_one_len(fidname, mds->mds_pending_dir, fidlen);
1746         if (IS_ERR(pending_child))
1747                 RETURN(PTR_ERR(pending_child));
1748
1749         if (pending_child->d_inode != NULL) {
1750                 CERROR("re-destroying orphan file %s?\n", rec->ur_name);
1751                 LASSERT(pending_child->d_inode == inode);
1752                 GOTO(out_dput, rc = 0);
1753         }
1754
1755         /* link() is semanticaly-wrong for S_IFDIR, so we set S_IFREG
1756          * for linking and return real mode back then -bzzz */
1757         mode = inode->i_mode;
1758         inode->i_mode = S_IFREG;
1759         /* avoid vfs_link upon 0 nlink inode, inc by 2 instead of 1 because
1760          * ext3_inc_count() can reset i_nlink for indexed directory */
1761         inode->i_nlink += 2;
1762         rc = ll_vfs_link(dentry, mds->mds_vfsmnt, pending_dir, pending_child,
1763                          mds->mds_vfsmnt);
1764         if (rc)
1765                 CERROR("error linking orphan %s %s to PENDING: rc = %d\n",
1766                        S_ISDIR(mode) ? "dir" : S_ISREG(mode) ? "file" : "other",
1767                        rec->ur_name, rc);
1768         else
1769                 mds_inode_set_orphan(inode);
1770
1771         /* return mode and correct i_nlink if inode is directory */
1772         inode->i_mode = mode;
1773         LASSERTF(rc || inode->i_nlink == 3, "%s nlink == %d\n",
1774                  S_ISDIR(mode) ? "dir" : S_ISREG(mode) ? "file" : "other",
1775                  inode->i_nlink);
1776
1777         if (S_ISDIR(mode)) {
1778                 pending_dir->i_nlink++;
1779                 if (pending_dir->i_sb->s_op->dirty_inode)
1780                         pending_dir->i_sb->s_op->dirty_inode(pending_dir);
1781         }
1782
1783         inode->i_nlink -= 2;
1784         if (inode->i_sb->s_op->dirty_inode)
1785                 inode->i_sb->s_op->dirty_inode(inode);
1786
1787         if (rc)
1788                 GOTO(out_dput, rc);
1789
1790         GOTO(out_dput, rc = 1);
1791 out_dput:
1792         l_dput(pending_child);
1793         RETURN(rc);
1794 }
1795
1796 int mds_get_cookie_size(struct obd_device *obd, struct lov_mds_md *lmm)
1797 {
1798         int count = le32_to_cpu(lmm->lmm_stripe_count);
1799         int real_csize = count * sizeof(struct llog_cookie);
1800         return real_csize;
1801 }
1802
1803 static void mds_shrink_reply(struct ptlrpc_request *req,
1804                            int reply_mdoff, int have_md, int have_acl)
1805 {
1806         struct obd_device *obd = req->rq_export->exp_obd;
1807         struct mds_body *reply_body;
1808         int cookie_size = 0, md_size = 0;
1809         ENTRY;
1810
1811         /* LSM and cookie is always placed after mds_body */
1812         reply_body =  lustre_msg_buf(req->rq_repmsg, reply_mdoff,
1813                                      sizeof(*reply_body));
1814         if (reply_body == NULL)
1815                 /* if there is no mds_body, no point in shrinking the reply */
1816                  return;
1817
1818         reply_mdoff++;
1819
1820         if (have_md || have_acl) {
1821                 if (reply_body->valid & (OBD_MD_FLEASIZE | OBD_MD_FLDIREA)) {
1822                         md_size = reply_body->eadatasize;
1823                 } else if (reply_body->valid & OBD_MD_LINKNAME)
1824                         md_size = reply_body->eadatasize;
1825
1826                 if (reply_body->valid & OBD_MD_FLCOOKIE) {
1827                         LASSERT(reply_body->valid & OBD_MD_FLEASIZE);
1828                         cookie_size = mds_get_cookie_size(obd, lustre_msg_buf(
1829                                                           req->rq_repmsg,
1830                                                           reply_mdoff, 0));
1831                 } else if (reply_body->valid & OBD_MD_FLACL) {
1832                         cookie_size = reply_body->aclsize;
1833                 }
1834         }
1835         CDEBUG(D_INFO, "Shrink %d/%d to md_size %d cookie_size %d \n",
1836                have_md, have_acl, md_size, cookie_size);
1837
1838         if (likely(have_md))
1839                 lustre_shrink_reply(req, reply_mdoff, md_size, 1);
1840
1841         if (likely(have_acl))
1842                 lustre_shrink_reply(req, reply_mdoff + (md_size > 0),
1843                                     cookie_size, 1);
1844 }
1845
1846 void mds_shrink_body_reply(struct ptlrpc_request *req,
1847                            int req_mdoff,
1848                            int reply_mdoff)
1849 {
1850         struct mds_body *rq_body;
1851         const long long have_acl = OBD_MD_FLCOOKIE | OBD_MD_FLACL;
1852         const long have_md = OBD_MD_FLEASIZE | OBD_MD_FLDIREA;
1853         ENTRY;
1854
1855         /* LSM and cookie is always placed after mds_body */
1856         rq_body =  lustre_msg_buf(req->rq_reqmsg, req_mdoff,
1857                                   sizeof(*rq_body));
1858         LASSERT(rq_body);
1859
1860         /* this check is need for avoid hit asset in case
1861          * OBD_MDS_FLFLAGS */
1862         mds_shrink_reply(req, reply_mdoff,
1863                          rq_body->valid & have_md,
1864                          rq_body->valid & have_acl);
1865 }
1866
1867 void mds_shrink_intent_reply(struct ptlrpc_request *req,
1868                              int opc, int reply_mdoff)
1869 {
1870         switch (opc) {
1871                 case REINT_UNLINK:
1872                 case REINT_RENAME:
1873                         mds_shrink_reply(req, reply_mdoff, 1, 1);
1874                         break;
1875                 case REINT_OPEN:
1876                         mds_shrink_reply(req, reply_mdoff, 1, 0);
1877                         break;
1878                 default:
1879                         break;
1880         }
1881 }
1882
1883 static int mds_reint_unlink(struct mds_update_record *rec, int offset,
1884                             struct ptlrpc_request *req,
1885                             struct lustre_handle *lh)
1886 {
1887         struct dentry *dparent = NULL, *dchild;
1888         struct mds_obd *mds = mds_req2mds(req);
1889         struct obd_device *obd = req->rq_export->exp_obd;
1890         struct mds_body *body = NULL;
1891         struct inode *inodes[PTLRPC_NUM_VERSIONS] = { NULL };
1892         struct inode *child_inode = NULL;
1893         struct lustre_handle parent_lockh, child_lockh, child_reuse_lockh;
1894         void *handle = NULL;
1895         int rc = 0, cleanup_phase = 0;
1896         unsigned int qcids[MAXQUOTAS] = { 0, 0 };
1897         unsigned int qpids[MAXQUOTAS] = { 0, 0 };
1898         ENTRY;
1899
1900         LASSERT(offset == REQ_REC_OFF); /*  || offset == DLM_INTENT_REC_OFF); */
1901         offset = REPLY_REC_OFF;
1902
1903         DEBUG_REQ(D_INODE, req, "parent ino "LPU64"/%u, child %s",
1904                   rec->ur_fid1->id, rec->ur_fid1->generation, rec->ur_name);
1905
1906         MDS_CHECK_RESENT(req, mds_reconstruct_generic(req));
1907
1908         if (OBD_FAIL_CHECK(OBD_FAIL_MDS_REINT_UNLINK))
1909                 GOTO(cleanup, rc = -ENOENT);
1910
1911         if (rec->ur_dlm)
1912                 ldlm_request_cancel(req, rec->ur_dlm, 0);
1913
1914         rc = mds_get_parent_child_locked(obd, mds, rec->ur_fid1,
1915                                          &parent_lockh, &dparent, LCK_EX,
1916                                          MDS_INODELOCK_UPDATE,
1917                                          rec->ur_name, rec->ur_namelen,
1918                                          &child_lockh, &dchild, LCK_EX,
1919                                          MDS_INODELOCK_FULL);
1920         if (rc)
1921                 GOTO(cleanup, rc);
1922
1923         cleanup_phase = 1; /* dchild, dparent, locks */
1924
1925         dget(dchild);
1926         child_inode = dchild->d_inode;
1927         if (child_inode == NULL) {
1928                 CDEBUG(D_INODE, "child doesn't exist (dir %lu, name %s)\n",
1929                        dparent->d_inode->i_ino, rec->ur_name);
1930                 GOTO(cleanup, rc = -ENOENT);
1931         }
1932
1933         /* save uid/gid for quota acquire/release */
1934         qcids[USRQUOTA] = child_inode->i_uid;
1935         qcids[GRPQUOTA] = child_inode->i_gid;
1936         qpids[USRQUOTA] = dparent->d_inode->i_uid;
1937         qpids[GRPQUOTA] = dparent->d_inode->i_gid;
1938
1939         cleanup_phase = 2; /* dchild has a lock */
1940
1941         /* VBR: version recovery check for parent */
1942         rc = mds_version_get_check(req, dparent->d_inode, 0);
1943         if (rc)
1944                 GOTO(cleanup_no_trans, rc);
1945
1946         /* version recovery check */
1947         rc = mds_version_get_check(req, child_inode, 1);
1948         if (rc)
1949                 GOTO(cleanup_no_trans, rc);
1950
1951         /* We have to do these checks ourselves, in case we are making an
1952          * orphan.  The client tells us whether rmdir() or unlink() was called,
1953          * so we need to return appropriate errors (bug 72). */
1954         if ((rec->ur_mode & S_IFMT) == S_IFDIR) {
1955                 if (!S_ISDIR(child_inode->i_mode))
1956                         GOTO(cleanup, rc = -ENOTDIR);
1957         } else {
1958                 if (S_ISDIR(child_inode->i_mode))
1959                         GOTO(cleanup, rc = -EISDIR);
1960         }
1961
1962         /* Check for EROFS after we check ENODENT, ENOTDIR, and EISDIR */
1963         if (req->rq_export->exp_connect_flags & OBD_CONNECT_RDONLY)
1964                 GOTO(cleanup, rc = -EROFS);
1965
1966         /* Step 3: Get a lock on the ino to sync with creation WRT inode
1967          * reuse (see bug 2029). */
1968         rc = mds_lock_new_child(obd, child_inode, &child_reuse_lockh);
1969         if (rc != ELDLM_OK)
1970                 GOTO(cleanup, rc);
1971
1972         cleanup_phase = 3; /* child inum lock */
1973
1974         OBD_FAIL_WRITE(obd, OBD_FAIL_MDS_REINT_UNLINK_WRITE, dparent->d_inode->i_sb);
1975
1976         /* ldlm_reply in buf[0] if called via intent */
1977         if (offset == DLM_INTENT_REC_OFF)
1978                 offset = DLM_REPLY_REC_OFF;
1979
1980         body = lustre_msg_buf(req->rq_repmsg, offset, sizeof(*body));
1981         LASSERT(body != NULL);
1982
1983         /* child orphan sem protects orphan_dec_test && is_orphan race */
1984         MDS_DOWN_READ_ORPHAN_SEM(child_inode);
1985         cleanup_phase = 4; /* MDS_UP_READ_ORPHAN_SEM(new_inode) when finished */
1986
1987         /* If this is potentially the last reference to this inode, get the
1988          * OBD EA data first so the client can destroy OST objects.  We
1989          * only do the object removal later if no open files/links remain. */
1990         if ((S_ISDIR(child_inode->i_mode) && child_inode->i_nlink == 2) ||
1991             child_inode->i_nlink == 1) {
1992                 if (mds_orphan_needed(obd, child_inode)) {
1993                         /* need to lock pending_dir before transaction */
1994                         LOCK_INODE_MUTEX(mds->mds_pending_dir->d_inode);
1995                         cleanup_phase = 5; /* UNLOCK_INODE_MUTEX(mds->mds_pending_dir->d_inode); */
1996                 } else if (S_ISREG(child_inode->i_mode)) {
1997                         mds_pack_inode2body(body, child_inode);
1998                         mds_pack_md(obd, req->rq_repmsg, offset + 1, body,
1999                                     child_inode, MDS_PACK_MD_LOCK, 0,
2000                                     req->rq_export->exp_connect_flags);
2001                 }
2002         }
2003
2004         /* Step 4: Do the unlink: we already verified ur_mode above (bug 72) */
2005         switch (child_inode->i_mode & S_IFMT) {
2006         case S_IFDIR:
2007                 /* Drop any lingering child directories before we start our
2008                  * transaction, to avoid doing multiple inode dirty/delete
2009                  * in our compound transaction (bug 1321). */
2010                 shrink_dcache_parent(dchild);
2011                 handle = fsfilt_start(obd, dparent->d_inode, FSFILT_OP_RMDIR,
2012                                       NULL);
2013                 if (IS_ERR(handle))
2014                         GOTO(cleanup, rc = PTR_ERR(handle));
2015                 LOCK_INODE_MUTEX(dparent->d_inode);
2016                 rc = ll_vfs_rmdir(dparent->d_inode, dchild, mds->mds_vfsmnt);
2017                 UNLOCK_INODE_MUTEX(dparent->d_inode);
2018                 mds_counter_incr(req->rq_export, LPROC_MDS_RMDIR);
2019                 break;
2020         case S_IFREG: {
2021                 struct lov_mds_md *lmm = lustre_msg_buf(req->rq_repmsg,
2022                                                         offset + 1, 0);
2023                 int sz = lustre_msg_buflen(req->rq_repmsg, offset + 1) > 0 ?
2024                          le32_to_cpu(lmm->lmm_stripe_count) : 0;
2025
2026                 handle = fsfilt_start_log(obd, dparent->d_inode,
2027                                           FSFILT_OP_UNLINK, NULL, sz);
2028                 if (IS_ERR(handle))
2029                         GOTO(cleanup, rc = PTR_ERR(handle));
2030                 LOCK_INODE_MUTEX(dparent->d_inode);
2031                 rc = ll_vfs_unlink(dparent->d_inode, dchild, mds->mds_vfsmnt);
2032                 UNLOCK_INODE_MUTEX(dparent->d_inode);
2033                 mds_counter_incr(req->rq_export, LPROC_MDS_UNLINK);
2034                 break;
2035         }
2036         case S_IFLNK:
2037         case S_IFCHR:
2038         case S_IFBLK:
2039         case S_IFIFO:
2040         case S_IFSOCK:
2041                 handle = fsfilt_start(obd, dparent->d_inode, FSFILT_OP_UNLINK,
2042                                       NULL);
2043                 if (IS_ERR(handle))
2044                         GOTO(cleanup, rc = PTR_ERR(handle));
2045                 LOCK_INODE_MUTEX(dparent->d_inode);
2046                 rc = ll_vfs_unlink(dparent->d_inode, dchild, mds->mds_vfsmnt);
2047                 UNLOCK_INODE_MUTEX(dparent->d_inode);
2048                 mds_counter_incr(req->rq_export, LPROC_MDS_UNLINK);
2049                 break;
2050         default:
2051                 CERROR("bad file type %o unlinking %s\n", rec->ur_mode,
2052                        rec->ur_name);
2053                 LBUG();
2054                 GOTO(cleanup, rc = -EINVAL);
2055         }
2056
2057         if (rc == 0 && child_inode->i_nlink == 0) {
2058                 if (mds_orphan_needed(obd, child_inode)) {
2059                         rc = mds_orphan_add_link(rec, obd, dchild);
2060                         if (rc == 1)
2061                                 /* child inode was successfully linked
2062                                  * to PENDING */
2063                                 GOTO(cleanup, rc = 0);
2064                         else
2065                                 /* we failed to move the file to PENDING,
2066                                  * really unlink the file as if there were
2067                                  * no more openers */
2068                                 rc = 0;
2069                 }
2070
2071                 if (!S_ISREG(child_inode->i_mode))
2072                         GOTO(cleanup, rc);
2073
2074                 if (!(body->valid & OBD_MD_FLEASIZE)) {
2075                         body->valid |=(OBD_MD_FLSIZE | OBD_MD_FLBLOCKS |
2076                                        OBD_MD_FLATIME | OBD_MD_FLMTIME);
2077                 } else if (mds_log_op_unlink(obd,
2078                                 lustre_msg_buf(req->rq_repmsg, offset + 1, 0),
2079                                 lustre_msg_buflen(req->rq_repmsg, offset + 1),
2080                                 lustre_msg_buf(req->rq_repmsg, offset + 2, 0),
2081                                 lustre_msg_buflen(req->rq_repmsg, offset+2)) >
2082                            0) {
2083                         body->valid |= OBD_MD_FLCOOKIE;
2084                 }
2085         }
2086
2087         GOTO(cleanup, rc);
2088 cleanup:
2089         if (rc == 0) {
2090                 struct iattr iattr;
2091                 int err;
2092
2093                 if (child_inode->i_nlink > 0 ||
2094                     mds_orphan_open_count(child_inode) > 0) {
2095                         /* update ctime of unlinked file only if it is still
2096                          * opened or a link still exists */
2097                         iattr.ia_valid = ATTR_CTIME;
2098                         LTIME_S(iattr.ia_ctime) = rec->ur_time;
2099                         err = fsfilt_setattr(obd, dchild, handle, &iattr, 0);
2100                         if (err)
2101                                 CERROR("error on unlinked inode time update: "
2102                                        "rc = %d\n", err);
2103                 }
2104
2105                 /* update mtime and ctime of parent directory*/
2106                 iattr.ia_valid = ATTR_MTIME | ATTR_CTIME;
2107                 LTIME_S(iattr.ia_mtime) = rec->ur_time;
2108                 LTIME_S(iattr.ia_ctime) = rec->ur_time;
2109                 err = fsfilt_setattr(obd, dparent, handle, &iattr, 0);
2110                 if (err)
2111                         CERROR("error on parent setattr: rc = %d\n", err);
2112         }
2113         inodes[0] = dparent ? dparent->d_inode : NULL;
2114         inodes[1] = child_inode;
2115         rc = mds_finish_transno(mds, inodes, handle, req, rc, 0, 0);
2116
2117 cleanup_no_trans:
2118         switch(cleanup_phase) {
2119         case 5: /* pending_dir semaphore */
2120                 UNLOCK_INODE_MUTEX(mds->mds_pending_dir->d_inode);
2121         case 4: /* child inode semaphore */
2122                 MDS_UP_READ_ORPHAN_SEM(child_inode);
2123         case 3: /* child ino-reuse lock */
2124                 if (rc && body != NULL) {
2125                         // Don't unlink the OST objects if the MDS unlink failed
2126                         body->valid = 0;
2127                 }
2128                 if (rc)
2129                         ldlm_lock_decref(&child_reuse_lockh, LCK_EX);
2130                 else
2131                         ptlrpc_save_lock(req, &child_reuse_lockh, LCK_EX);
2132         case 2: /* child lock */
2133                 ldlm_lock_decref(&child_lockh, LCK_EX);
2134         case 1: /* child and parent dentry, parent lock */
2135                 if (rc)
2136                         ldlm_lock_decref(&parent_lockh, LCK_EX);
2137                 else
2138                         ptlrpc_save_lock(req, &parent_lockh, LCK_EX);
2139                 l_dput(dchild);
2140                 l_dput(dchild);
2141                 l_dput(dparent);
2142         case 0:
2143                 break;
2144         default:
2145                 CERROR("invalid cleanup_phase %d\n", cleanup_phase);
2146                 LBUG();
2147         }
2148         req->rq_status = rc;
2149
2150         /* trigger dqrel on the owner of child and parent */
2151         lquota_adjust(mds_quota_interface_ref, obd, qcids, qpids, rc,
2152                       FSFILT_OP_UNLINK);
2153         return 0;
2154 }
2155
2156 static int mds_reint_link(struct mds_update_record *rec, int offset,
2157                           struct ptlrpc_request *req,
2158                           struct lustre_handle *lh)
2159 {
2160         struct obd_device *obd = req->rq_export->exp_obd;
2161         struct dentry *de_src = NULL;
2162         struct dentry *de_tgt_dir = NULL;
2163         struct dentry *dchild = NULL;
2164         struct inode *inodes[PTLRPC_NUM_VERSIONS] = { NULL };
2165         struct mds_obd *mds = mds_req2mds(req);
2166         struct lustre_handle *handle = NULL, tgt_dir_lockh, src_lockh;
2167         struct ldlm_res_id src_res_id = { .name = {0} };
2168         struct ldlm_res_id tgt_dir_res_id = { .name = {0} };
2169         ldlm_policy_data_t src_policy ={.l_inodebits = {MDS_INODELOCK_UPDATE}};
2170         ldlm_policy_data_t tgt_dir_policy =
2171                                        {.l_inodebits = {MDS_INODELOCK_UPDATE}};
2172         int rc = 0, cleanup_phase = 0;
2173         ENTRY;
2174
2175         LASSERT(offset == REQ_REC_OFF);
2176
2177         DEBUG_REQ(D_INODE, req, "original "LPU64"/%u to "LPU64"/%u %s",
2178                   rec->ur_fid1->id, rec->ur_fid1->generation,
2179                   rec->ur_fid2->id, rec->ur_fid2->generation, rec->ur_name);
2180
2181         mds_counter_incr(req->rq_export, LPROC_MDS_LINK);
2182
2183         MDS_CHECK_RESENT(req, mds_reconstruct_generic(req));
2184
2185         if (OBD_FAIL_CHECK(OBD_FAIL_MDS_REINT_LINK))
2186                 GOTO(cleanup, rc = -ENOENT);
2187
2188         if (rec->ur_dlm)
2189                 ldlm_request_cancel(req, rec->ur_dlm, 0);
2190
2191         /* Step 1: Lookup the source inode and target directory by FID */
2192         de_src = mds_fid2dentry(mds, rec->ur_fid1, NULL);
2193         if (IS_ERR(de_src))
2194                 GOTO(cleanup, rc = PTR_ERR(de_src));
2195
2196         cleanup_phase = 1; /* source dentry */
2197
2198         de_tgt_dir = mds_fid2dentry(mds, rec->ur_fid2, NULL);
2199         if (IS_ERR(de_tgt_dir)) {
2200                 rc = PTR_ERR(de_tgt_dir);
2201                 de_tgt_dir = NULL;
2202                 GOTO(cleanup, rc);
2203         }
2204
2205         cleanup_phase = 2; /* target directory dentry */
2206
2207         CDEBUG(D_INODE, "linking %.*s/%s to inode %lu\n",
2208                de_tgt_dir->d_name.len, de_tgt_dir->d_name.name, rec->ur_name,
2209                de_src->d_inode->i_ino);
2210
2211         /* Step 2: Take the two locks */
2212         src_res_id.name[0] = de_src->d_inode->i_ino;
2213         src_res_id.name[1] = de_src->d_inode->i_generation;
2214         tgt_dir_res_id.name[0] = de_tgt_dir->d_inode->i_ino;
2215         tgt_dir_res_id.name[1] = de_tgt_dir->d_inode->i_generation;
2216
2217         rc = enqueue_ordered_locks(obd, &src_res_id, &src_lockh, LCK_EX,
2218                                    &src_policy,
2219                                    &tgt_dir_res_id, &tgt_dir_lockh, LCK_EX,
2220                                    &tgt_dir_policy);
2221         if (rc)
2222                 GOTO(cleanup, rc);
2223
2224         cleanup_phase = 3; /* locks */
2225
2226         /* version recovery check */
2227         /* directory check */
2228         rc = mds_version_get_check(req, de_tgt_dir->d_inode, 0);
2229         if (rc)
2230                 GOTO(cleanup_no_trans, rc);
2231         /* inode version check */
2232         rc = mds_version_get_check(req, de_src->d_inode, 1);
2233         if (rc)
2234                 GOTO(cleanup_no_trans, rc);
2235
2236         if (mds_inode_is_orphan(de_src->d_inode)) {
2237                 CDEBUG(D_INODE, "an attempt to link an orphan inode %lu/%u\n",
2238                        de_src->d_inode->i_ino,
2239                        de_src->d_inode->i_generation);
2240                 GOTO(cleanup, rc = -ENOENT);
2241         }
2242
2243         /* Step 3: Lookup the child */
2244         dchild = mds_lookup(obd, rec->ur_name, de_tgt_dir, rec->ur_namelen-1);
2245         if (IS_ERR(dchild)) {
2246                 rc = PTR_ERR(dchild);
2247                 dchild = NULL;
2248                 if (rc != -EPERM && rc != -EACCES && rc != -ENAMETOOLONG)
2249                         CERROR("child lookup error %d\n", rc);
2250                 GOTO(cleanup, rc);
2251         }
2252
2253         cleanup_phase = 4; /* child dentry */
2254
2255         if (dchild->d_inode) {
2256                 CDEBUG(D_INODE, "child exists (dir %lu, name %s)\n",
2257                        de_tgt_dir->d_inode->i_ino, rec->ur_name);
2258                 rc = -EEXIST;
2259                 GOTO(cleanup, rc);
2260         }
2261
2262         /* Step 4: Do it. */
2263         OBD_FAIL_WRITE(obd, OBD_FAIL_MDS_REINT_LINK_WRITE, de_src->d_inode->i_sb);
2264
2265         if (req->rq_export->exp_connect_flags & OBD_CONNECT_RDONLY)
2266                 GOTO(cleanup, rc = -EROFS);
2267
2268         handle = fsfilt_start(obd, de_tgt_dir->d_inode, FSFILT_OP_LINK, NULL);
2269         if (IS_ERR(handle))
2270                 GOTO(cleanup, rc = PTR_ERR(handle));
2271
2272         LOCK_INODE_MUTEX(de_tgt_dir->d_inode);
2273         rc = ll_vfs_link(de_src, mds->mds_vfsmnt, de_tgt_dir->d_inode, dchild,
2274                          mds->mds_vfsmnt);
2275         UNLOCK_INODE_MUTEX(de_tgt_dir->d_inode);
2276         if (rc && rc != -EPERM && rc != -EACCES)
2277                 CERROR("vfs_link error %d\n", rc);
2278         if (rc == 0) {
2279                 struct iattr iattr;
2280                 int err;
2281
2282                 /* update ctime of old file */
2283                 iattr.ia_valid = ATTR_CTIME;
2284                 LTIME_S(iattr.ia_ctime) = rec->ur_time;
2285                 err = fsfilt_setattr(obd, de_src, handle, &iattr, 0);
2286                 if (err)
2287                         CERROR("error on old inode time update: "
2288                                "rc = %d\n", err);
2289
2290                 /* update mtime and ctime of target directory */
2291                 iattr.ia_valid = ATTR_MTIME | ATTR_CTIME;
2292                 LTIME_S(iattr.ia_mtime) = rec->ur_time;
2293                 LTIME_S(iattr.ia_ctime) = rec->ur_time;
2294                 err = fsfilt_setattr(obd, de_tgt_dir, handle, &iattr, 0);
2295                 if (err)
2296                         CERROR("error on target dir inode time update: "
2297                                "rc = %d\n", err);
2298         }
2299 cleanup:
2300         inodes[0] = de_tgt_dir ? de_tgt_dir->d_inode : NULL;
2301         inodes[1] = (dchild && !IS_ERR(dchild)) ? dchild->d_inode : NULL;
2302         rc = mds_finish_transno(mds, inodes, handle, req, rc, 0, 0);
2303         EXIT;
2304
2305 cleanup_no_trans:
2306         switch (cleanup_phase) {
2307         case 4: /* child dentry */
2308                 l_dput(dchild);
2309         case 3: /* locks */
2310                 if (rc) {
2311                         ldlm_lock_decref(&src_lockh, LCK_EX);
2312                         ldlm_lock_decref(&tgt_dir_lockh, LCK_EX);
2313                 } else {
2314                         ptlrpc_save_lock(req, &src_lockh, LCK_EX);
2315                         ptlrpc_save_lock(req, &tgt_dir_lockh, LCK_EX);
2316                 }
2317         case 2: /* target dentry */
2318                 l_dput(de_tgt_dir);
2319         case 1: /* source dentry */
2320                 l_dput(de_src);
2321         case 0:
2322                 break;
2323         default:
2324                 CERROR("invalid cleanup_phase %d\n", cleanup_phase);
2325                 LBUG();
2326         }
2327         req->rq_status = rc;
2328         return 0;
2329 }
2330
2331 /* The idea here is that we need to get four locks in the end:
2332  * one on each parent directory, one on each child.  We need to take
2333  * these locks in some kind of order (to avoid deadlocks), and the order
2334  * I selected is "increasing resource number" order.  We need to look up
2335  * the children, however, before we know what the resource number(s) are.
2336  * Thus the following plan:
2337  *
2338  * 1,2. Look up the parents
2339  * 3,4. Look up the children
2340  * 5. Take locks on the parents and children, in order
2341  * 6. Verify that the children haven't changed since they were looked up
2342  *
2343  * If there was a race and the children changed since they were first looked
2344  * up, it is possible that mds_verify_child() will be able to just grab the
2345  * lock on the new child resource (if it has a higher resource than any other)
2346  * but we need to compare against not only its parent, but also against the
2347  * parent and child of the "other half" of the rename, hence maxres_{src,tgt}.
2348  *
2349  * We need the fancy igrab() on the child inodes because we aren't holding a
2350  * lock on the parent after the lookup is done, so dentry->d_inode may change
2351  * at any time, and igrab() itself doesn't like getting passed a NULL argument.
2352  */
2353 int mds_get_parents_children_locked(struct obd_device *obd,
2354                                     struct mds_obd *mds,
2355                                     struct ll_fid *p1_fid,
2356                                     struct dentry **de_srcdirp,
2357                                     struct ll_fid *p2_fid,
2358                                     struct dentry **de_tgtdirp,
2359                                     int parent_mode,
2360                                     const char *old_name, int old_len,
2361                                     struct dentry **de_oldp,
2362                                     const char *new_name, int new_len,
2363                                     struct dentry **de_newp,
2364                                     struct lustre_handle *dlm_handles,
2365                                     int child_mode)
2366 {
2367         struct ldlm_res_id p1_res_id = { .name = {0} };
2368         struct ldlm_res_id p2_res_id = { .name = {0} };
2369         struct ldlm_res_id c1_res_id = { .name = {0} };
2370         struct ldlm_res_id c2_res_id = { .name = {0} };
2371         ldlm_policy_data_t p_policy = {.l_inodebits = {MDS_INODELOCK_UPDATE}};
2372         /* Only dentry should disappear, but the inode itself would be
2373            intact otherwise. */
2374         ldlm_policy_data_t c1_policy = {.l_inodebits = {MDS_INODELOCK_LOOKUP}};
2375         /* If something is going to be replaced, both dentry and inode locks are
2376          * needed */
2377         ldlm_policy_data_t c2_policy = {.l_inodebits = {MDS_INODELOCK_FULL}};
2378         struct ldlm_res_id *maxres_src, *maxres_tgt;
2379         struct inode *inode;
2380         int rc = 0, cleanup_phase = 0;
2381         ENTRY;
2382
2383         /* Step 1: Lookup the source directory */
2384         *de_srcdirp = mds_fid2dentry(mds, p1_fid, NULL);
2385         if (IS_ERR(*de_srcdirp))
2386                 GOTO(cleanup, rc = PTR_ERR(*de_srcdirp));
2387
2388         cleanup_phase = 1; /* source directory dentry */
2389
2390         p1_res_id.name[0] = (*de_srcdirp)->d_inode->i_ino;
2391         p1_res_id.name[1] = (*de_srcdirp)->d_inode->i_generation;
2392
2393         /* Step 2: Lookup the target directory */
2394         if (memcmp(p1_fid, p2_fid, sizeof(*p1_fid)) == 0) {
2395                 *de_tgtdirp = dget(*de_srcdirp);
2396         } else {
2397                 *de_tgtdirp = mds_fid2dentry(mds, p2_fid, NULL);
2398                 if (IS_ERR(*de_tgtdirp)) {
2399                         rc = PTR_ERR(*de_tgtdirp);
2400                         *de_tgtdirp = NULL;
2401                         GOTO(cleanup, rc);
2402                 }
2403         }
2404
2405         cleanup_phase = 2; /* target directory dentry */
2406
2407         p2_res_id.name[0] = (*de_tgtdirp)->d_inode->i_ino;
2408         p2_res_id.name[1] = (*de_tgtdirp)->d_inode->i_generation;
2409
2410         /* Step 3: Lookup the source child entry */
2411         *de_oldp = mds_lookup(obd, old_name, *de_srcdirp, old_len - 1);
2412         if (IS_ERR(*de_oldp)) {
2413                 rc = PTR_ERR(*de_oldp);
2414                 CDEBUG(D_INODE, "old child lookup error (%.*s): rc %d\n",
2415                        old_len - 1, old_name, rc);
2416                 GOTO(cleanup, rc);
2417         }
2418
2419         cleanup_phase = 3; /* original name dentry */
2420
2421         inode = (*de_oldp)->d_inode;
2422         if (inode != NULL)
2423                 inode = igrab(inode);
2424         if (inode == NULL)
2425                 GOTO(cleanup, rc = -ENOENT);
2426
2427         c1_res_id.name[0] = inode->i_ino;
2428         c1_res_id.name[1] = inode->i_generation;
2429
2430         iput(inode);
2431
2432         /* Step 4: Lookup the target child entry */
2433         if (!new_name)
2434                 GOTO(retry_locks, rc);
2435         *de_newp = mds_lookup(obd, new_name, *de_tgtdirp, new_len - 1);
2436         if (IS_ERR(*de_newp)) {
2437                 rc = PTR_ERR(*de_newp);
2438                 CDEBUG(D_DENTRY, "new child lookup error (%.*s): rc %d\n",
2439                        old_len - 1, old_name, rc);
2440                 GOTO(cleanup, rc);
2441         }
2442
2443         cleanup_phase = 4; /* target dentry */
2444
2445         inode = (*de_newp)->d_inode;
2446         if (inode != NULL) {
2447                 if (is_bad_inode(inode)) {
2448                         CERROR("bad inode returned %lu/%u\n",
2449                                inode->i_ino, inode->i_generation);
2450                         GOTO(cleanup, rc = -ENOENT);
2451                 }
2452                 inode = igrab(inode);
2453         }
2454         if (inode == NULL)
2455                 goto retry_locks;
2456
2457         c2_res_id.name[0] = inode->i_ino;
2458         c2_res_id.name[1] = inode->i_generation;
2459         iput(inode);
2460
2461 retry_locks:
2462         /* Step 5: Take locks on the parents and child(ren) */
2463         maxres_src = &p1_res_id;
2464         maxres_tgt = &p2_res_id;
2465         cleanup_phase = 4; /* target dentry */
2466
2467         if (c2_res_id.name[0] != 0 && res_gt(&c2_res_id, &p2_res_id,NULL,NULL))
2468                 maxres_tgt = &c2_res_id;
2469
2470         rc = enqueue_4ordered_locks(obd, &p1_res_id,&dlm_handles[0],parent_mode,
2471                                     &p_policy,
2472                                     &p2_res_id, &dlm_handles[1], parent_mode,
2473                                     &p_policy,
2474                                     &c1_res_id, &dlm_handles[2], child_mode,
2475                                     &c1_policy,
2476                                     &c2_res_id, &dlm_handles[3], child_mode,
2477                                     &c2_policy);
2478         if (rc)
2479                 GOTO(cleanup, rc);
2480
2481         cleanup_phase = 6; /* parent and child(ren) locks */
2482
2483         /* Step 6a: Re-lookup source child to verify it hasn't changed */
2484         rc = mds_verify_child(obd, &p1_res_id, &dlm_handles[0], *de_srcdirp,
2485                               parent_mode, &c1_res_id, &dlm_handles[2], de_oldp,
2486                               child_mode, &c1_policy, old_name, old_len,
2487                               maxres_tgt);
2488         if (rc) {
2489                 if (c2_res_id.name[0] != 0)
2490                         ldlm_lock_decref(&dlm_handles[3], child_mode);
2491                 ldlm_lock_decref(&dlm_handles[1], parent_mode);
2492                 cleanup_phase = 4;
2493                 if (rc > 0)
2494                         goto retry_locks;
2495                 GOTO(cleanup, rc);
2496         }
2497
2498         if ((*de_oldp)->d_inode == NULL)
2499                 GOTO(cleanup, rc = -ENOENT);
2500
2501         if (!new_name)
2502                 GOTO(cleanup, rc);
2503
2504         /* Safe to skip check for child res being all zero */
2505         if (res_gt(&c1_res_id, maxres_src, NULL, NULL))
2506                 maxres_src = &c1_res_id;
2507
2508         /* Step 6b: Re-lookup target child to verify it hasn't changed */
2509         rc = mds_verify_child(obd, &p2_res_id, &dlm_handles[1], *de_tgtdirp,
2510                               parent_mode, &c2_res_id, &dlm_handles[3], de_newp,
2511                               child_mode, &c2_policy, new_name, new_len,
2512                               maxres_src);
2513         if (rc) {
2514                 ldlm_lock_decref(&dlm_handles[2], child_mode);
2515                 ldlm_lock_decref(&dlm_handles[0], parent_mode);
2516                 cleanup_phase = 4;
2517                 if (rc > 0)
2518                         goto retry_locks;
2519                 GOTO(cleanup, rc);
2520         }
2521
2522         EXIT;
2523 cleanup:
2524         if (rc) {
2525                 switch (cleanup_phase) {
2526                 case 6: /* child lock(s) */
2527                         if (c2_res_id.name[0] != 0)
2528                                 ldlm_lock_decref(&dlm_handles[3], child_mode);
2529                         if (c1_res_id.name[0] != 0)
2530                                 ldlm_lock_decref(&dlm_handles[2], child_mode);
2531                 case 5: /* parent locks */
2532                         ldlm_lock_decref(&dlm_handles[1], parent_mode);
2533                         ldlm_lock_decref(&dlm_handles[0], parent_mode);
2534                 case 4: /* target dentry */
2535                         l_dput(*de_newp);
2536                 case 3: /* source dentry */
2537                         l_dput(*de_oldp);
2538                 case 2: /* target directory dentry */
2539                         l_dput(*de_tgtdirp);
2540                 case 1: /* source directry dentry */
2541                         l_dput(*de_srcdirp);
2542                 }
2543         }
2544
2545         return rc;
2546 }
2547
2548 static int mds_reint_rename(struct mds_update_record *rec, int offset,
2549                             struct ptlrpc_request *req,
2550                             struct lustre_handle *lockh)
2551 {
2552         struct obd_device *obd = req->rq_export->exp_obd;
2553         struct dentry *de_srcdir = NULL;
2554         struct dentry *de_tgtdir = NULL;
2555         struct dentry *de_old = NULL;
2556         struct dentry *de_new = NULL;
2557         struct dentry *trap;
2558         struct inode *old_inode = NULL, *new_inode = NULL;
2559         struct inode *inodes[PTLRPC_NUM_VERSIONS] = { NULL };
2560         struct mds_obd *mds = mds_req2mds(req);
2561         struct lustre_handle dlm_handles[4];
2562         struct mds_body *body = NULL;
2563         struct lov_mds_md *lmm = NULL;
2564         int rc = 0, lock_count = 3, cleanup_phase = 0, sz;
2565         void *handle = NULL;
2566         unsigned int qcids[MAXQUOTAS] = { 0, 0 };
2567         unsigned int qpids[4] = { 0, 0, 0, 0 };
2568         ENTRY;
2569
2570         LASSERT(offset == REQ_REC_OFF);
2571         offset = REPLY_REC_OFF;
2572
2573         DEBUG_REQ(D_INODE, req, "parent "LPU64"/%u %s to "LPU64"/%u %s",
2574                   rec->ur_fid1->id, rec->ur_fid1->generation, rec->ur_name,
2575                   rec->ur_fid2->id, rec->ur_fid2->generation, rec->ur_tgt);
2576
2577         mds_counter_incr(req->rq_export, LPROC_MDS_RENAME);
2578
2579         MDS_CHECK_RESENT(req, mds_reconstruct_generic(req));
2580
2581         if (rec->ur_dlm)
2582                 ldlm_request_cancel(req, rec->ur_dlm, 0);
2583
2584         rc = mds_get_parents_children_locked(obd, mds, rec->ur_fid1, &de_srcdir,
2585                                              rec->ur_fid2, &de_tgtdir, LCK_EX,
2586                                              rec->ur_name, rec->ur_namelen,
2587                                              &de_old, rec->ur_tgt,
2588                                              rec->ur_tgtlen, &de_new,
2589                                              dlm_handles, LCK_EX);
2590         if (rc)
2591                 GOTO(cleanup, rc);
2592
2593         cleanup_phase = 1; /* parent(s), children, locks */
2594
2595         old_inode = de_old->d_inode;
2596         new_inode = de_new->d_inode;
2597
2598         if (new_inode != NULL)
2599                 lock_count = 4;
2600
2601         /* version recovery check */
2602         rc = mds_version_get_check(req, de_srcdir->d_inode, 0);
2603         if (rc)
2604                 GOTO(cleanup_no_trans, rc);
2605         rc = mds_version_get_check(req, old_inode, 1);
2606         if (rc)
2607                 GOTO(cleanup_no_trans, rc);
2608         rc = mds_version_get_check(req, de_tgtdir->d_inode, 2);
2609         if (rc)
2610                 GOTO(cleanup_no_trans, rc);
2611         rc = mds_version_get_check(req, new_inode, 3);
2612         if (rc)
2613                 GOTO(cleanup_no_trans, rc);
2614
2615         /* sanity check for src inode */
2616         if (old_inode->i_ino == de_srcdir->d_inode->i_ino ||
2617             old_inode->i_ino == de_tgtdir->d_inode->i_ino)
2618                 GOTO(cleanup, rc = -EINVAL);
2619
2620         if (req->rq_export->exp_connect_flags & OBD_CONNECT_RDONLY)
2621                 GOTO(cleanup, rc = -EROFS);
2622
2623         if (new_inode == NULL)
2624                 goto no_unlink;
2625
2626         igrab(new_inode);
2627         cleanup_phase = 2; /* iput(new_inode) when finished */
2628
2629         /* sanity check for dest inode */
2630         if (new_inode->i_ino == de_srcdir->d_inode->i_ino ||
2631             new_inode->i_ino == de_tgtdir->d_inode->i_ino)
2632                 GOTO(cleanup, rc = -EINVAL);
2633
2634         if (old_inode == new_inode)
2635                 GOTO(cleanup, rc = 0);
2636
2637         /* save uids/gids for qunit acquire/release */
2638         qcids[USRQUOTA] = old_inode->i_uid;
2639         qcids[GRPQUOTA] = old_inode->i_gid;
2640         qpids[USRQUOTA] = de_tgtdir->d_inode->i_uid;
2641         qpids[GRPQUOTA] = de_tgtdir->d_inode->i_gid;
2642         qpids[2] = de_srcdir->d_inode->i_uid;
2643         qpids[3] = de_srcdir->d_inode->i_gid;
2644
2645         /* if we are about to remove the target at first, pass the EA of
2646          * that inode to client to perform and cleanup on OST */
2647         body = lustre_msg_buf(req->rq_repmsg, offset, sizeof(*body));
2648         LASSERT(body != NULL);
2649
2650         /* child orphan sem protects orphan_dec_test && is_orphan race */
2651         MDS_DOWN_READ_ORPHAN_SEM(new_inode);
2652         cleanup_phase = 3; /* MDS_UP_READ_ORPHAN_SEM(new_inode) when finished */
2653
2654         if ((S_ISDIR(new_inode->i_mode) && new_inode->i_nlink == 2) ||
2655             new_inode->i_nlink == 1) {
2656                 if (mds_orphan_needed(obd, new_inode)) {
2657                         /* need to lock pending_dir before transaction */
2658                         LOCK_INODE_MUTEX(mds->mds_pending_dir->d_inode);
2659                         cleanup_phase = 4; /* UNLOCK_INODE_MUTEX(mds->mds_pending_dir->d_inode); */
2660                 } else if (S_ISREG(new_inode->i_mode)) {
2661                         mds_pack_inode2body(body, new_inode);
2662                         mds_pack_md(obd, req->rq_repmsg, offset + 1, body,
2663                                     new_inode, MDS_PACK_MD_LOCK, 0,
2664                                     req->rq_export->exp_connect_flags);
2665                 }
2666         }
2667
2668 no_unlink:
2669         OBD_FAIL_WRITE(obd, OBD_FAIL_MDS_REINT_RENAME_WRITE,
2670                        de_srcdir->d_inode->i_sb);
2671
2672         lmm = lustre_msg_buf(req->rq_repmsg, offset + 1, 0);
2673         /* check that lmm size is not 0 */
2674         sz = lustre_msg_buflen(req->rq_repmsg, offset + 1) > 0 ?
2675              le32_to_cpu(lmm->lmm_stripe_count) : 0;
2676
2677         handle = fsfilt_start_log(obd, de_tgtdir->d_inode, FSFILT_OP_RENAME,
2678                                   NULL, sz);
2679         if (IS_ERR(handle))
2680                 GOTO(cleanup, rc = PTR_ERR(handle));
2681
2682         trap = lock_rename(de_tgtdir, de_srcdir);
2683         /* source should not be ancestor of target */
2684         if (de_old == trap) {
2685                 unlock_rename(de_tgtdir, de_srcdir);
2686                 GOTO(cleanup, rc = -EINVAL);
2687         }
2688         /* target should not be an ancestor of source */
2689         if (de_new == trap) {
2690                 unlock_rename(de_tgtdir, de_srcdir);
2691                 GOTO(cleanup, rc = -ENOTEMPTY);
2692         }
2693
2694         de_old->d_fsdata = req;
2695         de_new->d_fsdata = req;
2696
2697         rc = ll_vfs_rename(de_srcdir->d_inode, de_old, mds->mds_vfsmnt,
2698                            de_tgtdir->d_inode, de_new, mds->mds_vfsmnt);
2699
2700         unlock_rename(de_tgtdir, de_srcdir);
2701
2702         if (rc == 0) {
2703                 struct iattr iattr;
2704                 int err;
2705
2706                 /* update ctime of renamed file */
2707                 iattr.ia_valid = ATTR_CTIME;
2708                 LTIME_S(iattr.ia_ctime) = rec->ur_time;
2709                 if (S_ISDIR(de_old->d_inode->i_mode) &&
2710                     de_srcdir->d_inode != de_tgtdir->d_inode) {
2711                         /* cross directory rename of a directory, ".."
2712                            changed, update mtime also */
2713                         iattr.ia_valid |= ATTR_MTIME;
2714                         LTIME_S(iattr.ia_mtime) = rec->ur_time;
2715                 }
2716                 err = fsfilt_setattr(obd, de_old, handle, &iattr, 0);
2717                 if (err)
2718                         CERROR("error on old inode time update: "
2719                                "rc = %d\n", err);
2720
2721                 if (de_new->d_inode) {
2722                         /* target file exists, update its ctime as it
2723                            gets unlinked */
2724                         iattr.ia_valid = ATTR_CTIME;
2725                         LTIME_S(iattr.ia_ctime) = rec->ur_time;
2726                         err = fsfilt_setattr(obd, de_new, handle, &iattr, 0);
2727                         if (err)
2728                                 CERROR("error on target inode time update: "
2729                                        "rc = %d\n", err);
2730                 }
2731
2732                 /* update mtime and ctime of old directory */
2733                 iattr.ia_valid = ATTR_MTIME | ATTR_CTIME;
2734                 LTIME_S(iattr.ia_mtime) = rec->ur_time;
2735                 LTIME_S(iattr.ia_ctime) = rec->ur_time;
2736                 err = fsfilt_setattr(obd, de_srcdir, handle, &iattr, 0);
2737                 if (err)
2738                         CERROR("error on old dir inode update: "
2739                                "rc = %d\n", err);
2740
2741                 if (de_srcdir->d_inode != de_tgtdir->d_inode) {
2742                         /* cross directory rename, update
2743                            mtime and ctime of new directory */
2744                         iattr.ia_valid = ATTR_MTIME | ATTR_CTIME;
2745                         LTIME_S(iattr.ia_mtime) = rec->ur_time;
2746                         LTIME_S(iattr.ia_ctime) = rec->ur_time;
2747                         err = fsfilt_setattr(obd, de_tgtdir, handle, &iattr, 0);
2748                         if (err)
2749                                 CERROR("error on new dir inode time update: "
2750                                        "rc = %d\n", err);
2751                 }
2752         }
2753
2754         if (rc == 0 && new_inode != NULL && new_inode->i_nlink == 0) {
2755                 if (mds_orphan_needed(obd, new_inode)) {
2756                         rc = mds_orphan_add_link(rec, obd, de_new);
2757
2758                         if (rc == 1)
2759                                 /* inode successfully linked to PENDING */
2760                                 GOTO(cleanup, rc = 0);
2761                         else
2762                                 /* we failed to move the file to PENDING,
2763                                  * really unlink the file as if there were
2764                                  * no more openers */
2765                                 rc = 0;
2766                 }
2767
2768                 if (!S_ISREG(new_inode->i_mode))
2769                         GOTO(cleanup, rc);
2770
2771                 if (!(body->valid & OBD_MD_FLEASIZE)) {
2772                         body->valid |= (OBD_MD_FLSIZE | OBD_MD_FLBLOCKS |
2773                                         OBD_MD_FLATIME | OBD_MD_FLMTIME);
2774                 } else if (mds_log_op_unlink(obd,
2775                                              lustre_msg_buf(req->rq_repmsg,
2776                                                             offset + 1, 0),
2777                                              lustre_msg_buflen(req->rq_repmsg,
2778                                                                offset + 1),
2779                                              lustre_msg_buf(req->rq_repmsg,
2780                                                             offset + 2, 0),
2781                                              lustre_msg_buflen(req->rq_repmsg,
2782                                                                offset + 2))
2783                            > 0) {
2784                         body->valid |= OBD_MD_FLCOOKIE;
2785                 }
2786         }
2787
2788         GOTO(cleanup, rc);
2789 cleanup:
2790         inodes[0] = de_srcdir && !IS_ERR(de_srcdir) ? de_srcdir->d_inode : NULL;
2791         inodes[1] = old_inode;
2792         inodes[2] = de_tgtdir && !IS_ERR(de_tgtdir) ? de_tgtdir->d_inode : NULL;
2793         inodes[3] = new_inode;
2794         rc = mds_finish_transno(mds, inodes, handle, req, rc, 0, 0);
2795
2796 cleanup_no_trans:
2797         switch (cleanup_phase) {
2798         case 4:
2799                 UNLOCK_INODE_MUTEX(mds->mds_pending_dir->d_inode);
2800         case 3:
2801                 MDS_UP_READ_ORPHAN_SEM(new_inode);
2802         case 2:
2803                 iput(new_inode);
2804         case 1:
2805                 if (rc) {
2806                         if (lock_count == 4)
2807                                 ldlm_lock_decref(&(dlm_handles[3]), LCK_EX);
2808                         ldlm_lock_decref(&(dlm_handles[2]), LCK_EX);
2809                         ldlm_lock_decref(&(dlm_handles[1]), LCK_EX);
2810                         ldlm_lock_decref(&(dlm_handles[0]), LCK_EX);
2811                 } else {
2812                         if (lock_count == 4)
2813                                 ptlrpc_save_lock(req,&(dlm_handles[3]), LCK_EX);
2814                         ptlrpc_save_lock(req, &(dlm_handles[2]), LCK_EX);
2815                         ptlrpc_save_lock(req, &(dlm_handles[1]), LCK_EX);
2816                         ptlrpc_save_lock(req, &(dlm_handles[0]), LCK_EX);
2817                 }
2818                 l_dput(de_new);
2819                 l_dput(de_old);
2820                 l_dput(de_tgtdir);
2821                 l_dput(de_srcdir);
2822         case 0:
2823                 break;
2824         default:
2825                 CERROR("invalid cleanup_phase %d\n", cleanup_phase);
2826                 LBUG();
2827         }
2828         req->rq_status = rc;
2829
2830         /* acquire/release qunit */
2831         lquota_adjust(mds_quota_interface_ref, obd, qcids, qpids, rc,
2832                       FSFILT_OP_RENAME);
2833         return 0;
2834 }
2835
2836 typedef int (*mds_reinter)(struct mds_update_record *, int offset,
2837                            struct ptlrpc_request *, struct lustre_handle *);
2838
2839 static mds_reinter reinters[REINT_MAX] = {
2840         [REINT_SETATTR] mds_reint_setattr,
2841         [REINT_CREATE] mds_reint_create,
2842         [REINT_LINK] mds_reint_link,
2843         [REINT_UNLINK] mds_reint_unlink,
2844         [REINT_RENAME] mds_reint_rename,
2845         [REINT_OPEN] mds_open
2846 };
2847
2848 int mds_reint_rec(struct mds_update_record *rec, int offset,
2849                   struct ptlrpc_request *req, struct lustre_handle *lockh)
2850 {
2851         struct obd_device *obd = req->rq_export->exp_obd;
2852         struct mds_obd *mds = &obd->u.mds;
2853         struct lvfs_run_ctxt saved;
2854         int rc;
2855 #ifdef CRAY_XT3
2856         gid_t fsgid = rec->ur_uc.luc_fsgid;
2857 #endif
2858         ENTRY;
2859
2860 #ifdef CRAY_XT3
2861         if (req->rq_uid != LNET_UID_ANY) {
2862                 /* non-root local cluster client
2863                  * NB root's creds are believed... */
2864                 LASSERT (req->rq_uid != 0);
2865                 rec->ur_uc.luc_fsuid = req->rq_uid;
2866                 cfs_kernel_cap_unpack(&rec->ur_uc.luc_cap, 0);
2867         }
2868 #endif
2869
2870         /* get group info of this user */
2871         rec->ur_uc.luc_uce = upcall_cache_get_entry(mds->mds_group_hash,
2872                                                     rec->ur_uc.luc_fsuid,
2873                                                     rec->ur_uc.luc_fsgid, 2,
2874                                                     &rec->ur_uc.luc_suppgid1);
2875
2876         if (IS_ERR(rec->ur_uc.luc_uce)) {
2877                 rc = PTR_ERR(rec->ur_uc.luc_uce);
2878                 rec->ur_uc.luc_uce = NULL;
2879                 RETURN(rc);
2880         }
2881
2882         /* checked by unpacker */
2883         LASSERT(rec->ur_opcode < REINT_MAX && reinters[rec->ur_opcode] != NULL);
2884
2885 #ifdef CRAY_XT3
2886         if (rec->ur_uc.luc_uce)
2887                 rec->ur_uc.luc_fsgid = rec->ur_uc.luc_uce->ue_primary;
2888 #endif
2889
2890         push_ctxt(&saved, &obd->obd_lvfs_ctxt, &rec->ur_uc);
2891
2892 #ifdef CRAY_XT3
2893         if (rec->ur_uc.luc_uce && fsgid != rec->ur_uc.luc_fsgid &&
2894             in_group_p(fsgid)) {
2895                 rec->ur_uc.luc_fsgid = fsgid;
2896                 current->fsgid = saved.luc.luc_fsgid = fsgid;
2897         }
2898 #endif
2899
2900         rc = reinters[rec->ur_opcode] (rec, offset, req, lockh);
2901         pop_ctxt(&saved, &obd->obd_lvfs_ctxt, &rec->ur_uc);
2902
2903         upcall_cache_put_entry(mds->mds_group_hash, rec->ur_uc.luc_uce);
2904         RETURN(rc);
2905 }