[Devel] [RFC PATCH vz9 v5 41/49] 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 | 44 +++++++++++++++++++++++++--------------
 1 file changed, 28 insertions(+), 16 deletions(-)

diff --git a/drivers/md/dm-ploop-map.c b/drivers/md/dm-ploop-map.c
index ba9ad42ead26..ab87f67537f5 100644
--- a/drivers/md/dm-ploop-map.c
+++ b/drivers/md/dm-ploop-map.c
@@ -910,35 +910,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);
 	}
-- 
2.43.0