[Devel] [RFC PATCH vz9 v3 07/11] ploop: convert wait_list and wb_batch_llist to use lockless lists

[Alexander Atanasov]

Convert to lockless lists - intermix with regular list due to
that next pointer in both list_head and llist_head is the first
field, and prev is not used. Do this so we can make babysteps
forward.

Signed-off-by: Alexander Atanasov <alexander.atanasov at virtuozzo.com>
---
 drivers/md/dm-ploop-bat.c    |  3 +--
 drivers/md/dm-ploop-map.c    | 48 +++++++++++++++++++-----------------
 drivers/md/dm-ploop-target.c |  2 +-
 drivers/md/dm-ploop.h        |  6 ++---
 4 files changed, 31 insertions(+), 28 deletions(-)

diff --git a/drivers/md/dm-ploop-bat.c b/drivers/md/dm-ploop-bat.c
index 886a05cdd23a..655d0e4c91ab 100644
--- a/drivers/md/dm-ploop-bat.c
+++ b/drivers/md/dm-ploop-bat.c
@@ -80,8 +80,7 @@ static struct md_page *ploop_alloc_md_page(u32 id)
 	page = alloc_page(GFP_KERNEL);
 	if (!page)
 		goto err_page;
-	INIT_LIST_HEAD(&md->wait_list);
-	INIT_LIST_HEAD(&md->wb_link);
+	init_llist_head(&md->wait_llist);
 
 	md->status = 0;
 	md->bat_levels = levels;
diff --git a/drivers/md/dm-ploop-map.c b/drivers/md/dm-ploop-map.c
index e50e4ff58902..f848c66ab8e7 100644
--- a/drivers/md/dm-ploop-map.c
+++ b/drivers/md/dm-ploop-map.c
@@ -382,13 +382,14 @@ static bool ploop_delay_if_md_busy(struct ploop *ploop, struct md_page *md,
 
 	WARN_ON_ONCE(!list_empty(&pio->list));
 
-	write_lock_irqsave(&ploop->bat_rwlock, flags);
+	/* lock protects piwb */
+	read_lock_irqsave(&ploop->bat_rwlock, flags);
 	piwb = md->piwb;
 	if (piwb && (piwb->type != type || test_bit(MD_WRITEBACK, &md->status))) {
-		list_add_tail(&pio->list, &md->wait_list);
+		llist_add((struct llist_node *)(&pio->list), &md->wait_llist);
 		busy = true;
 	}
-	write_unlock_irqrestore(&ploop->bat_rwlock, flags);
+	read_unlock_irqrestore(&ploop->bat_rwlock, flags);
 
 	return busy;
 }
@@ -547,16 +548,13 @@ static void ploop_unlink_completed_pio(struct ploop *ploop, struct pio *pio)
 
 static bool ploop_md_make_dirty(struct ploop *ploop, struct md_page *md)
 {
-	unsigned long flags;
 	bool new = false;
 
-	write_lock_irqsave(&ploop->bat_rwlock, flags);
 	WARN_ON_ONCE(test_bit(MD_WRITEBACK, &md->status));
 	if (!test_and_set_bit(MD_DIRTY, &md->status)) {
-		list_add_tail(&md->wb_link, &ploop->wb_batch_list);
+		llist_add(&md->wb_llink, &ploop->wb_batch_llist);
 		new = true;
 	}
-	write_unlock_irqrestore(&ploop->bat_rwlock, flags);
 
 	return new;
 }
@@ -758,7 +756,9 @@ static void ploop_advance_local_after_bat_wb(struct ploop *ploop,
 	map_index_t *dst_clu, off;
 	unsigned long flags;
 	LIST_HEAD(list);
-
+	struct llist_node *wait_llist_pending;
+	struct pio *pio;
+	struct llist_node *pos, *t;
 	BUG_ON(!md);
 	bat_entries = md->kmpage;
 
@@ -775,7 +775,6 @@ static void ploop_advance_local_after_bat_wb(struct ploop *ploop,
 		i = PLOOP_MAP_OFFSET;
 
 	dst_clu = piwb->kmpage;
-	write_lock_irqsave(&ploop->bat_rwlock, flags);
 
 	for (; i < last; i++) {
 		if (piwb->type == PIWB_TYPE_DISCARD) {
@@ -802,10 +801,20 @@ static void ploop_advance_local_after_bat_wb(struct ploop *ploop,
 
 	WARN_ON_ONCE(!test_bit(MD_WRITEBACK, &md->status));
 	clear_bit(MD_WRITEBACK, &md->status);
+	/* protect piwb */
+	write_lock_irqsave(&ploop->bat_rwlock, flags);
 	md->piwb = NULL;
-	list_splice_tail_init(&md->wait_list, &list);
 	write_unlock_irqrestore(&ploop->bat_rwlock, flags);
 
+	wait_llist_pending = llist_del_all(&md->wait_llist);
+	if (wait_llist_pending) {
+		wait_llist_pending = llist_reverse_order(wait_llist_pending);
+		llist_for_each_safe(pos, t, wait_llist_pending) {
+			pio = list_entry((struct list_head *)pos, typeof(*pio), list);
+			list_add(&pio->list, &list);
+		}
+	}
+
 	if (!list_empty(&list))
 		ploop_dispatch_pios(ploop, NULL, &list);
 }
@@ -1758,22 +1767,17 @@ static void ploop_process_resubmit_pios(struct ploop *ploop,
 static void ploop_submit_metadata_writeback(struct ploop *ploop)
 {
 	struct md_page *md;
-	while (1) {
-		write_lock_irq(&ploop->bat_rwlock);
-		md = list_first_entry_or_null(&ploop->wb_batch_list,
-				struct md_page, wb_link);
-		if (!md) {
-			write_unlock_irq(&ploop->bat_rwlock);
-			break;
-		}
-		list_del_init(&md->wb_link);
-		/* L1L2 mustn't be redirtyed, when wb in-flight! */
+	struct md_page *t;
+	struct llist_node *wbl;
+	wbl = llist_del_all(&ploop->wb_batch_llist);
+	if (!wbl)
+		return;
+	wbl = llist_reverse_order(wbl);
+	llist_for_each_entry_safe(md, t, wbl, wb_llink) {
 		WARN_ON_ONCE(!test_bit(MD_DIRTY, &md->status));
 		WARN_ON_ONCE(test_bit(MD_WRITEBACK, &md->status));
 		set_bit(MD_WRITEBACK, &md->status);
 		clear_bit(MD_DIRTY, &md->status);
-		write_unlock_irq(&ploop->bat_rwlock);
-
 		ploop_index_wb_submit(ploop, md->piwb);
 	}
 }
diff --git a/drivers/md/dm-ploop-target.c b/drivers/md/dm-ploop-target.c
index 8b3b840dea66..b66ae3a9a4b0 100644
--- a/drivers/md/dm-ploop-target.c
+++ b/drivers/md/dm-ploop-target.c
@@ -378,7 +378,7 @@ static int ploop_ctr(struct dm_target *ti, unsigned int argc, char **argv)
 	INIT_LIST_HEAD(&ploop->resubmit_pios);
 	INIT_LIST_HEAD(&ploop->enospc_pios);
 	INIT_LIST_HEAD(&ploop->cluster_lk_list);
-	INIT_LIST_HEAD(&ploop->wb_batch_list);
+	init_llist_head(&ploop->wb_batch_llist);
 	ploop->bat_entries = RB_ROOT;
 	timer_setup(&ploop->enospc_timer, ploop_enospc_timer, 0);
 
diff --git a/drivers/md/dm-ploop.h b/drivers/md/dm-ploop.h
index 5fa176affc4e..4a64a06d0713 100644
--- a/drivers/md/dm-ploop.h
+++ b/drivers/md/dm-ploop.h
@@ -121,9 +121,9 @@ struct md_page {
 	struct page *page;
 	void *kmpage;
 	u8 *bat_levels;
-	struct list_head wait_list;
+	struct llist_head wait_llist;
 
-	struct list_head wb_link;
+	struct llist_node wb_llink;
 	struct ploop_index_wb *piwb;
 };
 
@@ -166,7 +166,7 @@ struct ploop {
 	u32 hb_nr; /* holes_bitmap size in bits */
 	rwlock_t bat_rwlock;
 
-	struct list_head wb_batch_list;
+	struct llist_head wb_batch_llist;
 
 	/*
 	 * Hash table to link non-exclusive submitted bios.
-- 
2.43.0