mds_lov.c

lustre 1.6.5 source code

/* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
 * vim:expandtab:shiftwidth=8:tabstop=8:
 *
 *  linux/mds/mds_lov.c
 *  Lustre Metadata Server (mds) handling of striped file data
 *
 *  Copyright (C) 2001-2003 Cluster File Systems, Inc.
 *   Author: Peter Braam <[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 <lustre_mds.h>
#include <lustre/lustre_idl.h>
#include <obd_class.h>
#include <obd_lov.h>
#include <lustre_lib.h>
#include <lustre_fsfilt.h>

#include "mds_internal.h"

static void mds_lov_dump_objids(const char *label, struct obd_device *obd)
{
        struct mds_obd *mds = &obd->u.mds;
        unsigned int i=0, j;

        CDEBUG(D_INFO, "dump from %s\n", label);
        if (mds->mds_lov_page_dirty == NULL) {
                CERROR("NULL bitmap!\n");
                GOTO(skip_bitmap, i);
        }

        for(i=0; i < ((mds->mds_lov_page_dirty->size/BITS_PER_LONG)+1) ;i++)
                CDEBUG(D_INFO, "%u - %lx\n", i, mds->mds_lov_page_dirty->data[i]);
skip_bitmap:
        if (mds->mds_lov_page_array == NULL) {
                CERROR("not init page array!\n");
                GOTO(skip_array, i);

        }
        for(i=0; i < MDS_LOV_OBJID_PAGES_COUNT ;i++) {
                obd_id *data = mds->mds_lov_page_array[i];

                if (data == NULL)
                        continue;

                for(j=0; j < OBJID_PER_PAGE(); j++) {
                        if (data[j] == 0)
                                continue;
                        CDEBUG(D_INFO,"objid page %u idx %u - "LPU64" \n",
                               i,j,data[j]);
                }
        }
skip_array:
        EXIT;
}

int mds_lov_init_objids(struct obd_device *obd)
{
        struct mds_obd *mds = &obd->u.mds;
        int size = MDS_LOV_OBJID_PAGES_COUNT*sizeof(void *);
        struct file *file;
        int rc;
        ENTRY;

        CLASSERT(((MDS_LOV_ALLOC_SIZE % sizeof(obd_id)) == 0));

        mds->mds_lov_page_dirty = ALLOCATE_BITMAP(MDS_LOV_OBJID_PAGES_COUNT);
        if (mds->mds_lov_page_dirty == NULL)
                RETURN(-ENOMEM);


        OBD_ALLOC(mds->mds_lov_page_array, size);
        if (mds->mds_lov_page_array == NULL)
                GOTO(err_free_bitmap, rc = -ENOMEM);

        /* open and test the lov objd file */
        file = filp_open(LOV_OBJID, O_RDWR | O_CREAT, 0644);
        if (IS_ERR(file)) {
                rc = PTR_ERR(file);
                CERROR("cannot open/create %s file: rc = %d\n", LOV_OBJID, rc);
                GOTO(err_free, rc = PTR_ERR(file));
        }
        if (!S_ISREG(file->f_dentry->d_inode->i_mode)) {
                CERROR("%s is not a regular file!: mode = %o\n", LOV_OBJID,
                       file->f_dentry->d_inode->i_mode);
                GOTO(err_open, rc = -ENOENT);
        }
        mds->mds_lov_objid_filp = file;

        RETURN (0);
err_open:
        if (filp_close((struct file *)file, 0))
                CERROR("can't close %s after error\n", LOV_OBJID);
err_free:
        OBD_FREE(mds->mds_lov_page_array, size);
err_free_bitmap:
        FREE_BITMAP(mds->mds_lov_page_dirty);

        RETURN(rc);
}
EXPORT_SYMBOL(mds_lov_init_objids);

void mds_lov_destroy_objids(struct obd_device *obd)
{
        struct mds_obd *mds = &obd->u.mds;
        int i, rc;
        ENTRY;

        if (mds->mds_lov_page_array != NULL) {
                for(i=0;i<MDS_LOV_OBJID_PAGES_COUNT;i++) {
                        obd_id *data = mds->mds_lov_page_array[i];
                        if (data != NULL)
                                OBD_FREE(data, MDS_LOV_ALLOC_SIZE);
                }
                OBD_FREE(mds->mds_lov_page_array,
                         MDS_LOV_OBJID_PAGES_COUNT*sizeof(void *));
        }

        if (mds->mds_lov_objid_filp) {
                rc = filp_close((struct file *)mds->mds_lov_objid_filp, 0);
                mds->mds_lov_objid_filp = NULL;
                if (rc)
                        CERROR("%s file won't close, rc=%d\n", LOV_OBJID, rc);
        }

        FREE_BITMAP(mds->mds_lov_page_dirty);
        EXIT;
}
EXPORT_SYMBOL(mds_lov_destroy_objids);

void mds_lov_update_objids(struct obd_device *obd, struct lov_mds_md *lmm)
{
        struct mds_obd *mds = &obd->u.mds;
        int j;
        ENTRY;

        /* if we create file without objects - lmm is NULL */
        if (lmm == NULL)
                return;

        for (j = 0; j < le32_to_cpu(lmm->lmm_stripe_count); j++) {
                int i = le32_to_cpu(lmm->lmm_objects[j].l_ost_idx);
                obd_id id = le64_to_cpu(lmm->lmm_objects[j].l_object_id);
                int page = i / OBJID_PER_PAGE();
                int idx = i % OBJID_PER_PAGE();
                obd_id *data = mds->mds_lov_page_array[page];

                CDEBUG(D_INODE,"update last object for ost %d"
                       " - new "LPU64" old "LPU64"\n", i, id, data[idx]);
                if (id > data[idx]) {
                        data[idx] = id;
                        cfs_bitmap_set(mds->mds_lov_page_dirty, page);
                }
        }
        EXIT;
}
EXPORT_SYMBOL(mds_lov_update_objids);

static int mds_lov_read_objids(struct obd_device *obd)
{
        struct mds_obd *mds = &obd->u.mds;
        loff_t off = 0;
        int i, rc = 0, count = 0, page = 0;
        size_t size;
        ENTRY;

        /* Read everything in the file, even if our current lov desc
           has fewer targets. Old targets not in the lov descriptor
           during mds setup may still have valid objids. */
        size = i_size_read(mds->mds_lov_objid_filp->f_dentry->d_inode);
        if (size == 0)
                RETURN(0);

        page = (size/(OBJID_PER_PAGE()*sizeof(obd_id)))+1;
        CDEBUG(D_INFO, "file size %d pages %d\n", (int)size, page);
        for(i=0; i < page; i++) {
                obd_id *data =  mds->mds_lov_page_array[i];
                loff_t off_old = off;

                LASSERT(data == NULL);
                OBD_ALLOC(data, MDS_LOV_ALLOC_SIZE);
                if (data == NULL)
                        GOTO(out, rc = -ENOMEM);

                mds->mds_lov_page_array[i] = data;

                rc = fsfilt_read_record(obd, mds->mds_lov_objid_filp, data,
                                        OBJID_PER_PAGE()*sizeof(obd_id), &off);
                if (rc < 0) {
                        CERROR("Error reading objids %d\n", rc);
                        GOTO(out, rc);
                }
                if (off == off_old)
                        break; /* eof */

                count += (off-off_old)/sizeof(obd_id);
        }
        mds->mds_lov_objid_count = count;
        if (count) {
                count --;
                mds->mds_lov_objid_lastpage = count / OBJID_PER_PAGE();
                mds->mds_lov_objid_lastidx = count % OBJID_PER_PAGE();
        }
        CDEBUG(D_INFO, "Read %u - %u %u objid\n", count,
               mds->mds_lov_objid_lastpage, mds->mds_lov_objid_lastidx);
out:
        mds_lov_dump_objids("read",obd);

        RETURN(rc);
}

int mds_lov_write_objids(struct obd_device *obd)
{
        struct mds_obd *mds = &obd->u.mds;
        int i, rc = 0;
        ENTRY;

        if (cfs_bitmap_check_empty(mds->mds_lov_page_dirty))
                RETURN(0);

        mds_lov_dump_objids("write", obd);

        cfs_foreach_bit(mds->mds_lov_page_dirty, i) {
                obd_id *data =  mds->mds_lov_page_array[i];
                unsigned int size = OBJID_PER_PAGE()*sizeof(obd_id);
                loff_t off = i * size;

                LASSERT(data != NULL);

                /* check for particaly filled last page */
                if (i == mds->mds_lov_objid_lastpage)
                        size = (mds->mds_lov_objid_lastidx + 1) * sizeof(obd_id);

                rc = fsfilt_write_record(obd, mds->mds_lov_objid_filp, data,
                                         size, &off, 0);
                if (rc < 0)
                        break;
                cfs_bitmap_clear(mds->mds_lov_page_dirty, i);
        }
        if (rc >= 0)
                rc = 0;

        RETURN(rc);
}
EXPORT_SYMBOL(mds_lov_write_objids);

static int mds_lov_get_objid(struct obd_device * obd,
                             __u32 idx)
{
        struct mds_obd *mds = &obd->u.mds;
        unsigned int page;
        unsigned int off;
        obd_id *data;
        int rc = 0;
        ENTRY;

        page = idx / OBJID_PER_PAGE();
        off = idx % OBJID_PER_PAGE();
        data = mds->mds_lov_page_array[page];
        if (data == NULL) {
                OBD_ALLOC(data, MDS_LOV_ALLOC_SIZE);
                if (data == NULL)
                        GOTO(out, rc = -ENOMEM);

                mds->mds_lov_page_array[page] = data;
        }

        if (data[off] == 0) {
                /* We never read this lastid; ask the osc */
                struct obd_id_info lastid;
                __u32 size = sizeof(lastid);

                lastid.idx = idx;
                lastid.data = &data[off];
                rc = obd_get_info(mds->mds_osc_exp, sizeof(KEY_LAST_ID),
                                  KEY_LAST_ID, &size, &lastid);
                if (rc)
                        GOTO(out, rc);

                if (idx > mds->mds_lov_objid_count) {
                        mds->mds_lov_objid_count = idx;
                        mds->mds_lov_objid_lastpage = page;
                        mds->mds_lov_objid_lastidx = off;
                }
                cfs_bitmap_set(mds->mds_lov_page_dirty, page);
        }
        CDEBUG(D_INFO, "idx %d - %p - %d/%d - "LPU64"\n",
               idx, data, page, off, data[off]);
out:
        RETURN(rc);
}

int mds_lov_clear_orphans(struct mds_obd *mds, struct obd_uuid *ost_uuid)
{
        int rc;
        struct obdo oa;
        struct obd_trans_info oti = {0};
        struct lov_stripe_md  *empty_ea = NULL;
        ENTRY;

        LASSERT(mds->mds_lov_page_array != NULL);

        /* This create will in fact either create or destroy:  If the OST is
         * missing objects below this ID, they will be created.  If it finds
         * objects above this ID, they will be removed. */
        memset(&oa, 0, sizeof(oa));
        oa.o_flags = OBD_FL_DELORPHAN;
        oa.o_valid = OBD_MD_FLFLAGS | OBD_MD_FLGROUP;
        if (ost_uuid != NULL)
                oti.oti_ost_uuid = ost_uuid;

        rc = obd_create(mds->mds_osc_exp, &oa, &empty_ea, &oti);

        RETURN(rc);
}

/* for one target */
static int mds_lov_set_one_nextid(struct obd_device * obd, __u32 idx, obd_id *id)
{
        struct mds_obd *mds = &obd->u.mds;
        int rc;
        struct obd_id_info info;
        ENTRY;

        LASSERT(!obd->obd_recovering);

        /* obd->obd_dev_sem must be held so mds_lov_objids doesn't change */
        LASSERT_SEM_LOCKED(&obd->obd_dev_sem);

        info.idx = idx;
        info.data = id;

        rc = obd_set_info_async(mds->mds_osc_exp, sizeof(KEY_NEXT_ID),
                                KEY_NEXT_ID, sizeof(info), &info, NULL);
        if (rc)
                CERROR ("%s: mds_lov_set_nextid failed (%d)\n",
                        obd->obd_name, rc);

        RETURN(rc);
}

static __u32 mds_lov_get_idx(struct obd_export *lov,
                             struct obd_uuid *ost_uuid)
{
        int rc;