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  *  linux/mds/mds_reint.c
   5  *  Lustre Metadata Server (mds) reintegration routines
   6  *
   7  *  Copyright (C) 2002 Cluster File Systems, Inc.
   8  *
   9  *   This file is part of Lustre, http://www.lustre.org.
  10  *
  11  *   Lustre is free software; you can redistribute it and/or
  12  *   modify it under the terms of version 2 of the GNU General Public
  13  *   License as published by the Free Software Foundation.
  14  *
  15  *   Lustre is distributed in the hope that it will be useful,
  16  *   but WITHOUT ANY WARRANTY; without even the implied warranty of
  17  *   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  18  *   GNU General Public License for more details.
  19  *
  20  *   You should have received a copy of the GNU General Public License
  21  *   along with Lustre; if not, write to the Free Software
  22  *   Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
  23  *
  24  *  Author: Peter Braam <braam@clusterfs.com>
  25  *  Author: Andreas Dilger <adilger@clusterfs.com>
  26  */
  27
  28 #define EXPORT_SYMTAB
  29 #define DEBUG_SUBSYSTEM S_MDS
  30
  31 #include <linux/obd_support.h>
  32 #include <linux/obd_class.h>
  33 #include <linux/obd.h>
  34 #include <linux/lustre_lib.h>
  35 #include <linux/lustre_idl.h>
  36 #include <linux/lustre_mds.h>
  37 #include <linux/lustre_dlm.h>
  38 #include <linux/obd_class.h>
  39
  40 extern inline struct mds_obd *mds_req2mds(struct ptlrpc_request *req);
  41
  42 void mds_start_transno(struct mds_obd *mds)
  43 {
  44         ENTRY;
  45         down(&mds->mds_transno_sem);
  46 }
  47
  48 /* Assumes caller has already pushed us into the kernel context. */
  49 int mds_finish_transno(struct mds_obd *mds, void *handle,
  50                        struct ptlrpc_request *req, int rc)
  51 {
  52         struct mds_export_data *med = &req->rq_export->exp_mds_data;
  53         struct mds_client_data *mcd = med->med_mcd;
  54         __u64 last_rcvd;
  55         loff_t off;
  56         ssize_t written;
  57
  58         /* Propagate error code. */
  59         if (rc)
  60                 goto out;
  61
  62         /* we don't allocate new transnos for replayed requests */
  63         if (req->rq_level == LUSTRE_CONN_RECOVD) {
  64                 rc = 0;
  65                 goto out;
  66         }
  67
  68         off = MDS_LR_CLIENT + med->med_off * MDS_LR_SIZE;
  69
  70         last_rcvd = ++mds->mds_last_rcvd;
  71         req->rq_repmsg->transno = HTON__u64(last_rcvd);
  72         mcd->mcd_last_rcvd = cpu_to_le64(last_rcvd);
  73         mcd->mcd_mount_count = cpu_to_le64(mds->mds_mount_count);
  74         mcd->mcd_last_xid = cpu_to_le64(req->rq_xid);
  75
  76         mds_fs_set_last_rcvd(mds, handle);
  77         written = lustre_fwrite(mds->mds_rcvd_filp, (char *)mcd, sizeof(*mcd),
  78                                 &off);
  79         CDEBUG(D_INODE, "wrote trans #"LPD64" for client %s at #%d: written = "
  80                "%d\n", last_rcvd, mcd->mcd_uuid, med->med_off, written);
  81
  82         if (written == sizeof(*mcd))
  83                 GOTO(out, rc = 0);
  84         CERROR("error writing to last_rcvd file: rc = %d\n", rc);
  85         if (written >= 0)
  86                 GOTO(out, rc = -EIO);
  87
  88         rc = 0;
  89
  90  out:
  91         EXIT;
  92         up(&mds->mds_transno_sem);
  93         return rc;
  94 }
  95
  96 /* In the write-back case, the client holds a lock on a subtree.
  97  * In the intent case, the client holds a lock on the child inode.
  98  * In the pathname case, the client (may) hold a lock on the child inode. */
  99 static int mds_reint_setattr(struct mds_update_record *rec, int offset,
 100                              struct ptlrpc_request *req)
 101 {
 102         struct mds_obd *mds = mds_req2mds(req);
 103         struct obd_device *obd = req->rq_export->exp_obd;
 104         struct mds_body *body;
 105         struct dentry *de;
 106         struct inode *inode;
 107         void *handle;
 108         struct lustre_handle child_lockh;
 109         int rc = 0, err;
 110
 111         if (req->rq_reqmsg->bufcount > offset + 1) {
 112                 struct dentry *dir;
 113                 struct lustre_handle dir_lockh;
 114                 char *name;
 115                 int namelen;
 116
 117                 /* a name was supplied by the client; fid1 is the directory */
 118                 dir = mds_fid2locked_dentry(obd, rec->ur_fid1, NULL, LCK_PR,
 119                                             &dir_lockh);
 120                 if (IS_ERR(dir)) {
 121                         LBUG();
 122                         GOTO(out_setattr, rc = PTR_ERR(dir));
 123                 }
 124
 125                 name = lustre_msg_buf(req->rq_reqmsg, offset + 1);
 126                 namelen = req->rq_reqmsg->buflens[offset + 1] - 1;
 127                 de = mds_name2locked_dentry(obd, dir, NULL, name, namelen,
 128                                             0, &child_lockh, LCK_PR);
 129                 l_dput(dir);
 130                 if (IS_ERR(de)) {
 131                         LBUG();
 132                         GOTO(out_setattr_de, rc = PTR_ERR(de));
 133                 }
 134         } else {
 135                 de = mds_fid2dentry(mds, rec->ur_fid1, NULL);
 136                 if (!de || IS_ERR(de)) {
 137                         GOTO(out_setattr_de, rc = PTR_ERR(de));
 138                 }
 139         }
 140         inode = de->d_inode;
 141         CDEBUG(D_INODE, "ino %ld\n", inode->i_ino);
 142
 143         OBD_FAIL_WRITE(OBD_FAIL_MDS_REINT_SETATTR_WRITE,
 144                        to_kdev_t(inode->i_sb->s_dev));
 145
 146         mds_start_transno(mds);
 147         handle = mds_fs_start(mds, inode, MDS_FSOP_SETATTR);
 148         if (IS_ERR(handle)) {
 149                 rc = PTR_ERR(handle);
 150                 (void)mds_finish_transno(mds, handle, req, rc);
 151                 GOTO(out_setattr_de, rc);
 152         }
 153
 154         rc = mds_fs_setattr(mds, de, handle, &rec->ur_iattr);
 155
 156         if (offset) {
 157                 body = lustre_msg_buf(req->rq_repmsg, 1);
 158                 mds_pack_inode2fid(&body->fid1, inode);
 159                 mds_pack_inode2body(body, inode);
 160         }
 161
 162         rc = mds_finish_transno(mds, handle, req, rc);
 163
 164         err = mds_fs_commit(mds, de->d_inode, handle);
 165         if (err) {
 166                 CERROR("error on commit: err = %d\n", err);
 167                 if (!rc)
 168                         rc = err;
 169         }
 170
 171         EXIT;
 172 out_setattr_de:
 173         l_dput(de);
 174 out_setattr:
 175         req->rq_status = rc;
 176         return 0;
 177 }
 178
 179 static int mds_reint_create(struct mds_update_record *rec, int offset,
 180                             struct ptlrpc_request *req)
 181 {
 182         struct dentry *de = NULL;
 183         struct mds_obd *mds = mds_req2mds(req);
 184         struct obd_device *obd = req->rq_export->exp_obd;
 185         struct dentry *dchild = NULL;
 186         struct inode *dir;
 187         void *handle;
 188         struct lustre_handle lockh;
 189         int rc = 0, err, lock_mode, type = rec->ur_mode & S_IFMT;
 190         ENTRY;
 191
 192         /* requests were at offset 2, replies go back at 1 */
 193         if (offset)
 194                 offset = 1;
 195
 196         LASSERT(!strcmp(req->rq_export->exp_obd->obd_type->typ_name, "mds"));
 197
 198         lock_mode = (req->rq_reqmsg->opc == MDS_REINT) ? LCK_CW : LCK_PW;
 199
 200         if (OBD_FAIL_CHECK(OBD_FAIL_MDS_REINT_CREATE))
 201                 GOTO(out_create, rc = -ESTALE);
 202
 203         de = mds_fid2locked_dentry(obd, rec->ur_fid1, NULL, lock_mode, &lockh);
 204         if (IS_ERR(de)) {
 205                 rc = PTR_ERR(de);
 206                 CERROR("parent lookup error %d\n", rc);
 207                 LBUG();
 208                 GOTO(out_create, rc);
 209         }
 210         dir = de->d_inode;
 211         CDEBUG(D_INODE, "parent ino %ld name %s mode %o\n",
 212                dir->i_ino, rec->ur_name, rec->ur_mode);
 213
 214         ldlm_lock_dump((void *)(unsigned long)lockh.addr);
 215
 216         down(&dir->i_sem);
 217         dchild = lookup_one_len(rec->ur_name, de, rec->ur_namelen - 1);
 218         if (IS_ERR(dchild)) {
 219                 CERROR("child lookup error %ld\n", PTR_ERR(dchild));
 220                 LBUG();
 221                 GOTO(out_create_de, rc = -ESTALE);
 222         }
 223
 224         if (dchild->d_inode) {
 225                 struct mds_body *body;
 226                 struct inode *inode = dchild->d_inode;
 227
 228                 CDEBUG(D_INODE, "child exists (dir %ld, name %s, ino %ld)\n",
 229                        dir->i_ino, rec->ur_name, dchild->d_inode->i_ino);
 230
 231                 /* XXX check that mode is correct? */
 232
 233                 body = lustre_msg_buf(req->rq_repmsg, offset);
 234                 mds_pack_inode2fid(&body->fid1, inode);
 235                 mds_pack_inode2body(body, inode);
 236                 if (S_ISREG(inode->i_mode))
 237                         rc = mds_pack_md(mds, req, offset + 1, body, inode);
 238
 239                 /* This isn't an error for RECREATE. */
 240                 if (rec->ur_opcode & REINT_REPLAYING) {
 241                         CDEBUG(D_INODE, "EEXIST suppressed for REPLAYING\n");
 242                         rc = 0;
 243                 } else {
 244                         rc = -EEXIST;
 245                 }
 246                 GOTO(out_create_dchild, rc);
 247         }
 248
 249         OBD_FAIL_WRITE(OBD_FAIL_MDS_REINT_CREATE_WRITE,
 250                        to_kdev_t(dir->i_sb->s_dev));
 251
 252         if (dir->i_mode & S_ISGID) {
 253                 rec->ur_gid = dir->i_gid;
 254                 if (S_ISDIR(rec->ur_mode))
 255                         rec->ur_mode |= S_ISGID;
 256         }
 257
 258         /* From here on, we must exit via a path that calls mds_finish_transno,
 259          * so that we release the mds_transno_sem (and, in the case of success,
 260          * update the transno correctly).  out_create_commit and
 261          * out_transno_dchild are good candidates.
 262          */
 263         mds_start_transno(mds);
 264
 265         switch (type) {
 266         case S_IFREG:{
 267                 handle = mds_fs_start(mds, dir, MDS_FSOP_CREATE);
 268                 if (IS_ERR(handle))
 269                         GOTO(out_transno_dchild, rc = PTR_ERR(handle));
 270                 rc = vfs_create(dir, dchild, rec->ur_mode);
 271                 EXIT;
 272                 break;
 273         }
 274         case S_IFDIR:{
 275                 handle = mds_fs_start(mds, dir, MDS_FSOP_MKDIR);
 276                 if (IS_ERR(handle))
 277                         GOTO(out_transno_dchild, rc = PTR_ERR(handle));
 278                 rc = vfs_mkdir(dir, dchild, rec->ur_mode);
 279                 EXIT;
 280                 break;
 281         }
 282         case S_IFLNK:{
 283                 handle = mds_fs_start(mds, dir, MDS_FSOP_SYMLINK);
 284                 if (IS_ERR(handle))
 285                         GOTO(out_transno_dchild, rc = PTR_ERR(handle));
 286                 rc = vfs_symlink(dir, dchild, rec->ur_name);
 287                 EXIT;
 288                 break;
 289         }
 290         case S_IFCHR:
 291         case S_IFBLK:
 292         case S_IFIFO:
 293         case S_IFSOCK:{
 294                 int rdev = rec->ur_rdev;
 295                 handle = mds_fs_start(mds, dir, MDS_FSOP_MKNOD);
 296                 if (IS_ERR(handle))
 297                         GOTO(out_transno_dchild, rc = PTR_ERR(handle));
 298                 rc = vfs_mknod(dir, dchild, rec->ur_mode, rdev);
 299                 EXIT;
 300                 break;
 301         }
 302         default:
 303                 CERROR("bad file type %o creating %s\n", type, rec->ur_name);
 304                 handle = NULL; /* quell uninitialized warning */
 305                 GOTO(out_transno_dchild, rc = -EINVAL);
 306         }
 307
 308         if (rc) {
 309                 CDEBUG(D_INODE, "error during create: %d\n", rc);
 310                 GOTO(out_create_commit, rc);
 311         } else {
 312                 struct iattr iattr;
 313                 struct inode *inode = dchild->d_inode;
 314                 struct mds_body *body;
 315
 316                 iattr.ia_atime = rec->ur_time;
 317                 iattr.ia_ctime = rec->ur_time;
 318                 iattr.ia_mtime = rec->ur_time;
 319                 iattr.ia_uid = rec->ur_uid;
 320                 iattr.ia_gid = rec->ur_gid;
 321                 iattr.ia_valid = ATTR_UID | ATTR_GID | ATTR_ATIME |
 322                         ATTR_MTIME | ATTR_CTIME;
 323
 324                 if (rec->ur_fid2->id) {
 325                         LASSERT(rec->ur_opcode & REINT_REPLAYING);
 326                         inode->i_generation = rec->ur_fid2->generation;
 327                         /* Dirtied and committed by the upcoming setattr. */
 328                         CDEBUG(D_INODE, "recreated ino %ld with gen %ld\n",
 329                                inode->i_ino, inode->i_generation);
 330                 } else {
 331                         CDEBUG(D_INODE, "created ino %ld\n", inode->i_ino);
 332                 }
 333
 334                 rc = mds_fs_setattr(mds, dchild, handle, &iattr);
 335                 if (rc) {
 336                         CERROR("error on setattr: rc = %d\n", rc);
 337                         /* XXX should we abort here in case of error? */
 338                 }
 339
 340                 body = lustre_msg_buf(req->rq_repmsg, offset);
 341                 mds_pack_inode2fid(&body->fid1, inode);
 342                 mds_pack_inode2body(body, inode);
 343         }
 344         EXIT;
 345 out_create_commit:
 346         if (rc) {
 347                 rc = mds_finish_transno(mds, handle, req, rc);
 348         } else {
 349                 rc = mds_finish_transno(mds, handle, req, rc);
 350                 if (rc)
 351                         GOTO(out_create_unlink, rc);
 352         }
 353         err = mds_fs_commit(mds, dir, handle);
 354         if (err) {
 355                 CERROR("error on commit: err = %d\n", err);
 356                 if (!rc)
 357                         rc = err;
 358         }
 359 out_create_dchild:
 360         l_dput(dchild);
 361         ldlm_lock_decref(&lockh, lock_mode);
 362 out_create_de:
 363         up(&dir->i_sem);
 364         l_dput(de);
 365 out_create:
 366         req->rq_status = rc;
 367         return 0;
 368
 369 out_transno_dchild:
 370         /* Need to release the transno lock, and then put the dchild. */
 371         LASSERT(rc);
 372         mds_finish_transno(mds, handle, req, rc);
 373         goto out_create_dchild;
 374
 375 out_create_unlink:
 376         /* Destroy the file we just created.  This should not need extra
 377          * journal credits, as we have already modified all of the blocks
 378          * needed in order to create the file in the first place.
 379          */
 380         switch (type) {
 381         case S_IFDIR:
 382                 err = vfs_rmdir(dir, dchild);
 383                 if (err)
 384                         CERROR("failed rmdir in error path: rc = %d\n", err);
 385                 break;
 386         default:
 387                 err = vfs_unlink(dir, dchild);
 388                 if (err)
 389                         CERROR("failed unlink in error path: rc = %d\n", err);
 390                 break;
 391         }
 392
 393         goto out_create_commit;
 394 }
 395
 396 static int mds_reint_unlink(struct mds_update_record *rec, int offset,
 397                             struct ptlrpc_request *req)
 398 {
 399         struct dentry *de = NULL;
 400         struct dentry *dchild = NULL;
 401         struct mds_obd *mds = mds_req2mds(req);
 402         struct obd_device *obd = req->rq_export->exp_obd;
 403         struct mds_body *body = NULL;
 404         char *name;
 405         struct inode *dir, *inode;
 406         struct lustre_handle lockh, child_lockh;
 407         void *handle;
 408         int namelen, lock_mode, err, rc = 0;
 409         ENTRY;
 410
 411         /* a name was supplied by the client; fid1 is the directory */
 412         lock_mode = (req->rq_reqmsg->opc == MDS_REINT) ? LCK_PW : LCK_PW;
 413         de = mds_fid2locked_dentry(obd, rec->ur_fid1, NULL, lock_mode, &lockh);
 414         if (IS_ERR(de)) {
 415                 LBUG();
 416                 RETURN(PTR_ERR(de));
 417         }
 418
 419         if (OBD_FAIL_CHECK(OBD_FAIL_MDS_REINT_UNLINK))
 420                 GOTO(out_unlink, rc = -ENOENT);
 421
 422         name = lustre_msg_buf(req->rq_reqmsg, offset + 1);
 423         namelen = req->rq_reqmsg->buflens[offset + 1] - 1;
 424 #warning "FIXME: if mds_name2locked_dentry decrefs this lock, we must not"
 425         memcpy(&child_lockh, &lockh, sizeof(child_lockh));
 426         dchild = mds_name2locked_dentry(obd, de, NULL, name, namelen,
 427                                         LCK_EX, &child_lockh, lock_mode);
 428
 429         if (IS_ERR(dchild)) {
 430                 LBUG();
 431                 GOTO(out_unlink, rc = PTR_ERR(dchild));
 432         }
 433
 434         dir = de->d_inode;
 435         inode = dchild->d_inode;
 436         CDEBUG(D_INODE, "parent ino %ld\n", dir->i_ino);
 437
 438         if (!inode) {
 439                 if (rec->ur_opcode & REINT_REPLAYING) {
 440                         CDEBUG(D_INODE,
 441                                "child missing (%ld/%s); OK for REPLAYING\n",
 442                                dir->i_ino, rec->ur_name);
 443                         rc = 0;
 444                 } else {
 445                         CDEBUG(D_INODE,
 446                                "child doesn't exist (dir %ld, name %s)\n",
 447                                dir->i_ino, rec->ur_name);
 448                         rc = -ENOENT;
 449                 }
 450                 /* going to out_unlink_cancel causes an LBUG, don't know why */
 451                 GOTO(out_unlink_dchild, rc);
 452         }
 453
 454         if (offset) {
 455                 /* XXX offset? */
 456                 offset = 1;
 457
 458                 body = lustre_msg_buf(req->rq_repmsg, offset);
 459                 mds_pack_inode2fid(&body->fid1, inode);
 460                 mds_pack_inode2body(body, inode);
 461         }
 462
 463         OBD_FAIL_WRITE(OBD_FAIL_MDS_REINT_UNLINK_WRITE,
 464                        to_kdev_t(dir->i_sb->s_dev));
 465
 466         mds_start_transno(mds);
 467         switch (rec->ur_mode /* & S_IFMT ? */) {
 468         case S_IFDIR:
 469                 handle = mds_fs_start(mds, dir, MDS_FSOP_RMDIR);
 470                 if (IS_ERR(handle))
 471                         GOTO(out_unlink_cancel_transno, rc = PTR_ERR(handle));
 472                 rc = vfs_rmdir(dir, dchild);
 473                 break;
 474         case S_IFREG:
 475                 /* get OBD EA data first so client can also destroy object */
 476                 if ((inode->i_mode & S_IFMT) == S_IFREG && offset)
 477                         rc = mds_pack_md(mds, req, offset + 1, body, inode);
 478                 /* no break */
 479         case S_IFLNK:
 480         case S_IFCHR:
 481         case S_IFBLK:
 482         case S_IFIFO:
 483         case S_IFSOCK:
 484                 handle = mds_fs_start(mds, dir, MDS_FSOP_UNLINK);
 485                 if (IS_ERR(handle))
 486                         GOTO(out_unlink_cancel_transno, rc = PTR_ERR(handle));
 487                 rc = vfs_unlink(dir, dchild);
 488                 break;
 489         default:
 490                 CERROR("bad file type %o unlinking %s\n", rec->ur_mode, name);
 491                 handle = NULL;
 492                 LBUG();
 493                 GOTO(out_unlink_cancel_transno, rc = -EINVAL);
 494         }
 495
 496         rc = mds_finish_transno(mds, handle, req, rc);
 497         err = mds_fs_commit(mds, dir, handle);
 498         if (err) {
 499                 CERROR("error on commit: err = %d\n", err);
 500                 if (!rc)
 501                         rc = err;
 502         }
 503
 504         EXIT;
 505
 506 out_unlink_cancel:
 507         ldlm_lock_decref(&child_lockh, LCK_EX);
 508         err = ldlm_cli_cancel(&child_lockh);
 509         if (err < 0) {
 510                 CERROR("failed to cancel child inode lock: err = %d\n", err);
 511                 if (!rc)
 512                         rc = -ENOLCK;   /*XXX translate LDLM lock error */
 513         }
 514 out_unlink_dchild:
 515         l_dput(dchild);
 516         up(&dir->i_sem);
 517 out_unlink:
 518         ldlm_lock_decref(&lockh, lock_mode);
 519         l_dput(de);
 520         req->rq_status = rc;
 521         return 0;
 522
 523 out_unlink_cancel_transno:
 524         rc = mds_finish_transno(mds, handle, req, rc);
 525         goto out_unlink_cancel;
 526 }
 527
 528 static int mds_reint_link(struct mds_update_record *rec, int offset,
 529                           struct ptlrpc_request *req)
 530 {
 531         struct obd_device *obd = req->rq_export->exp_obd;
 532         struct dentry *de_src = NULL;
 533         struct dentry *de_tgt_dir = NULL;
 534         struct dentry *dchild = NULL;
 535         struct mds_obd *mds = mds_req2mds(req);
 536         struct lustre_handle *handle, tgtlockh, srclockh;
 537         int lock_mode;
 538         __u64 res_id[3] = { 0 };
 539         int flags = 0;
 540         int rc = 0, err;
 541
 542         ENTRY;
 543         de_src = mds_fid2dentry(mds, rec->ur_fid1, NULL);
 544         if (IS_ERR(de_src) || OBD_FAIL_CHECK(OBD_FAIL_MDS_REINT_LINK)) {
 545                 GOTO(out_link, rc = -ESTALE);
 546         }
 547
 548         /* plan to change the link count on this inode: write lock */
 549         lock_mode = (req->rq_reqmsg->opc == MDS_REINT) ? LCK_PW : LCK_PW;
 550         res_id[0] = de_src->d_inode->i_ino;
 551         res_id[1] = de_src->d_inode->i_generation;
 552
 553         rc = ldlm_lock_match(obd->obd_namespace, res_id, LDLM_PLAIN,
 554                              NULL, 0, lock_mode, &srclockh);
 555         if (rc == 0) {
 556                 LDLM_DEBUG_NOLOCK("enqueue res "LPU64, res_id[0]);
 557                 rc = ldlm_cli_enqueue(NULL, NULL, obd->obd_namespace, NULL,
 558                                       res_id, LDLM_PLAIN, NULL, 0, lock_mode,
 559                                       &flags, ldlm_completion_ast,
 560                                       mds_blocking_ast, NULL, 0, &srclockh);
 561                 if (rc != ELDLM_OK) {
 562                         CERROR("lock enqueue: err: %d\n", rc);
 563                         GOTO(out_link_src_put, rc = -EIO);
 564                 }
 565         } else
 566                 ldlm_lock_dump((void *)(unsigned long)srclockh.addr);
 567
 568         de_tgt_dir = mds_fid2dentry(mds, rec->ur_fid2, NULL);
 569         if (IS_ERR(de_tgt_dir)) {
 570                 GOTO(out_link_src, rc = -ESTALE);
 571         }
 572
 573         lock_mode = (req->rq_reqmsg->opc == MDS_REINT) ? LCK_PW : LCK_PW;
 574         res_id[0] = de_tgt_dir->d_inode->i_ino;
 575         res_id[1] = de_tgt_dir->d_inode->i_generation;
 576
 577         rc = ldlm_lock_match(obd->obd_namespace, res_id, LDLM_PLAIN,
 578                              NULL, 0, lock_mode, &tgtlockh);
 579         if (rc == 0) {
 580                 LDLM_DEBUG_NOLOCK("enqueue res "LPU64, res_id[0]);
 581                 rc = ldlm_cli_enqueue(NULL, NULL, obd->obd_namespace, NULL,
 582                                       res_id, LDLM_PLAIN, NULL, 0, lock_mode,
 583                                       &flags, ldlm_completion_ast,
 584                                       mds_blocking_ast, NULL, 0, &tgtlockh);
 585                 if (rc != ELDLM_OK) {
 586                         CERROR("lock enqueue: err: %d\n", rc);
 587                         GOTO(out_link_tgt_dir_put, rc = -EIO);
 588                 }
 589         } else
 590                 ldlm_lock_dump((void *)(unsigned long)tgtlockh.addr);
 591
 592         down(&de_tgt_dir->d_inode->i_sem);
 593         dchild = lookup_one_len(rec->ur_name, de_tgt_dir, rec->ur_namelen - 1);
 594         if (IS_ERR(dchild)) {
 595                 CERROR("child lookup error %ld\n", PTR_ERR(dchild));
 596                 GOTO(out_link_tgt_dir, rc = -ESTALE);
 597         }
 598
 599         if (dchild->d_inode) {
 600                 struct inode *inode = dchild->d_inode;
 601                 /* in intent case ship back attributes to client */
 602                 if (offset) {
 603                         struct mds_body *body =
 604                                 lustre_msg_buf(req->rq_repmsg, 1);
 605
 606                         mds_pack_inode2fid(&body->fid1, inode);
 607                         mds_pack_inode2body(body, inode);
 608                         if (S_ISREG(inode->i_mode))
 609                                 rc = mds_pack_md(mds, req, 2, body, inode);
 610                 }
 611                 if (rec->ur_opcode & REINT_REPLAYING) {
 612                         /* XXX verify that the link is to the the right file? */
 613                         rc = 0;
 614                         CDEBUG(D_INODE,
 615                                "child exists (dir %ld, name %s) (REPLAYING)\n",
 616                                de_tgt_dir->d_inode->i_ino, rec->ur_name);
 617                 } else {
 618                         rc = -EEXIST;
 619                         CERROR("child exists (dir %ld, name %s)\n",
 620                                de_tgt_dir->d_inode->i_ino, rec->ur_name);
 621                 }
 622                 GOTO(out_link_dchild, rc);
 623         }
 624
 625         OBD_FAIL_WRITE(OBD_FAIL_MDS_REINT_LINK_WRITE,
 626                        to_kdev_t(de_src->d_inode->i_sb->s_dev));
 627
 628         mds_start_transno(mds);
 629         handle = mds_fs_start(mds, de_tgt_dir->d_inode, MDS_FSOP_LINK);
 630         if (IS_ERR(handle)) {
 631                 rc = PTR_ERR(handle);
 632                 mds_finish_transno(mds, handle, req, rc);
 633                 GOTO(out_link_dchild, rc);
 634         }
 635
 636         rc = vfs_link(de_src, de_tgt_dir->d_inode, dchild);
 637         if (rc)
 638                 CERROR("link error %d\n", rc);
 639         rc = mds_finish_transno(mds, handle, req, rc);
 640
 641         err = mds_fs_commit(mds, de_tgt_dir->d_inode, handle);
 642         if (err) {
 643                 CERROR("error on commit: err = %d\n", err);
 644                 if (!rc)
 645                         rc = err;
 646         }
 647         EXIT;
 648
 649 out_link_dchild:
 650         l_dput(dchild);
 651 out_link_tgt_dir:
 652         ldlm_lock_decref(&tgtlockh, lock_mode);
 653 out_link_tgt_dir_put:
 654         up(&de_tgt_dir->d_inode->i_sem);
 655         l_dput(de_tgt_dir);
 656 out_link_src:
 657         ldlm_lock_decref(&srclockh, lock_mode);
 658 out_link_src_put:
 659         l_dput(de_src);
 660 out_link:
 661         req->rq_status = rc;
 662         return 0;
 663 }
 664
 665 static int mds_reint_rename(struct mds_update_record *rec, int offset,
 666                             struct ptlrpc_request *req)
 667 {
 668         struct obd_device *obd = req->rq_export->exp_obd;
 669         struct dentry *de_srcdir = NULL;
 670         struct dentry *de_tgtdir = NULL;
 671         struct dentry *de_old = NULL;
 672         struct dentry *de_new = NULL;
 673         struct mds_obd *mds = mds_req2mds(req);
 674         struct lustre_handle tgtlockh, srclockh, oldhandle;
 675         int flags = 0, lock_mode, rc = 0, err;
 676         void *handle;
 677         __u64 res_id[3] = { 0 };
 678         ENTRY;
 679
 680         de_srcdir = mds_fid2dentry(mds, rec->ur_fid1, NULL);
 681         if (IS_ERR(de_srcdir))
 682                 GOTO(out_rename, rc = -ESTALE);
 683
 684         lock_mode = (req->rq_reqmsg->opc == MDS_REINT) ? LCK_PW : LCK_PW;
 685         res_id[0] = de_srcdir->d_inode->i_ino;
 686         res_id[1] = de_srcdir->d_inode->i_generation;
 687
 688         rc = ldlm_lock_match(obd->obd_namespace, res_id, LDLM_PLAIN,
 689                              NULL, 0, lock_mode, &srclockh);
 690         if (rc == 0) {
 691                 LDLM_DEBUG_NOLOCK("enqueue res "LPU64, res_id[0]);
 692                 rc = ldlm_cli_enqueue(NULL, NULL, obd->obd_namespace, NULL,
 693                                       res_id, LDLM_PLAIN, NULL, 0, lock_mode,
 694                                       &flags, ldlm_completion_ast,
 695                                       mds_blocking_ast, NULL, 0, &srclockh);
 696                 if (rc != ELDLM_OK) {
 697                         CERROR("lock enqueue: err: %d\n", rc);
 698                         GOTO(out_rename_srcput, rc = -EIO);
 699                 }
 700         } else
 701                 ldlm_lock_dump((void *)(unsigned long)srclockh.addr);
 702
 703         de_tgtdir = mds_fid2dentry(mds, rec->ur_fid2, NULL);
 704         if (IS_ERR(de_tgtdir))
 705                 GOTO(out_rename_srcdir, rc = -ESTALE);
 706
 707         lock_mode = (req->rq_reqmsg->opc == MDS_REINT) ? LCK_PW : LCK_PW;
 708         res_id[0] = de_tgtdir->d_inode->i_ino;
 709         res_id[1] = de_tgtdir->d_inode->i_generation;
 710
 711         rc = ldlm_lock_match(obd->obd_namespace, res_id, LDLM_PLAIN,
 712                              NULL, 0, lock_mode, &tgtlockh);
 713         if (rc == 0) {
 714                 flags = 0;
 715                 LDLM_DEBUG_NOLOCK("enqueue res "LPU64, res_id[0]);
 716                 rc = ldlm_cli_enqueue(NULL, NULL, obd->obd_namespace, NULL,
 717                                       res_id, LDLM_PLAIN, NULL, 0, lock_mode,
 718                                       &flags, ldlm_completion_ast,
 719                                       mds_blocking_ast, NULL, 0, &tgtlockh);
 720                 if (rc != ELDLM_OK) {
 721                         CERROR("lock enqueue: err: %d\n", rc);
 722                         GOTO(out_rename_tgtput, rc = -EIO);
 723                 }
 724         } else
 725                 ldlm_lock_dump((void *)(unsigned long)tgtlockh.addr);
 726
 727 #if (LINUX_VERSION_CODE < KERNEL_VERSION(2,5,0))
 728         double_lock(de_tgtdir, de_srcdir);
 729 #endif
 730         de_old = lookup_one_len(rec->ur_name, de_srcdir, rec->ur_namelen - 1);
 731         if (IS_ERR(de_old)) {
 732                 CERROR("old child lookup error (%*s): %ld\n",
 733                        rec->ur_namelen - 1, rec->ur_name, PTR_ERR(de_old));
 734                 GOTO(out_rename_tgtdir, rc = -ENOENT);
 735         }
 736
 737         de_new = lookup_one_len(rec->ur_tgt, de_tgtdir, rec->ur_tgtlen - 1);
 738         if (IS_ERR(de_new)) {
 739                 CERROR("new child lookup error (%*s): %ld\n",
 740                        rec->ur_tgtlen - 1, rec->ur_tgt, PTR_ERR(de_new));
 741                 GOTO(out_rename_deold, rc = -ENOENT);
 742         }
 743
 744         /* in intent case ship back attributes to client */
 745         if (offset) {
 746                 struct mds_body *body = lustre_msg_buf(req->rq_repmsg, 1);
 747                 struct inode *inode = de_new->d_inode;
 748
 749                 if (!inode) {
 750                         body->valid = 0;
 751                 } else {
 752                         mds_pack_inode2fid(&body->fid1, inode);
 753                         mds_pack_inode2body(body, inode);
 754                         if (S_ISREG(inode->i_mode))
 755                                 rc = mds_pack_md(mds, req, 2, body, inode);
 756                 }
 757         }
 758
 759         OBD_FAIL_WRITE(OBD_FAIL_MDS_REINT_RENAME_WRITE,
 760                        to_kdev_t(de_srcdir->d_inode->i_sb->s_dev));
 761
 762         mds_start_transno(mds);
 763         handle = mds_fs_start(mds, de_tgtdir->d_inode, MDS_FSOP_RENAME);
 764         if (IS_ERR(handle)) {
 765                 rc = PTR_ERR(handle);
 766                 mds_finish_transno(mds, handle, req, rc);
 767                 GOTO(out_rename_denew, rc);
 768         }
 769
 770         lock_kernel();
 771         rc = vfs_rename(de_srcdir->d_inode, de_old, de_tgtdir->d_inode, de_new,
 772                         NULL);
 773         unlock_kernel();
 774
 775         rc = mds_finish_transno(mds, handle, req, rc);
 776
 777         err = mds_fs_commit(mds, de_tgtdir->d_inode, handle);
 778         if (err) {
 779                 CERROR("error on commit: err = %d\n", err);
 780                 if (!rc)
 781                         rc = err;
 782         }
 783         EXIT;
 784
 785 out_rename_denew:
 786         l_dput(de_new);
 787 out_rename_deold:
 788         if (!rc) {
 789                 res_id[0] = de_old->d_inode->i_ino;
 790                 res_id[1] = de_old->d_inode->i_generation;
 791                 flags = 0;
 792                 /* Take an exclusive lock on the resource that we're
 793                  * about to free, to force everyone to drop their
 794                  * locks. */
 795                 LDLM_DEBUG_NOLOCK("getting EX lock res "LPU64, res_id[0]);
 796                 rc = ldlm_cli_enqueue(NULL, NULL, obd->obd_namespace, NULL,
 797                                       res_id, LDLM_PLAIN, NULL, 0, LCK_EX,
 798                                       &flags, ldlm_completion_ast,
 799                                       mds_blocking_ast, NULL, 0, &oldhandle);
 800                 if (rc)
 801                         CERROR("failed to get child inode lock (child ino "
 802                                LPD64" dir ino %ld)\n",
 803                                res_id[0], de_old->d_inode->i_ino);
 804         }
 805
 806         l_dput(de_old);
 807
 808         if (!rc) {
 809                 ldlm_lock_decref(&oldhandle, LCK_EX);
 810                 rc = ldlm_cli_cancel(&oldhandle);
 811                 if (rc < 0)
 812                         CERROR("failed to cancel child inode lock ino "
 813                                LPD64": %d\n", res_id[0], rc);
 814         }
 815 out_rename_tgtdir:
 816 #if (LINUX_VERSION_CODE < KERNEL_VERSION(2,5,0))
 817         double_up(&de_srcdir->d_inode->i_sem, &de_tgtdir->d_inode->i_sem);
 818 #endif
 819         ldlm_lock_decref(&tgtlockh, lock_mode);
 820 out_rename_tgtput:
 821         l_dput(de_tgtdir);
 822 out_rename_srcdir:
 823         ldlm_lock_decref(&srclockh, lock_mode);
 824 out_rename_srcput:
 825         l_dput(de_srcdir);
 826 out_rename:
 827         req->rq_status = rc;
 828         return 0;
 829 }
 830
 831 typedef int (*mds_reinter) (struct mds_update_record *, int offset,
 832                             struct ptlrpc_request *);
 833
 834 static mds_reinter reinters[REINT_MAX + 1] = {
 835         [REINT_SETATTR] mds_reint_setattr,
 836         [REINT_CREATE] mds_reint_create,
 837         [REINT_UNLINK] mds_reint_unlink,
 838         [REINT_LINK] mds_reint_link,
 839         [REINT_RENAME] mds_reint_rename,
 840 };
 841
 842 int mds_reint_rec(struct mds_update_record *rec, int offset,
 843                   struct ptlrpc_request *req)
 844 {
 845         struct mds_obd *mds = mds_req2mds(req);
 846         struct obd_run_ctxt saved;
 847         struct obd_ucred uc;
 848         int realop = rec->ur_opcode & REINT_OPCODE_MASK;
 849         int rc;
 850
 851         if (realop < 1 || realop > REINT_MAX) {
 852                 CERROR("opcode %d not valid (%sREPLAYING)\n", realop,
 853                        rec->ur_opcode & REINT_REPLAYING ? "" : "not ");
 854                 rc = req->rq_status = -EINVAL;
 855                 RETURN(rc);
 856         }
 857
 858         uc.ouc_fsuid = rec->ur_fsuid;
 859         uc.ouc_fsgid = rec->ur_fsgid;
 860         uc.ouc_cap = rec->ur_cap;
 861
 862         push_ctxt(&saved, &mds->mds_ctxt, &uc);
 863         rc = reinters[realop] (rec, offset, req);
 864         pop_ctxt(&saved);
 865
 866         return rc;
 867 }