struct ll_sa_entry {
/* link into sai->sai_entries */
- cfs_list_t se_link;
+ struct list_head se_link;
/* link into sai->sai_entries_{received,stated} */
- cfs_list_t se_list;
+ struct list_head se_list;
/* link into sai hash table locally */
- cfs_list_t se_hash;
+ struct list_head se_hash;
/* entry reference count */
- cfs_atomic_t se_refcount;
+ atomic_t se_refcount;
/* entry index in the sai */
__u64 se_index;
/* low layer ldlm lock handle */
static inline int ll_sa_entry_unhashed(struct ll_sa_entry *entry)
{
- return cfs_list_empty(&entry->se_hash);
+ return list_empty(&entry->se_hash);
}
/*
int i = ll_sa_entry_hash(entry->se_qstr.hash);
spin_lock(&sai->sai_cache_lock[i]);
- cfs_list_add_tail(&entry->se_hash, &sai->sai_cache[i]);
+ list_add_tail(&entry->se_hash, &sai->sai_cache[i]);
spin_unlock(&sai->sai_cache_lock[i]);
}
int i = ll_sa_entry_hash(entry->se_qstr.hash);
spin_lock(&sai->sai_cache_lock[i]);
- cfs_list_del_init(&entry->se_hash);
+ list_del_init(&entry->se_hash);
spin_unlock(&sai->sai_cache_lock[i]);
}
static inline struct ll_sa_entry *
sa_first_received_entry(struct ll_statahead_info *sai)
{
- return cfs_list_entry(sai->sai_entries_received.next,
- struct ll_sa_entry, se_list);
+ return list_entry(sai->sai_entries_received.next,
+ struct ll_sa_entry, se_list);
}
static inline struct ll_inode_info *
agl_first_entry(struct ll_statahead_info *sai)
{
- return cfs_list_entry(sai->sai_entries_agl.next,
- struct ll_inode_info, lli_agl_list);
+ return list_entry(sai->sai_entries_agl.next,
+ struct ll_inode_info, lli_agl_list);
}
static inline int sa_sent_full(struct ll_statahead_info *sai)
{
- return cfs_atomic_read(&sai->sai_cache_count) >= sai->sai_max;
+ return atomic_read(&sai->sai_cache_count) >= sai->sai_max;
}
static inline int sa_received_empty(struct ll_statahead_info *sai)
{
- return cfs_list_empty(&sai->sai_entries_received);
+ return list_empty(&sai->sai_entries_received);
}
static inline int agl_list_empty(struct ll_statahead_info *sai)
{
- return cfs_list_empty(&sai->sai_entries_agl);
+ return list_empty(&sai->sai_entries_agl);
}
/**
entry->se_index = index;
- /*
- * Statahead entry reference rules:
- *
- * 1) When statahead entry is initialized, its reference is set as 2.
- * One reference is used by the directory scanner. When the scanner
- * searches the statahead cache for the given name, it can perform
- * lockless hash lookup (only the scanner can remove entry from hash
- * list), and once found, it needn't to call "atomic_inc()" for the
- * entry reference. So the performance is improved. After using the
- * statahead entry, the scanner will call "atomic_dec()" to drop the
- * reference held when initialization. If it is the last reference,
- * the statahead entry will be freed.
- *
- * 2) All other threads, including statahead thread and ptlrpcd thread,
- * when they process the statahead entry, the reference for target
- * should be held to guarantee the entry will not be released by the
- * directory scanner. After processing the entry, these threads will
- * drop the entry reference. If it is the last reference, the entry
- * will be freed.
- *
- * The second reference when initializes the statahead entry is used
- * by the statahead thread, following the rule 2).
- */
- cfs_atomic_set(&entry->se_refcount, 2);
- entry->se_stat = SA_ENTRY_INIT;
- entry->se_size = entry_size;
- dname = (char *)entry + sizeof(struct ll_sa_entry);
- memcpy(dname, name, len);
- dname[len] = 0;
- entry->se_qstr.hash = full_name_hash(name, len);
- entry->se_qstr.len = len;
- entry->se_qstr.name = dname;
-
- lli = ll_i2info(sai->sai_inode);
+ /*
+ * Statahead entry reference rules:
+ *
+ * 1) When statahead entry is initialized, its reference is set as 2.
+ * One reference is used by the directory scanner. When the scanner
+ * searches the statahead cache for the given name, it can perform
+ * lockless hash lookup (only the scanner can remove entry from hash
+ * list), and once found, it needn't to call "atomic_inc()" for the
+ * entry reference. So the performance is improved. After using the
+ * statahead entry, the scanner will call "atomic_dec()" to drop the
+ * reference held when initialization. If it is the last reference,
+ * the statahead entry will be freed.
+ *
+ * 2) All other threads, including statahead thread and ptlrpcd thread,
+ * when they process the statahead entry, the reference for target
+ * should be held to guarantee the entry will not be released by the
+ * directory scanner. After processing the entry, these threads will
+ * drop the entry reference. If it is the last reference, the entry
+ * will be freed.
+ *
+ * The second reference when initializes the statahead entry is used
+ * by the statahead thread, following the rule 2).
+ */
+ atomic_set(&entry->se_refcount, 2);
+ entry->se_stat = SA_ENTRY_INIT;
+ entry->se_size = entry_size;
+ dname = (char *)entry + sizeof(struct ll_sa_entry);
+ memcpy(dname, name, len);
+ dname[len] = 0;
+ entry->se_qstr.hash = full_name_hash(name, len);
+ entry->se_qstr.len = len;
+ entry->se_qstr.name = dname;
+
+ lli = ll_i2info(sai->sai_inode);
spin_lock(&lli->lli_sa_lock);
- cfs_list_add_tail(&entry->se_link, &sai->sai_entries);
- CFS_INIT_LIST_HEAD(&entry->se_list);
+ list_add_tail(&entry->se_link, &sai->sai_entries);
+ INIT_LIST_HEAD(&entry->se_list);
ll_sa_entry_enhash(sai, entry);
spin_unlock(&lli->lli_sa_lock);
- cfs_atomic_inc(&sai->sai_cache_count);
+ atomic_inc(&sai->sai_cache_count);
RETURN(entry);
}
struct ll_sa_entry *entry;
int i = ll_sa_entry_hash(qstr->hash);
- cfs_list_for_each_entry(entry, &sai->sai_cache[i], se_hash) {
+ list_for_each_entry(entry, &sai->sai_cache[i], se_hash) {
if (entry->se_qstr.hash == qstr->hash &&
entry->se_qstr.len == qstr->len &&
memcmp(entry->se_qstr.name, qstr->name, qstr->len) == 0)
{
struct ll_sa_entry *entry;
- cfs_list_for_each_entry(entry, &sai->sai_entries, se_link) {
+ list_for_each_entry(entry, &sai->sai_entries, se_link) {
if (entry->se_index == index) {
LASSERT(atomic_read(&entry->se_refcount) > 0);
- cfs_atomic_inc(&entry->se_refcount);
+ atomic_inc(&entry->se_refcount);
return entry;
}
if (entry->se_index > index)
static void ll_sa_entry_put(struct ll_statahead_info *sai,
struct ll_sa_entry *entry)
{
- if (cfs_atomic_dec_and_test(&entry->se_refcount)) {
+ if (atomic_dec_and_test(&entry->se_refcount)) {
CDEBUG(D_READA, "free sa entry %.*s(%p) index "LPU64"\n",
entry->se_qstr.len, entry->se_qstr.name, entry,
entry->se_index);
- LASSERT(cfs_list_empty(&entry->se_link));
- LASSERT(cfs_list_empty(&entry->se_list));
+ LASSERT(list_empty(&entry->se_link));
+ LASSERT(list_empty(&entry->se_list));
LASSERT(ll_sa_entry_unhashed(entry));
ll_sa_entry_cleanup(sai, entry);
iput(entry->se_inode);
OBD_FREE(entry, entry->se_size);
- cfs_atomic_dec(&sai->sai_cache_count);
+ atomic_dec(&sai->sai_cache_count);
}
}
struct ll_inode_info *lli = ll_i2info(sai->sai_inode);
LASSERT(!ll_sa_entry_unhashed(entry));
- LASSERT(!cfs_list_empty(&entry->se_link));
+ LASSERT(!list_empty(&entry->se_link));
ll_sa_entry_unhash(sai, entry);
spin_lock(&lli->lli_sa_lock);
entry->se_stat = SA_ENTRY_DEST;
- cfs_list_del_init(&entry->se_link);
- if (likely(!cfs_list_empty(&entry->se_list)))
- cfs_list_del_init(&entry->se_list);
+ list_del_init(&entry->se_link);
+ if (likely(!list_empty(&entry->se_list)))
+ list_del_init(&entry->se_list);
spin_unlock(&lli->lli_sa_lock);
ll_sa_entry_put(sai, entry);
do_sa_entry_fini(sai, entry);
/* drop old entry, only 'scanner' process does this, no need to lock */
- cfs_list_for_each_entry_safe(pos, next, &sai->sai_entries, se_link) {
+ list_for_each_entry_safe(pos, next, &sai->sai_entries, se_link) {
if (!is_omitted_entry(sai, pos->se_index))
break;
do_sa_entry_fini(sai, pos);
struct ll_sa_entry *entry, se_stat_t stat)
{
struct ll_sa_entry *se;
- cfs_list_t *pos = &sai->sai_entries_stated;
+ struct list_head *pos = &sai->sai_entries_stated;
- if (!cfs_list_empty(&entry->se_list))
- cfs_list_del_init(&entry->se_list);
+ if (!list_empty(&entry->se_list))
+ list_del_init(&entry->se_list);
- cfs_list_for_each_entry_reverse(se, &sai->sai_entries_stated, se_list) {
+ list_for_each_entry_reverse(se, &sai->sai_entries_stated, se_list) {
if (se->se_index < entry->se_index) {
pos = &se->se_list;
break;
}
}
- cfs_list_add(&entry->se_list, pos);
+ list_add(&entry->se_list, pos);
entry->se_stat = stat;
}
child->lli_agl_index = index;
spin_unlock(&child->lli_agl_lock);
- LASSERT(cfs_list_empty(&child->lli_agl_list));
+ LASSERT(list_empty(&child->lli_agl_list));
igrab(inode);
spin_lock(&parent->lli_agl_lock);
if (agl_list_empty(sai))
added = 1;
- cfs_list_add_tail(&child->lli_agl_list, &sai->sai_entries_agl);
+ list_add_tail(&child->lli_agl_list, &sai->sai_entries_agl);
spin_unlock(&parent->lli_agl_lock);
} else {
spin_unlock(&child->lli_agl_lock);
if (!sai)
RETURN(NULL);
- cfs_atomic_set(&sai->sai_refcount, 1);
+ atomic_set(&sai->sai_refcount, 1);
spin_lock(&sai_generation_lock);
sai->sai_generation = ++sai_generation;
init_waitqueue_head(&sai->sai_thread.t_ctl_waitq);
init_waitqueue_head(&sai->sai_agl_thread.t_ctl_waitq);
- CFS_INIT_LIST_HEAD(&sai->sai_entries);
- CFS_INIT_LIST_HEAD(&sai->sai_entries_received);
- CFS_INIT_LIST_HEAD(&sai->sai_entries_stated);
- CFS_INIT_LIST_HEAD(&sai->sai_entries_agl);
+ INIT_LIST_HEAD(&sai->sai_entries);
+ INIT_LIST_HEAD(&sai->sai_entries_received);
+ INIT_LIST_HEAD(&sai->sai_entries_stated);
+ INIT_LIST_HEAD(&sai->sai_entries_agl);
for (i = 0; i < LL_SA_CACHE_SIZE; i++) {
- CFS_INIT_LIST_HEAD(&sai->sai_cache[i]);
+ INIT_LIST_HEAD(&sai->sai_cache[i]);
spin_lock_init(&sai->sai_cache_lock[i]);
}
- cfs_atomic_set(&sai->sai_cache_count, 0);
+ atomic_set(&sai->sai_cache_count, 0);
RETURN(sai);
}
static inline struct ll_statahead_info *
ll_sai_get(struct ll_statahead_info *sai)
{
- cfs_atomic_inc(&sai->sai_refcount);
- return sai;
+ atomic_inc(&sai->sai_refcount);
+ return sai;
}
static void ll_sai_put(struct ll_statahead_info *sai)
{
- struct inode *inode = sai->sai_inode;
- struct ll_inode_info *lli = ll_i2info(inode);
- ENTRY;
+ struct inode *inode = sai->sai_inode;
+ struct ll_inode_info *lli = ll_i2info(inode);
+ ENTRY;
- if (cfs_atomic_dec_and_lock(&sai->sai_refcount, &lli->lli_sa_lock)) {
- struct ll_sa_entry *entry, *next;
+ if (atomic_dec_and_lock(&sai->sai_refcount, &lli->lli_sa_lock)) {
+ struct ll_sa_entry *entry, *next;
- if (unlikely(cfs_atomic_read(&sai->sai_refcount) > 0)) {
- /* It is race case, the interpret callback just hold
- * a reference count */
+ if (unlikely(atomic_read(&sai->sai_refcount) > 0)) {
+ /* It is race case, the interpret callback just hold
+ * a reference count */
spin_unlock(&lli->lli_sa_lock);
RETURN_EXIT;
}
PFID(&lli->lli_fid),
sai->sai_sent, sai->sai_replied);
- cfs_list_for_each_entry_safe(entry, next,
- &sai->sai_entries, se_link)
+ list_for_each_entry_safe(entry, next,
+ &sai->sai_entries, se_link)
do_sa_entry_fini(sai, entry);
LASSERT(list_empty(&sai->sai_entries));
LASSERT(sa_received_empty(sai));
LASSERT(list_empty(&sai->sai_entries_stated));
- LASSERT(cfs_atomic_read(&sai->sai_cache_count) == 0);
- LASSERT(agl_list_empty(sai));
+ LASSERT(atomic_read(&sai->sai_cache_count) == 0);
+ LASSERT(agl_list_empty(sai));
- iput(inode);
- OBD_FREE_PTR(sai);
- }
+ iput(inode);
+ OBD_FREE_PTR(sai);
+ }
- EXIT;
+ EXIT;
}
/* Do NOT forget to drop inode refcount when into sai_entries_agl. */
int rc;
ENTRY;
- LASSERT(cfs_list_empty(&lli->lli_agl_list));
+ LASSERT(list_empty(&lli->lli_agl_list));
/* AGL maybe fall behind statahead with one entry */
if (is_omitted_entry(sai, index + 1)) {
RETURN_EXIT;
}
entry = sa_first_received_entry(sai);
- cfs_atomic_inc(&entry->se_refcount);
- cfs_list_del_init(&entry->se_list);
+ atomic_inc(&entry->se_refcount);
+ list_del_init(&entry->se_list);
spin_unlock(&lli->lli_sa_lock);
LASSERT(entry->se_handle != 0);
/* XXX: No fid in reply, this is probaly cross-ref case.
* SA can't handle it yet. */
- if (body->valid & OBD_MD_MDS)
- GOTO(out, rc = -EAGAIN);
- } else {
- /*
- * revalidate.
- */
- /* unlinked and re-created with the same name */
- if (unlikely(!lu_fid_eq(&minfo->mi_data.op_fid2, &body->fid1))){
+ if (body->mbo_valid & OBD_MD_MDS)
+ GOTO(out, rc = -EAGAIN);
+ } else {
+ /*
+ * revalidate.
+ */
+ /* unlinked and re-created with the same name */
+ if (unlikely(!lu_fid_eq(&minfo->mi_data.op_fid2,
+ &body->mbo_fid1))) {
entry->se_inode = NULL;
iput(child);
child = NULL;
entry->se_handle = it->d.lustre.it_lock_handle;
ll_intent_drop_lock(it);
wakeup = sa_received_empty(sai);
- cfs_list_add_tail(&entry->se_list,
+ list_add_tail(&entry->se_list,
&sai->sai_entries_received);
}
sai->sai_replied++;
struct l_wait_info lwi = { 0 };
ENTRY;
- CDEBUG(D_READA, "agl thread started: [pid %d] [parent %.*s]\n",
- current_pid(), parent->d_name.len, parent->d_name.name);
+ thread->t_pid = current_pid();
+ CDEBUG(D_READA, "agl thread started: sai %p, parent %.*s\n",
+ sai, parent->d_name.len, parent->d_name.name);
atomic_inc(&sbi->ll_agl_total);
spin_lock(&plli->lli_agl_lock);
sai->sai_agl_valid = 1;
- thread_set_flags(thread, SVC_RUNNING);
+ if (thread_is_init(thread))
+ /* If someone else has changed the thread state
+ * (e.g. already changed to SVC_STOPPING), we can't just
+ * blindly overwrite that setting. */
+ thread_set_flags(thread, SVC_RUNNING);
spin_unlock(&plli->lli_agl_lock);
wake_up(&thread->t_ctl_waitq);
* so check whether list empty again. */
if (!agl_list_empty(sai)) {
clli = agl_first_entry(sai);
- cfs_list_del_init(&clli->lli_agl_list);
+ list_del_init(&clli->lli_agl_list);
spin_unlock(&plli->lli_agl_lock);
ll_agl_trigger(&clli->lli_vfs_inode, sai);
} else {
sai->sai_agl_valid = 0;
while (!agl_list_empty(sai)) {
clli = agl_first_entry(sai);
- cfs_list_del_init(&clli->lli_agl_list);
+ list_del_init(&clli->lli_agl_list);
spin_unlock(&plli->lli_agl_lock);
clli->lli_agl_index = 0;
iput(&clli->lli_vfs_inode);
spin_unlock(&plli->lli_agl_lock);
wake_up(&thread->t_ctl_waitq);
ll_sai_put(sai);
- CDEBUG(D_READA, "agl thread stopped: [pid %d] [parent %.*s]\n",
- current_pid(), parent->d_name.len, parent->d_name.name);
+ CDEBUG(D_READA, "agl thread stopped: sai %p, parent %.*s\n",
+ sai, parent->d_name.len, parent->d_name.name);
RETURN(0);
}
struct task_struct *task;
ENTRY;
- CDEBUG(D_READA, "start agl thread: [pid %d] [parent %.*s]\n",
- current_pid(), parent->d_name.len, parent->d_name.name);
+ CDEBUG(D_READA, "start agl thread: sai %p, parent %.*s\n",
+ sai, parent->d_name.len, parent->d_name.name);
plli = ll_i2info(parent->d_inode);
task = kthread_run(ll_agl_thread, parent,
struct md_op_data *op_data;
struct ll_dir_chain chain;
struct l_wait_info lwi = { 0 };
- struct lu_dirent *ent;
struct page *page = NULL;
+ __u64 pos = 0;
ENTRY;
- CDEBUG(D_READA, "statahead thread started: [pid %d] [parent %.*s]\n",
- current_pid(), parent->d_name.len, parent->d_name.name);
+ thread->t_pid = current_pid();
+ CDEBUG(D_READA, "statahead thread starting: sai %p, parent %.*s\n",
+ sai, parent->d_name.len, parent->d_name.name);
op_data = ll_prep_md_op_data(NULL, dir, dir, NULL, 0, 0,
LUSTRE_OPC_ANY, dir);
if (IS_ERR(op_data))
- RETURN(PTR_ERR(op_data));
+ GOTO(out_put, rc = PTR_ERR(op_data));
- op_data->op_hash_offset = 0;
- op_data->op_max_pages =
- ll_i2sbi(dir)->ll_md_brw_size >> PAGE_CACHE_SHIFT;
+ op_data->op_max_pages = ll_i2sbi(dir)->ll_md_brw_pages;
if (sbi->ll_flags & LL_SBI_AGL_ENABLED)
ll_start_agl(parent, sai);
atomic_inc(&sbi->ll_sa_total);
spin_lock(&plli->lli_sa_lock);
- thread_set_flags(thread, SVC_RUNNING);
+ if (thread_is_init(thread))
+ /* If someone else has changed the thread state
+ * (e.g. already changed to SVC_STOPPING), we can't just
+ * blindly overwrite that setting. */
+ thread_set_flags(thread, SVC_RUNNING);
spin_unlock(&plli->lli_sa_lock);
wake_up(&thread->t_ctl_waitq);
ll_dir_chain_init(&chain);
- for (ent = ll_dir_entry_start(dir, op_data, &page);
- ent != NULL && !IS_ERR(ent);
- ent = ll_dir_entry_next(dir, op_data, ent, &page)) {
- __u64 hash;
- int namelen;
- char *name;
-
- hash = le64_to_cpu(ent->lde_hash);
- if (unlikely(hash < op_data->op_hash_offset))
- /*
- * Skip until we find target hash value.
- */
- continue;
+ page = ll_get_dir_page(dir, op_data, pos, &chain);
+ while (1) {
+ struct lu_dirpage *dp;
+ struct lu_dirent *ent;
+
+ if (IS_ERR(page)) {
+ rc = PTR_ERR(page);
+ CDEBUG(D_READA, "error reading dir "DFID" at "LPU64
+ "/"LPU64" opendir_pid = %u: rc = %d\n",
+ PFID(ll_inode2fid(dir)), pos, sai->sai_index,
+ plli->lli_opendir_pid, rc);
+ GOTO(out, rc);
+ }
- namelen = le16_to_cpu(ent->lde_namelen);
- if (unlikely(namelen == 0))
- /*
- * Skip dummy record.
- */
- continue;
+ dp = page_address(page);
+ for (ent = lu_dirent_start(dp); ent != NULL;
+ ent = lu_dirent_next(ent)) {
+ __u64 hash;
+ int namelen;
+ char *name;
- name = ent->lde_name;
- if (name[0] == '.') {
- if (namelen == 1) {
- /*
- * skip "."
- */
- continue;
- } else if (name[1] == '.' && namelen == 2) {
+ hash = le64_to_cpu(ent->lde_hash);
+ if (unlikely(hash < pos))
/*
- * skip ".."
+ * Skip until we find target hash value.
*/
continue;
- } else if (!sai->sai_ls_all) {
+
+ namelen = le16_to_cpu(ent->lde_namelen);
+ if (unlikely(namelen == 0))
/*
- * skip hidden files.
+ * Skip dummy record.
*/
- sai->sai_skip_hidden++;
continue;
+
+ name = ent->lde_name;
+ if (name[0] == '.') {
+ if (namelen == 1) {
+ /*
+ * skip "."
+ */
+ continue;
+ } else if (name[1] == '.' && namelen == 2) {
+ /*
+ * skip ".."
+ */
+ continue;
+ } else if (!sai->sai_ls_all) {
+ /*
+ * skip hidden files.
+ */
+ sai->sai_skip_hidden++;
+ continue;
+ }
}
- }
- /*
- * don't stat-ahead first entry.
- */
- if (unlikely(++first == 1))
- continue;
+ /*
+ * don't stat-ahead first entry.
+ */
+ if (unlikely(++first == 1))
+ continue;
keep_it:
- l_wait_event(thread->t_ctl_waitq,
- !sa_sent_full(sai) ||
- !sa_received_empty(sai) ||
- !agl_list_empty(sai) ||
- !thread_is_running(thread),
- &lwi);
+ l_wait_event(thread->t_ctl_waitq,
+ !sa_sent_full(sai) ||
+ !sa_received_empty(sai) ||
+ !agl_list_empty(sai) ||
+ !thread_is_running(thread),
+ &lwi);
interpret_it:
- while (!sa_received_empty(sai))
- ll_post_statahead(sai);
+ while (!sa_received_empty(sai))
+ ll_post_statahead(sai);
- if (unlikely(!thread_is_running(thread)))
- GOTO(out, rc = 0);
+ if (unlikely(!thread_is_running(thread))) {
+ ll_release_page(dir, page, false);
+ GOTO(out, rc = 0);
+ }
- /* If no window for metadata statahead, but there are
- * some AGL entries to be triggered, then try to help
- * to process the AGL entries. */
- if (sa_sent_full(sai)) {
- spin_lock(&plli->lli_agl_lock);
- while (!agl_list_empty(sai)) {
- clli = agl_first_entry(sai);
- cfs_list_del_init(&clli->lli_agl_list);
+ /* If no window for metadata statahead, but there are
+ * some AGL entries to be triggered, then try to help
+ * to process the AGL entries. */
+ if (sa_sent_full(sai)) {
+ spin_lock(&plli->lli_agl_lock);
+ while (!agl_list_empty(sai)) {
+ clli = agl_first_entry(sai);
+ list_del_init(&clli->lli_agl_list);
+ spin_unlock(&plli->lli_agl_lock);
+ ll_agl_trigger(&clli->lli_vfs_inode,
+ sai);
+
+ if (!sa_received_empty(sai))
+ goto interpret_it;
+
+ if (unlikely(
+ !thread_is_running(thread))) {
+ ll_release_page(dir, page,
+ false);
+ GOTO(out, rc = 0);
+ }
+
+ if (!sa_sent_full(sai))
+ goto do_it;
+
+ spin_lock(&plli->lli_agl_lock);
+ }
spin_unlock(&plli->lli_agl_lock);
- ll_agl_trigger(&clli->lli_vfs_inode,
- sai);
- if (!sa_received_empty(sai))
- goto interpret_it;
+ goto keep_it;
+ }
+do_it:
+ ll_statahead_one(parent, name, namelen);
+ }
- if (unlikely(
- !thread_is_running(thread)))
+ pos = le64_to_cpu(dp->ldp_hash_end);
+ if (pos == MDS_DIR_END_OFF) {
+ /*
+ * End of directory reached.
+ */
+ ll_release_page(dir, page, false);
+ while (1) {
+ l_wait_event(thread->t_ctl_waitq,
+ !sa_received_empty(sai) ||
+ sai->sai_sent == sai->sai_replied ||
+ !thread_is_running(thread),
+ &lwi);
+
+ while (!sa_received_empty(sai))
+ ll_post_statahead(sai);
+
+ if (unlikely(!thread_is_running(thread)))
GOTO(out, rc = 0);
- if (!sa_sent_full(sai))
- goto do_it;
+ if (sai->sai_sent == sai->sai_replied &&
+ sa_received_empty(sai))
+ break;
+ }
+ spin_lock(&plli->lli_agl_lock);
+ while (!agl_list_empty(sai) &&
+ thread_is_running(thread)) {
+ clli = agl_first_entry(sai);
+ list_del_init(&clli->lli_agl_list);
+ spin_unlock(&plli->lli_agl_lock);
+ ll_agl_trigger(&clli->lli_vfs_inode, sai);
spin_lock(&plli->lli_agl_lock);
}
spin_unlock(&plli->lli_agl_lock);
- goto keep_it;
- }
-
-do_it:
- ll_statahead_one(parent, name, namelen);
- }
-
- if (page != NULL) {
- kunmap(page);
- page_cache_release(page);
- }
-
- /*
- * End of directory reached.
- */
- while (1) {
- l_wait_event(thread->t_ctl_waitq,
- !sa_received_empty(sai) ||
- sai->sai_sent == sai->sai_replied ||
- !thread_is_running(thread),
- &lwi);
-
- while (!sa_received_empty(sai))
- ll_post_statahead(sai);
-
- if (unlikely(!thread_is_running(thread)))
GOTO(out, rc = 0);
-
- if (sai->sai_sent == sai->sai_replied &&
- sa_received_empty(sai))
- break;
- }
-
- spin_lock(&plli->lli_agl_lock);
- while (!agl_list_empty(sai) &&
- thread_is_running(thread)) {
- clli = agl_first_entry(sai);
- cfs_list_del_init(&clli->lli_agl_list);
- spin_unlock(&plli->lli_agl_lock);
- ll_agl_trigger(&clli->lli_vfs_inode, sai);
- spin_lock(&plli->lli_agl_lock);
+ } else {
+ /*
+ * chain is exhausted.
+ * Normal case: continue to the next page.
+ */
+ ll_release_page(dir, page, le32_to_cpu(dp->ldp_flags) &
+ LDF_COLLIDE);
+ sai->sai_in_readpage = 1;
+ page = ll_get_dir_page(dir, op_data, pos, &chain);
+ sai->sai_in_readpage = 0;
+ }
}
- spin_unlock(&plli->lli_agl_lock);
out:
EXIT;
+ ll_dir_chain_fini(&chain);
ll_finish_md_op_data(op_data);
+out_put:
if (sai->sai_agl_valid) {
spin_lock(&plli->lli_agl_lock);
thread_set_flags(agl_thread, SVC_STOPPING);
spin_unlock(&plli->lli_agl_lock);
wake_up(&agl_thread->t_ctl_waitq);
- CDEBUG(D_READA, "stop agl thread: [pid %d]\n",
- current_pid());
+ CDEBUG(D_READA, "stop agl thread: sai %p pid %u\n",
+ sai, (unsigned int)agl_thread->t_pid);
l_wait_event(agl_thread->t_ctl_waitq,
thread_is_stopped(agl_thread),
&lwi);
/* Set agl_thread flags anyway. */
thread_set_flags(&sai->sai_agl_thread, SVC_STOPPED);
}
- ll_dir_chain_fini(&chain);
spin_lock(&plli->lli_sa_lock);
if (!sa_received_empty(sai)) {
thread_set_flags(thread, SVC_STOPPING);
wake_up(&sai->sai_waitq);
wake_up(&thread->t_ctl_waitq);
ll_sai_put(sai);
- dput(parent);
- CDEBUG(D_READA, "statahead thread stopped: [pid %d] [parent %.*s]\n",
- current_pid(), parent->d_name.len, parent->d_name.name);
+ CDEBUG(D_READA, "statahead thread stopped: sai %p, parent %.*s\n",
+ sai, parent->d_name.len, parent->d_name.name);
+ dput(parent);
return rc;
}
spin_unlock(&lli->lli_sa_lock);
wake_up(&thread->t_ctl_waitq);
- CDEBUG(D_READA, "stop statahead thread: [pid %d]\n",
- current_pid());
+ CDEBUG(D_READA, "stop statahead thread: sai %p pid %u\n",
+ lli->lli_sai, (unsigned int)thread->t_pid);
l_wait_event(thread->t_ctl_waitq,
thread_is_stopped(thread),
&lwi);
struct qstr *target = &dentry->d_name;
struct md_op_data *op_data;
int dot_de;
- struct lu_dirent *ent;
struct page *page = NULL;
int rc = LS_NONE_FIRST_DE;
+ __u64 pos = 0;
ENTRY;
- ll_dir_chain_init(&chain);
-
op_data = ll_prep_md_op_data(NULL, dir, dir, NULL, 0, 0,
LUSTRE_OPC_ANY, dir);
if (IS_ERR(op_data))
- GOTO(out, rc = PTR_ERR(op_data));
+ RETURN(PTR_ERR(op_data));
/**
*FIXME choose the start offset of the readdir
*/
op_data->op_stripe_offset = 0;
- op_data->op_hash_offset = 0;
- op_data->op_max_pages =
- ll_i2sbi(dir)->ll_md_brw_size >> PAGE_CACHE_SHIFT;
-
- for (ent = ll_dir_entry_start(dir, op_data, &page);
- ent != NULL && !IS_ERR(ent);
- ent = ll_dir_entry_next(dir, op_data, ent, &page)) {
- __u64 hash;
- int namelen;
- char *name;
-
- hash = le64_to_cpu(ent->lde_hash);
- /* The ll_get_dir_page() can return any page containing
- * the given hash which may be not the start hash. */
- if (unlikely(hash < op_data->op_hash_offset))
- continue;
-
- namelen = le16_to_cpu(ent->lde_namelen);
- if (unlikely(namelen == 0))
- /*
- * skip dummy record.
- */
- continue;
+ op_data->op_max_pages = ll_i2sbi(dir)->ll_md_brw_pages;
- name = ent->lde_name;
- if (name[0] == '.') {
- if (namelen == 1)
- /*
- * skip "."
- */
+ ll_dir_chain_init(&chain);
+ page = ll_get_dir_page(dir, op_data, 0, &chain);
+
+ while (1) {
+ struct lu_dirpage *dp;
+ struct lu_dirent *ent;
+
+ if (IS_ERR(page)) {
+ struct ll_inode_info *lli = ll_i2info(dir);
+
+ rc = PTR_ERR(page);
+ CERROR("%s: reading dir "DFID" at "LPU64
+ "opendir_pid = %u : rc = %d\n",
+ ll_get_fsname(dir->i_sb, NULL, 0),
+ PFID(ll_inode2fid(dir)), pos,
+ lli->lli_opendir_pid, rc);
+ break;
+ }
+
+ dp = page_address(page);
+ for (ent = lu_dirent_start(dp); ent != NULL;
+ ent = lu_dirent_next(ent)) {
+ __u64 hash;
+ int namelen;
+ char *name;
+
+ hash = le64_to_cpu(ent->lde_hash);
+ /* The ll_get_dir_page() can return any page containing
+ * the given hash which may be not the start hash. */
+ if (unlikely(hash < pos))
continue;
- else if (name[1] == '.' && namelen == 2)
+
+ namelen = le16_to_cpu(ent->lde_namelen);
+ if (unlikely(namelen == 0))
/*
- * skip ".."
+ * skip dummy record.
*/
continue;
- else
- dot_de = 1;
- } else {
- dot_de = 0;
- }
- if (dot_de && target->name[0] != '.') {
- CDEBUG(D_READA, "%.*s skip hidden file %.*s\n",
- target->len, target->name,
- namelen, name);
- continue;
- }
+ name = ent->lde_name;
+ if (name[0] == '.') {
+ if (namelen == 1)
+ /*
+ * skip "."
+ */
+ continue;
+ else if (name[1] == '.' && namelen == 2)
+ /*
+ * skip ".."
+ */
+ continue;
+ else
+ dot_de = 1;
+ } else {
+ dot_de = 0;
+ }
- if (target->len != namelen ||
- memcmp(target->name, name, namelen) != 0)
- rc = LS_NONE_FIRST_DE;
- else if (!dot_de)
- rc = LS_FIRST_DE;
- else
- rc = LS_FIRST_DOT_DE;
+ if (dot_de && target->name[0] != '.') {
+ CDEBUG(D_READA, "%.*s skip hidden file %.*s\n",
+ target->len, target->name,
+ namelen, name);
+ continue;
+ }
- break;
- }
- EXIT;
+ if (target->len != namelen ||
+ memcmp(target->name, name, namelen) != 0)
+ rc = LS_NONE_FIRST_DE;
+ else if (!dot_de)
+ rc = LS_FIRST_DE;
+ else
+ rc = LS_FIRST_DOT_DE;
- if (page != NULL) {
- kunmap(page);
- page_cache_release(page);
+ ll_release_page(dir, page, false);
+ GOTO(out, rc);
+ }
+ pos = le64_to_cpu(dp->ldp_hash_end);
+ if (pos == MDS_DIR_END_OFF) {
+ /*
+ * End of directory reached.
+ */
+ ll_release_page(dir, page, false);
+ GOTO(out, rc);
+ } else {
+ /*
+ * chain is exhausted
+ * Normal case: continue to the next page.
+ */
+ ll_release_page(dir, page, le32_to_cpu(dp->ldp_flags) &
+ LDF_COLLIDE);
+ page = ll_get_dir_page(dir, op_data, pos, &chain);
+ }
}
- ll_finish_md_op_data(op_data);
+ EXIT;
out:
ll_dir_chain_fini(&chain);
+ ll_finish_md_op_data(op_data);
return rc;
}
} else {
struct ll_inode_info *lli = ll_i2info(sai->sai_inode);
- sai->sai_miss++;
- sai->sai_consecutive_miss++;
- if (sa_low_hit(sai) && thread_is_running(thread)) {
- atomic_inc(&sbi->ll_sa_wrong);
+ sai->sai_miss++;
+ sai->sai_consecutive_miss++;
+ if (sa_low_hit(sai) && thread_is_running(thread)) {
+ atomic_inc(&sbi->ll_sa_wrong);
CDEBUG(D_READA, "Statahead for dir "DFID" hit "
"ratio too low: hit/miss "LPU64"/"LPU64
", sent/replied "LPU64"/"LPU64", stopping "
- "statahead thread: pid %d\n",
+ "statahead thread\n",
PFID(&lli->lli_fid), sai->sai_hit,
sai->sai_miss, sai->sai_sent,
- sai->sai_replied, current_pid());
+ sai->sai_replied);
spin_lock(&lli->lli_sa_lock);
if (!thread_is_stopped(thread))
thread_set_flags(thread, SVC_STOPPING);
struct ll_sa_entry *entry;
struct ptlrpc_thread *thread;
struct l_wait_info lwi = { 0 };
+ struct task_struct *task;
int rc = 0;
struct ll_inode_info *plli;
ENTRY;
if (sai) {
thread = &sai->sai_thread;
if (unlikely(thread_is_stopped(thread) &&
- cfs_list_empty(&sai->sai_entries_stated))) {
+ list_empty(&sai->sai_entries_stated))) {
/* to release resource */
ll_stop_statahead(dir, lli->lli_opendir_key);
RETURN(-EAGAIN);
}
/* if statahead is busy in readdir, help it do post-work */
- while (!ll_sa_entry_stated(entry) &&
- sai->sai_in_readpage &&
- !sa_received_empty(sai))
+ while (!ll_sa_entry_stated(entry) &&
+ sai->sai_in_readpage && !sa_received_empty(sai))
ll_post_statahead(sai);
if (!ll_sa_entry_stated(entry)) {
(*dentryp)->d_name.name,
PFID(ll_inode2fid((*dentryp)->d_inode)),
PFID(ll_inode2fid(inode)));
- ll_sai_unplug(sai, entry);
+ ll_intent_release(&it);
+ ll_sai_unplug(sai, entry);
RETURN(-ESTALE);
} else {
iput(inode);
GOTO(out, rc = -EAGAIN);
}
- CDEBUG(D_READA, "start statahead thread: [pid %d] [parent %.*s]\n",
- current_pid(), parent->d_name.len, parent->d_name.name);
+ CDEBUG(D_READA, "start statahead thread: sai %p, parent %.*s\n",
+ sai, parent->d_name.len, parent->d_name.name);
+ /* The sai buffer already has one reference taken at allocation time,
+ * but as soon as we expose the sai by attaching it to the lli that
+ * default reference can be dropped by another thread calling
+ * ll_stop_statahead. We need to take a local reference to protect
+ * the sai buffer while we intend to access it. */
+ ll_sai_get(sai);
lli->lli_sai = sai;
plli = ll_i2info(parent->d_inode);
- rc = PTR_ERR(kthread_run(ll_statahead_thread, parent,
- "ll_sa_%u", plli->lli_opendir_pid));
+ task = kthread_run(ll_statahead_thread, parent, "ll_sa_%u",
+ plli->lli_opendir_pid);
thread = &sai->sai_thread;
- if (IS_ERR_VALUE(rc)) {
- CERROR("can't start ll_sa thread, rc: %d\n", rc);
+ if (IS_ERR(task)) {
+ rc = PTR_ERR(task);
+ CERROR("cannot start ll_sa thread: rc = %d\n", rc);
dput(parent);
lli->lli_opendir_key = NULL;
thread_set_flags(thread, SVC_STOPPED);
thread_set_flags(&sai->sai_agl_thread, SVC_STOPPED);
+ /* Drop both our own local reference and the default
+ * reference from allocation time. */
ll_sai_put(sai);
+ ll_sai_put(sai);
LASSERT(lli->lli_sai == NULL);
RETURN(-EAGAIN);
}
l_wait_event(thread->t_ctl_waitq,
thread_is_running(thread) || thread_is_stopped(thread),
&lwi);
+ ll_sai_put(sai);
/*
* We don't stat-ahead for the first dirent since we are already in
git://git.whamcloud.com / fs / lustre-release.git / blobdiff