diff --git a/lustre/ChangeLog b/lustre/ChangeLog
index 00474bd5d3c334be0701624e24311f6cf81ee908..81abad7adc3656a45f18a3125e3e3d461ca59704 100644
--- a/lustre/ChangeLog
+++ b/lustre/ChangeLog
@@ -14,6 +14,13 @@ tbd Cluster File Systems, Inc. <info@clusterfs.com>
* Recommended e2fsprogs version: 1.40.2-cfs5
* Note that reiserfs quotas are disabled on SLES 10 in this kernel.
+Severity : normal
+Bugzilla : 12702
+Frequency : rare, in recovery and (or) destroy lovobjid file.
+Description: rewrite lov objid code.
+Details : Cleanup for lov objid code, remove scability problems and wrong
+ locking. Fix sending last_id into ost.
+
Severity : enhancement
Bugzilla : 14388
Description: Update to SLES10 SP1 latest kernel-2.6.16.54-0.2.3.
diff --git a/lustre/include/liblustre.h b/lustre/include/liblustre.h
index e4ab4ffb7cfc0123f2e92e36fbb46014cc489e59..4e80252f9ec097edb00892159dd750f694866027 100644
--- a/lustre/include/liblustre.h
+++ b/lustre/include/liblustre.h
@@ -64,6 +64,7 @@
#include <fcntl.h>
#include <libcfs/list.h>
+#include <libcfs/user-bitops.h>
#include <lnet/lnet.h>
#include <libcfs/kp30.h>
diff --git a/lustre/include/obd.h b/lustre/include/obd.h
index 8c708819f8b61557315da8f082e95264d39666c2..6af873403f1885e15ece0c1f0fbc85de09081393 100644
--- a/lustre/include/obd.h
+++ b/lustre/include/obd.h
@@ -33,6 +33,9 @@
#include <lustre_quota.h>
#include <class_hash.h>
+#include <libcfs/bitmap.h>
+
+
#define MAX_OBD_DEVICES 8192
/* this is really local to the OSC */
@@ -456,11 +459,17 @@ struct mds_obd {
char *mds_profile;
struct obd_export *mds_osc_exp; /* XXX lov_exp */
struct lov_desc mds_lov_desc;
- 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;
@@ -468,9 +477,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_fl_cfglog:1,
mds_fl_synced:1,
@@ -481,6 +488,24 @@ struct mds_obd {
lnet_nid_t mds_nosquash_nid;
};
+/* lov objid */
+#define mds_max_ost_index (0xFFFF)
+#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;
@@ -604,7 +629,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;
@@ -842,6 +866,8 @@ enum obd_cleanup_stage {
#define KEY_LOVDESC "lovdesc"
#define KEY_INIT_RECOV "initial_recov"
#define KEY_INIT_RECOV_BACKUP "init_recov_bk"
+#define KEY_LOV_IDX "lov_idx"
+#define KEY_LAST_ID "last_id"
struct obd_ops {
struct module *o_owner;
diff --git a/lustre/llite/file.c b/lustre/llite/file.c
index a5613aa760e1d6cd77f8acbd158b2c0ede5d0fe0..d736fadcf494a3fd0fb053d4117fc3d5b7efe8b3 100644
--- a/lustre/llite/file.c
+++ b/lustre/llite/file.c
@@ -1569,7 +1569,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 0f614284e2d4583743b49fca22ac2e5153bd7069..2975dabc1e6d98d5c3de9739b27f4f47d23b73bb 100644
--- a/lustre/lov/lov_obd.c
+++ b/lustre/lov/lov_obd.c
@@ -2496,23 +2496,32 @@ static int lov_get_info(struct obd_export *exp, __u32 keylen,
LDLM_ERROR(data->lock, "lock on inode without such object");
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);
- }
+ } else if (KEY_IS(KEY_LAST_ID)) {
+ 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;
*desc_ret = lov->desc;
GOTO(out, rc = 0);
+ } else if (KEY_IS(KEY_LOV_IDX)) {
+ struct lov_tgt_desc *tgt;
+
+ for(i = 0; i < lov->desc.ld_tgt_count; i++) {
+ tgt = lov->lov_tgts[i];
+ if (obd_uuid_equals(val, &tgt->ltd_uuid))
+ GOTO(out, rc = i);
+ }
}
rc = -EINVAL;
@@ -2530,6 +2539,9 @@ static int lov_set_info_async(struct obd_export *exp, obd_count keylen,
obd_count count;
int i, rc = 0, err, incr = 0, check_uuid = 0, do_inactive = 0;
int no_set = !set;
+ unsigned next_id = 0;
+ struct lov_tgt_desc *tgt;
+ void *data;
ENTRY;
if (no_set) {
@@ -2542,13 +2554,11 @@ 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);
+ count = vallen / sizeof(struct obd_id_info);
vallen = sizeof(obd_id);
- incr = sizeof(obd_id);
+ incr = sizeof(struct obd_id_info);
do_inactive = 1;
+ next_id = 1;
} else if (KEY_IS("checksum")) {
do_inactive = 1;
} else if (KEY_IS(KEY_MDS_CONN) || KEY_IS("unlinked")) {
@@ -2560,21 +2570,28 @@ static int lov_set_info_async(struct obd_export *exp, obd_count keylen,
}
for (i = 0; i < count; i++, val = (char *)val + incr) {
+ if (next_id) {
+ tgt = lov->lov_tgts[((struct obd_id_info*)val)->idx];
+ data = ((struct obd_id_info*)val)->data;
+ } else {
+ tgt = lov->lov_tgts[i];
+ data = val;
+ }
/* 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;
/* 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,
- keylen, key, vallen, val, set);
+ err = obd_set_info_async(tgt->ltd_exp,
+ keylen, key, vallen, data, set);
if (!rc)
rc = err;
}
diff --git a/lustre/lov/lov_request.c b/lustre/lov/lov_request.c
index 858894c93d0ce7501bd7cb0c65a14d3c596ea063..82153b7a46966b3005f44da9bb8fa2b487bfb03a 100644
--- a/lustre/lov/lov_request.c
+++ b/lustre/lov/lov_request.c
@@ -687,9 +687,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_ost_idx = req->rq_idx;
CDEBUG(D_INODE, "objid "LPX64" has subobj "LPX64"/"LPU64" at idx %d\n",
diff --git a/lustre/mds/handler.c b/lustre/mds/handler.c
index de7f4eaf5fde0852f36afb5b7f0838877c425652..696ac4da28b07a841206b70d0e04ee401e7fcdfc 100644
--- a/lustre/mds/handler.c
+++ b/lustre/mds/handler.c
@@ -505,7 +505,7 @@ static int mds_destroy_export(struct obd_export *export)
static int mds_disconnect(struct obd_export *exp)
{
- int rc;
+ int rc = 0;
ENTRY;
LASSERT(exp);
@@ -2172,16 +2172,6 @@ int mds_postrecov(struct obd_device *obd)
LASSERT(ctxt != NULL);
llog_ctxt_put(ctxt);
- /* set nextid first, so we are sure it happens */
- mutex_down(&obd->obd_dev_sem);
- rc = mds_lov_set_nextid(obd);
- mutex_up(&obd->obd_dev_sem);
- if (rc) {
- CERROR("%s: mds_lov_set_nextid failed %d\n",
- obd->obd_name, rc);
- GOTO(out, rc);
- }
-
/* clean PENDING dir */
rc = mds_cleanup_pending(obd);
if (rc < 0)
@@ -2263,8 +2253,6 @@ 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);
mds_fs_cleanup(obd);
upcall_cache_cleanup(mds->mds_group_hash);
diff --git a/lustre/mds/mds_fs.c b/lustre/mds/mds_fs.c
index 1b94af2b57e6068449c54ad739c469b796940cbc..b91d624c3af2ea256ecc57960ac2d775fa43753e 100644
--- a/lustre/mds/mds_fs.c
+++ b/lustre/mds/mds_fs.c
@@ -587,18 +587,10 @@ int mds_fs_setup(struct obd_device *obd, struct vfsmount *mnt)
GOTO(err_last_rcvd, 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_client, 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_client, rc );
}
/* open and test the check io file junk */
@@ -627,8 +619,7 @@ err_health_check:
filp_close(mds->mds_health_check_filp, 0))
CERROR("can't close %s after error\n", HEALTH_CHECK);
err_lov_objid:
- if (mds->mds_lov_objid_filp && filp_close(mds->mds_lov_objid_filp, 0))
- CERROR("can't close %s after error\n", LOV_OBJID);
+ mds_lov_destroy_objids(obd);
err_client:
class_disconnect_exports(obd);
err_last_rcvd:
@@ -665,12 +656,9 @@ int mds_fs_cleanup(struct obd_device *obd)
if (rc)
CERROR("%s file won't close, rc=%d\n", LAST_RCVD, rc);
}
- if (mds->mds_lov_objid_filp) {
- rc = filp_close(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_health_check_filp) {
rc = filp_close(mds->mds_health_check_filp, 0);
mds->mds_health_check_filp = NULL;
diff --git a/lustre/mds/mds_internal.h b/lustre/mds/mds_internal.h
index dc71c4510897f73d40b45e1c89ed04521bb7abd2..59f901456dc5a575ef4fb8242fbdc8e1705c52b5 100644
--- a/lustre/mds/mds_internal.h
+++ b/lustre/mds/mds_internal.h
@@ -198,10 +198,11 @@ int mds_llog_finish(struct obd_device *obd, int count);
/* mds/mds_lov.c */
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);
-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_clear_orphans(struct mds_obd *mds, struct obd_uuid *ost_uuid);
-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);
@@ -212,8 +213,6 @@ int mds_get_default_md(struct obd_device *obd, struct lov_mds_md *lmm,
int *lmmsize);
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 6b67e213433a0415a8895460334378cf38dfe188..33a19f0ce1430cd05e35615bdb5e9ec63f642272 100644
--- a/lustre/mds/mds_lov.c
+++ b/lustre/mds/mds_lov.c
@@ -41,33 +41,152 @@
#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; 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;
- 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 (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);
}
- unlock_kernel();
+ 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;
+ 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;
- obd_id *ids;
loff_t off = 0;
- int i, rc, size;
+ int i, rc = 0, 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. */
@@ -75,54 +194,122 @@ 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;
+
+ 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 */
- 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);
+ count += (off-off_old)/sizeof(obd_id);
}
- RETURN(0);
+ 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;
- 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;
+ /* 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;
+ 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;
+ }
+ 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);
}
@@ -134,14 +321,14 @@ 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
* objects above this ID, they will be removed. */
memset(&oa, 0, sizeof(oa));
- oa.o_valid = OBD_MD_FLFLAGS;
oa.o_flags = OBD_FL_DELORPHAN;
+ oa.o_valid = OBD_MD_FLFLAGS | OBD_MD_FLGROUP;
if (ost_uuid != NULL) {
memcpy(&oa.o_inline, ost_uuid, sizeof(*ost_uuid));
oa.o_valid |= OBD_MD_FLINLINE;
@@ -151,25 +338,24 @@ 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 */
-int mds_lov_set_nextid(struct obd_device *obd)
+/* 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);
@@ -177,12 +363,25 @@ int mds_lov_set_nextid(struct obd_device *obd)
RETURN(rc);
}
+static __u32 mds_lov_get_idx(struct obd_export *lov,
+ struct obd_uuid *ost_uuid)
+{
+ int rc;
+ int valsize = sizeof(ost_uuid);
+
+ rc = obd_get_info(lov, sizeof(KEY_LOV_IDX), KEY_LOV_IDX,
+ &valsize, ost_uuid);
+ LASSERT(rc >= 0);
+
+ RETURN(rc);
+}
+
/* Update the lov desc for a new size lov. */
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);
+ __u32 stripes, valsize = sizeof(mds->mds_lov_desc);
int rc = 0;
ENTRY;
@@ -190,38 +389,11 @@ static int mds_lov_update_desc(struct obd_device *obd, struct obd_export *lov)
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)) {
- 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);
- 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;
- }
-
/* Don't change the mds_lov_desc until the objids size matches the
count (paranoia) */
mds->mds_lov_desc = *ld;
@@ -229,8 +401,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: "
@@ -240,7 +412,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, mds->mds_lov_desc.ld_tgt_count, NULL);
+ rc = llog_cat_initialize(obd, mds->mds_lov_desc.ld_tgt_count, NULL);
out:
OBD_FREE(ld, sizeof(*ld));
@@ -248,16 +420,18 @@ out:
}
-#define MDSLOV_NO_INDEX -1
-
/* Inform MDS about new/updated target */
static int mds_lov_update_mds(struct obd_device *obd,
struct obd_device *watched,
- __u32 idx, struct obd_uuid *uuid)
+ __u32 idx)
{
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 */
@@ -265,7 +439,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",
@@ -273,39 +447,35 @@ 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 (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);
- } else {
- /* We have read this lastid from disk; tell the osc.
- Don't call this during recovery. */
- rc = mds_lov_set_nextid(obd);
- if (rc) {
- CERROR("Failed to set next id, idx=%d rc=%d\n", idx,rc);
- /* Don't abort the rest of the sync */
- rc = 0;
- }
+ rc = mds_lov_get_objid(obd, idx);
+ if (rc) {
+ CERROR("Failed to get objid - %d\n", rc);
+ GOTO(out, rc);
+ }
+
+ page = idx / OBJID_PER_PAGE();
+ off = idx % OBJID_PER_PAGE();
+ data = mds->mds_lov_page_array[page];
+ /* We have read this lastid from disk; tell the osc.
+ Don't call this during recovery. */
+ 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 */
+ rc = 0;
}
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);
@@ -317,7 +487,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))
@@ -358,7 +528,7 @@ int mds_lov_connect(struct obd_device *obd, char * lov_name)
lov_name, rc);
GOTO(err_discon, rc);
}
-
+
/* Deny new client connections until we are sure we have some OSTs */
obd->obd_no_conn = 1;
@@ -373,30 +543,22 @@ 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, 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, 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);
/* I want to see a callback happen when the OBD moves to a
@@ -612,7 +774,7 @@ int mds_iocontrol(unsigned int cmd, struct obd_export *exp, int len,
llog_cat_initialize(obd, mds->mds_lov_desc.ld_tgt_count, NULL);
llog_ctxt_put(ctxt);
rc2 = obd_set_info_async(mds->mds_osc_exp,
- strlen(KEY_MDS_CONN), KEY_MDS_CONN,
+ sizeof(KEY_MDS_CONN), KEY_MDS_CONN,
0, NULL, NULL);
if (!rc)
rc = rc2;
@@ -688,25 +850,25 @@ static int __mds_lov_synchronize(void *data)
OBD_RACE(OBD_FAIL_MDS_LOV_SYNC_RACE);
- rc = mds_lov_update_mds(obd, watched, idx, uuid);
+ rc = mds_lov_update_mds(obd, watched, idx);
if (rc != 0) {
CERROR("%s failed at update_mds: %d\n", obd_uuid2str(uuid), rc);
GOTO(out, rc);
}
- 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, 0, uuid, NULL);
if (rc != 0)
GOTO(out, rc);
ctxt = llog_get_context(obd, LLOG_MDS_OST_ORIG_CTXT);
- if (!ctxt)
- RETURN(-ENODEV);
-
+ if (!ctxt)
+ RETURN(-ENODEV);
+
OBD_FAIL_TIMEOUT(OBD_FAIL_MDS_LLOG_SYNC_TIMEOUT, 60);
- rc = llog_connect(ctxt, obd->u.mds.mds_lov_desc.ld_tgt_count,
- NULL, NULL, uuid);
+ rc = llog_connect(ctxt, obd->u.mds.mds_lov_desc.ld_tgt_count,
+ NULL, NULL, uuid);
llog_ctxt_put(ctxt);
if (rc != 0) {
@@ -753,12 +915,7 @@ int mds_lov_synchronize(void *data)
struct mds_lov_sync_info *mlsi = data;
char name[20];
- if (mlsi->mlsi_index == MDSLOV_NO_INDEX)
- /* There is still a watched target,
- but we don't know its index */
- sprintf(name, "ll_sync_tgt");
- else
- snprintf(name, sizeof(name), "ll_sync_%02u", mlsi->mlsi_index);
+ snprintf(name, sizeof(name), "ll_sync_%02u", mlsi->mlsi_index);
ptlrpc_daemonize(name);
RETURN(__mds_lov_synchronize(data));
@@ -769,11 +926,13 @@ int mds_lov_start_synchronize(struct obd_device *obd,
void *data, int nonblock)
{
struct mds_lov_sync_info *mlsi;
+ struct mds_obd *mds = &obd->u.mds;
int rc;
-
+ struct obd_uuid *uuid;
ENTRY;
LASSERT(watched);
+ uuid = &watched->u.cli.cl_target_uuid;
OBD_ALLOC(mlsi, sizeof(*mlsi));
if (mlsi == NULL)
@@ -784,7 +943,7 @@ int mds_lov_start_synchronize(struct obd_device *obd,
if (data)
mlsi->mlsi_index = *(__u32 *)data;
else
- mlsi->mlsi_index = MDSLOV_NO_INDEX;
+ mlsi->mlsi_index = mds_lov_get_idx(mds->mds_osc_exp, uuid);
/* Although class_export_get(obd->obd_self_export) would lock
the MDS in place, since it's only a self-export
@@ -942,14 +1101,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);
- }
-}
-
diff --git a/lustre/mds/mds_open.c b/lustre/mds/mds_open.c
index d038f111c3d45e2e1766557f336b0ca558488c3f..a2dda300684a1a4926e57e7da9bdbf768a91461a 100644
--- a/lustre/mds/mds_open.c
+++ b/lustre/mds/mds_open.c
@@ -307,7 +307,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 };
@@ -318,6 +318,8 @@ static int mds_create_objects(struct ptlrpc_request *req, int offset,
void *lmm_buf;
ENTRY;
+ *objid = NULL;
+
if (!S_ISREG(inode->i_mode))
RETURN(0);
if (rec->ur_flags & MDS_OPEN_DELAY_CREATE ||
@@ -329,11 +331,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) {
@@ -355,8 +353,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);
@@ -364,6 +360,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);
}
@@ -481,16 +478,14 @@ 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);
@@ -673,7 +668,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 = NULL; /* object IDs created */
int rc = 0;
ENTRY;
@@ -706,7 +701,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);
@@ -746,10 +741,8 @@ 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);
- }
+ mds_lov_update_objids(obd, lmm);
+
if (rc) /* coverity[deadcode] */
mds_mfd_unlink(mfd, 1);
diff --git a/lustre/tests/conf-sanity.sh b/lustre/tests/conf-sanity.sh
index ed863b5461ec3ab6d4590b3398999c53b5b5b702..8aad7de125fb04ea3ee4f5a29771610a29c55d66 100644
--- a/lustre/tests/conf-sanity.sh
+++ b/lustre/tests/conf-sanity.sh
@@ -1468,5 +1468,62 @@ test_37() {
}
run_test 37 "verify set tunables works for symlink device"
+test_38() { # bug 14222
+ setup
+ # like runtests
+ COUNT=10
+ SRC="/etc /bin"
+ FILES=`find $SRC -type f -mtime +1 | head -n $COUNT`
+ log "copying $(echo $FILES | wc -w) files to $DIR/$tdir"
+ mkdir -p $DIR/$tdir
+ tar cf - $FILES | tar xf - -C $DIR/$tdir || \
+ error "copying $SRC to $DIR/$tdir"
+ sync
+ umount_client $MOUNT
+ stop_mds
+ log "rename lov_objid file on MDS"
+ rm -f $TMP/lov_objid.orig
+ do_facet mds "debugfs -c -R \\\"dump lov_objid $TMP/lov_objid.orig\\\" $MDSDEV"
+ do_facet mds "debugfs -w -R \\\"rm lov_objid\\\" $MDSDEV"
+
+ do_facet mds "od -Ax -td8 $TMP/lov_objid.orig"
+ # check create in mds_lov_connect
+ start_mds
+ mount_client $MOUNT
+ for f in $FILES; do
+ [ $V ] && log "verifying $DIR/$tdir/$f"
+ diff -q $f $DIR/$tdir/$f || ERROR=y
+ done
+ do_facet mds "debugfs -c -R \\\"dump lov_objid $TMP/lov_objid.new\\\" $MDSDEV"
+ do_facet mds "od -Ax -td8 $TMP/lov_objid.new"
+ [ "$ERROR" = "y" ] && error "old and new files are different after connect" || true
+
+
+ # check it's updates in sync
+ umount_client $MOUNT
+ stop_mds
+
+ do_facet mds dd if=/dev/zero of=$TMP/lov_objid.clear bs=4096 count=1
+ do_facet mds "debugfs -w -R \\\"rm lov_objid\\\" $MDSDEV"
+ do_facet mds "debugfs -w -R \\\"write $TMP/lov_objid.clear lov_objid\\\" $MDSDEV "
+
+ start_mds
+ mount_client $MOUNT
+ for f in $FILES; do
+ [ $V ] && log "verifying $DIR/$tdir/$f"
+ diff -q $f $DIR/$tdir/$f || ERROR=y
+ done
+ do_facet mds "debugfs -c -R \\\"dump lov_objid $TMP/lov_objid.new1\\\" $MDSDEV"
+ do_facet mds "od -Ax -td8 $TMP/lov_objid.new1"
+ umount_client $MOUNT
+ stop_mds
+ [ "$ERROR" = "y" ] && error "old and new files are different after sync" || true
+
+ log "files compared the same"
+ #cleanup
+}
+
+run_test 38 "MDS recreates missing lov_objid file from OST data"
+
equals_msg `basename $0`: test complete
[ -f "$TESTSUITELOG" ] && cat $TESTSUITELOG || true