X-Git-Url: https://git.whamcloud.com/?a=blobdiff_plain;f=lustre%2Fllite%2Fstatahead.c;h=315e7621d777693eb82e16a2ed18ff6d0ea5745b;hb=5d4ae6c905b8e8a38a6e4d3195f550a1e2b37ba8;hp=ca65c0b410b5dd0c21f930d30d594578fafce648;hpb=1c2eb904b1545eb9fc48f910dd5648a6aa174f6a;p=fs%2Flustre-release.git diff --git a/lustre/llite/statahead.c b/lustre/llite/statahead.c index ca65c0b..315e762 100644 --- a/lustre/llite/statahead.c +++ b/lustre/llite/statahead.c @@ -26,7 +26,7 @@ * GPL HEADER END */ /* - * Copyright 2008 Sun Microsystems, Inc. All rights reserved + * Copyright (c) 2008, 2010, Oracle and/or its affiliates. All rights reserved. * Use is subject to license terms. */ /* @@ -55,8 +55,6 @@ struct ll_sai_entry { int se_stat; struct ptlrpc_request *se_req; struct md_enqueue_info *se_minfo; - struct dentry *se_dentry; - struct inode *se_inode; }; enum { @@ -94,7 +92,7 @@ static inline int sa_received_empty(struct ll_statahead_info *sai) static inline int sa_not_full(struct ll_statahead_info *sai) { - return (sai->sai_index < sai->sai_hit + sai->sai_miss + sai->sai_max); + return !!(sai->sai_index < sai->sai_index_next + sai->sai_max); } static inline int sa_is_running(struct ll_statahead_info *sai) @@ -123,38 +121,6 @@ static inline int sa_low_hit(struct ll_statahead_info *sai) (sai->sai_consecutive_miss > 8)); } -static inline int sa_skip_nolock(struct ll_statahead_info *sai) -{ - return (sai->sai_nolock >= 3); -} - -static void ll_sai_entry_free(struct ll_sai_entry *entry) -{ - struct dentry *dentry = entry->se_dentry; - struct inode *inode = entry->se_inode; - - if (dentry) { - struct ll_dentry_data *lld = ll_d2d(dentry); - struct ll_inode_info *lli; - - entry->se_dentry = NULL; - LASSERT(inode != NULL); - lli = ll_i2info(inode); - if (!cfs_list_empty(&lli->lli_sa_dentry)) { - cfs_spin_lock(&lli->lli_sa_lock); - cfs_list_del_init(&lld->lld_sa_alias); - cfs_spin_unlock(&lli->lli_sa_lock); - } - dput(dentry); - } - if (inode) { - entry->se_inode = NULL; - iput(inode); - } - LASSERT(cfs_list_empty(&entry->se_list)); - OBD_FREE_PTR(entry); -} - /** * process the deleted entry's member and free the entry. * (1) release intent @@ -179,8 +145,10 @@ static void ll_sai_entry_cleanup(struct ll_sai_entry *entry, int free) entry->se_req = NULL; ptlrpc_req_finished(req); } - if (free) - ll_sai_entry_free(entry); + if (free) { + LASSERT(cfs_list_empty(&entry->se_list)); + OBD_FREE_PTR(entry); + } EXIT; } @@ -226,16 +194,14 @@ static void ll_sai_put(struct ll_statahead_info *sai) lli = ll_i2info(inode); LASSERT(lli->lli_sai == sai); - if (cfs_atomic_dec_and_test(&sai->sai_refcount)) { + if (cfs_atomic_dec_and_lock(&sai->sai_refcount, &lli->lli_sa_lock)) { struct ll_sai_entry *entry, *next; - cfs_spin_lock(&lli->lli_sa_lock); if (unlikely(cfs_atomic_read(&sai->sai_refcount) > 0)) { /* It is race case, the interpret callback just hold * a reference count */ cfs_spin_unlock(&lli->lli_sa_lock); - EXIT; - return; + RETURN_EXIT; } LASSERT(lli->lli_opendir_key == NULL); @@ -319,7 +285,7 @@ static int ll_sai_entry_fini(struct ll_statahead_info *sai) if (entry->se_index < sai->sai_index_next) { cfs_list_del_init(&entry->se_list); rc = entry->se_stat; - ll_sai_entry_free(entry); + OBD_FREE_PTR(entry); } } else { LASSERT(sa_is_stopped(sai)); @@ -390,7 +356,7 @@ ll_sai_entry_to_stated(struct ll_statahead_info *sai, struct ll_sai_entry *entry /* stale entry */ if (unlikely(entry->se_index < sai->sai_index_next)) { cfs_spin_unlock(&lli->lli_sa_lock); - ll_sai_entry_free(entry); + OBD_FREE_PTR(entry); RETURN(0); } @@ -456,12 +422,9 @@ static int do_statahead_interpret(struct ll_statahead_info *sai) * lookup. */ struct dentry *save = dentry; - __u32 bits = 0; struct it_cb_data icbd = { .icbd_parent = minfo->mi_dir, - .icbd_childp = &dentry, - .icbd_alias = &entry->se_inode, - .bits = &bits + .icbd_childp = &dentry }; LASSERT(fid_is_zero(&minfo->mi_data.op_fid2)); @@ -473,45 +436,14 @@ static int do_statahead_interpret(struct ll_statahead_info *sai) /* Here dentry->d_inode might be NULL, because the entry may * have been removed before we start doing stat ahead. */ - - /* BUG 15962, 21739: since statahead thread does not hold - * parent's i_mutex, it can not alias the dentry to inode. - * Here we just create/update inode in memory, and let the - * main "ls -l" thread to alias such dentry to the inode with - * parent's i_mutex held. - * On the other hand, we hold ldlm ibits lock for the inode - * yet, to allow other operations to cancel such lock in time, - * we should drop the ldlm lock reference count, then the main - * "ls -l" thread should check/get such ldlm ibits lock before - * aliasing such dentry to the inode later. If we don't do such - * drop here, it maybe cause deadlock with i_muext held by - * others, just like bug 21739. */ rc = ll_lookup_it_finish(req, it, &icbd); - if (entry->se_inode != NULL) { - struct ll_dentry_data *lld = ll_d2d(dentry); - struct ll_inode_info *sei = ll_i2info(entry->se_inode); - - /* For statahead lookup case, both MDS_INODELOCK_LOOKUP - * and MDS_INODELOCK_UPDATE should be granted */ - if (likely(bits & MDS_INODELOCK_LOOKUP && - bits & MDS_INODELOCK_UPDATE)) { - /* the first dentry ref_count will be dropped by - * ll_sai_entry_to_stated(), so hold another ref - * in advance */ - entry->se_dentry = dget(dentry); - cfs_spin_lock(&sei->lli_sa_lock); - cfs_list_add(&lld->lld_sa_alias, - &sei->lli_sa_dentry); - cfs_spin_unlock(&sei->lli_sa_lock); - sai->sai_nolock = 0; - } else { - iput(entry->se_inode); - entry->se_inode = NULL; - sai->sai_nolock++; - } + if (!rc) + ll_lookup_finish_locks(it, dentry); + + if (dentry != save) { + minfo->mi_dentry = dentry; + dput(save); } - LASSERT(dentry == save); - ll_intent_drop_lock(it); } else { /* * revalidate. @@ -722,8 +654,7 @@ static int do_sa_revalidate(struct inode *dir, struct dentry *dentry) if (unlikely(dentry == dentry->d_sb->s_root)) RETURN(1); - rc = md_revalidate_lock(ll_i2mdexp(dir), &it, ll_inode2fid(inode), - NULL); + rc = md_revalidate_lock(ll_i2mdexp(dir), &it, ll_inode2fid(inode)); if (rc == 1) { ll_intent_release(&it); RETURN(1); @@ -758,7 +689,7 @@ static int ll_statahead_one(struct dentry *parent, const char* entry_name, struct ll_inode_info *lli = ll_i2info(dir); struct ll_statahead_info *sai = lli->lli_sai; struct qstr name; - struct dentry *dentry; + struct dentry *dentry = NULL; struct ll_sai_entry *se; int rc; ENTRY; @@ -777,36 +708,24 @@ static int ll_statahead_one(struct dentry *parent, const char* entry_name, ll_name2qstr(&name, entry_name, entry_name_len); dentry = d_lookup(parent, &name); if (!dentry) { - if (unlikely(sa_skip_nolock(sai))) { - CWARN("can not obtain lookup lock, skip the succeedent " - "lookup cases, will cause statahead miss, and " - "statahead maybe exit for that.\n"); - GOTO(out, rc = -EAGAIN); - } - dentry = d_alloc(parent, &name); - if (dentry) { - rc = ll_dops_init(dentry, 1); - if (!rc) - rc = do_sa_lookup(dir, dentry); - if (rc) - dput(dentry); - } else { + if (dentry) + rc = do_sa_lookup(dir, dentry); + else GOTO(out, rc = -ENOMEM); - } } else { rc = do_sa_revalidate(dir, dentry); - if (rc) - dput(dentry); } EXIT; out: if (rc) { + if (dentry != NULL) + dput(dentry); + se->se_stat = rc < 0 ? rc : SA_ENTRY_STATED; CDEBUG(D_READA, "set sai entry %p index %u stat %d rc %d\n", se, se->se_index, se->se_stat, rc); - se->se_stat = rc < 0 ? rc : SA_ENTRY_STATED; if (ll_sai_entry_to_stated(sai, se)) cfs_waitq_signal(&sai->sai_waitq); } else { @@ -845,8 +764,10 @@ static int ll_statahead_thread(void *arg) cfs_waitq_signal(&thread->t_ctl_waitq); CDEBUG(D_READA, "start doing statahead for %s\n", parent->d_name.name); + sai->sai_pid = cfs_curproc_pid(); + lli->lli_sa_pos = 0; ll_dir_chain_init(&chain); - page = ll_get_dir_page(dir, pos, 0, &chain); + page = ll_get_dir_page(NULL, dir, pos, 0, &chain); while (1) { struct l_wait_info lwi = { 0 }; @@ -865,15 +786,25 @@ static int ll_statahead_thread(void *arg) dp = page_address(page); for (ent = lu_dirent_start(dp); ent != NULL; ent = lu_dirent_next(ent)) { - char *name = ent->lde_name; - int namelen = le16_to_cpu(ent->lde_namelen); + __u64 hash; + int namelen; + char *name; + + hash = le64_to_cpu(ent->lde_hash); + if (unlikely(hash < pos)) + /* + * Skip until we find target hash value. + */ + continue; + namelen = le16_to_cpu(ent->lde_namelen); if (unlikely(namelen == 0)) /* * Skip dummy record. */ continue; + name = ent->lde_name; if (name[0] == '.') { if (namelen == 1) { /* @@ -951,7 +882,8 @@ keep_de: * chain is exhausted. * Normal case: continue to the next page. */ - page = ll_get_dir_page(dir, pos, 1, &chain); + lli->lli_sa_pos = pos; + page = ll_get_dir_page(NULL, dir, pos, 1, &chain); } else { /* * go into overflow page. @@ -977,9 +909,9 @@ out: /** * called in ll_file_release(). */ -void ll_stop_statahead(struct inode *inode, void *key) +void ll_stop_statahead(struct inode *dir, void *key) { - struct ll_inode_info *lli = ll_i2info(inode); + struct ll_inode_info *lli = ll_i2info(dir); if (unlikely(key == NULL)) return; @@ -1039,6 +971,7 @@ enum { static int is_first_dirent(struct inode *dir, struct dentry *dentry) { + struct ll_inode_info *lli = ll_i2info(dir); struct ll_dir_chain chain; struct qstr *target = &dentry->d_name; struct page *page; @@ -1047,8 +980,9 @@ static int is_first_dirent(struct inode *dir, struct dentry *dentry) int rc = LS_NONE_FIRST_DE; ENTRY; + lli->lli_sa_pos = 0; ll_dir_chain_init(&chain); - page = ll_get_dir_page(dir, pos, 0, &chain); + page = ll_get_dir_page(NULL, dir, pos, 0, &chain); while (1) { struct lu_dirpage *dp; @@ -1068,15 +1002,17 @@ static int is_first_dirent(struct inode *dir, struct dentry *dentry) dp = page_address(page); for (ent = lu_dirent_start(dp); ent != NULL; ent = lu_dirent_next(ent)) { - char *name = ent->lde_name; - int namelen = le16_to_cpu(ent->lde_namelen); + int namelen; + char *name; - if (namelen == 0) + namelen = le16_to_cpu(ent->lde_namelen); + if (unlikely(namelen == 0)) /* * skip dummy record. */ continue; + name = ent->lde_name; if (name[0] == '.') { if (namelen == 1) /* @@ -1124,7 +1060,8 @@ static int is_first_dirent(struct inode *dir, struct dentry *dentry) * chain is exhausted * Normal case: continue to the next page. */ - page = ll_get_dir_page(dir, pos, 1, &chain); + lli->lli_sa_pos = pos; + page = ll_get_dir_page(NULL, dir, pos, 1, &chain); } else { /* * go into overflow page. @@ -1138,23 +1075,11 @@ out: return rc; } -static int is_same_dentry(struct dentry *d1, struct dentry *d2) -{ - if (unlikely(d1 == d2)) - return 1; - if (d1->d_parent == d2->d_parent && - d1->d_name.hash == d2->d_name.hash && - d1->d_name.len == d2->d_name.len && - memcmp(d1->d_name.name, d2->d_name.name, d1->d_name.len) == 0) - return 1; - return 0; -} - /** * Start statahead thread if this is the first dir entry. * Otherwise if a thread is started already, wait it until it is ahead of me. - * \retval 0 -- stat ahead thread process such dentry, miss for lookup - * \retval 1 -- stat ahead thread process such dentry, hit for any case + * \retval 0 -- stat ahead thread process such dentry, for lookup, it miss + * \retval 1 -- stat ahead thread process such dentry, for lookup, it hit * \retval -EEXIST -- stat ahead thread started, and this is the first dentry * \retval -EBADFD -- statahead thread exit and not dentry available * \retval -EAGAIN -- try to stat by caller @@ -1219,119 +1144,6 @@ int do_statahead_enter(struct inode *dir, struct dentry **dentryp, int lookup) RETURN(rc); } - if (ll_sai_entry_stated(sai)) { - struct ll_sai_entry *entry; - - entry = cfs_list_entry(sai->sai_entries_stated.next, - struct ll_sai_entry, se_list); - /* This is for statahead lookup */ - if (entry->se_inode != NULL) { - struct lookup_intent it = {.it_op = IT_LOOKUP}; - struct dentry *dchild = entry->se_dentry; - struct inode *ichild = entry->se_inode; - struct ll_dentry_data *lld = ll_d2d(dchild); - struct ll_inode_info *sei = ll_i2info(ichild); - struct dentry *save = dchild; - int invalid = 0; - __u32 bits = MDS_INODELOCK_LOOKUP | - MDS_INODELOCK_UPDATE; - int found = 0; - - LASSERT(dchild != *dentryp); - - if (!lookup) - mutex_lock(&dir->i_mutex); - - /* - * Make sure dentry is still valid. - * For statahead lookup case, we need both - * LOOKUP lock and UPDATE lock which obtained - * by statahead thread originally. - * - * Consider following racer case: - * 1. statahead thread on client1 get lock with - * both LOOKUK and UPDATE bits for "aaa" - * 2. rename thread on client2 cancel such lock - * from client1, then rename "aaa" to "bbb" - * 3. ls thread on client1 obtain LOOKUP lock - * for "bbb" again - * 4. here the dentry "aaa" created by statahead - * thread should be invalid even related - * LOOKUP lock valid for the same inode - */ - rc = md_revalidate_lock(ll_i2mdexp(dir), &it, - ll_inode2fid(ichild), - &bits); - cfs_spin_lock(&sei->lli_sa_lock); - if (!cfs_list_empty(&lld->lld_sa_alias)) - cfs_list_del_init(&lld->lld_sa_alias); - else - invalid = 1; - cfs_spin_unlock(&sei->lli_sa_lock); - if (rc != 1) - /* Someone has cancelled the original - * lock before the real "revalidate" - * using it. Drop it. */ - goto out_mutex; - - if (invalid) { - /* Someone has cancelled the original - * lock, and reobtained it, the dentry - * maybe invalid anymore, Drop it. */ - ll_intent_drop_lock(&it); - goto out_mutex; - } - - ll_lookup_it_alias(&dchild, ichild, bits); - ll_lookup_finish_locks(&it, dchild); - if (dchild != save) - dput(save); - found = is_same_dentry(dchild, *dentryp); - ichild = NULL; - -out_mutex: - if (!lookup) - mutex_unlock(&dir->i_mutex); - /* Drop the inode reference count held by - * interpreter. */ - if (ichild != NULL) - iput(ichild); - - entry->se_dentry = NULL; - entry->se_inode = NULL; - if (found) { - if (lookup) { - LASSERT(dchild != *dentryp); - /* VFS will drop the reference - * count for dchild and *dentryp - * by itself. */ - *dentryp = dchild; - } else { - LASSERTF(dchild == *dentryp, - "[%.*s/%.*s] " - "[%p "DFID"] " - "[%p "DFID"]\n", - dchild->d_parent->d_name.len, - dchild->d_parent->d_name.name, - dchild->d_name.len, - dchild->d_name.name, - dchild, - PFID(ll_inode2fid(dchild->d_inode)), - *dentryp, - PFID(ll_inode2fid((*dentryp)->d_inode))); - /* Drop the dentry reference - * count held by statahead. */ - dput(dchild); - } - RETURN(1); - } else { - /* Drop the dentry reference count held - * by statahead. */ - dput(dchild); - } - } - } - if (lookup) { struct dentry *result;