[Devel] [PATCH vz9 v1 36/63] dm-ploop: make ploop_bat_write_complete ready for parallel pio completion

[Alexander Atanasov]

There are some requirements listed in the comment inside
ploop_bat_write_complete:
	* Success: now update local BAT copy. We could do this
	* from our delayed work, but we want to publish new
	* mapping in the fastest way. This must be done before
	* data bios completion, since right after we complete
	* a bio, subsequent read wants to see written data
	* (ploop_map() wants to see not zero bat_entries[.])

Currently it assumes sequential pio completion but with threads
it must be ready to have them in any order.

To meet that only ploop_advance_local_after_bat_wb when if it is
the last call to ploop_bat_write_complete.

Complete ready data and cow pios - protect lists with piwb->lock
and deferred_lock - as per other users do.
Code assumes that no one can touch the lists since it is assumed to
be complete which is not valid when pios are processed in parallel.

https://virtuozzo.atlassian.net/browse/VSTOR-91821
Signed-off-by: Alexander Atanasov <alexander.atanasov at virtuozzo.com>
---
 drivers/md/dm-ploop-map.c | 47 +++++++++++++++++++++++++--------------
 1 file changed, 30 insertions(+), 17 deletions(-)

diff --git a/drivers/md/dm-ploop-map.c b/drivers/md/dm-ploop-map.c
index 129447510033..86c7b7b946b2 100644
--- a/drivers/md/dm-ploop-map.c
+++ b/drivers/md/dm-ploop-map.c
@@ -895,35 +895,47 @@ static void ploop_bat_write_complete(struct pio *pio, void *piwb_ptr,
 	struct ploop_cow *cow;
 	struct pio *data_pio;
 	unsigned long flags;
-
-	if (!bi_status) {
-		/*
-		 * Success: now update local BAT copy. We could do this
-		 * from our delayed work, but we want to publish new
-		 * mapping in the fastest way. This must be done before
-		 * data bios completion, since right after we complete
-		 * a bio, subsequent read wants to see written data
-		 * (ploop_map() wants to see not zero bat_entries[.]).
-		 */
-		ploop_advance_local_after_bat_wb(ploop, piwb, true);
+	LIST_HEAD(lready_pios);
+	LIST_HEAD(lcow_pios);
+	int completed = atomic_read(&piwb->count) == 1;
+
+	if (completed) {
+		/* We are the last count so it is safe to advance bat */
+		if (!bi_status) {
+			/*
+			 * Success: now update local BAT copy. We could do this
+			 * from our delayed work, but we want to publish new
+			 * mapping in the fastest way. This must be done before
+			 * data bios completion, since right after we complete
+			 * a bio, subsequent read wants to see written data
+			 * (ploop_map() wants to see not zero bat_entries[.]).
+			 */
+			ploop_advance_local_after_bat_wb(ploop, piwb, true);
+		}
 	}
 
 	spin_lock_irqsave(&piwb->lock, flags);
-	piwb->completed = true;
+	if (completed)
+		piwb->completed = completed;
 	piwb->bi_status = bi_status;
+	list_splice_init(&piwb->ready_data_pios, &lready_pios);
 	spin_unlock_irqrestore(&piwb->lock, flags);
 
+	spin_lock_irqsave(&ploop->deferred_lock, flags);
+	list_splice_init(&piwb->cow_list, &lcow_pios);
+	spin_unlock_irqrestore(&ploop->deferred_lock, flags);
+
 	/*
-	 * End pending data bios. Unlocked, as nobody can
-	 * add a new element after piwc->completed is true.
+	 * End pending data bios.
 	 */
-	while ((data_pio = ploop_pio_list_pop(&piwb->ready_data_pios)) != NULL) {
+
+	while ((data_pio = ploop_pio_list_pop(&lready_pios)) != NULL) {
 		if (bi_status)
 			data_pio->bi_status = bi_status;
 		ploop_pio_endio(data_pio);
 	}
 
-	while ((aux_pio = ploop_pio_list_pop(&piwb->cow_list))) {
+	while ((aux_pio = ploop_pio_list_pop(&lcow_pios))) {
 		cow = aux_pio->endio_cb_data;
 		ploop_complete_cow(cow, bi_status);
 	}
@@ -1901,7 +1913,6 @@ static void ploop_submit_metadata_writeback(struct ploop *ploop)
 	 */
 	llist_for_each_safe(pos, t, ll_wb_batch) {
 		md = list_entry((struct list_head *)pos, typeof(*md), wb_link);
-		INIT_LIST_HEAD(&md->wb_link);
 		if (!llist_empty(&md->wait_llist) ||
 		    test_bit(MD_HIGHPRIO, &md->status) ||
 		    time_before(md->dirty_timeout, timeout) ||
@@ -1913,7 +1924,9 @@ static void ploop_submit_metadata_writeback(struct ploop *ploop)
 			clear_bit(MD_DIRTY, &md->status);
 			clear_bit(MD_HIGHPRIO, &md->status);
 			ploop_index_wb_submit(ploop, md->piwb);
+			ret++;
 		} else {
+			INIT_LIST_HEAD(&md->wb_link);
 			llist_add((struct llist_node *)&md->wb_link, &ploop->wb_batch_llist);
 		}
 	}
-- 
2.43.0