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