From: shadow Date: Thu, 29 Nov 2007 12:37:33 +0000 (+0000) Subject: Fixes some scability and access to not inited memory problems X-Git-Tag: v1_7_0_51~448 X-Git-Url: https://git.whamcloud.com/?p=fs%2Flustre-release.git;a=commitdiff_plain;h=5b26bbe0f452713ddd54c75781153adc54a4dafb Fixes some scability and access to not inited memory problems in work with lov objdid file. b=12702 i=umka i=nathan --- diff --git a/lnet/include/libcfs/Makefile.am b/lnet/include/libcfs/Makefile.am index 5bdb653..f62d2c77 100644 --- a/lnet/include/libcfs/Makefile.am +++ b/lnet/include/libcfs/Makefile.am @@ -6,4 +6,4 @@ DIST_SUBDIRS := $(SUBDIRS) EXTRA_DIST := curproc.h kp30.h libcfs.h list.h lltrace.h \ portals_utils.h types.h user-lock.h user-prim.h user-time.h \ - user-tcpip.h + user-tcpip.h bitmap.h diff --git a/lnet/include/libcfs/bitmap.h b/lnet/include/libcfs/bitmap.h new file mode 100644 index 0000000..b9e6e18 --- /dev/null +++ b/lnet/include/libcfs/bitmap.h @@ -0,0 +1,82 @@ +/* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*- + * vim:expandtab:shiftwidth=8:tabstop=8: + * + * Copyright (C) 2007 Cluster File Systems, Inc. + * + * 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 _LIBCFS_BITMAP_H_ +#define _LIBCFS_BITMAP_H_ + + +typedef struct { + int size; + unsigned long data[0]; +} bitmap_t; + +#define BITMAP_SIZE(nbits) (((nbits/BITS_PER_LONG)+1)*sizeof(long)+sizeof(bitmap_t)) + +static inline +bitmap_t *ALLOCATE_BITMAP(int size) +{ + bitmap_t *ptr; + + OBD_ALLOC(ptr, BITMAP_SIZE(size)); + if (ptr == NULL) + RETURN(ptr); + + ptr->size = size; + memset(ptr->data, 0, BITMAP_SIZE(size)); + + RETURN (ptr); +} + +#define FREE_BITMAP(ptr) OBD_FREE(ptr, BITMAP_SIZE(ptr->size)) + +static inline +void bitmap_set(bitmap_t *bitmap, int nbit) +{ + set_bit(nbit, bitmap->data); +} + +static inline +void bitmap_clear(bitmap_t *bitmap, int nbit) +{ + clear_bit(nbit, bitmap->data); +} + +static inline +int bitmap_check(bitmap_t *bitmap, int nbit) +{ + int pos = nbit % BITS_PER_LONG; + return test_bit(pos, bitmap->data+(nbit/BITS_PER_LONG)); +} + +/* return 0 is bitmap has none set bits */ +static inline +int bitmap_check_empty(bitmap_t *bitmap) +{ + return find_first_bit(bitmap->data, bitmap->size) == bitmap->size; +} + + +#define foreach_bit(bitmap, pos) \ + for((pos)=find_first_bit((bitmap)->data, bitmap->size); \ + (pos) < (bitmap)->size; \ + (pos) = find_next_bit((bitmap)->data, (bitmap)->size, (pos))) + +#endif diff --git a/lnet/include/libcfs/user-bitops.h b/lnet/include/libcfs/user-bitops.h new file mode 100644 index 0000000..f6034c3 --- /dev/null +++ b/lnet/include/libcfs/user-bitops.h @@ -0,0 +1,38 @@ +/* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*- + * vim:expandtab:shiftwidth=8:tabstop=8: + * + * Copyright (C) 2004 Cluster File Systems, Inc. + * Author: Nikita Danilov + * + * 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. + * + * Implementation of portable time API for user-level. + * + */ + +#ifndef __LIBCFS_USER_BITOPS_H__ +#define __LIBCFS_USER_BITOPS_H__ + +unsigned long find_next_bit(const unsigned int *addr, + unsigned long size, unsigned long offset); + +unsigned long find_next_zero_bit(const unsigned int *addr, + unsigned long size, unsigned long offset); + +#define find_first_bit(addr,size) (find_next_bit((addr),(size),0)) +#define find_first_zero_bit(addr,size) (find_next_zero_bit((addr),(size),0)) + +#endif diff --git a/lnet/libcfs/autoMakefile.am b/lnet/libcfs/autoMakefile.am index 628bd61..b20cd33 100644 --- a/lnet/libcfs/autoMakefile.am +++ b/lnet/libcfs/autoMakefile.am @@ -11,7 +11,7 @@ DIST_SUBDIRS := $(SUBDIRS) if LIBLUSTRE noinst_LIBRARIES= libcfs.a -libcfs_a_SOURCES= debug.c user-prim.c user-lock.c user-tcpip.c +libcfs_a_SOURCES= debug.c user-prim.c user-lock.c user-tcpip.c user-bitops.c libcfs_a_CPPFLAGS = $(LLCPPFLAGS) libcfs_a_CFLAGS = $(LLCFLAGS) endif diff --git a/lnet/libcfs/user-bitops.c b/lnet/libcfs/user-bitops.c new file mode 100644 index 0000000..2c252e8 --- /dev/null +++ b/lnet/libcfs/user-bitops.c @@ -0,0 +1,100 @@ +/* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*- + * vim:expandtab:shiftwidth=8:tabstop=8: + * + * Copyright (C) 2007 Cluster File Systems, Inc. + * + * 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 __KERNEL__ + +#include +#include + +#include /* for ffs - confirm POSIX */ + +#define BITS_PER_WORD 32 +#define OFF_BY_START(start) ((start)/BITS_PER_WORD) + +unsigned long find_next_bit(const unsigned int *addr, + unsigned long size, unsigned long offset) +{ + uint32_t *word, *last; + unsigned int first_bit, bit, base; + + word = addr + OFF_BY_START(offset); + last = addr + OFF_BY_START(size-1); + first_bit = offset % BITS_PER_WORD; + base = offset - first_bit; + + if (offset >= size) + return size; + if (first_bit != 0) { + int tmp = (*word++) & (~0UL << first_bit); + bit = ffs(tmp); + if (bit < BITS_PER_WORD) + goto found; + word++; + base += BITS_PER_WORD; + } + while (word <= last) { + if (*word != 0ul) { + bit = ffs(*word); + goto found; + } + word++; + base += BITS_PER_WORD; + } + return size; +found: + return base + bit; +} + +unsigned long find_next_zero_bit(const unsigned int *addr, + unsigned long size, unsigned long offset) +{ + uint32_t *word, *last; + unsigned int first_bit, bit, base; + + word = addr + OFF_BY_START(offset); + last = addr + OFF_BY_START(size-1); + first_bit = offset % BITS_PER_WORD; + base = offset - first_bit; + + if (offset >= size) + return size; + if (first_bit != 0) { + int tmp = (*word++) & (~0UL << first_bit); + bit = ffs(~tmp); + if (bit < BITS_PER_WORD) + goto found; + word++; + base += BITS_PER_WORD; + } + while (word <= last) { + if (*word != ~0ul) { + bit = ffs(*word); + goto found; + } + word++; + base += BITS_PER_WORD; + } + return size; +found: + return base + bit; +} + +#endif diff --git a/lustre/ChangeLog b/lustre/ChangeLog index 821a4f5..2cb5668 100644 --- a/lustre/ChangeLog +++ b/lustre/ChangeLog @@ -12,6 +12,13 @@ * Recommended e2fsprogs version: 1.40.2-cfs4 * Note that reiserfs quotas are disabled on SLES 10 in this kernel. + +Severity : enhancement +Bugzilla : 12702 +Description: lost problems with lov objid file +Details : Fixes some scability and access to not inited memory problems + in work with lov objdid file. + Severity : major Frequency : always Bugzilla : 14270 diff --git a/lustre/include/liblustre.h b/lustre/include/liblustre.h index a9c3a91..3e773b9 100644 --- a/lustre/include/liblustre.h +++ b/lustre/include/liblustre.h @@ -65,6 +65,7 @@ #include #include +#include #include #include diff --git a/lustre/include/lustre_mds.h b/lustre/include/lustre_mds.h index f2ecea8..5655d5b 100644 --- a/lustre/include/lustre_mds.h +++ b/lustre/include/lustre_mds.h @@ -93,8 +93,7 @@ int mds_log_op_setattr(struct obd_device *obd, __u32 uid, __u32 gid, struct llog_cookie *logcookies, int cookies_size); int mds_lov_write_objids(struct obd_device *obd); -void mds_lov_update_objids(struct obd_device *obd, obd_id *ids); -void mds_objids_from_lmm(obd_id *, struct lov_mds_md *, struct lov_desc *); +void mds_lov_update_objids(struct obd_device *obd, struct lov_mds_md *lmm); /* ioctls for trying requests */ #define IOC_REQUEST_TYPE 'f' diff --git a/lustre/include/obd.h b/lustre/include/obd.h index 0950330..ef4c7ff 100644 --- a/lustre/include/obd.h +++ b/lustre/include/obd.h @@ -36,6 +36,9 @@ #include #include +#include + + #define MAX_OBD_DEVICES 8192 /* this is really local to the OSC */ @@ -524,11 +527,17 @@ struct mds_obd { struct obd_export *mds_osc_exp; /* XXX lov_exp */ struct lov_desc mds_lov_desc; __u32 mds_id; - obd_id *mds_lov_objids; - int mds_lov_objids_size; - __u32 mds_lov_objids_in_file; - int mds_lov_nextid_set; + + /* mark pages dirty for write. */ + bitmap_t *mds_lov_page_dirty; + /* array for store pages with obd_id */ + void **mds_lov_page_array; + /* file for store objid */ struct file *mds_lov_objid_filp; + __u32 mds_lov_objid_count; + __u32 mds_lov_objid_lastpage; + __u32 mds_lov_objid_lastidx; + struct file *mds_health_check_filp; unsigned long *mds_client_bitmap; // struct upcall_cache *mds_group_hash; @@ -536,9 +545,7 @@ struct mds_obd { struct lustre_quota_info mds_quota_info; struct semaphore mds_qonoff_sem; struct semaphore mds_health_sem; - unsigned long mds_lov_objids_valid:1, - mds_lov_objids_dirty:1, - mds_fl_user_xattr:1, + unsigned long mds_fl_user_xattr:1, mds_fl_acl:1, mds_evict_ost_nids:1, mds_fl_cfglog:1, @@ -554,6 +561,25 @@ struct mds_obd { struct lustre_capa_key *mds_capa_keys; }; +/* lov objid */ +extern __u32 mds_max_ost_index; + +#define MDS_LOV_ALLOC_SIZE (CFS_PAGE_SIZE) + +#define OBJID_PER_PAGE() (MDS_LOV_ALLOC_SIZE / sizeof(obd_id)) + +#define MDS_LOV_OBJID_PAGES_COUNT (mds_max_ost_index/OBJID_PER_PAGE()) + +extern int mds_lov_init_objids(struct obd_device *obd); +extern void mds_lov_destroy_objids(struct obd_device *obd); + +struct obd_id_info { + __u32 idx; + obd_id *data; +}; + +/* */ + struct echo_obd { struct obdo eo_oa; spinlock_t eo_lock; @@ -715,7 +741,6 @@ struct niobuf_local { struct obd_trans_info { __u64 oti_transno; __u64 oti_xid; - __u64 *oti_objid; /* Only used on the server side for tracking acks. */ struct oti_req_ack_lock { struct lustre_handle lock; diff --git a/lustre/llite/file.c b/lustre/llite/file.c index e6c8d39..dfaddb8 100644 --- a/lustre/llite/file.c +++ b/lustre/llite/file.c @@ -1717,7 +1717,6 @@ static int ll_lov_recreate_obj(struct inode *inode, struct file *file, obdo_from_inode(oa, inode, OBD_MD_FLTYPE | OBD_MD_FLATIME | OBD_MD_FLMTIME | OBD_MD_FLCTIME); - oti.oti_objid = NULL; memcpy(lsm2, lsm, lsm_size); rc = obd_create(exp, oa, &lsm2, &oti); diff --git a/lustre/lov/lov_obd.c b/lustre/lov/lov_obd.c index 7fb2f21..96d6e88 100644 --- a/lustre/lov/lov_obd.c +++ b/lustre/lov/lov_obd.c @@ -2368,16 +2368,17 @@ static int lov_get_info(struct obd_export *exp, __u32 keylen, dump_lsm(D_ERROR, data->lsm); GOTO(out, rc = -ENXIO); } else if (KEY_IS("last_id")) { - obd_id *ids = val; - unsigned int size = sizeof(obd_id); - for (i = 0; i < lov->desc.ld_tgt_count; i++) { - if (!lov->lov_tgts[i] || !lov->lov_tgts[i]->ltd_active) - continue; - rc = obd_get_info(lov->lov_tgts[i]->ltd_exp, - keylen, key, &size, ids + i); - if (rc != 0) - GOTO(out, rc); - } + struct obd_id_info *info = val; + int size = sizeof(obd_id); + struct lov_tgt_desc *tgt; + + LASSERT(*vallen == sizeof(struct obd_id_info)); + tgt = lov->lov_tgts[info->idx]; + + if (!tgt || !tgt->ltd_active) + GOTO(out, rc = -ESRCH); + + rc = obd_get_info(tgt->ltd_exp, keylen, key, &size, info->data); GOTO(out, rc = 0); } else if (KEY_IS(KEY_LOVDESC)) { struct lov_desc *desc_ret = val; @@ -2400,11 +2401,15 @@ static int lov_set_info_async(struct obd_export *exp, obd_count keylen, struct lov_obd *lov = &obddev->u.lov; obd_count count; int i, rc = 0, err; - int no_set = !set; - int incr = 0, check_uuid = 0, do_inactive = 0; + struct lov_tgt_desc *tgt; + unsigned incr, check_uuid, + do_inactive, no_set; + unsigned next_id = 0, mds_con = 0; ENTRY; - if (no_set) { + incr = check_uuid = do_inactive = no_set = 0; + if (set == NULL) { + no_set = 1; set = ptlrpc_prep_set(); if (!set) RETURN(-ENOMEM); @@ -2414,54 +2419,63 @@ static int lov_set_info_async(struct obd_export *exp, obd_count keylen, count = lov->desc.ld_tgt_count; if (KEY_IS(KEY_NEXT_ID)) { - /* We must use mds's idea of # osts for indexing into - mds->mds_lov_objids */ - count = vallen / sizeof(obd_id); - LASSERT(count <= lov->desc.ld_tgt_count); - vallen = sizeof(obd_id); - incr = sizeof(obd_id); + count = vallen / sizeof(struct obd_id_info); + vallen = sizeof(struct obd_id_info); + incr = sizeof(struct obd_id_info); do_inactive = 1; + next_id = 1; } else if (KEY_IS("checksum")) { do_inactive = 1; } else if (KEY_IS("unlinked")) { check_uuid = val ? 1 : 0; } else if (KEY_IS("evict_by_nid")) { /* use defaults: do_inactive = incr = 0; */ + } else if (KEY_IS(KEY_MDS_CONN)) { + mds_con = 1; } for (i = 0; i < count; i++, val = (char *)val + incr) { + if (next_id) { + tgt = lov->lov_tgts[((struct obd_id_info*)val)->idx]; + } else { + tgt = lov->lov_tgts[i]; + } /* OST was disconnected */ - if (!lov->lov_tgts[i] || !lov->lov_tgts[i]->ltd_exp) + if (!tgt || !tgt->ltd_exp) continue; /* OST is inactive and we don't want inactive OSCs */ - if (!lov->lov_tgts[i]->ltd_active && !do_inactive) + if (!tgt->ltd_active && !do_inactive) continue; - if (KEY_IS(KEY_MDS_CONN)) { + if (mds_con) { struct mds_group_info *mgi; LASSERT(vallen == sizeof(*mgi)); mgi = (struct mds_group_info *)val; - + /* Only want a specific OSC */ if (mgi->uuid && !obd_uuid_equals(mgi->uuid, - &lov->lov_tgts[i]->ltd_uuid)) + &tgt->ltd_uuid)) continue; - err = obd_set_info_async(lov->lov_tgts[i]->ltd_exp, - keylen, key, sizeof(int), + err = obd_set_info_async(tgt->ltd_exp, + keylen, key, sizeof(int), &mgi->group, set); - } else { + } else if (next_id) { + err = obd_set_info_async(tgt->ltd_exp, + keylen, key, vallen, + ((struct obd_id_info*)val)->data, set); + } else { /* Only want a specific OSC */ if (check_uuid && - !obd_uuid_equals(val, &lov->lov_tgts[i]->ltd_uuid)) + !obd_uuid_equals(val, &tgt->ltd_uuid)) continue; - - err = obd_set_info_async(lov->lov_tgts[i]->ltd_exp, + + err = obd_set_info_async(tgt->ltd_exp, keylen, key, vallen, val, set); } - + if (!rc) rc = err; } diff --git a/lustre/lov/lov_request.c b/lustre/lov/lov_request.c index f800a18..5d80fcb 100644 --- a/lustre/lov/lov_request.c +++ b/lustre/lov/lov_request.c @@ -690,9 +690,6 @@ int lov_update_create_set(struct lov_request_set *set, if (rc) RETURN(rc); - if (oti && oti->oti_objid) - oti->oti_objid[req->rq_idx] = req->rq_oi.oi_oa->o_id; - loi->loi_id = req->rq_oi.oi_oa->o_id; loi->loi_gr = req->rq_oi.oi_oa->o_gr; loi->loi_ost_idx = req->rq_idx; diff --git a/lustre/mdd/mdd_dir.c b/lustre/mdd/mdd_dir.c index dc24003..c90f83b 100644 --- a/lustre/mdd/mdd_dir.c +++ b/lustre/mdd/mdd_dir.c @@ -1015,7 +1015,7 @@ static int mdd_create_data(const struct lu_env *env, struct md_object *pobj, /* update lov_objid data, must be before transaction stop! */ if (rc == 0) - mdd_lov_objid_update(env, mdd); + mdd_lov_objid_update(mdd, lmm); mdd_trans_stop(env, mdd, rc, handle); out_free: @@ -1389,7 +1389,7 @@ cleanup: /* update lov_objid data, must be before transaction stop! */ if (rc == 0) - mdd_lov_objid_update(env, mdd); + mdd_lov_objid_update(mdd, lmm); mdd_pdo_write_unlock(env, mdd_pobj, dlh); out_trans: diff --git a/lustre/mdd/mdd_internal.h b/lustre/mdd/mdd_internal.h index 3cbe6d0..82efb15 100644 --- a/lustre/mdd/mdd_internal.h +++ b/lustre/mdd/mdd_internal.h @@ -141,7 +141,7 @@ int mdd_lov_create(const struct lu_env *env, struct mdd_device *mdd, struct mdd_object *parent, struct mdd_object *child, struct lov_mds_md **lmm, int *lmm_size, const struct md_op_spec *spec, struct lu_attr *la); -void mdd_lov_objid_update(const struct lu_env *env, struct mdd_device *mdd); +void mdd_lov_objid_update(struct mdd_device *mdd, struct lov_mds_md *lmm); void mdd_lov_create_finish(const struct lu_env *env, struct mdd_device *mdd, struct lov_mds_md *lmm, int lmm_size, const struct md_op_spec *spec); diff --git a/lustre/mdd/mdd_lov.c b/lustre/mdd/mdd_lov.c index 687137a..42d8fb5 100644 --- a/lustre/mdd/mdd_lov.c +++ b/lustre/mdd/mdd_lov.c @@ -342,56 +342,17 @@ static obd_id mdd_lov_create_id(const struct lu_fid *fid) return fid_flatten(fid); } -static int mdd_lov_objid_alloc(const struct lu_env *env, - struct mdd_device *mdd) +void mdd_lov_objid_update(struct mdd_device *mdd, struct lov_mds_md *lmm) { - struct mdd_thread_info *info = mdd_env_info(env); - struct mds_obd *mds = &mdd->mdd_obd_dev->u.mds; - - OBD_ALLOC(info->mti_oti.oti_objid, - mds->mds_lov_desc.ld_tgt_count * sizeof(obd_id)); - return (info->mti_oti.oti_objid == NULL ? -ENOMEM : 0); -} - -void mdd_lov_objid_update(const struct lu_env *env, struct mdd_device *mdd) -{ - struct mdd_thread_info *info = mdd_env_info(env); - if (info->mti_oti.oti_objid != NULL) - mds_lov_update_objids(mdd->mdd_obd_dev, - info->mti_oti.oti_objid); -} - -static void mdd_lov_objid_from_lmm(const struct lu_env *env, - struct mdd_device *mdd, - struct lov_mds_md *lmm) -{ - struct mds_obd *mds = &mdd->mdd_obd_dev->u.mds; - struct mdd_thread_info *info = mdd_env_info(env); - mds_objids_from_lmm(info->mti_oti.oti_objid, lmm, &mds->mds_lov_desc); -} - -static void mdd_lov_objid_free(const struct lu_env *env, - struct mdd_device *mdd) -{ - struct mdd_thread_info *info = mdd_env_info(env); - struct mds_obd *mds = &mdd->mdd_obd_dev->u.mds; - - OBD_FREE(info->mti_oti.oti_objid, - mds->mds_lov_desc.ld_tgt_count * sizeof(obd_id)); - info->mti_oti.oti_objid = NULL; + mds_lov_update_objids(mdd->mdd_obd_dev, lmm); } void mdd_lov_create_finish(const struct lu_env *env, struct mdd_device *mdd, struct lov_mds_md *lmm, int lmm_size, const struct md_op_spec *spec) { - struct mdd_thread_info *info = mdd_env_info(env); - if (lmm && !spec->u.sp_ea.no_lov_create) OBD_FREE(lmm, lmm_size); - - if (info->mti_oti.oti_objid != NULL) - mdd_lov_objid_free(env, mdd); } int mdd_lov_create(const struct lu_env *env, struct mdd_device *mdd, @@ -413,15 +374,11 @@ int mdd_lov_create(const struct lu_env *env, struct mdd_device *mdd, RETURN(0); oti_init(oti, NULL); - rc = mdd_lov_objid_alloc(env, mdd); - if (rc != 0) - RETURN(rc); /* replay case, has objects already, only get lov from eadata */ if (spec->u.sp_ea.no_lov_create != 0) { *lmm = (struct lov_mds_md *)spec->u.sp_ea.eadata; *lmm_size = spec->u.sp_ea.eadatalen; - mdd_lov_objid_from_lmm(env, mdd, *lmm); RETURN(0); } @@ -553,13 +510,10 @@ out_oti: out_ids: if (lsm) obd_free_memmd(lov_exp, &lsm); - if (rc != 0) - mdd_lov_objid_free(env, mdd); return rc; } - /* * used when destroying orphans and from mds_reint_unlink() when MDS wants to * destroy objects on OSS. diff --git a/lustre/mds/handler.c b/lustre/mds/handler.c index 2a9c066..e6d2a96 100644 --- a/lustre/mds/handler.c +++ b/lustre/mds/handler.c @@ -62,6 +62,10 @@ int mds_num_threads; CFS_MODULE_PARM(mds_num_threads, "i", int, 0444, "number of MDS service threads to start"); +__u32 mds_max_ost_index=0xFFFF; +CFS_MODULE_PARM(mds_max_ost_index, "i", int, 0444, + "maximal OST index"); + static int mds_intent_policy(struct ldlm_namespace *ns, struct ldlm_lock **lockp, void *req_cookie, ldlm_mode_t mode, int flags, void *data); @@ -2236,8 +2240,9 @@ static int mds_cleanup(struct obd_device *obd) lquota_cleanup(mds_quota_interface_ref, obd); mds_update_server_data(obd, 1); - if (mds->mds_lov_objids != NULL) - OBD_FREE(mds->mds_lov_objids, mds->mds_lov_objids_size); + /* XXX + mds_lov_destroy_objids(obd); + */ mds_fs_cleanup(obd); #if 0 @@ -2778,7 +2783,6 @@ static int mds_cmd_setup(struct obd_device *obd, struct lustre_cfg *lcfg) struct lustre_sb_info *lsi; struct lustre_mount_info *lmi; struct dentry *dentry; - struct file *file; int rc = 0; ENTRY; @@ -2829,19 +2833,10 @@ static int mds_cmd_setup(struct obd_device *obd, struct lustre_cfg *lcfg) CERROR("__iopen__ directory has no inode? rc = %d\n", rc); GOTO(err_fid, rc); } - - /* 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_fid, rc = PTR_ERR(file)); - } - mds->mds_lov_objid_filp = 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_lov_objid, rc = -ENOENT); + rc = mds_lov_init_objids(obd); + if (rc != 0) { + CERROR("cannot init lov objid rc = %d\n", rc); + GOTO(err_fid, rc ); } rc = mds_lov_presetup(mds, lcfg); @@ -2865,10 +2860,6 @@ static int mds_cmd_setup(struct obd_device *obd, struct lustre_cfg *lcfg) err_pop: pop_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL); RETURN(rc); -err_lov_objid: - if (mds->mds_lov_objid_filp && - filp_close((struct file *)mds->mds_lov_objid_filp, 0)) - CERROR("can't close %s after error\n", LOV_OBJID); err_fid: dput(mds->mds_fid_de); err_objects: @@ -2890,20 +2881,14 @@ static int mds_cmd_cleanup(struct obd_device *obd) "will be preserved.\n", obd->obd_name); push_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL); - 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); - } + + mds_lov_destroy_objids(obd); + if (mds->mds_objects_dir != NULL) { l_dput(mds->mds_objects_dir); mds->mds_objects_dir = NULL; } - if (mds->mds_lov_objids != NULL) - OBD_FREE(mds->mds_lov_objids, mds->mds_lov_objids_size); - shrink_dcache_parent(mds->mds_fid_de); dput(mds->mds_fid_de); LL_DQUOT_OFF(obd->u.obt.obt_sb); diff --git a/lustre/mds/mds_internal.h b/lustre/mds/mds_internal.h index cdcc0f0..8193d5b 100644 --- a/lustre/mds/mds_internal.h +++ b/lustre/mds/mds_internal.h @@ -186,9 +186,11 @@ int mds_llog_finish(struct obd_device *obd, int count); int mds_lov_connect(struct obd_device *obd, char * lov_name); int mds_lov_disconnect(struct obd_device *obd); int mds_lov_write_objids(struct obd_device *obd); + int mds_lov_clear_orphans(struct mds_obd *mds, struct obd_uuid *ost_uuid); -void mds_lov_update_objids(struct obd_device *obd, obd_id *ids); +void mds_lov_update_objids(struct obd_device *obd, struct lov_mds_md *lmm); int mds_lov_set_nextid(struct obd_device *obd); + int mds_lov_start_synchronize(struct obd_device *obd, struct obd_device *watched, void *data, int nonblock); @@ -197,8 +199,6 @@ int mds_notify(struct obd_device *obd, struct obd_device *watched, enum obd_notify_event ev, void *data); int mds_convert_lov_ea(struct obd_device *obd, struct inode *inode, struct lov_mds_md *lmm, int lmm_size); -void mds_objids_from_lmm(obd_id *ids, struct lov_mds_md *lmm, - struct lov_desc *desc); int mds_init_lov_desc(struct obd_device *obd, struct obd_export *osc_exp); /* mds/mds_open.c */ diff --git a/lustre/mds/mds_lov.c b/lustre/mds/mds_lov.c index f282de1..f4cbd24 100644 --- a/lustre/mds/mds_lov.c +++ b/lustre/mds/mds_lov.c @@ -41,19 +41,139 @@ #include "mds_internal.h" -void mds_lov_update_objids(struct obd_device *obd, obd_id *ids) +static void mds_lov_dump_objids(const char *label, struct obd_device *obd) { struct mds_obd *mds = &obd->u.mds; - int i; + 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;imds_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 - %llu ",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;imds_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; - lock_kernel(); - for (i = 0; i < mds->mds_lov_desc.ld_tgt_count; i++) - if (ids[i] > (mds->mds_lov_objids)[i]) { - (mds->mds_lov_objids)[i] = ids[i]; - mds->mds_lov_objids_dirty = 1; + /* 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 %llu" + " old %llu\n", i, id, data[idx]); + if (id > data[idx]) { + data[idx] = id; + bitmap_set(mds->mds_lov_page_dirty, page); } - unlock_kernel(); + } EXIT; } EXPORT_SYMBOL(mds_lov_update_objids); @@ -61,14 +181,11 @@ EXPORT_SYMBOL(mds_lov_update_objids); static int mds_lov_read_objids(struct obd_device *obd) { struct mds_obd *mds = &obd->u.mds; - obd_id *ids; loff_t off = 0; - int i, rc, size; + int i, rc, count = 0, page = 0; + size_t size; ENTRY; - LASSERT(!mds->mds_lov_objids_size); - LASSERT(!mds->mds_lov_objids_dirty); - /* 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. */ @@ -76,58 +193,121 @@ static int mds_lov_read_objids(struct obd_device *obd) if (size == 0) RETURN(0); - OBD_ALLOC(ids, size); - if (ids == NULL) - RETURN(-ENOMEM); - mds->mds_lov_objids = ids; - mds->mds_lov_objids_size = size; + page = (size/(OBJID_PER_PAGE()*sizeof(obd_id)))+1; + CDEBUG(D_INFO, "file size %d pages %d\n", size, page); + for(i=0; i < page; i++) { + obd_id *data = mds->mds_lov_page_array[i]; + loff_t off_old = off; - rc = fsfilt_read_record(obd, mds->mds_lov_objid_filp, ids, size, &off); - if (rc < 0) { - CERROR("Error reading objids %d\n", rc); - RETURN(rc); - } + LASSERT(data == NULL); + OBD_ALLOC(data, MDS_LOV_ALLOC_SIZE); + if (data == NULL) + GOTO(out, rc = -ENOMEM); - mds->mds_lov_objids_in_file = size / sizeof(*ids); + mds->mds_lov_page_array[i] = data; - for (i = 0; i < mds->mds_lov_objids_in_file; i++) { - CDEBUG(D_INFO, "read last object "LPU64" for idx %d\n", - mds->mds_lov_objids[i], i); + 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)-1)/sizeof(obd_id); } + mds->mds_lov_objid_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 objid\n", count); +out: + mds_lov_dump_objids("read",obd); + RETURN(0); } int mds_lov_write_objids(struct obd_device *obd) { struct mds_obd *mds = &obd->u.mds; - loff_t off = 0; - int i, rc, tgts; + int i, rc = 0; ENTRY; - if (!mds->mds_lov_objids_dirty) + if (bitmap_check_empty(mds->mds_lov_page_dirty)) RETURN(0); - tgts = max(mds->mds_lov_desc.ld_tgt_count, mds->mds_lov_objids_in_file); + mds_lov_dump_objids("write", obd); - if (!tgts) - RETURN(0); + 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; - for (i = 0; i < tgts; i++) - CDEBUG(D_INFO, "writing last object "LPU64" for idx %d\n", - mds->mds_lov_objids[i], i); + LASSERT(data != NULL); - rc = fsfilt_write_record(obd, mds->mds_lov_objid_filp, - mds->mds_lov_objids, tgts * sizeof(obd_id), - &off, 0); - if (rc >= 0) { - mds->mds_lov_objids_dirty = 0; - rc = 0; + /* check for particaly filled last page */ + if (i == mds->mds_lov_objid_lastpage) { + size = mds->mds_lov_objid_lastidx * sizeof(obd_id); + } + + rc = fsfilt_write_record(obd, mds->mds_lov_objid_filp, data, + size, &off, 0); + if (rc < 0) + break; + 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, struct obd_export *export, + __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; + } + + printk("get %d - %p - %d/%d\n", idx, data, page, off); + 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(export, sizeof("last_id"), + "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; + } + bitmap_set(mds->mds_lov_page_dirty, page); + } +out: + RETURN(rc); +} + int mds_lov_clear_orphans(struct mds_obd *mds, struct obd_uuid *ost_uuid) { int rc; @@ -136,7 +316,7 @@ int mds_lov_clear_orphans(struct mds_obd *mds, struct obd_uuid *ost_uuid) struct lov_stripe_md *empty_ea = NULL; ENTRY; - LASSERT(mds->mds_lov_objids != NULL); + 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 @@ -153,26 +333,73 @@ int mds_lov_clear_orphans(struct mds_obd *mds, struct obd_uuid *ost_uuid) RETURN(rc); } - /* update the LOV-OSC knowledge of the last used object id's */ +/* for all targets */ +/* is we realy need this ? all osc's should be pass via __mds_lov_synchronize + * and call */ +#define MDS_LOV_SETID_COUNT (CFS_PAGE_SIZE / sizeof(struct obd_id_info)) + int mds_lov_set_nextid(struct obd_device *obd) { struct mds_obd *mds = &obd->u.mds; + int i = 0, j, rc = 0; + 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); + + OBD_ALLOC(info, CFS_PAGE_SIZE); + if (info == NULL) + RETURN(-ENOMEM); + + while(i < mds->mds_lov_desc.ld_tgt_count) { + for(j=0; j < MDS_LOV_SETID_COUNT; i++, j++) { + int page = i / OBJID_PER_PAGE(); + int idx = i % OBJID_PER_PAGE(); + obd_id *data = mds->mds_lov_page_array[page]; + + if (i == mds->mds_lov_desc.ld_tgt_count) + break; + + info[j].idx = i; + info[j].data = &data[idx]; + } + + 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); + break; + } + } + OBD_FREE(info, CFS_PAGE_SIZE); + + 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); - LASSERT(mds->mds_lov_objids != NULL); - + /* obd->obd_dev_sem must be held so mds_lov_objids doesn't change */ LASSERT_SEM_LOCKED(&obd->obd_dev_sem); - rc = obd_set_info_async(mds->mds_osc_exp, strlen(KEY_NEXT_ID), - KEY_NEXT_ID, - mds->mds_lov_desc.ld_tgt_count * - sizeof(*mds->mds_lov_objids), - mds->mds_lov_objids, NULL); + 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); @@ -185,44 +412,29 @@ static int mds_lov_update_desc(struct obd_device *obd, struct obd_export *lov) { struct mds_obd *mds = &obd->u.mds; struct lov_desc *ld; - __u32 size, stripes, valsize = sizeof(mds->mds_lov_desc); - int rc = 0; + __u32 stripes, valsize = sizeof(mds->mds_lov_desc); + int page, rc = 0; ENTRY; OBD_ALLOC(ld, sizeof(*ld)); if (!ld) RETURN(-ENOMEM); - rc = obd_get_info(lov, strlen(KEY_LOVDESC) + 1, KEY_LOVDESC, + rc = obd_get_info(lov, sizeof(KEY_LOVDESC), KEY_LOVDESC, &valsize, ld); if (rc) GOTO(out, rc); /* The size of the LOV target table may have increased. */ - size = ld->ld_tgt_count * sizeof(obd_id); - if ((mds->mds_lov_objids_size == 0) || - (size > mds->mds_lov_objids_size)) { + page = ld->ld_tgt_count / OBJID_PER_PAGE(); + if (mds->mds_lov_page_array[page] == NULL) { obd_id *ids; - /* add room by powers of 2 */ - size = 1; - while (size < ld->ld_tgt_count) - size = size << 1; - size = size * sizeof(obd_id); - - OBD_ALLOC(ids, size); + OBD_ALLOC(ids, MDS_LOV_ALLOC_SIZE); if (ids == NULL) GOTO(out, rc = -ENOMEM); - memset(ids, 0, size); - if (mds->mds_lov_objids_size) { - obd_id *old_ids = mds->mds_lov_objids; - memcpy(ids, mds->mds_lov_objids, - mds->mds_lov_objids_size); - mds->mds_lov_objids = ids; - OBD_FREE(old_ids, mds->mds_lov_objids_size); - } - mds->mds_lov_objids = ids; - mds->mds_lov_objids_size = size; + + mds->mds_lov_page_array[page] = ids; } /* Don't change the mds_lov_desc until the objids size matches the @@ -232,8 +444,8 @@ static int mds_lov_update_desc(struct obd_device *obd, struct obd_export *lov) mds->mds_lov_desc.ld_tgt_count); stripes = min_t(__u32, LOV_MAX_STRIPE_COUNT, - max(mds->mds_lov_desc.ld_tgt_count, - mds->mds_lov_objids_in_file)); + mds->mds_lov_desc.ld_tgt_count); + mds->mds_max_mdsize = lov_mds_md_size(stripes); mds->mds_max_cookiesize = stripes * sizeof(struct llog_cookie); CDEBUG(D_CONFIG, "updated max_mdsize/max_cookiesize for %d stripes: " @@ -243,7 +455,7 @@ static int mds_lov_update_desc(struct obd_device *obd, struct obd_export *lov) /* If we added a target we have to reconnect the llogs */ /* We only _need_ to do this at first add (idx), or the first time after recovery. However, it should now be safe to call anytime. */ - llog_cat_initialize(obd, NULL, mds->mds_lov_desc.ld_tgt_count, NULL); + rc = llog_cat_initialize(obd, NULL, mds->mds_lov_desc.ld_tgt_count, NULL); /*XXX this notifies the MDD until lov handling use old mds code */ if (obd->obd_upcall.onu_owner) { @@ -265,8 +477,12 @@ static int mds_lov_update_mds(struct obd_device *obd, __u32 idx, struct obd_uuid *uuid) { struct mds_obd *mds = &obd->u.mds; - int old_count; + __u32 old_count; int rc = 0; + int page; + int off; + obd_id *data; + ENTRY; /* Don't let anyone else mess with mds_lov_objids now */ @@ -274,7 +490,7 @@ static int mds_lov_update_mds(struct obd_device *obd, old_count = mds->mds_lov_desc.ld_tgt_count; rc = mds_lov_update_desc(obd, mds->mds_osc_exp); - if (rc) + if (rc) GOTO(out, rc); CDEBUG(D_CONFIG, "idx=%d, recov=%d/%d, cnt=%d/%d\n", @@ -282,30 +498,25 @@ static int mds_lov_update_mds(struct obd_device *obd, mds->mds_lov_desc.ld_tgt_count); /* idx is set as data from lov_notify. */ - if (idx == MDSLOV_NO_INDEX || obd->obd_recovering) + if (idx == MDSLOV_NO_INDEX || obd->obd_recovering) GOTO(out, rc); - + if (idx >= mds->mds_lov_desc.ld_tgt_count) { - CERROR("index %d > count %d!\n", idx, + CERROR("index %d > count %d!\n", idx, mds->mds_lov_desc.ld_tgt_count); GOTO(out, rc = -EINVAL); } - if (idx >= mds->mds_lov_objids_in_file) { - /* We never read this lastid; ask the osc */ - obd_id lastid; - __u32 size = sizeof(lastid); - rc = obd_get_info(watched->obd_self_export, strlen("last_id"), - "last_id", &size, &lastid); - if (rc) - GOTO(out, rc); - mds->mds_lov_objids[idx] = lastid; - mds->mds_lov_objids_dirty = 1; - mds_lov_write_objids(obd); + page = idx / OBJID_PER_PAGE(); + off = idx % OBJID_PER_PAGE(); + data = mds->mds_lov_page_array[page]; + CDEBUG(D_CONFIG, "idx %d - %p - %d/%d\n", idx, data, page, off); + if (data[off] == 0) { + rc = mds_lov_get_objid(obd, watched->obd_self_export, idx); } else { /* We have read this lastid from disk; tell the osc. Don't call this during recovery. */ - rc = mds_lov_set_nextid(obd); + rc = mds_lov_set_one_nextid(obd, idx, &data[off]); if (rc) { CERROR("Failed to set next id, idx=%d rc=%d\n", idx,rc); /* Don't abort the rest of the sync */ @@ -314,7 +525,7 @@ static int mds_lov_update_mds(struct obd_device *obd, } CDEBUG(D_CONFIG, "last object "LPU64" from OST %d rc=%d\n", - mds->mds_lov_objids[idx], idx, rc); + data[off], idx, rc); out: mutex_up(&obd->obd_dev_sem); RETURN(rc); @@ -326,7 +537,7 @@ int mds_lov_connect(struct obd_device *obd, char * lov_name) struct mds_obd *mds = &obd->u.mds; struct lustre_handle conn = {0,}; struct obd_connect_data *data; - int rc, i; + int rc; ENTRY; if (IS_ERR(mds->mds_osc_obd)) @@ -384,29 +595,20 @@ int mds_lov_connect(struct obd_device *obd, char * lov_name) if (rc) GOTO(err_reg, rc); - /* tgt_count may be 0! */ - rc = llog_cat_initialize(obd, NULL, mds->mds_lov_desc.ld_tgt_count, NULL); - if (rc) { - CERROR("failed to initialize catalog %d\n", rc); - GOTO(err_reg, rc); - } - /* If we're mounting this code for the first time on an existing FS, * we need to populate the objids array from the real OST values */ - if (mds->mds_lov_desc.ld_tgt_count > mds->mds_lov_objids_in_file) { - int size = sizeof(obd_id) * mds->mds_lov_desc.ld_tgt_count; - rc = obd_get_info(mds->mds_osc_exp, strlen("last_id"), - "last_id", &size, mds->mds_lov_objids); - if (!rc) { - for (i = 0; i < mds->mds_lov_desc.ld_tgt_count; i++) - CWARN("got last object "LPU64" from OST %d\n", - mds->mds_lov_objids[i], i); - mds->mds_lov_objids_dirty = 1; - rc = mds_lov_write_objids(obd); - if (rc) - CERROR("got last objids from OSTs, but error " - "writing objids file: %d\n", rc); + if (mds->mds_lov_desc.ld_tgt_count > mds->mds_lov_objid_count) { + __u32 i = mds->mds_lov_objid_count; + for(; i <= mds->mds_lov_desc.ld_tgt_count; i++) { + rc = mds_lov_get_objid(obd, mds->mds_osc_exp, i); + if (rc != 0) + break; } + if (rc == 0) + rc = mds_lov_write_objids(obd); + if (rc) + CERROR("got last objids from OSTs, but error " + "in update objids file: %d\n", rc); } mutex_up(&obd->obd_dev_sem); @@ -728,7 +930,7 @@ static int __mds_lov_synchronize(void *data) mgi.group = FILTER_GROUP_MDS0 + mds->mds_id; mgi.uuid = uuid; - rc = obd_set_info_async(mds->mds_osc_exp, strlen(KEY_MDS_CONN), + rc = obd_set_info_async(mds->mds_osc_exp, sizeof(KEY_MDS_CONN), KEY_MDS_CONN, sizeof(mgi), &mgi, NULL); if (rc != 0) GOTO(out, rc); @@ -763,7 +965,7 @@ static int __mds_lov_synchronize(void *data) obd_uuid2str(uuid), rc); GOTO(out, rc); } - + if (obd->obd_upcall.onu_owner) { /* * This is a hack for mds_notify->mdd_notify. When the mds obd @@ -969,14 +1171,3 @@ conv_free: conv_end: return rc; } - -void mds_objids_from_lmm(obd_id *ids, struct lov_mds_md *lmm, - struct lov_desc *desc) -{ - int i; - for (i = 0; i < le32_to_cpu(lmm->lmm_stripe_count); i++) { - ids[le32_to_cpu(lmm->lmm_objects[i].l_ost_idx)] = - le64_to_cpu(lmm->lmm_objects[i].l_object_id); - } -} -EXPORT_SYMBOL(mds_objids_from_lmm); diff --git a/lustre/mds/mds_open.c b/lustre/mds/mds_open.c index 0bce9e6..f4e86be 100644 --- a/lustre/mds/mds_open.c +++ b/lustre/mds/mds_open.c @@ -303,7 +303,7 @@ static int mds_create_objects(struct ptlrpc_request *req, int offset, struct mds_update_record *rec, struct mds_obd *mds, struct obd_device *obd, struct dentry *dchild, void **handle, - obd_id **ids) + struct lov_mds_md **objid) { struct inode *inode = dchild->d_inode; struct obd_trans_info oti = { 0 }; @@ -324,11 +324,7 @@ static int mds_create_objects(struct ptlrpc_request *req, int offset, if (body->valid & OBD_MD_FLEASIZE) RETURN(0); - OBD_ALLOC(*ids, mds->mds_lov_desc.ld_tgt_count * sizeof(**ids)); - if (*ids == NULL) - RETURN(-ENOMEM); oti_init(&oti, req); - oti.oti_objid = *ids; /* replay case */ if (lustre_msg_get_flags(req->rq_reqmsg) & MSG_REPLAY) { @@ -350,8 +346,6 @@ static int mds_create_objects(struct ptlrpc_request *req, int offset, GOTO(out_ids, rc); } - mds_objids_from_lmm(*ids, lmm, &mds->mds_lov_desc); - rc = fsfilt_set_md(obd, inode, *handle, lmm, lmm_size, "lov"); if (rc) CERROR("open replay failed to set md:%d\n", rc); @@ -359,6 +353,7 @@ static int mds_create_objects(struct ptlrpc_request *req, int offset, LASSERT(lmm_buf); memcpy(lmm_buf, lmm, lmm_size); + *objid = lmm_buf; RETURN(rc); } @@ -477,16 +472,15 @@ static int mds_create_objects(struct ptlrpc_request *req, int offset, lmm_buf = lustre_msg_buf(req->rq_repmsg, offset, lmm_size); LASSERT(lmm_buf); memcpy(lmm_buf, lmm, lmm_size); + + *objid = lmm_buf; // save for mds_lov_update_objid + free_diskmd: obd_free_diskmd(mds->mds_osc_exp, &lmm); out_oa: oti_free_cookies(&oti); OBDO_FREE(oinfo.oi_oa); out_ids: - if (rc) { - OBD_FREE(*ids, mds->mds_lov_desc.ld_tgt_count * sizeof(**ids)); - *ids = NULL; - } if (oinfo.oi_md) obd_free_memmd(mds->mds_osc_exp, &oinfo.oi_md); RETURN(rc); @@ -669,7 +663,7 @@ static int mds_finish_open(struct ptlrpc_request *req, struct dentry *dchild, struct mds_obd *mds = mds_req2mds(req); struct obd_device *obd = req->rq_export->exp_obd; struct mds_file_data *mfd = NULL; - obd_id *ids = NULL; /* object IDs created */ + struct lov_mds_md *lmm; /* object IDs created */ int rc = 0; ENTRY; @@ -702,7 +696,7 @@ static int mds_finish_open(struct ptlrpc_request *req, struct dentry *dchild, !(body->valid & OBD_MD_FLMODEASIZE)) { /* no EA: create objects */ rc = mds_create_objects(req, DLM_REPLY_REC_OFF + 1, rec, - mds, obd, dchild, handle, &ids); + mds, obd, dchild, handle, &lmm); if (rc) { CERROR("mds_create_objects: rc = %d\n", rc); UNLOCK_INODE_MUTEX(dchild->d_inode); @@ -742,10 +736,9 @@ static int mds_finish_open(struct ptlrpc_request *req, struct dentry *dchild, CDEBUG(D_INODE, "mfd %p, cookie "LPX64"\n", mfd, mfd->mfd_handle.h_cookie); - if (ids != NULL) { - mds_lov_update_objids(obd, ids); - OBD_FREE(ids, sizeof(*ids) * mds->mds_lov_desc.ld_tgt_count); - } + if (lmm != NULL) + mds_lov_update_objids(obd, lmm); + if (rc) /* coverity[deadcode] */ mds_mfd_unlink(mfd, 1);