lustre/mds/mds_fs.c

   1 /* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
   2  * vim:expandtab:shiftwidth=8:tabstop=8:
   3  *
   4  *  linux/mds/mds_fs.c
   5  *
   6  *  Lustre Metadata Server (MDS) filesystem interface code
   7  *
   8  *  Copyright (C) 2002 Cluster File Systems, Inc.
   9  *
  10  *  This code is issued under the GNU General Public License.
  11  *  See the file COPYING in this distribution
  12  *
  13  *  by Andreas Dilger <adilger@clusterfs.com>
  14  *
  15  */
  16
  17 #define EXPORT_SYMTAB
  18 #define DEBUG_SUBSYSTEM S_MDS
  19
  20 #include <linux/module.h>
  21 #include <linux/kmod.h>
  22 #include <linux/lustre_mds.h>
  23 #include <linux/obd_class.h>
  24 #include <linux/obd_support.h>
  25 #include <linux/lustre_lib.h>
  26
  27 LIST_HEAD(mds_fs_types);
  28
  29 struct mds_fs_type {
  30         struct list_head                 mft_list;
  31         struct mds_fs_operations        *mft_ops;
  32         char                            *mft_name;
  33 };
  34
  35 /* This limit is arbitrary, but for now we fit it in 1 page (32k clients) */
  36 #define MDS_MAX_CLIENTS (PAGE_SIZE * 8)
  37 #define MDS_MAX_CLIENT_WORDS (MDS_MAX_CLIENTS / sizeof(unsigned long))
  38
  39 static unsigned long last_rcvd_slots[MDS_MAX_CLIENT_WORDS];
  40
  41 #define LAST_RCVD "last_rcvd"
  42
  43 /* Add client data to the MDS.  We use a bitmap to locate a free space
  44  * in the last_rcvd file if cl_off is -1 (i.e. a new client).
  45  * Otherwise, we have just read the data from the last_rcvd file and
  46  * we know its offset.
  47  */
  48 int mds_client_add(struct mds_obd *mds, struct mds_export_data *med, int cl_off)
  49 {
  50         int new_client = (cl_off == -1);
  51
  52         /* the bitmap operations can handle cl_off > sizeof(long) * 8, so
  53          * there's no need for extra complication here
  54          */
  55         if (new_client) {
  56                 cl_off = find_first_zero_bit(last_rcvd_slots, MDS_MAX_CLIENTS);
  57         repeat:
  58                 if (cl_off >= MDS_MAX_CLIENTS) {
  59                         CERROR("no room for clients - fix MDS_MAX_CLIENTS\n");
  60                         return -ENOMEM;
  61                 }
  62                 if (test_and_set_bit(cl_off, last_rcvd_slots)) {
  63                         CERROR("MDS client %d: found bit is set in bitmap\n",
  64                                cl_off);
  65                         cl_off = find_next_zero_bit(last_rcvd_slots,
  66                                                     MDS_MAX_CLIENTS, cl_off);
  67                         goto repeat;
  68                 }
  69         } else {
  70                 if (test_and_set_bit(cl_off, last_rcvd_slots)) {
  71                         CERROR("MDS client %d: bit already set in bitmap!!\n",
  72                                cl_off);
  73                         LBUG();
  74                 }
  75         }
  76
  77         CDEBUG(D_INFO, "client at offset %d with UUID '%s' added\n",
  78                cl_off, med->med_mcd->mcd_uuid);
  79
  80         med->med_off = cl_off;
  81
  82         if (new_client) {
  83                 struct obd_run_ctxt saved;
  84                 loff_t off = MDS_LR_CLIENT + (cl_off * MDS_LR_SIZE);
  85                 ssize_t written;
  86
  87                 push_ctxt(&saved, &mds->mds_ctxt, NULL);
  88                 written = lustre_fwrite(mds->mds_rcvd_filp,
  89                                                 (char *)med->med_mcd,
  90                                                 sizeof(*med->med_mcd), &off);
  91                 pop_ctxt(&saved, &mds->mds_ctxt, NULL);
  92
  93                 if (written != sizeof(*med->med_mcd)) {
  94                         if (written < 0)
  95                                 RETURN(written);
  96                         RETURN(-EIO);
  97                 }
  98         }
  99         return 0;
 100 }
 101
 102 int mds_client_free(struct obd_export *exp)
 103 {
 104         struct mds_export_data *med = &exp->exp_mds_data;
 105         struct mds_obd *mds = &exp->exp_obd->u.mds;
 106         struct mds_client_data zero_mcd;
 107         struct obd_run_ctxt saved;
 108         int written;
 109         loff_t off;
 110
 111         if (!med->med_mcd)
 112                 RETURN(0);
 113
 114         CDEBUG(D_INFO, "freeing client at offset %d with UUID '%s'\n",
 115                med->med_off, med->med_mcd->mcd_uuid);
 116
 117         if (!test_and_clear_bit(med->med_off, last_rcvd_slots)) {
 118                 CERROR("MDS client %d: bit already clear in bitmap!!\n",
 119                        med->med_off);
 120                 LBUG();
 121         }
 122
 123         off = med->med_off;
 124
 125         memset(&zero_mcd, 0, sizeof zero_mcd);
 126         push_ctxt(&saved, &mds->mds_ctxt, NULL);
 127         written = lustre_fwrite(mds->mds_rcvd_filp, (const char *)&zero_mcd,
 128                                 sizeof zero_mcd, &off);
 129         pop_ctxt(&saved, &mds->mds_ctxt, NULL);
 130
 131         if (written != sizeof zero_mcd) {
 132                 CERROR("error zeroing out client %s off %d in %s: %d\n",
 133                        med->med_mcd->mcd_uuid, med->med_off, LAST_RCVD,
 134                        written);
 135                 LBUG();
 136         } else {
 137                 CDEBUG(D_INFO, "zeroed out disconnecting client %s at off %d\n",
 138                        med->med_mcd->mcd_uuid, med->med_off);
 139         }
 140
 141         OBD_FREE(med->med_mcd, sizeof(*med->med_mcd));
 142
 143         return 0;
 144 }
 145
 146 static int mds_server_free_data(struct mds_obd *mds)
 147 {
 148         OBD_FREE(mds->mds_server_data, sizeof(*mds->mds_server_data));
 149         mds->mds_server_data = NULL;
 150
 151         return 0;
 152 }
 153
 154 static int mds_read_last_rcvd(struct obd_device *obddev, struct file *f)
 155 {
 156         struct mds_obd *mds = &obddev->u.mds;
 157         struct mds_server_data *msd;
 158         struct mds_client_data *mcd = NULL;
 159         loff_t off = 0;
 160         int cl_off;
 161         int max_off = f->f_dentry->d_inode->i_size / sizeof(*mcd);
 162         __u64 last_rcvd = 0;
 163         __u64 last_mount;
 164         int rc = 0;
 165
 166         OBD_ALLOC(msd, sizeof(*msd));
 167         if (!msd)
 168                 RETURN(-ENOMEM);
 169         rc = lustre_fread(f, (char *)msd, sizeof(*msd), &off);
 170
 171         mds->mds_server_data = msd;
 172         if (rc == 0) {
 173                 CERROR("empty MDS %s, new MDS?\n", LAST_RCVD);
 174                 RETURN(0);
 175         }
 176
 177         if (rc != sizeof(*msd)) {
 178                 CERROR("error reading MDS %s: rc = %d\n", LAST_RCVD, rc);
 179                 if (rc > 0) {
 180                         rc = -EIO;
 181                 }
 182                 GOTO(err_msd, rc);
 183         }
 184
 185         /*
 186          * When we do a clean MDS shutdown, we save the last_rcvd into
 187          * the header.  If we find clients with higher last_rcvd values
 188          * then those clients may need recovery done.
 189          */
 190         last_rcvd = le64_to_cpu(msd->msd_last_rcvd);
 191         mds->mds_last_rcvd = last_rcvd;
 192         CDEBUG(D_INODE, "got %Lu for server last_rcvd value\n",
 193                (unsigned long long)last_rcvd);
 194
 195         last_mount = le64_to_cpu(msd->msd_mount_count);
 196         mds->mds_mount_count = last_mount;
 197         CDEBUG(D_INODE, "got %Lu for server last_mount value\n",
 198                (unsigned long long)last_mount);
 199
 200         for (off = MDS_LR_CLIENT, cl_off = 0;
 201              off < max_off;
 202              off += MDS_LR_SIZE, cl_off++) {
 203                 int mount_age;
 204
 205                 if (!mcd) {
 206                         OBD_ALLOC(mcd, sizeof(*mcd));
 207                         if (!mcd)
 208                                 GOTO(err_msd, rc = -ENOMEM);
 209                 }
 210
 211                 rc = lustre_fread(f, (char *)mcd, sizeof(*mcd), &off);
 212                 if (rc != sizeof(*mcd)) {
 213                         CERROR("error reading MDS %s offset %d: rc = %d\n",
 214                                LAST_RCVD, cl_off, rc);
 215                         if (rc > 0)
 216                                 rc = -EIO;
 217                         break;
 218                 }
 219
 220                 if (mcd->mcd_uuid[0] == '\0') {
 221                         CDEBUG(D_INFO, "skipping zeroed client at offset %d\n",
 222                                cl_off);
 223                         continue;
 224                 }
 225
 226                 last_rcvd = le64_to_cpu(mcd->mcd_last_rcvd);
 227
 228                 /* The exports are cleaned up by mds_disconnect, so they
 229                  * need to be set up like real exports also.
 230                  */
 231                 mount_age = last_mount - le64_to_cpu(mcd->mcd_mount_count);
 232                 if (last_rcvd && mount_age < MDS_MOUNT_RECOV) {
 233                         struct obd_export *exp = class_new_export(obddev);
 234                         struct mds_export_data *med;
 235
 236                         if (!exp) {
 237                                 rc = -ENOMEM;
 238                                 break;
 239                         }
 240
 241                         med = &exp->exp_mds_data;
 242                         med->med_mcd = mcd;
 243                         mds_client_add(mds, med, cl_off);
 244                         /* XXX put this in a helper if it gets more complex */
 245                         INIT_LIST_HEAD(&med->med_open_head);
 246                         spin_lock_init(&med->med_open_lock);
 247
 248                         mcd = NULL;
 249                         mds->mds_recoverable_clients++;
 250                         MOD_INC_USE_COUNT;
 251                 } else {
 252                         CDEBUG(D_INFO,
 253                                "discarded client %d, UUID '%s', count %Ld\n",
 254                                cl_off, mcd->mcd_uuid,
 255                                (long long)le64_to_cpu(mcd->mcd_mount_count));
 256                 }
 257
 258                 if (last_rcvd > mds->mds_last_rcvd) {
 259                         CDEBUG(D_OTHER,
 260                                "client at offset %d has last_rcvd = %Lu\n",
 261                                cl_off, (unsigned long long)last_rcvd);
 262                         mds->mds_last_rcvd = last_rcvd;
 263                 }
 264         }
 265
 266         mds->mds_last_committed = mds->mds_last_rcvd;
 267         if (mds->mds_recoverable_clients) {
 268                 CERROR("need recovery: %d recoverable clients, last_rcvd %Lu\n",
 269                        mds->mds_recoverable_clients, mds->mds_last_rcvd);
 270         }
 271
 272         if (mcd)
 273                 OBD_FREE(mcd, sizeof(*mcd));
 274
 275         return 0;
 276
 277 err_msd:
 278         mds_server_free_data(mds);
 279         return rc;
 280 }
 281
 282 static int mds_fs_prep(struct obd_device *obddev)
 283 {
 284         struct mds_obd *mds = &obddev->u.mds;
 285         struct obd_run_ctxt saved;
 286         struct dentry *dentry;
 287         struct file *f;
 288         int rc;
 289
 290         push_ctxt(&saved, &mds->mds_ctxt, NULL);
 291         dentry = simple_mkdir(current->fs->pwd, "ROOT", 0755);
 292         if (IS_ERR(dentry)) {
 293                 rc = PTR_ERR(dentry);
 294                 CERROR("cannot create ROOT directory: rc = %d\n", rc);
 295                 GOTO(err_pop, rc);
 296         }
 297
 298         mds->mds_rootfid.id = dentry->d_inode->i_ino;
 299         mds->mds_rootfid.generation = dentry->d_inode->i_generation;
 300         mds->mds_rootfid.f_type = S_IFDIR;
 301
 302         dput(dentry);
 303
 304         dentry = simple_mkdir(current->fs->pwd, "FH", 0700);
 305         if (IS_ERR(dentry)) {
 306                 rc = PTR_ERR(dentry);
 307                 CERROR("cannot create FH directory: rc = %d\n", rc);
 308                 GOTO(err_pop, rc);
 309         }
 310         /* XXX probably want to hold on to this later... */
 311         dput(dentry);
 312
 313         f = filp_open(LAST_RCVD, O_RDWR | O_CREAT, 0644);
 314         if (IS_ERR(f)) {
 315                 rc = PTR_ERR(f);
 316                 CERROR("cannot open/create %s file: rc = %d\n", LAST_RCVD, rc);
 317                 GOTO(err_pop, rc = PTR_ERR(f));
 318         }
 319         if (!S_ISREG(f->f_dentry->d_inode->i_mode)) {
 320                 CERROR("%s is not a regular file!: mode = %o\n", LAST_RCVD,
 321                        f->f_dentry->d_inode->i_mode);
 322                 GOTO(err_pop, rc = -ENOENT);
 323         }
 324
 325         rc = mds_fs_journal_data(mds, f);
 326         if (rc) {
 327                 CERROR("cannot journal data on %s: rc = %d\n", LAST_RCVD, rc);
 328                 GOTO(err_filp, rc);
 329         }
 330
 331         rc = mds_read_last_rcvd(obddev, f);
 332         if (rc) {
 333                 CERROR("cannot read %s: rc = %d\n", LAST_RCVD, rc);
 334                 GOTO(err_client, rc);
 335         }
 336         mds->mds_rcvd_filp = f;
 337 err_pop:
 338         pop_ctxt(&saved, &mds->mds_ctxt, NULL);
 339
 340         return rc;
 341
 342 err_client:
 343         class_disconnect_all(obddev);
 344 err_filp:
 345         if (filp_close(f, 0))
 346                 CERROR("can't close %s after error\n", LAST_RCVD);
 347         goto err_pop;
 348 }
 349
 350 static struct mds_fs_operations *mds_search_fs_type(const char *name)
 351 {
 352         struct list_head *p;
 353         struct mds_fs_type *type;
 354
 355         /* lock mds_fs_types list */
 356         list_for_each(p, &mds_fs_types) {
 357                 type = list_entry(p, struct mds_fs_type, mft_list);
 358                 if (!strcmp(type->mft_name, name)) {
 359                         /* unlock mds_fs_types list */
 360                         return type->mft_ops;
 361                 }
 362         }
 363         /* unlock mds_fs_types list */
 364         return NULL;
 365 }
 366
 367 int mds_register_fs_type(struct mds_fs_operations *ops, const char *name)
 368 {
 369         struct mds_fs_operations *found;
 370         struct mds_fs_type *type;
 371
 372         if ((found = mds_search_fs_type(name))) {
 373                 if (found != ops) {
 374                         CERROR("different operations for type %s\n", name);
 375                         RETURN(-EEXIST);
 376                 }
 377                 return 0;
 378         }
 379         OBD_ALLOC(type, sizeof(*type));
 380         if (!type)
 381                 RETURN(-ENOMEM);
 382
 383         INIT_LIST_HEAD(&type->mft_list);
 384         type->mft_ops = ops;
 385         type->mft_name = strdup(name);
 386         if (!type->mft_name) {
 387                 OBD_FREE(type, sizeof(*type));
 388                 RETURN(-ENOMEM);
 389         }
 390         MOD_INC_USE_COUNT;
 391         list_add(&type->mft_list, &mds_fs_types);
 392
 393         return 0;
 394 }
 395
 396 void mds_unregister_fs_type(const char *name)
 397 {
 398         struct list_head *p;
 399
 400         /* lock mds_fs_types list */
 401         list_for_each(p, &mds_fs_types) {
 402                 struct mds_fs_type *type;
 403
 404                 type = list_entry(p, struct mds_fs_type, mft_list);
 405                 if (!strcmp(type->mft_name, name)) {
 406                         list_del(p);
 407                         kfree(type->mft_name);
 408                         OBD_FREE(type, sizeof(*type));
 409                         MOD_DEC_USE_COUNT;
 410                         break;
 411                 }
 412         }
 413         /* unlock mds_fs_types list */
 414 }
 415
 416 struct mds_fs_operations *mds_fs_get_ops(char *fstype)
 417 {
 418         struct mds_fs_operations *fs_ops;
 419
 420         if (!(fs_ops = mds_search_fs_type(fstype))) {
 421                 char name[32];
 422                 int rc;
 423
 424                 snprintf(name, sizeof(name) - 1, "mds_%s", fstype);
 425                 name[sizeof(name) - 1] = '\0';
 426
 427                 if ((rc = request_module(name))) {
 428                         fs_ops = mds_search_fs_type(fstype);
 429                         CDEBUG(D_INFO, "Loaded module '%s'\n", name);
 430                         if (!fs_ops)
 431                                 rc = -ENOENT;
 432                 }
 433
 434                 if (rc) {
 435                         CERROR("Can't find MDS fs interface '%s'\n", name);
 436                         RETURN(ERR_PTR(rc));
 437                 }
 438         }
 439         __MOD_INC_USE_COUNT(fs_ops->fs_owner);
 440
 441         return fs_ops;
 442 }
 443
 444 void mds_fs_put_ops(struct mds_fs_operations *fs_ops)
 445 {
 446         __MOD_DEC_USE_COUNT(fs_ops->fs_owner);
 447 }
 448
 449 int mds_fs_setup(struct obd_device *obddev, struct vfsmount *mnt)
 450 {
 451         struct mds_obd *mds = &obddev->u.mds;
 452         int rc;
 453
 454         mds->mds_fsops = mds_fs_get_ops(mds->mds_fstype);
 455         if (IS_ERR(mds->mds_fsops))
 456                 RETURN(PTR_ERR(mds->mds_fsops));
 457
 458         mds->mds_vfsmnt = mnt;
 459
 460         OBD_SET_CTXT_MAGIC(&mds->mds_ctxt);
 461         mds->mds_ctxt.pwdmnt = mnt;
 462         mds->mds_ctxt.pwd = mnt->mnt_root;
 463         mds->mds_ctxt.fs = get_ds();
 464
 465         /*
 466          * Replace the client filesystem delete_inode method with our own,
 467          * so that we can clear the object ID before the inode is deleted.
 468          * The fs_delete_inode method will call cl_delete_inode for us.
 469          * We need to do this for the MDS superblock only, hence we install
 470          * a modified copy of the original superblock method table.
 471          *
 472          * We still assume that there is only a single MDS client filesystem
 473          * type, as we don't have access to the mds struct in delete_inode
 474          * and store the client delete_inode method in a global table.  This
 475          * will only become a problem if/when multiple MDSs are running on a
 476          * single host with different underlying filesystems.
 477          */
 478         OBD_ALLOC(mds->mds_sop, sizeof(*mds->mds_sop));
 479         if (!mds->mds_sop)
 480                 GOTO(out_dec, rc = -ENOMEM);
 481
 482         memcpy(mds->mds_sop, mds->mds_sb->s_op, sizeof(*mds->mds_sop));
 483         mds->mds_fsops->cl_delete_inode = mds->mds_sop->delete_inode;
 484         mds->mds_sop->delete_inode = mds->mds_fsops->fs_delete_inode;
 485         mds->mds_sb->s_op = mds->mds_sop;
 486
 487         rc = mds_fs_prep(obddev);
 488
 489         if (rc)
 490                 GOTO(out_free, rc);
 491
 492         return 0;
 493
 494 out_free:
 495         OBD_FREE(mds->mds_sop, sizeof(*mds->mds_sop));
 496 out_dec:
 497         mds_fs_put_ops(mds->mds_fsops);
 498         return rc;
 499 }
 500
 501 void mds_fs_cleanup(struct obd_device *obddev)
 502 {
 503         struct mds_obd *mds = &obddev->u.mds;
 504
 505         class_disconnect_all(obddev); /* this cleans up client info too */
 506         mds_server_free_data(mds);
 507
 508         OBD_FREE(mds->mds_sop, sizeof(*mds->mds_sop));
 509         mds_fs_put_ops(mds->mds_fsops);
 510 }
 511
 512 EXPORT_SYMBOL(mds_register_fs_type);
 513 EXPORT_SYMBOL(mds_unregister_fs_type);