X-Git-Url: https://git.whamcloud.com/?a=blobdiff_plain;f=lustre%2Fmds%2Fmds_lmv.c;h=df80c75c9afc991f5fae55fc7132f6886a255d9a;hb=c39489126f88bb5b30643ebb11c72fbe9f9d2241;hp=3267e6fdbe181932cc4ef99a3103ca7fcd45a484;hpb=b2c344740d0c8324565b1243099b612a665aec1f;p=fs%2Flustre-release.git diff --git a/lustre/mds/mds_lmv.c b/lustre/mds/mds_lmv.c index 3267e6f..df80c75 100644 --- a/lustre/mds/mds_lmv.c +++ b/lustre/mds/mds_lmv.c @@ -54,23 +54,29 @@ int mds_lmv_connect(struct obd_device *obd, char * lmv_name) if (IS_ERR(mds->mds_lmv_obd)) RETURN(PTR_ERR(mds->mds_lmv_obd)); - if (mds->mds_lmv_obd) + if (mds->mds_lmv_connected) RETURN(0); + down(&mds->mds_lmv_sem); + if (mds->mds_lmv_connected) { + up(&mds->mds_lmv_sem); + RETURN(0); + } + mds->mds_lmv_obd = class_name2obd(lmv_name); if (!mds->mds_lmv_obd) { CERROR("MDS cannot locate LMV %s\n", lmv_name); mds->mds_lmv_obd = ERR_PTR(-ENOTCONN); - RETURN(-ENOTCONN); + GOTO(err_last, rc = -ENOTCONN); } - rc = obd_connect(&conn, mds->mds_lmv_obd, &obd->obd_uuid); + rc = obd_connect(&conn, mds->mds_lmv_obd, &obd->obd_uuid, OBD_OPT_MDS_CONNECTION); if (rc) { CERROR("MDS cannot connect to LMV %s (%d)\n", lmv_name, rc); mds->mds_lmv_obd = ERR_PTR(rc); - RETURN(rc); + GOTO(err_last, rc); } mds->mds_lmv_exp = class_conn2export(&conn); if (mds->mds_lmv_exp == NULL) @@ -104,7 +110,9 @@ int mds_lmv_connect(struct obd_device *obd, char * lmv_name) "inter_mds", 0, NULL); if (rc) GOTO(err_reg, rc); - + + mds->mds_lmv_connected = 1; + up(&mds->mds_lmv_sem); RETURN(0); err_reg: @@ -113,6 +121,8 @@ err_discon: obd_disconnect(mds->mds_lmv_exp, 0); mds->mds_lmv_exp = NULL; mds->mds_lmv_obd = ERR_PTR(rc); +err_last: + up(&mds->mds_lmv_sem); RETURN(rc); } @@ -135,7 +145,10 @@ int mds_lmv_disconnect(struct obd_device *obd, int flags) int rc = 0; ENTRY; + down(&mds->mds_lmv_sem); if (!IS_ERR(mds->mds_lmv_obd) && mds->mds_lmv_exp != NULL) { + LASSERT(mds->mds_lmv_connected != 0); + mds->mds_lmv_connected = 0; obd_register_observer(mds->mds_lmv_obd, NULL); /* if obd_disconnect fails (probably because the export was @@ -148,6 +161,7 @@ int mds_lmv_disconnect(struct obd_device *obd, int flags) mds->mds_lmv_exp = NULL; mds->mds_lmv_obd = NULL; } + up(&mds->mds_lmv_sem); RETURN(rc); } @@ -161,6 +175,8 @@ int mds_get_lmv_attr(struct obd_device *obd, struct inode *inode, if (!mds->mds_lmv_obd) RETURN(0); + if (!S_ISDIR(inode->i_mode)) + RETURN(0); /* first calculate mea size */ *mea_size = obd_alloc_diskmd(mds->mds_lmv_exp, @@ -168,16 +184,14 @@ int mds_get_lmv_attr(struct obd_device *obd, struct inode *inode, if (*mea_size < 0 || *mea == NULL) return *mea_size < 0 ? *mea_size : -EINVAL; - down(&inode->i_sem); - rc = fsfilt_get_md(obd, inode, *mea, *mea_size); - up(&inode->i_sem); + rc = mds_get_md(obd, inode, *mea, mea_size, 1); if (rc <= 0) { OBD_FREE(*mea, *mea_size); *mea = NULL; } else rc = 0; - + RETURN(rc); } @@ -224,7 +238,7 @@ static int dc_new_page_to_cache(struct dir_cache * dirc) page = alloc_page(GFP_KERNEL); if (page == NULL) return -ENOMEM; - list_add_tail(&page->list, &dirc->list); + list_add_tail(&page->lru, &dirc->list); dirc->cur = page_address(page); dirc->free = PAGE_SIZE; return 0; @@ -287,13 +301,13 @@ static int flush_buffer_onto_mds(struct dirsplit_control *dc, int mdsnum) list_for_each_safe(cur, tmp, &ca->list) { struct page *page; - page = list_entry(cur, struct page, list); + page = list_entry(cur, struct page, lru); LASSERT(page != NULL); retrieve_generation_numbers(dc, page_address(page)); ca->brwc.pg = page; - ca->brwc.off = 0; + ca->brwc.disk_offset = ca->brwc.page_offset = 0; ca->brwc.count = PAGE_SIZE; ca->brwc.flag = 0; ca->oa.o_mds = mdsnum; @@ -320,7 +334,7 @@ static int remove_entries_from_orig_dir(struct dirsplit_control *dc, int mdsnum) ca = dc->cache + mdsnum; list_for_each_safe(cur, tmp, &ca->list) { - page = list_entry(cur, struct page, list); + page = list_entry(cur, struct page, lru); buf = page_address(page); end = buf + PAGE_SIZE; @@ -464,8 +478,8 @@ cleanup: continue; list_for_each_safe(cur, tmp, &dc.cache[i].list) { struct page *page; - page = list_entry(cur, struct page, list); - list_del(&page->list); + page = list_entry(cur, struct page, lru); + list_del(&page->lru); __free_page(page); } } @@ -477,6 +491,8 @@ cleanup: #define MAX_DIR_SIZE (64 * 1024) +#define I_NON_SPLITTABLE 256 + int mds_splitting_expected(struct obd_device *obd, struct dentry *dentry) { struct mds_obd *mds = &obd->u.mds; @@ -495,6 +511,10 @@ int mds_splitting_expected(struct obd_device *obd, struct dentry *dentry) if (!S_ISDIR(dentry->d_inode->i_mode)) return MDS_NO_SPLITTABLE; + /* already splittied or slave directory (part of splitted dir) */ + if (dentry->d_inode->i_flags & I_NON_SPLITTABLE) + return MDS_NO_SPLITTABLE; + /* don't split root directory */ if (dentry->d_inode->i_ino == mds->mds_rootfid.id) return MDS_NO_SPLITTABLE; @@ -507,6 +527,8 @@ int mds_splitting_expected(struct obd_device *obd, struct dentry *dentry) if (mea) { /* already splitted or slave object: shouldn't be splitted */ rc = MDS_NO_SPLITTABLE; + /* mark to skip subsequent checks */ + dentry->d_inode->i_flags |= I_NON_SPLITTABLE; } else { /* may be splitted */ rc = MDS_EXPECT_SPLIT; @@ -521,7 +543,7 @@ int mds_splitting_expected(struct obd_device *obd, struct dentry *dentry) * must not be called on already splitted directories. */ int mds_try_to_split_dir(struct obd_device *obd, struct dentry *dentry, - struct mea **mea, int nstripes) + struct mea **mea, int nstripes, int update_mode) { struct inode *dir = dentry->d_inode; struct mds_obd *mds = &obd->u.mds; @@ -531,12 +553,16 @@ int mds_try_to_split_dir(struct obd_device *obd, struct dentry *dentry, void *handle; ENTRY; + if (update_mode != LCK_EX) + return 0; /* TODO: optimization possible - we already may have mea here */ rc = mds_splitting_expected(obd, dentry); if (rc == MDS_NO_SPLITTABLE) return 0; if (rc == MDS_NO_SPLIT_EXPECTED && nstripes == 0) return 0; + if (nstripes && nstripes == 1) + return 0; LASSERT(mea == NULL || *mea == NULL); @@ -673,7 +699,11 @@ int mds_preprw(int cmd, struct obd_export *exp, struct obdo *oa, fid.id = obj->ioo_id; fid.generation = obj->ioo_gr; dentry = mds_fid2dentry(mds, &fid, NULL); - LASSERT(!IS_ERR(dentry)); + if (IS_ERR(dentry)) { + CERROR("can't get dentry for "LPU64"/%u: %d\n", + fid.id, fid.generation, (int) PTR_ERR(dentry)); + GOTO(cleanup, rc = (int) PTR_ERR(dentry)); + } if (dentry->d_inode == NULL) { CERROR("trying to BRW to non-existent file "LPU64"\n", @@ -748,6 +778,18 @@ int mds_commitrw(int cmd, struct obd_export *exp, struct obdo *oa, err = fsfilt_add_dir_entry(obd, res->dentry, de->name, de->namelen, de->ino, de->generation, de->mds); + if (err) { + CERROR("can't add dir entry %*s->%u/%u/%u" + " to %lu/%u: %d\n", + de->namelen, de->name, + de->mds, (unsigned) de->ino, + (unsigned) de->generation, + res->dentry->d_inode->i_ino, + res->dentry->d_inode->i_generation, + err); + rc = err; + break; + } LASSERT(err == 0); de = (struct dir_entry *) ((char *) de + DIR_REC_LEN(de->namelen)); @@ -773,7 +815,7 @@ int mds_choose_mdsnum(struct obd_device *obd, const char *name, int len, int fla i = mds->mds_num; } else if (mds->mds_lmv_exp) { lmv = &mds->mds_lmv_exp->exp_obd->u.lmv; - i = raw_name2idx(lmv->desc.ld_tgt_count, name, len); + i = raw_name2idx(MEA_MAGIC_LAST_CHAR, lmv->desc.ld_tgt_count, name, len); } RETURN(i); } @@ -974,11 +1016,19 @@ int mds_lock_and_check_slave(int offset, struct ptlrpc_request *req, struct dentry *dentry = NULL; struct lvfs_run_ctxt saved; int cleanup_phase = 0; + struct mds_req_sec_desc *rsd; struct mds_body *body; struct lvfs_ucred uc; int rc, update_mode; ENTRY; + rsd = lustre_swab_mds_secdesc(req, MDS_REQ_SECDESC_OFF); + if (!rsd) { + CERROR("Can't unpack security desc\n"); + GOTO(cleanup, rc = -EFAULT); + } + mds_squash_root(&obd->u.mds, rsd, &req->rq_peer.peer_id.nid); + body = lustre_swab_reqbuf(req, offset, sizeof(*body), lustre_swab_mds_body); if (body == NULL) { @@ -999,11 +1049,12 @@ int mds_lock_and_check_slave(int offset, struct ptlrpc_request *req, LASSERT(S_ISDIR(dentry->d_inode->i_mode)); - uc.luc_fsuid = body->fsuid; - uc.luc_fsgid = body->fsgid; - uc.luc_cap = body->capability; - uc.luc_suppgid1 = body->suppgid; - uc.luc_suppgid2 = -1; + rc = mds_init_ucred(&uc, rsd); + if (rc) { + CERROR("can't init ucred\n"); + GOTO(cleanup, rc); + } + push_ctxt(&saved, &obd->obd_lvfs_ctxt, &uc); rc = 0; @@ -1017,9 +1068,69 @@ cleanup: ldlm_lock_decref(lockh, LCK_EX); l_dput(dentry); pop_ctxt(&saved, &obd->obd_lvfs_ctxt, &uc); + mds_exit_ucred(&uc); default: break; } RETURN(rc); } +int mds_convert_mea_ea(struct obd_device *obd, struct inode *inode, + struct lov_mds_md *lmm, int lmmsize) +{ + int i, rc, err, size; + struct mea_old *old; + struct mea *mea; + struct mea *new; + void *handle; + ENTRY; + + mea = (struct mea *) lmm; + if (mea->mea_magic == MEA_MAGIC_LAST_CHAR || + mea->mea_magic == MEA_MAGIC_ALL_CHARS) + RETURN(0); + + old = (struct mea_old *) lmm; + rc = sizeof(struct ll_fid) * old->mea_count + sizeof(struct mea_old); + if (old->mea_count > 256 || old->mea_master > 256 || lmmsize < rc + || old->mea_master > old->mea_count) { + CWARN("unknown MEA format, dont convert it\n"); + CWARN(" count %u, master %u, size %u\n", + old->mea_count, old->mea_master, rc); + RETURN(0); + } + + CWARN("converting MEA EA on %lu/%u from V0 to V1 (%u/%u)\n", + inode->i_ino, inode->i_generation, old->mea_count, old->mea_master); + + size = sizeof(struct ll_fid) * old->mea_count + sizeof(struct mea); + OBD_ALLOC(new, size); + if (new == NULL) + RETURN(-ENOMEM); + + new->mea_magic = MEA_MAGIC_LAST_CHAR; + new->mea_count = old->mea_count; + new->mea_master = old->mea_master; + for (i = 0; i < new->mea_count; i++) + new->mea_fids[i] = old->mea_fids[i]; + + handle = fsfilt_start(obd, inode, FSFILT_OP_SETATTR, NULL); + if (IS_ERR(handle)) { + rc = PTR_ERR(handle); + GOTO(conv_free, rc); + } + + rc = fsfilt_set_md(obd, inode, handle, (struct lov_mds_md *) new, size); + if (rc > lmmsize) + size = lmmsize; + memcpy(lmm, new, size); + + err = fsfilt_commit(obd, obd->u.mds.mds_sb, inode, handle, 0); + if (!rc) + rc = err ? err : size; + GOTO(conv_free, rc); +conv_free: + OBD_FREE(new, size); + return rc; +} +