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  *
   6  *  Lustre Metadata Server (mds) reintegration routines
   7  *
   8  *  Copyright (C) 2002  Cluster File Systems, Inc.
   9  *  author: Peter Braam <braam@clusterfs.com>
  10  *
  11  *  This code is issued under the GNU General Public License.
  12  *  See the file COPYING in this distribution
  13  *
  14  */
  15
  16 // XXX - add transaction sequence numbers
  17
  18 #define EXPORT_SYMTAB
  19 #define DEBUG_SUBSYSTEM S_MDS
  20
  21 #include <linux/obd_support.h>
  22 #include <linux/obd_class.h>
  23 #include <linux/obd.h>
  24 #include <linux/lustre_lib.h>
  25 #include <linux/lustre_idl.h>
  26 #include <linux/lustre_mds.h>
  27 #include <linux/obd_class.h>
  28
  29 struct mds_client_info *mds_uuid_to_mci(struct mds_obd *mds, __u8 *uuid)
  30 {
  31         struct list_head *p;
  32
  33         if (!uuid)
  34                 return NULL;
  35
  36         list_for_each(p, &mds->mds_client_info) {
  37                 struct mds_client_info *mci;
  38
  39                 mci = list_entry(p, struct mds_client_info, mci_list);
  40                 CDEBUG(D_INFO, "checking client UUID '%s'\n",
  41                        mci->mci_mcd->mcd_uuid);
  42                 if (!strncmp(mci->mci_mcd->mcd_uuid, uuid,
  43                              sizeof(mci->mci_mcd->mcd_uuid)))
  44                         return mci;
  45         }
  46         CDEBUG(D_INFO, "no mds client info found for  UUID '%s'\n", uuid);
  47         return NULL;
  48 }
  49
  50 int mds_update_last_rcvd(struct mds_obd *mds, void *handle,
  51                          struct ptlrpc_request *req)
  52 {
  53         /* get from req->rq_connection-> or req->rq_client */
  54         struct obd_run_ctxt saved;
  55         struct mds_client_info *mci;
  56         loff_t off;
  57         int rc;
  58
  59         mci = mds_uuid_to_mci(mds, req->rq_connection->c_remote_uuid);
  60         if (!mci) {
  61                 CERROR("unable to locate MDS client data for UUID '%s'\n",
  62                        ptlrpc_req_to_uuid(req));
  63                 /* This will be a real error once everything is working */
  64                 //LBUG();
  65                 RETURN(0);
  66         }
  67
  68         off = MDS_LR_CLIENT + mci->mci_off * MDS_LR_SIZE;
  69
  70         ++mds->mds_last_rcvd;   /* lock this, or make it an LDLM function? */
  71         req->rq_repmsg->transno = HTON__u64(mds->mds_last_rcvd);
  72         mci->mci_mcd->mcd_last_rcvd = cpu_to_le64(mds->mds_last_rcvd);
  73         mci->mci_mcd->mcd_mount_count = cpu_to_le64(mds->mds_mount_count);
  74         mci->mci_mcd->mcd_last_xid = cpu_to_le32(req->rq_reqmsg->xid);
  75
  76         mds_fs_set_last_rcvd(mds, handle);
  77         push_ctxt(&saved, &mds->mds_ctxt);
  78         rc = lustre_fwrite(mds->mds_rcvd_filp, (char *)mci->mci_mcd,
  79                            sizeof(*mci->mci_mcd), &off);
  80         pop_ctxt(&saved);
  81         CDEBUG(D_INODE, "wrote trans #%Ld for client '%s' at #%d: rc = %d\n",
  82                mds->mds_last_rcvd, mci->mci_mcd->mcd_uuid, mci->mci_off, rc);
  83         // store new value and last committed value in req struct
  84
  85         if (rc == sizeof(*mci->mci_mcd))
  86                 rc = 0;
  87         else {
  88                 CERROR("error writing to last_rcvd file: rc = %d\n", rc);
  89                 if (rc >= 0)
  90                         rc = -EIO;
  91         }
  92
  93         return rc;
  94 }
  95
  96 static int mds_reint_setattr(struct mds_update_record *rec,
  97                              struct ptlrpc_request *req)
  98 {
  99         struct mds_obd *mds = &req->rq_obd->u.mds;
 100         struct dentry *de;
 101         void *handle;
 102         int rc = 0;
 103         int err;
 104
 105         de = mds_fid2dentry(mds, rec->ur_fid1, NULL);
 106         if (IS_ERR(de) || OBD_FAIL_CHECK(OBD_FAIL_MDS_REINT_SETATTR)) {
 107                 GOTO(out_setattr, rc = -ESTALE);
 108         }
 109
 110         CDEBUG(D_INODE, "ino %ld\n", de->d_inode->i_ino);
 111
 112         OBD_FAIL_WRITE(OBD_FAIL_MDS_REINT_SETATTR_WRITE,
 113                        de->d_inode->i_sb->s_dev);
 114
 115         handle = mds_fs_start(mds, de->d_inode, MDS_FSOP_SETATTR);
 116         if (!handle)
 117                 GOTO(out_setattr_de, rc = PTR_ERR(handle));
 118         rc = mds_fs_setattr(mds, de, handle, &rec->ur_iattr);
 119
 120         if (!rc)
 121                 rc = mds_update_last_rcvd(mds, handle, req);
 122
 123         err = mds_fs_commit(mds, de->d_inode, handle);
 124         if (err) {
 125                 CERROR("error on commit: err = %d\n", err);
 126                 if (!rc)
 127                         rc = err;
 128         }
 129         EXIT;
 130 out_setattr_de:
 131         l_dput(de);
 132 out_setattr:
 133         req->rq_status = rc;
 134         return(0);
 135 }
 136
 137 static int mds_reint_recreate(struct mds_update_record *rec,
 138                             struct ptlrpc_request *req)
 139 {
 140         struct dentry *de = NULL;
 141         struct mds_obd *mds = &req->rq_obd->u.mds;
 142         struct dentry *dchild = NULL;
 143         struct inode *dir;
 144         int rc = 0;
 145         ENTRY;
 146
 147         de = mds_fid2dentry(mds, rec->ur_fid1, NULL);
 148         if (IS_ERR(de) || OBD_FAIL_CHECK(OBD_FAIL_MDS_REINT_CREATE)) {
 149                 LBUG();
 150                 GOTO(out_create_de, rc = -ESTALE);
 151         }
 152         dir = de->d_inode;
 153         CDEBUG(D_INODE, "parent ino %ld\n", dir->i_ino);
 154
 155         down(&dir->i_sem);
 156         dchild = lookup_one_len(rec->ur_name, de, rec->ur_namelen - 1);
 157         if (IS_ERR(dchild)) {
 158                 CERROR("child lookup error %ld\n", PTR_ERR(dchild));
 159                 up(&dir->i_sem);
 160                 LBUG();
 161                 GOTO(out_create_dchild, rc = -ESTALE);
 162         }
 163
 164         if (dchild->d_inode) {
 165                 struct mds_body *body;
 166                 rc = 0;
 167                 body = lustre_msg_buf(req->rq_repmsg, 0);
 168                 body->ino = dchild->d_inode->i_ino;
 169                 body->generation = dchild->d_inode->i_generation;
 170         } else {
 171                 CERROR("child doesn't exist (dir %ld, name %s)\n",
 172                        dir->i_ino, rec->ur_name);
 173                 rc = -ENOENT;
 174                 LBUG();
 175         }
 176
 177 out_create_dchild:
 178         l_dput(dchild);
 179         up(&dir->i_sem);
 180 out_create_de:
 181         l_dput(de);
 182         req->rq_status = rc;
 183         return 0;
 184 }
 185
 186 static int mds_reint_create(struct mds_update_record *rec,
 187                             struct ptlrpc_request *req)
 188 {
 189         struct dentry *de = NULL;
 190         struct mds_obd *mds = &req->rq_obd->u.mds;
 191         struct dentry *dchild = NULL;
 192         struct inode *dir;
 193         void *handle;
 194         int rc = 0, type = rec->ur_mode & S_IFMT;
 195         int err;
 196         ENTRY;
 197
 198         de = mds_fid2dentry(mds, rec->ur_fid1, NULL);
 199         if (IS_ERR(de) || OBD_FAIL_CHECK(OBD_FAIL_MDS_REINT_CREATE)) {
 200                 LBUG();
 201                 GOTO(out_create_de, rc = -ESTALE);
 202         }
 203         dir = de->d_inode;
 204         CDEBUG(D_INODE, "parent ino %ld name %s mode %o\n",
 205                dir->i_ino, rec->ur_name, rec->ur_mode);
 206
 207         down(&dir->i_sem);
 208         dchild = lookup_one_len(rec->ur_name, de, rec->ur_namelen - 1);
 209         if (IS_ERR(dchild)) {
 210                 CERROR("child lookup error %ld\n", PTR_ERR(dchild));
 211                 up(&dir->i_sem);
 212                 LBUG();
 213                 GOTO(out_create_dchild, rc = -ESTALE);
 214         }
 215
 216         if (dchild->d_inode) {
 217                 CERROR("child exists (dir %ld, name %s, ino %ld)\n",
 218                        dir->i_ino, rec->ur_name, dchild->d_inode->i_ino);
 219                 LBUG();
 220                 GOTO(out_create_dchild, rc = -EEXIST);
 221         }
 222
 223         OBD_FAIL_WRITE(OBD_FAIL_MDS_REINT_CREATE_WRITE, dir->i_sb->s_dev);
 224
 225         switch (type) {
 226         case S_IFREG: {
 227                 handle = mds_fs_start(mds, dir, MDS_FSOP_CREATE);
 228                 if (!handle)
 229                         GOTO(out_create_dchild, PTR_ERR(handle));
 230                 rc = vfs_create(dir, dchild, rec->ur_mode);
 231                 EXIT;
 232                 break;
 233         }
 234         case S_IFDIR: {
 235                 handle = mds_fs_start(mds, dir, MDS_FSOP_MKDIR);
 236                 if (!handle)
 237                         GOTO(out_create_dchild, PTR_ERR(handle));
 238                 rc = vfs_mkdir(dir, dchild, rec->ur_mode);
 239                 EXIT;
 240                 break;
 241         }
 242         case S_IFLNK: {
 243                 handle = mds_fs_start(mds, dir, MDS_FSOP_SYMLINK);
 244                 if (!handle)
 245                         GOTO(out_create_dchild, PTR_ERR(handle));
 246                 rc = vfs_symlink(dir, dchild, rec->ur_tgt);
 247                 EXIT;
 248                 break;
 249         }
 250         case S_IFCHR:
 251         case S_IFBLK:
 252         case S_IFIFO:
 253         case S_IFSOCK: {
 254                 int rdev = rec->ur_rdev;
 255                 handle = mds_fs_start(mds, dir, MDS_FSOP_MKNOD);
 256                 if (!handle)
 257                         GOTO(out_create_dchild, PTR_ERR(handle));
 258                 rc = vfs_mknod(dir, dchild, rec->ur_mode, rdev);
 259                 EXIT;
 260                 break;
 261         }
 262         default:
 263                 CERROR("bad file type %d for create of %s\n",type,rec->ur_name);
 264                 GOTO(out_create_dchild, rc = -EINVAL);
 265         }
 266
 267         if (rc) {
 268                 CERROR("error during create: %d\n", rc);
 269                 if (rc != -ENOSPC) LBUG();
 270                 GOTO(out_create_commit, rc);
 271         } else {
 272                 struct iattr iattr;
 273                 struct inode *inode = dchild->d_inode;
 274                 struct mds_body *body;
 275
 276                 CDEBUG(D_INODE, "created ino %ld\n", dchild->d_inode->i_ino);
 277                 if (type == S_IFREG) {
 278                         struct obdo *obdo;
 279                         obdo = lustre_msg_buf(req->rq_reqmsg, 2);
 280                         rc = mds_fs_set_obdo(mds, inode, handle, obdo);
 281                         if (rc)
 282                                 CERROR("error %d setting objid for %ld\n",
 283                                        rc, inode->i_ino);
 284                 }
 285
 286                 iattr.ia_atime = rec->ur_time;
 287                 iattr.ia_ctime = rec->ur_time;
 288                 iattr.ia_mtime = rec->ur_time;
 289                 iattr.ia_uid = rec->ur_uid;
 290                 iattr.ia_gid = rec->ur_gid;
 291                 iattr.ia_valid = ATTR_UID | ATTR_GID | ATTR_ATIME |
 292                         ATTR_MTIME | ATTR_CTIME;
 293
 294                 rc = mds_fs_setattr(mds, dchild, handle, &iattr);
 295                 if (rc) {
 296                         CERROR("error on setattr: rc = %d\n", rc);
 297                         /* XXX should we abort here in case of error? */
 298                 }
 299
 300                 rc = mds_update_last_rcvd(mds, handle, req);
 301                 if (rc) {
 302                         CERROR("error on update_last_rcvd: rc = %d\n", rc);
 303                         /* XXX should we abort here in case of error? */
 304                 }
 305
 306                 body = lustre_msg_buf(req->rq_repmsg, 0);
 307                 body->ino = inode->i_ino;
 308                 body->generation = inode->i_generation;
 309         }
 310
 311 out_create_commit:
 312         err = mds_fs_commit(mds, dir, handle);
 313         if (err) {
 314                 CERROR("error on commit: err = %d\n", err);
 315                 if (!rc)
 316                         rc = err;
 317         }
 318 out_create_dchild:
 319         l_dput(dchild);
 320         up(&dir->i_sem);
 321 out_create_de:
 322         l_dput(de);
 323         req->rq_status = rc;
 324         return 0;
 325 }
 326
 327 static int mds_reint_unlink(struct mds_update_record *rec,
 328                             struct ptlrpc_request *req)
 329 {
 330         struct dentry *de = NULL;
 331         struct dentry *dchild = NULL;
 332         struct mds_obd *mds = &req->rq_obd->u.mds;
 333         struct inode *dir, *inode;
 334         void *handle;
 335         int rc = 0;
 336         int err;
 337         ENTRY;
 338
 339         de = mds_fid2dentry(mds, rec->ur_fid1, NULL);
 340         if (IS_ERR(de) || OBD_FAIL_CHECK(OBD_FAIL_MDS_REINT_UNLINK)) {
 341                 LBUG();
 342                 GOTO(out_unlink, rc = -ESTALE);
 343         }
 344         dir = de->d_inode;
 345         CDEBUG(D_INODE, "parent ino %ld\n", dir->i_ino);
 346
 347         down(&dir->i_sem);
 348         dchild = lookup_one_len(rec->ur_name, de, rec->ur_namelen - 1);
 349         if (IS_ERR(dchild)) {
 350                 CERROR("child lookup error %ld\n", PTR_ERR(dchild));
 351                 LBUG();
 352                 GOTO(out_unlink_de, rc = -ESTALE);
 353         }
 354
 355         inode = dchild->d_inode;
 356         if (!inode) {
 357                 CERROR("child doesn't exist (dir %ld, name %s\n",
 358                        dir->i_ino, rec->ur_name);
 359                 LBUG();
 360                 GOTO(out_unlink_dchild, rc = -ESTALE);
 361         }
 362
 363         if (inode->i_ino != rec->ur_fid2->id) {
 364                 CERROR("inode and FID ID do not match (%ld != %Ld)\n",
 365                        inode->i_ino, rec->ur_fid2->id);
 366                 LBUG();
 367                 GOTO(out_unlink_dchild, rc = -ESTALE);
 368         }
 369         if (inode->i_generation != rec->ur_fid2->generation) {
 370                 CERROR("inode and FID GENERATION do not match (%d != %d)\n",
 371                        inode->i_generation, rec->ur_fid2->generation);
 372                 LBUG();
 373                 GOTO(out_unlink_dchild, rc = -ESTALE);
 374         }
 375
 376         OBD_FAIL_WRITE(OBD_FAIL_MDS_REINT_UNLINK_WRITE, dir->i_sb->s_dev);
 377
 378         switch (dchild->d_inode->i_mode & S_IFMT) {
 379         case S_IFDIR:
 380                 handle = mds_fs_start(mds, dir, MDS_FSOP_RMDIR);
 381                 if (!handle)
 382                         GOTO(out_unlink_dchild, rc = PTR_ERR(handle));
 383                 rc = vfs_rmdir(dir, dchild);
 384                 break;
 385         default:
 386                 handle = mds_fs_start(mds, dir, MDS_FSOP_UNLINK);
 387                 if (!handle)
 388                         GOTO(out_unlink_dchild, rc = PTR_ERR(handle));
 389                 rc = vfs_unlink(dir, dchild);
 390                 break;
 391         }
 392
 393         if (!rc)
 394                 rc = mds_update_last_rcvd(mds, handle, req);
 395         err = mds_fs_commit(mds, dir, handle);
 396         if (err) {
 397                 CERROR("error on commit: err = %d\n", err);
 398                 if (!rc)
 399                         rc = err;
 400         }
 401
 402         EXIT;
 403 out_unlink_dchild:
 404         l_dput(dchild);
 405 out_unlink_de:
 406         up(&dir->i_sem);
 407         l_dput(de);
 408 out_unlink:
 409         req->rq_status = rc;
 410         return 0;
 411 }
 412
 413 static int mds_reint_link(struct mds_update_record *rec,
 414                             struct ptlrpc_request *req)
 415 {
 416         struct dentry *de_src = NULL;
 417         struct dentry *de_tgt_dir = NULL;
 418         struct dentry *dchild = NULL;
 419         struct mds_obd *mds = &req->rq_obd->u.mds;
 420         void *handle;
 421         int rc = 0;
 422         int err;
 423
 424         ENTRY;
 425         de_src = mds_fid2dentry(mds, rec->ur_fid1, NULL);
 426         if (IS_ERR(de_src) || OBD_FAIL_CHECK(OBD_FAIL_MDS_REINT_LINK)) {
 427                 GOTO(out_link, rc = -ESTALE);
 428         }
 429
 430         de_tgt_dir = mds_fid2dentry(mds, rec->ur_fid2, NULL);
 431         if (IS_ERR(de_tgt_dir)) {
 432                 GOTO(out_link_de_src, rc = -ESTALE);
 433         }
 434
 435         down(&de_tgt_dir->d_inode->i_sem);
 436         dchild = lookup_one_len(rec->ur_name, de_tgt_dir, rec->ur_namelen - 1);
 437         if (IS_ERR(dchild)) {
 438                 CERROR("child lookup error %ld\n", PTR_ERR(dchild));
 439                 GOTO(out_link_de_tgt_dir, rc = -ESTALE);
 440         }
 441
 442         if (dchild->d_inode) {
 443                 CERROR("child exists (dir %ld, name %s\n",
 444                        de_tgt_dir->d_inode->i_ino, rec->ur_name);
 445                 GOTO(out_link_dchild, rc = -EEXIST);
 446         }
 447
 448         OBD_FAIL_WRITE(OBD_FAIL_MDS_REINT_LINK_WRITE,
 449                        dchild->d_inode->i_sb->s_dev);
 450
 451         handle = mds_fs_start(mds, de_tgt_dir->d_inode, MDS_FSOP_LINK);
 452         if (!handle)
 453                 GOTO(out_link_dchild, rc = PTR_ERR(handle));
 454
 455         rc = vfs_link(de_src, de_tgt_dir->d_inode, dchild);
 456
 457         if (!rc)
 458                 rc = mds_update_last_rcvd(mds, handle, req);
 459
 460         err = mds_fs_commit(mds, de_tgt_dir->d_inode, handle);
 461         if (err) {
 462                 CERROR("error on commit: err = %d\n", err);
 463                 if (!rc)
 464                         rc = err;
 465         }
 466         EXIT;
 467
 468 out_link_dchild:
 469         l_dput(dchild);
 470 out_link_de_tgt_dir:
 471         up(&de_tgt_dir->d_inode->i_sem);
 472         l_dput(de_tgt_dir);
 473 out_link_de_src:
 474         l_dput(de_src);
 475 out_link:
 476         req->rq_status = rc;
 477         return 0;
 478 }
 479
 480 static int mds_reint_rename(struct mds_update_record *rec,
 481                             struct ptlrpc_request *req)
 482 {
 483         struct dentry *de_srcdir = NULL;
 484         struct dentry *de_tgtdir = NULL;
 485         struct dentry *de_old = NULL;
 486         struct dentry *de_new = NULL;
 487         struct mds_obd *mds = &req->rq_obd->u.mds;
 488         void *handle;
 489         int rc = 0;
 490         int err;
 491         ENTRY;
 492
 493         de_srcdir = mds_fid2dentry(mds, rec->ur_fid1, NULL);
 494         if (IS_ERR(de_srcdir)) {
 495                 GOTO(out_rename, rc = -ESTALE);
 496         }
 497
 498         de_tgtdir = mds_fid2dentry(mds, rec->ur_fid2, NULL);
 499         if (IS_ERR(de_tgtdir)) {
 500                 GOTO(out_rename_srcdir, rc = -ESTALE);
 501         }
 502
 503         de_old = lookup_one_len(rec->ur_name, de_srcdir, rec->ur_namelen - 1);
 504         if (IS_ERR(de_old)) {
 505                 CERROR("old child lookup error %ld\n", PTR_ERR(de_old));
 506                 GOTO(out_rename_tgtdir, rc = -ESTALE);
 507         }
 508
 509         de_new = lookup_one_len(rec->ur_tgt, de_tgtdir, rec->ur_tgtlen - 1);
 510         if (IS_ERR(de_new)) {
 511                 CERROR("new child lookup error %ld\n", PTR_ERR(de_new));
 512                 GOTO(out_rename_deold, rc = -ESTALE);
 513         }
 514
 515         OBD_FAIL_WRITE(OBD_FAIL_MDS_REINT_RENAME_WRITE,
 516                        de_srcdir->d_inode->i_sb->s_dev);
 517
 518         handle = mds_fs_start(mds, de_tgtdir->d_inode, MDS_FSOP_RENAME);
 519         if (!handle)
 520                 GOTO(out_rename_denew, rc = PTR_ERR(handle));
 521         rc = vfs_rename(de_srcdir->d_inode, de_old, de_tgtdir->d_inode, de_new,
 522                         NULL);
 523
 524         if (!rc)
 525                 rc = mds_update_last_rcvd(mds, handle, req);
 526
 527         err = mds_fs_commit(mds, de_tgtdir->d_inode, handle);
 528         if (err) {
 529                 CERROR("error on commit: err = %d\n", err);
 530                 if (!rc)
 531                         rc = err;
 532         }
 533         EXIT;
 534
 535 out_rename_denew:
 536         l_dput(de_new);
 537 out_rename_deold:
 538         l_dput(de_old);
 539 out_rename_tgtdir:
 540         l_dput(de_tgtdir);
 541 out_rename_srcdir:
 542         l_dput(de_srcdir);
 543 out_rename:
 544         req->rq_status = rc;
 545         return 0;
 546 }
 547
 548 typedef int (*mds_reinter)(struct mds_update_record *, struct ptlrpc_request*);
 549
 550 static mds_reinter reinters[REINT_MAX+1] = {
 551         [REINT_SETATTR]   mds_reint_setattr,
 552         [REINT_CREATE]    mds_reint_create,
 553         [REINT_UNLINK]    mds_reint_unlink,
 554         [REINT_LINK]      mds_reint_link,
 555         [REINT_RENAME]    mds_reint_rename,
 556         [REINT_RECREATE]  mds_reint_recreate,
 557 };
 558
 559 int mds_reint_rec(struct mds_update_record *rec, struct ptlrpc_request *req)
 560 {
 561         int rc, size = sizeof(struct mds_body);
 562
 563         if (rec->ur_opcode < 1 || rec->ur_opcode > REINT_MAX) {
 564                 CERROR("opcode %d not valid\n", rec->ur_opcode);
 565                 rc = req->rq_status = -EINVAL;
 566                 RETURN(rc);
 567         }
 568
 569         rc = lustre_pack_msg(1, &size, NULL, &req->rq_replen, &req->rq_repmsg);
 570         if (rc) {
 571                 CERROR("mds: out of memory\n");
 572                 rc = req->rq_status = -ENOMEM;
 573                 RETURN(rc);
 574         }
 575
 576         rc = reinters[rec->ur_opcode](rec, req);
 577
 578         return rc;
 579 }