ocfs2: drop and retake orphan_dir.i_mutex in ocfs2_recover_orphans
PATCH 3/3 ocfs2: drop and retake orphan_dir.i_mutex in ocfs2_recover_orphans
drop and retake orphan_dir.i_mutex in ocfs2_recover_orphans.
--- fs/ocfs2/journal.c | 40 ++++++++++++++++++++++++++++++++++------ 1 files changed, 34 insertions(+), 6 deletions(-)
fs/ocfs2/journal.c +34 -6
@@ -56,6 +56,7 @@DEFINE_SPINLOCK (trans_inc_lock);#define ORPHAN_SCAN_SCHEDULE_TIMEOUT 300000 +#define ORPHAN_SCAN_ITEMS_PER_LOOP 200static int ocfs2_force_read_journal (struct inode *inode); static int ocfs2_recover_node(struct ocfs2_super *osb,
@@ -1949,6 +1950,8 @@ void ocfs2_orphan_scan_start(struct ocfs2_super *osb)struct ocfs2_orphan_filldir_priv {struct inode *head;struct ocfs2_super *osb; +int count; +int for_live:1 ; }; static int ocfs2_orphan_filldir(void *priv, const char *name, int name_len,
@@ -1957,6 +1960,9 @@ static int ocfs2_orphan_filldir(void *priv, const char *name, int name_len,struct ocfs2_orphan_filldir_priv *p = priv;struct inode *iter; +if (p->for_live) +BUG_ON (p->count >ORPHAN_SCAN_ITEMS_PER_LOOP ); +if (name_len ==1 && !strncmp ("." , name,1 ))return 0 ;if (name_len ==2 && !strncmp (".." , name,2 ))
@@ -1974,20 +1980,38 @@ static int ocfs2_orphan_filldir(void *priv, const char *name, int name_len,OCFS2_I (iter)->ip_next_orphan = p->head; p->head = iter; +/* +* we fill at most ORPHAN_SCAN_ITEMS_PER_LOOP direntry in a loop with +* the orphan_dir.i_mutex locked. +*/ +if (p->for_live) { +if (++p->count ==ORPHAN_SCAN_ITEMS_PER_LOOP ) +return -EAGAIN ; + }return 0 ; }static int ocfs2_queue_orphans (struct ocfs2_super *osb,int slot, -struct inode **head) +struct inode **head, +loff_t *pos) {int status;struct inode *orphan_dir_inode =NULL ;struct ocfs2_orphan_filldir_priv priv; -loff_t pos =0 ; +unsigned int node_num; priv.osb = osb; priv.head = *head; + priv.count =0 ; + +spin_lock (&osb->osb_lock); + status =ocfs2_slot_to_node_num_locked (osb, slot, &node_num); +spin_unlock (&osb->osb_lock); +if (!status && node_num == osb->node_num) + priv.for_live = true; +else + priv.for_live = false; orphan_dir_inode = ocfs2_get_system_file_inode(osb, ORPHAN_DIR_SYSTEM_INODE,
@@ -2005,9 +2029,9 @@ static int ocfs2_queue_orphans(struct ocfs2_super *osb, goto out; } - status =ocfs2_dir_foreach (orphan_dir_inode, &pos, &priv, + status =ocfs2_dir_foreach (orphan_dir_inode, pos, &priv, ocfs2_orphan_filldir); -if (status) { +if (status && status != -EAGAIN ) {mlog_errno (status); goto out_cluster; }
@@ -2083,16 +2107,18 @@ static int ocfs2_recover_orphans(struct ocfs2_super *osb,struct inode *inode =NULL ;struct inode *iter;struct ocfs2_inode_info *oi; +loff_t pos =0 ;trace_ocfs2_recover_orphans (slot); +cnt_scan:ocfs2_mark_recovering_orphan_dir (osb, slot); - ret =ocfs2_queue_orphans (osb, slot, &inode); + ret =ocfs2_queue_orphans (osb, slot, &inode, &pos);ocfs2_clear_recovering_orphan_dir (osb, slot);/* Error here should be noted, but we want to continue with as * many queued inodes as we've got. */ -if (ret) +if (ret && ret != -EAGAIN )mlog_errno (ret);while (inode) {
@@ -2119,6 +2145,8 @@ static int ocfs2_recover_orphans(struct ocfs2_super *osb, inode = iter; } +if (ret == -EAGAIN ) + goto cnt_scan;return ret; }
Signature · 1 line
1.7.5.2
PATCH 3/3 ocfs2: drop and retake orphan_dir.i_mutex in ocfs2_recover_orphans
This is too hacky for mainline. May work as a temp fix only.
This assumes that we only need to recover orphans in our own slot. If that is the case, then if nothing else, we can remove the orphan scan lock. We added the lock so as to ensure all nodes get a chance to replay it. If all nodes are doing there own, they why do we need the lock.
Quoted 66 lines · Wengang Wang
drop and retake orphan_dir.i_mutex in
ocfs2_recover_orphans.Signed-off-by Wengang Wang<[email protected]>--- fs/ocfs2/journal.c | 40 ++++++++++++++++++++++++++++++++++------ 1 files changed, 34 insertions(+), 6 deletions(-)fs/ocfs2/journal.c +17
-2@@ -56,6 +56,7 @@DEFINE_SPINLOCK (trans_inc_lock);#define ORPHAN_SCAN_SCHEDULE_TIMEOUT 300000 +#define ORPHAN_SCAN_ITEMS_PER_LOOP 200static int ocfs2_force_read_journal (struct inode *inode); static int ocfs2_recover_node(struct ocfs2_super *osb,@@ -1949,6 +1950,8 @@ void ocfs2_orphan_scan_start(struct ocfs2_super *osb)struct ocfs2_orphan_filldir_priv {struct inode *head;struct ocfs2_super *osb; +int count; +int for_live:1 ; }; static int ocfs2_orphan_filldir(void *priv, const char *name, int name_len,@@ -1957,6 +1960,9 @@ static int ocfs2_orphan_filldir(void *priv, const char *name, int name_len,struct ocfs2_orphan_filldir_priv *p = priv;struct inode *iter; +if (p->for_live) +BUG_ON (p->count>ORPHAN_SCAN_ITEMS_PER_LOOP ); +if (name_len ==1 && !strncmp ("." , name,1 ))return 0 ;if (name_len ==2 && !strncmp (".." , name,2 ))@@ -1974,20 +1980,38 @@ static int ocfs2_orphan_filldir(void *priv, const char *name, int name_len,OCFS2_I (iter)->ip_next_orphan = p->head; p->head = iter; +/* +* we fill at most ORPHAN_SCAN_ITEMS_PER_LOOP direntry in a loop with +* the orphan_dir.i_mutex locked. +*/ +if (p->for_live) { +if (++p->count ==ORPHAN_SCAN_ITEMS_PER_LOOP ) +return -EAGAIN ; + }return 0 ; }static int ocfs2_queue_orphans (struct ocfs2_super *osb,int slot, -struct inode **head) +struct inode **head, +loff_t *pos) {int status;struct inode *orphan_dir_inode =NULL ;struct ocfs2_orphan_filldir_priv priv; -loff_t pos =0 ; +unsigned int node_num; priv.osb = osb;
fs/ocfs2/journal.c static int ocfs2_orphan_filldir(void *priv, const char *name, int name_len, · line 2014
Quoted 10 lines
fs/ocfs2/journal.c +9
-0priv.head = *head; + priv.count =0 ; + +spin_lock (&osb->osb_lock); + status =ocfs2_slot_to_node_num_locked (osb, slot,&node_num); +spin_unlock (&osb->osb_lock); +if (!status&& node_num == osb->node_num) + priv.for_live = true; +else + priv.for_live = false;
You cannot return (save) pos and yet lookup the slotnum again. That's not the right order. Too hackish. All these functions will need to be re-prototyped if your approach is what we use.
Quoted 18 lines
fs/ocfs2/journal.c +2
-2orphan_dir_inode = ocfs2_get_system_file_inode(osb, ORPHAN_DIR_SYSTEM_INODE,@@ -2005,9 +2029,9 @@ static int ocfs2_queue_orphans(struct ocfs2_super *osb, goto out; } - status =ocfs2_dir_foreach (orphan_dir_inode,&pos,&priv, + status =ocfs2_dir_foreach (orphan_dir_inode, pos,&priv, ocfs2_orphan_filldir); -if (status) { +if (status&& status != -EAGAIN ) {mlog_errno (status); goto out_cluster; }@@ -2083,16 +2107,18 @@ static int ocfs2_recover_orphans(struct ocfs2_super *osb,struct inode *inode =NULL ;struct inode *iter;
fs/ocfs2/journal.c static int ocfs2_recover_orphans(struct ocfs2_super *osb, · line 2117
Quoted 10 lines
fs/ocfs2/journal.c +3
-1struct ocfs2_inode_info *oi; +loff_t pos =0 ;trace_ocfs2_recover_orphans (slot); +cnt_scan:ocfs2_mark_recovering_orphan_dir (osb, slot); - ret =ocfs2_queue_orphans (osb, slot,&inode); + ret =ocfs2_queue_orphans (osb, slot,&inode,&pos);ocfs2_clear_recovering_orphan_dir (osb, slot);
The slot is being marked and cleared multiple times. I am unsure what the ramifications are, but it does not look right.
Quoted 16 lines
fs/ocfs2/journal.c +3
-1/* Error here should be noted, but we want to continue with as * many queued inodes as we've got. */ -if (ret) +if (ret&& ret != -EAGAIN )mlog_errno (ret);while (inode) {@@ -2119,6 +2145,8 @@ static int ocfs2_recover_orphans(struct ocfs2_super *osb, inode = iter; } +if (ret == -EAGAIN ) + goto cnt_scan;return ret; }
PATCH 3/3 ocfs2: drop and retake orphan_dir.i_mutex in ocfs2_recover_orphans
Quoted 1 line · Sunil Mushran
This is too hacky for mainline. May work as a temp fix only.
:D
Quoted 6 lines
This assumes that we only need to recover orphans in our own slot. If that is the case, then if nothing else, we can remove the orphan scan lock. We added the lock so as to ensure all nodes get a chance to replay it. If all nodes are doing there own, they why do we need the lock.
Actually, for live nodes, yes we only recover our own slot. For the dead slots, we need the orphan scan lock.
Quoted 67 lines · Wengang Wang
Quoted 65 lines
drop and retake orphan_dir.i_mutex in
ocfs2_recover_orphans.Signed-off-by Wengang Wang<[email protected]>--- fs/ocfs2/journal.c | 40 ++++++++++++++++++++++++++++++++++------ 1 files changed, 34 insertions(+), 6 deletions(-)fs/ocfs2/journal.c +17
-2@@ -56,6 +56,7 @@DEFINE_SPINLOCK (trans_inc_lock);#define ORPHAN_SCAN_SCHEDULE_TIMEOUT 300000 +#define ORPHAN_SCAN_ITEMS_PER_LOOP 200static int ocfs2_force_read_journal (struct inode *inode); static int ocfs2_recover_node(struct ocfs2_super *osb,@@ -1949,6 +1950,8 @@ void ocfs2_orphan_scan_start(struct ocfs2_super *osb)struct ocfs2_orphan_filldir_priv {struct inode *head;struct ocfs2_super *osb; +int count; +int for_live:1 ; }; static int ocfs2_orphan_filldir(void *priv, const char *name, int name_len,@@ -1957,6 +1960,9 @@ static int ocfs2_orphan_filldir(void *priv, const char *name, int name_len,struct ocfs2_orphan_filldir_priv *p = priv;struct inode *iter; +if (p->for_live) +BUG_ON (p->count>ORPHAN_SCAN_ITEMS_PER_LOOP ); +if (name_len ==1 && !strncmp ("." , name,1 ))return 0 ;if (name_len ==2 && !strncmp (".." , name,2 ))@@ -1974,20 +1980,38 @@ static int ocfs2_orphan_filldir(void *priv, const char *name, int name_len,OCFS2_I (iter)->ip_next_orphan = p->head; p->head = iter; +/* +* we fill at most ORPHAN_SCAN_ITEMS_PER_LOOP direntry in a loop with +* the orphan_dir.i_mutex locked. +*/ +if (p->for_live) { +if (++p->count ==ORPHAN_SCAN_ITEMS_PER_LOOP ) +return -EAGAIN ; + }return 0 ; }static int ocfs2_queue_orphans (struct ocfs2_super *osb,int slot, -struct inode **head) +struct inode **head, +loff_t *pos) {int status;struct inode *orphan_dir_inode =NULL ;struct ocfs2_orphan_filldir_priv priv; -loff_t pos =0 ; +unsigned int node_num;
Quoted 15 lines
fs/ocfs2/journal.c static int ocfs2_orphan_filldir(void *priv, const char *name, int name_len, · line 2014
Quoted 11 lines
fs/ocfs2/journal.c +9
-0priv.osb = osb; priv.head = *head; + priv.count =0 ; + +spin_lock (&osb->osb_lock); + status =ocfs2_slot_to_node_num_locked (osb, slot,&node_num); +spin_unlock (&osb->osb_lock); +if (!status&& node_num == osb->node_num) + priv.for_live = true; +else + priv.for_live = false;You cannot return (save) pos and yet lookup the slotnum again. That's not the right order. Too hackish. All these functions will need to be re-prototyped if your approach is what we use.
Yes, I also think it's hackish.
But I don't have a better fix so far. While though it's hackish, it works. It is just like ocfs2_readdir(). If a new direntry is added before the position, we just leave there for next run; If one is added after the possition, we take it; If a diretry is removed, I means the file/inode is freed, OK; It's safe.
If we restart from offset 0, in case all the inodes in the first 200 or whatever other limit are not freed and won't be freed(say still opened), we are in bad loops.
Quoted 24 lines
Quoted 24 lines
fs/ocfs2/journal.c +3
-2orphan_dir_inode = ocfs2_get_system_file_inode(osb, ORPHAN_DIR_SYSTEM_INODE,@@ -2005,9 +2029,9 @@ static int ocfs2_queue_orphans(struct ocfs2_super *osb, goto out; } - status =ocfs2_dir_foreach (orphan_dir_inode,&pos,&priv, + status =ocfs2_dir_foreach (orphan_dir_inode, pos,&priv, ocfs2_orphan_filldir); -if (status) { +if (status&& status != -EAGAIN ) {mlog_errno (status); goto out_cluster; }@@ -2083,16 +2107,18 @@ static int ocfs2_recover_orphans(struct ocfs2_super *osb,struct inode *inode =NULL ;struct inode *iter;struct ocfs2_inode_info *oi; +loff_t pos =0 ;trace_ocfs2_recover_orphans (slot);
Quoted 8 lines
fs/ocfs2/journal.c static int ocfs2_recover_orphans(struct ocfs2_super *osb, · line 2117
Quoted 5 lines
fs/ocfs2/journal.c +2
-1+cnt_scan:ocfs2_mark_recovering_orphan_dir (osb, slot); - ret =ocfs2_queue_orphans (osb, slot,&inode); + ret =ocfs2_queue_orphans (osb, slot,&inode,&pos);ocfs2_clear_recovering_orphan_dir (osb, slot);The slot is being marked and cleared multiple times. I am unsure what the ramifications are, but it does not look right.
I will look at this more...
thanks,
wengang.
PATCH 3/3 ocfs2: drop and retake orphan_dir.i_mutex in ocfs2_recover_orphans
Quoted 2 lines · Wengang Wang
Quoted 1 line · Sunil Mushran
This is too hacky for mainline. May work as a temp fix only.
<snip>
Quoted 11 lines
Quoted 10 lines
fs/ocfs2/journal.c static int ocfs2_recover_orphans(struct ocfs2_super *osb, · line 2117
Quoted 7 lines
fs/ocfs2/journal.c +2
-1trace_ocfs2_recover_orphans (slot); +cnt_scan:ocfs2_mark_recovering_orphan_dir (osb, slot); - ret =ocfs2_queue_orphans (osb, slot,&inode); + ret =ocfs2_queue_orphans (osb, slot,&inode,&pos);ocfs2_clear_recovering_orphan_dir (osb, slot);The slot is being marked and cleared multiple times. I am unsure what the ramifications are, but it does not look right.
Quoted 1 line
I will look at this more...
Per my understand, ocfs2_mark_recovering_orphan_dir() is used to avoid dead lock only.
ABBA lock.
A: inode lock(cluster lock) on inode to delete
B: inode lock on orphan_dir inode
the ocfs2_delete_inode() path takes in AB.
the ocfs2_recover_orphans() path takes in BA on a different node.
Since AB are both required and then released within ocfs2_queue_orphans(), calling ocfs2_mark_recovering_orphan_dir() multiple times is not a problem.
Opition?
thanks,
wengang.