mds_fs.c

lustre 1.6.5 source code

/* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
 * vim:expandtab:shiftwidth=8:tabstop=8:
 *
 *  mds/mds_fs.c
 *  Lustre Metadata Server (MDS) filesystem interface code
 *
 *  Copyright (C) 2002, 2003 Cluster File Systems, Inc.
 *   Author: Andreas Dilger <[e-mail]>
 *
 *   This file is part of the Lustre file system, http://www.lustre.org
 *   Lustre is a trademark of Cluster File Systems, Inc.
 *
 *   You may have signed or agreed to another license before downloading
 *   this software.  If so, you are bound by the terms and conditions
 *   of that agreement, and the following does not apply to you.  See the
 *   LICENSE file included with this distribution for more information.
 *
 *   If you did not agree to a different license, then this copy of Lustre
 *   is open source software; you can redistribute it and/or modify it
 *   under the terms of version 2 of the GNU General Public License as
 *   published by the Free Software Foundation.
 *
 *   In either case, Lustre is distributed in the hope that it will be
 *   useful, but WITHOUT ANY WARRANTY; without even the implied warranty
 *   of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 *   license text for more details.
 */

#ifndef EXPORT_SYMTAB
# define EXPORT_SYMTAB
#endif
#define DEBUG_SUBSYSTEM S_MDS

#include <linux/module.h>
#include <linux/kmod.h>
#include <linux/version.h>
#include <linux/sched.h>
#include <lustre_quota.h>
#if (LINUX_VERSION_CODE >= KERNEL_VERSION(2,5,0))
#include <linux/mount.h>
#endif
#include <lustre_mds.h>
#include <obd_class.h>
#include <obd_support.h>
#include <lustre_lib.h>
#include <lustre_fsfilt.h>
#include <lustre_disk.h>
#include <libcfs/list.h>

#include "mds_internal.h"


static int mds_export_stats_init(struct obd_device *obd,
                                 struct obd_export *exp,
                                 lnet_nid_t client_nid)
  {
        int rc, num_stats, newnid;

        rc = lprocfs_exp_setup(exp, client_nid, &newnid);
        if (rc)
                return rc;

        if (client_nid && newnid) {
                struct nid_stat *tmp = exp->exp_nid_stats;
                LASSERT(tmp != NULL);

                num_stats = (sizeof(*obd->obd_type->typ_ops) / sizeof(void *)) +
                             LPROC_MDS_LAST - 1;
                tmp->nid_stats = lprocfs_alloc_stats(num_stats,
                                                     LPROCFS_STATS_FLAG_NOPERCPU);
                if (tmp->nid_stats == NULL)
                        return -ENOMEM;

                lprocfs_init_ops_stats(LPROC_MDS_LAST, tmp->nid_stats);
                rc = lprocfs_register_stats(tmp->nid_proc, "stats",
                                            tmp->nid_stats);
                if (rc)
                        return rc;

                mds_stats_counter_init(tmp->nid_stats);
        }

        return 0;
}

/* Add client data to the MDS.  We use a bitmap to locate a free space
 * in the last_rcvd file if cl_off is -1 (i.e. a new client).
 * Otherwise, we have just read the data from the last_rcvd file and
 * we know its offset.
 *
 * It should not be possible to fail adding an existing client - otherwise
 * mds_init_server_data() callsite needs to be fixed.
 */
int mds_client_add(struct obd_device *obd, struct obd_export *exp,
                   int cl_idx, lnet_nid_t client_nid)
{
        struct mds_obd *mds = &obd->u.mds;
        struct mds_export_data *med = &exp->exp_mds_data;
        unsigned long *bitmap = mds->mds_client_bitmap;
        int new_client = (cl_idx == -1);
        int rc;
        ENTRY;

        LASSERT(bitmap != NULL);
        LASSERTF(cl_idx > -2, "%d\n", cl_idx);

        /* XXX if mcd_uuid were a real obd_uuid, I could use obd_uuid_equals */
        if (!strcmp(med->med_mcd->mcd_uuid, obd->obd_uuid.uuid))
                RETURN(0);

        /* the bitmap operations can handle cl_idx > sizeof(long) * 8, so
         * there's no need for extra complication here
         */
        if (new_client) {
                cl_idx = find_first_zero_bit(bitmap, LR_MAX_CLIENTS);
        repeat:
                if (cl_idx >= LR_MAX_CLIENTS ||
                    OBD_FAIL_CHECK_ONCE(OBD_FAIL_MDS_CLIENT_ADD)) {
                        CERROR("no room for %u clients - fix LR_MAX_CLIENTS\n",
                               cl_idx);
                        return -EOVERFLOW;
                }
                if (test_and_set_bit(cl_idx, bitmap)) {
                        cl_idx = find_next_zero_bit(bitmap, LR_MAX_CLIENTS,
                                                    cl_idx);
                        goto repeat;
                }
        } else {
                if (test_and_set_bit(cl_idx, bitmap)) {
                        CERROR("MDS client %d: bit already set in bitmap!!\n",
                               cl_idx);
                        LBUG();
                }
        }

        CDEBUG(D_INFO, "client at idx %d with UUID '%s' added\n",
               cl_idx, med->med_mcd->mcd_uuid);

        med->med_lr_idx = cl_idx;
        med->med_lr_off = le32_to_cpu(mds->mds_server_data->lsd_client_start) +
                (cl_idx * le16_to_cpu(mds->mds_server_data->lsd_client_size));
        LASSERTF(med->med_lr_off > 0, "med_lr_off = %llu\n", med->med_lr_off);
        mds_export_stats_init(obd, exp, client_nid);

        if (new_client) {
                struct lvfs_run_ctxt saved;
                loff_t off = med->med_lr_off;
                struct file *file = mds->mds_rcvd_filp;
                void *handle;

                push_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL);
                handle = fsfilt_start(obd, file->f_dentry->d_inode,
                                      FSFILT_OP_SETATTR, NULL);
                if (IS_ERR(handle)) {
                        rc = PTR_ERR(handle);
                        CERROR("unable to start transaction: rc %d\n", rc);
                } else {
                        rc = fsfilt_add_journal_cb(obd, 0, handle,
                                                   target_client_add_cb, exp);
                        if (rc == 0) {
                                spin_lock(&exp->exp_lock);
                                exp->exp_need_sync = 1;
                                spin_unlock(&exp->exp_lock);
                        }
                        rc = fsfilt_write_record(obd, file, med->med_mcd,
                                                 sizeof(*med->med_mcd),
                                                 &off, rc /* sync if no cb */);
                        fsfilt_commit(obd, file->f_dentry->d_inode, handle, 0);
                }

                pop_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL);

                if (rc)
                        return rc;
                CDEBUG(D_INFO, "wrote client mcd at idx %u off %llu (len %u)\n",
                       med->med_lr_idx, med->med_lr_off,
                       (unsigned int)sizeof(*med->med_mcd));
        }
        return 0;
}

int mds_client_free(struct obd_export *exp)
{
        struct mds_export_data *med = &exp->exp_mds_data;
        struct mds_obd *mds = &exp->exp_obd->u.mds;
        struct obd_device *obd = exp->exp_obd;
        struct mds_client_data zero_mcd;
        struct lvfs_run_ctxt saved;
        int rc;
        loff_t off;
        ENTRY;

        if (!med->med_mcd)
                RETURN(0);

        /* XXX if mcd_uuid were a real obd_uuid, I could use obd_uuid_equals */
        if (!strcmp(med->med_mcd->mcd_uuid, obd->obd_uuid.uuid))
                GOTO(free, 0);

        CDEBUG(D_INFO, "freeing client at idx %u, offset %lld with UUID '%s'\n",
               med->med_lr_idx, med->med_lr_off, med->med_mcd->mcd_uuid);

        LASSERT(mds->mds_client_bitmap != NULL);

        lprocfs_exp_cleanup(exp);

        off = med->med_lr_off;

        /* Don't clear med_lr_idx here as it is likely also unset.  At worst
         * we leak a client slot that will be cleaned on the next recovery. */
        if (off <= 0) {
                CERROR("%s: client idx %d has offset %lld\n",
                        obd->obd_name, med->med_lr_idx, off);
                GOTO(free, rc = -EINVAL);
        }

        /* Clear the bit _after_ zeroing out the client so we don't
           race with mds_client_add and zero out new clients.*/
        if (!test_bit(med->med_lr_idx, mds->mds_client_bitmap)) {
                CERROR("MDS client %u: bit already clear in bitmap!!\n",
                       med->med_lr_idx);
                LBUG();
        }

        if (!(exp->exp_flags & OBD_OPT_FAILOVER)) {
                memset(&zero_mcd, 0, sizeof zero_mcd);
                push_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL);
                rc = fsfilt_write_record(obd, mds->mds_rcvd_filp, &zero_mcd,
                                         sizeof(zero_mcd), &off,
                                         (!exp->exp_libclient ||
                                          exp->exp_need_sync));
                pop_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL);

                CDEBUG(rc == 0 ? D_INFO : D_ERROR,
                       "zeroing out client %s idx %u in %s rc %d\n",
                       med->med_mcd->mcd_uuid, med->med_lr_idx, LAST_RCVD, rc);
        }

        if (!test_and_clear_bit(med->med_lr_idx, mds->mds_client_bitmap)) {
                CERROR("MDS client %u: bit already clear in bitmap!!\n",
                       med->med_lr_idx);
                LBUG();
        }


        /* Make sure the server's last_transno is up to date. Do this
         * after the client is freed so we know all the client's
         * transactions have been committed. */
        mds_update_server_data(exp->exp_obd, 0);

        EXIT;
 free:
        OBD_FREE(med->med_mcd, sizeof(*med->med_mcd));
        med->med_mcd = NULL;

        return 0;
}

static int mds_server_free_data(struct mds_obd *mds)
{
        OBD_FREE(mds->mds_client_bitmap, LR_MAX_CLIENTS / 8);
        OBD_FREE(mds->mds_server_data, sizeof(*mds->mds_server_data));
        mds->mds_server_data = NULL;

        return 0;
}

static int mds_init_server_data(struct obd_device *obd, struct file *file)
{
        struct mds_obd *mds = &obd->u.mds;
        struct lr_server_data *lsd;
        struct mds_client_data *mcd = NULL;
        loff_t off = 0;
        unsigned long last_rcvd_size = i_size_read(file->f_dentry->d_inode);
        __u64 mount_count;
        int cl_idx, rc = 0;
        ENTRY;

        /* ensure padding in the struct is the correct size */
        LASSERT(offsetof(struct lr_server_data, lsd_padding) +
                sizeof(lsd->lsd_padding) == LR_SERVER_SIZE);
        LASSERT(offsetof(struct mds_client_data, mcd_padding) +
                sizeof(mcd->mcd_padding) == LR_CLIENT_SIZE);

        OBD_ALLOC_WAIT(lsd, sizeof(*lsd));