[Devel] [PATCH rh7 6/6] ploop: push_backup: implement timeout functions

[Maxim Patlasov]

If timeout fired, the function must scan the pending/reported lists
searching for expired preq. If at least one expired preq found,
the function must abort the whole push_backup operation merging
initial CBT mask back to CBT.

It is not nice to call heavy-weight ploop_pb_stop() directly from
timeout handler, so we only change pbd->ppb_state in the handler,
and wake up ploop_pb_health_monitor(). The latter will do the rest.

The strategy of handling pending_set/reported_set timers consits of
three cases (having in mind that the first preq in the set list is
always the oldest):

1) The first preq coming to the set arms the timer.
2) If we delete the first preq from the list, we have to
modify the timer according to the next preq timestamp.
3) If timer fired, but no expired preq found, let's
re-schedule timer according to the oldest preq timestamp.

https://jira.sw.ru/browse/PSBM-48082

Signed-off-by: Maxim Patlasov <mpatlasov at virtuozzo.com>
---
 drivers/block/ploop/push_backup.c |   78 +++++++++++++++++++++++++++++++++++++
 1 file changed, 78 insertions(+)

diff --git a/drivers/block/ploop/push_backup.c b/drivers/block/ploop/push_backup.c
index 97bfa9f..4f7b225 100644
--- a/drivers/block/ploop/push_backup.c
+++ b/drivers/block/ploop/push_backup.c
@@ -463,6 +463,7 @@ static void ploop_pb_add_req_to_tree(struct ploop_request *preq,
 	struct rb_node ** p = &tree->rb_node;
 	struct rb_node *parent = NULL;
 	struct ploop_request * pr;
+	unsigned long timeout = preq->plo->tune.push_backup_timeout * HZ;
 
 	while (*p) {
 		parent = *p;
@@ -475,6 +476,11 @@ static void ploop_pb_add_req_to_tree(struct ploop_request *preq,
 			p = &(*p)->rb_right;
 	}
 
+	preq->tstamp = jiffies;
+	if (timeout && list_empty(&pbs->list) &&
+	    pbs->pbd->ppb_state == PLOOP_PB_ALIVE)
+		mod_timer(&pbs->timer, preq->tstamp + timeout + 1);
+
 	list_add_tail(&preq->list, &pbs->list);
 
 	rb_link_node(&preq->reloc_link, parent, p);
@@ -496,8 +502,21 @@ static void ploop_pb_add_req_to_reported(struct ploop_pushbackup_desc *pbd,
 static void remove_req_from_pbs(struct pb_set *pbs,
 					 struct ploop_request *preq)
 {
+	unsigned long timeout = preq->plo->tune.push_backup_timeout * HZ;
+	bool oldest_deleted = false;
+
+	if (preq == list_first_entry(&pbs->list, struct ploop_request, list))
+		oldest_deleted = true;
+
 	rb_erase(&preq->reloc_link, &pbs->tree);
 	list_del_init(&preq->list);
+
+	if (timeout && oldest_deleted && !list_empty(&pbs->list) &&
+	    pbs->pbd->ppb_state == PLOOP_PB_ALIVE) {
+		preq = list_first_entry(&pbs->list, struct ploop_request,
+					list);
+		mod_timer(&pbs->timer, preq->tstamp + timeout + 1);
+	}
 }
 
 
@@ -935,6 +954,65 @@ int ploop_pb_destroy(struct ploop_device *plo, __u32 *status)
 	return 0;
 }
 
+static bool ploop_pb_set_expired(struct pb_set *pbs)
+{
+	struct ploop_pushbackup_desc *pbd = pbs->pbd;
+	struct ploop_device          *plo = pbd->plo;
+	unsigned long timeout = plo->tune.push_backup_timeout * HZ;
+	unsigned long tstamp = 0;
+	cluster_t clu = 0;
+	bool ret = false;
+
+	if (!timeout)
+		return false;
+
+	spin_lock(&pbd->ppb_lock);
+
+	if (pbd->ppb_state != PLOOP_PB_ALIVE) {
+		spin_unlock(&pbd->ppb_lock);
+		return false;
+	}
+
+	/* No need to scan the whole list: the first preq is the oldest! */
+	if (!list_empty(&pbs->list)) {
+		struct ploop_request *preq = list_first_entry(&pbs->list,
+							      struct ploop_request, list);
+		if (time_before(preq->tstamp + timeout, jiffies)) {
+			tstamp = preq->tstamp;
+			clu = preq->req_cluster;
+			ret = true;
+		} else
+			mod_timer(&pbs->timer, preq->tstamp + timeout + 1);
+	}
+
+	spin_unlock(&pbd->ppb_lock);
+
+	if (ret)
+		printk(KERN_WARNING "Abort push_backup for ploop%d: found "
+		       "preq (clu=%d) in %s tree delayed for %u msecs\n",
+		       plo->index, clu, pbs->name,
+		       jiffies_to_msecs(jiffies - tstamp));
+
+	return ret;
+}
+
 static void ploop_pb_timeout_func(unsigned long data)
 {
+	struct pb_set                *pbs = (void*)data;
+	struct ploop_pushbackup_desc *pbd = pbs->pbd;
+	struct ploop_device          *plo = pbd->plo;
+
+	if (!plo->tune.push_backup_timeout ||
+	    !test_bit(PLOOP_S_RUNNING, &plo->state) ||
+	    !test_bit(PLOOP_S_PUSH_BACKUP, &plo->state) ||
+	    !ploop_pb_set_expired(pbs))
+		return;
+
+	spin_lock(&pbd->ppb_lock);
+	if (pbd->ppb_state == PLOOP_PB_ALIVE) {
+		pbd->ppb_state = PLOOP_PB_STOPPING;
+		if (waitqueue_active(&pbd->ppb_waitq))
+			wake_up_interruptible(&pbd->ppb_waitq);
+	}
+	spin_unlock(&pbd->ppb_lock);
 }