mds_join.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_join.c
 *  Lustre Metadata join handler file
 *
 *  Copyright (c) 2001-2005 Cluster File Systems, Inc.
 *   Author: Wang Di <[e-mail]>
 *   This file is part of Lustre, http://www.lustre.org.
 *
 *   Lustre is free 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.
 *
 *   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
 *   GNU General Public License for more details.
 *
 *   You should have received a copy of the GNU General Public License
 *   along with Lustre; if not, write to the Free Software
 *   Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
 */

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

#include <linux/fs.h>
#include <linux/jbd.h>
#include <linux/ext3_fs.h>
#include <obd_support.h>
#include <obd_class.h>
#include <obd.h>
#include <lustre_lib.h>
#include <lustre/lustre_idl.h>
#include <lustre_mds.h>
#include <lustre_dlm.h>
#include <lustre_log.h>
#include <lustre_fsfilt.h>
#include <lustre_lite.h>
#include <obd_lov.h>
#include "mds_internal.h"

struct mdsea_cb_data {
    struct llog_handle     *mc_llh;
    struct lov_mds_md      *mc_lmm;
    struct lov_mds_md_join *mc_lmm_join;
    __u64                   mc_offset;
    __u64                   mc_headfile_sz;
};

static int mdsea_iterate(struct llog_handle *llh_tail, llog_cb_t cb,
                         void *cbdata)
{
    return llog_process(llh_tail, cb, cbdata, NULL);
}

static int mds_insert_join_lmm(struct llog_handle *llh,
                               struct lov_mds_md *lmm,
                               __u64 start, __u64 len,
                               struct lov_mds_md_join *lmmj)
{
        struct llog_rec_hdr rec;
        struct mds_extent_desc *med;
        int sz_med, rc;
        ENTRY;


        sz_med = lov_mds_md_size(le32_to_cpu(lmm->lmm_stripe_count));
        sz_med += 2 * sizeof(__u64);
        sz_med = size_round(sz_med);

        rec.lrh_len  = cpu_to_le32(sz_med);
        rec.lrh_type = cpu_to_le32(LLOG_JOIN_REC);

        CDEBUG(D_INFO, "insert extent "LPU64":"LPU64" lmm \n", start, len);

        OBD_ALLOC(med, sz_med);
        if (med == NULL)
                RETURN(-ENOMEM);

        med->med_start = start;
        med->med_len = len;
        memcpy(&med->med_lmm, lmm,
                lov_mds_md_size(le32_to_cpu(lmm->lmm_stripe_count)));

        rc = llog_write_rec(llh, &rec, NULL, 0, med, -1);
        OBD_FREE(med, sz_med);

        if (lmmj) {
                /*modify lmmj for join stripe info*/
                lmmj->lmmj_md.lmm_stripe_count += lmm->lmm_stripe_count;
                lmmj->lmmj_extent_count ++;
        }

        RETURN(rc);
}

static int mdsea_append_extent(struct llog_handle *llh_tail,
                               struct llog_rec_hdr *rec_in_tail,
                               struct mdsea_cb_data *cbdata)
{
        struct mds_extent_desc *med =
                        &((struct llog_array_rec *)rec_in_tail)->lmr_med;
        int rc;
        ENTRY;

        CDEBUG(D_INODE, "insert lmm extent: "LPU64":"LPU64" \n",
               med->med_start, med->med_len);
        rc = mds_insert_join_lmm(cbdata->mc_llh, &med->med_lmm,
                                 med->med_start + cbdata->mc_headfile_sz,
                                 med->med_len, cbdata->mc_lmm_join);
        if (rc) {
                CERROR("error %d insert the lmm \n", rc);
                RETURN(rc);
        }
        RETURN(LLOG_DEL_RECORD);
}

static void mds_init_stripe_join(struct lov_mds_md_join *lmmj,
                                 struct lov_mds_md *lmm,
                                 struct llog_logid  *logid)
{
        lmmj->lmmj_md.lmm_magic = cpu_to_le32(LOV_MAGIC_JOIN);
        lmmj->lmmj_md.lmm_object_id = lmm->lmm_object_id;
        lmmj->lmmj_md.lmm_object_gr = lmm->lmm_object_gr;
        lmmj->lmmj_md.lmm_pattern = lmm->lmm_pattern;
        lmmj->lmmj_md.lmm_stripe_size = lmm->lmm_stripe_size;
        lmmj->lmmj_md.lmm_stripe_count = 0;
        lmmj->lmmj_extent_count = 0;
        lmmj->lmmj_array_id = *logid;
}

static int mdsea_cancel_last_extent(struct llog_handle *llh_tail,
                                    struct llog_rec_hdr *rec_in_tail,
                                    struct mdsea_cb_data *cbdata)
{
        struct mds_extent_desc *med =
                        &((struct llog_array_rec *)rec_in_tail)->lmr_med;

        CDEBUG(D_INODE, "extent: "LPU64":"LPU64" \n",  med->med_start,
               med->med_len);

        LASSERTF(cbdata->mc_offset == med->med_start,
                 "A hole in the extent "LPU64"--"LPU64"\n",
                 cbdata->mc_offset, med->med_start);

        if (med->med_len != -1)
                cbdata->mc_offset = med->med_start + med->med_len;

        if (med->med_start > cbdata->mc_headfile_sz || (med->med_len == -1)) {
                CDEBUG(D_INFO, "del rec offset"LPU64", head size "LPU64" \n",
                       med->med_start, cbdata->mc_headfile_sz);
                if (!cbdata->mc_lmm) {
                        int stripe = le32_to_cpu(med->med_lmm.lmm_stripe_count);
                        OBD_ALLOC(cbdata->mc_lmm, lov_mds_md_size(stripe));
                        if (!cbdata->mc_lmm)
                                RETURN(-ENOMEM);
                        memcpy(cbdata->mc_lmm, &med->med_lmm,
                               lov_mds_md_size(stripe));
                }
                RETURN(LLOG_DEL_RECORD);
        }
        RETURN(0);
}

static int  mds_adjust_last_extent(struct llog_handle *llh_head,
                                   __u64 head_size)
{
        struct mdsea_cb_data  *cbdata;
        int    rc;
        ENTRY;

        OBD_ALLOC_PTR(cbdata);

        if (!cbdata)
                RETURN(-ENOMEM);

        cbdata->mc_headfile_sz = head_size;
        /*Find the last extent and cancel the record in the lmm*/
        rc = mdsea_iterate(llh_head, (llog_cb_t)mdsea_cancel_last_extent,
                           cbdata);

        if (rc) {
                CERROR("can not find the last extent rc=%d\n", rc);
                GOTO(exit, rc);
        }

        LASSERT(cbdata->mc_lmm);

        CDEBUG(D_INODE, "insert lmm extent: "LPU64":"LPU64" \n",
               cbdata->mc_offset, (head_size - cbdata->mc_offset));

        rc = mds_insert_join_lmm(llh_head, cbdata->mc_lmm,
                                 cbdata->mc_offset,
                                 (head_size - cbdata->mc_offset),
                                 NULL);
        if (rc)
                CERROR("error insert the lmm rc %d \n", rc);
exit:
        if (cbdata && cbdata->mc_lmm)
                OBD_FREE(cbdata->mc_lmm,
                         lov_mds_md_size(cbdata->mc_lmm->lmm_stripe_count));
        if (cbdata)
                OBD_FREE_PTR(cbdata);

        RETURN(rc);
}

static void mds_finish_join(struct mds_obd *mds, struct ptlrpc_request *req,
                           struct inode *inode, struct lov_mds_md_join *lmmj)
{
        struct mds_body *body = (struct mds_body *)
                                lustre_msg_buf(req->rq_repmsg, 1, 0);
        int max_cookiesize = lmmj->lmmj_md.lmm_stripe_count *
                                sizeof(struct llog_cookie);
        int max_easize = sizeof(*lmmj);

        CDEBUG(D_INFO, "change the max md size from %d to "LPSZ"\n",
               mds->mds_max_mdsize, sizeof(*lmmj));

        if (mds->mds_max_mdsize < max_easize || 
            mds->mds_max_cookiesize < max_cookiesize) {
                body->max_mdsize = mds->mds_max_mdsize > max_easize ?
                                   mds->mds_max_mdsize : max_easize;
                mds->mds_max_mdsize = body->max_mdsize;
                body->max_cookiesize = mds->mds_max_cookiesize > max_cookiesize?
                                   mds->mds_max_cookiesize : max_cookiesize;
                mds->mds_max_cookiesize = body->max_cookiesize;
                body->valid |= OBD_MD_FLMODEASIZE;
        }

        if (body->valid & OBD_MD_FLMODEASIZE)
                CDEBUG(D_INODE, "updating max_mdsize/max_cookiesize: %d/%d\n",
                       mds->mds_max_mdsize, mds->mds_max_cookiesize);

        mds_pack_inode2fid(&body->fid1, inode);
        mds_pack_inode2body(body, inode);
}

static int mds_join_unlink_tail_inode(struct mds_update_record *rec,
                                      struct ptlrpc_request *req,
                                      struct mds_rec_join *join_rec,
                                      struct lov_mds_md *tail_lmm,
                                      int lmm_size, struct dentry *dchild,
                                      void **handle,struct lustre_handle *lockh)
{
        struct mds_obd *mds = mds_req2mds(req);
        struct obd_device *obd = req->rq_export->exp_obd;
        struct inode *tail_inode, *head_inode;
        struct dentry *de_tailparent = NULL, *de_tail = NULL, *de_head = NULL;
        struct lustre_handle dlm_handles[4] = {{0}, {0}, {0}, {0}};