[sheepdog] [PATCH 2/2] sheep: cleanup prepare_schedule_oid and oid_in_recovery

[MORITA Kazutaka]

From: MORITA Kazutaka <morita.kazutaka at lab.ntt.co.jp>

This moves all the check of recovering objects from
prepare_schedule_oid() to oid_in_recovery(), and makes the code easier
to read.

This also fixes a problem that sheep doesn't recover an object when
 - sys->cinfo.disable_recovery is true,
 - rinfo->state is RW_PREPARE_LIST, and
 - oid_in_recovery() is called against the object
and fixes an error of tests/functional/010, which happens in rare
cases.

Signed-off-by: MORITA Kazutaka <morita.kazutaka at lab.ntt.co.jp>
---
 include/sheep.h  |    5 ++++
 sheep/recovery.c |   77 ++++++++++++++++++++++++++++++------------------------
 2 files changed, 48 insertions(+), 34 deletions(-)

diff --git a/include/sheep.h b/include/sheep.h
index 68c852f..f9945c6 100644
--- a/include/sheep.h
+++ b/include/sheep.h
@@ -251,6 +251,11 @@ static inline const char *sd_strerror(int err)
 	return descs[err];
 }
 
+static inline int oid_cmp(const uint64_t *oid1, const uint64_t *oid2)
+{
+	return intcmp(*oid1, *oid2);
+}
+
 static inline int node_id_cmp(const struct node_id *node1,
 			      const struct node_id *node2)
 {
diff --git a/sheep/recovery.c b/sheep/recovery.c
index 6faaeab..4305e48 100644
--- a/sheep/recovery.c
+++ b/sheep/recovery.c
@@ -331,24 +331,11 @@ bool node_in_recovery(void)
 static inline void prepare_schedule_oid(uint64_t oid)
 {
 	struct recovery_info *rinfo = main_thread_get(current_rinfo);
-	uint64_t i;
 
-	for (i = 0; i < rinfo->nr_prio_oids; i++)
-		if (rinfo->prio_oids[i] == oid)
-			return;
-	/*
-	 * We need this check because oid might not be recovered.
-	 * Very much unlikely though, but it might happen indeed.
-	 */
-	for (i = 0; i < rinfo->done; i++)
-		if (rinfo->oids[i] == oid) {
-			sd_debug("%"PRIx64" not recovered, don't schedule it",
-				 oid);
-			return;
-		}
-	/* When recovery is not suspended, oid is currently being recovered */
-	if (!rinfo->suspended && rinfo->oids[rinfo->done] == oid)
+	if (xlfind(&oid, rinfo->prio_oids, rinfo->nr_prio_oids, oid_cmp)) {
+		sd_debug("%" PRIx64 " has been already in prio_oids", oid);
 		return;
+	}
 
 	rinfo->nr_prio_oids++;
 	rinfo->prio_oids = xrealloc(rinfo->prio_oids,
@@ -359,10 +346,9 @@ static inline void prepare_schedule_oid(uint64_t oid)
 	resume_suspended_recovery();
 }
 
-bool oid_in_recovery(uint64_t oid)
+main_fn bool oid_in_recovery(uint64_t oid)
 {
 	struct recovery_info *rinfo = main_thread_get(current_rinfo);
-	uint64_t i;
 
 	if (!node_in_recovery())
 		return false;
@@ -373,27 +359,50 @@ bool oid_in_recovery(uint64_t oid)
 	}
 
 	if (uatomic_read(&next_rinfo))
+		/*
+		 * The current recovery_info will be taken over by the next one
+		 * soon, so no need to call prepare_schedule_oid() now.
+		 */
 		return true;
 
-	/* If we are in preparation of object list, oid is not recovered yet */
-	if (rinfo->state == RW_PREPARE_LIST)
-		return true;
+	switch (rinfo->state) {
+	case RW_PREPARE_LIST:
+		/* oid is not recovered yet */
+		break;
+	case RW_RECOVER_OBJ:
+		if (xlfind(&oid, rinfo->oids, rinfo->done, oid_cmp)) {
+			sd_debug("%" PRIx64 " has been already recovered", oid);
+			return false;
+		}
 
-	/*
-	 * Check if oid is in the list that to be recovered later
-	 *
-	 * FIXME: do we need more efficient yet complex data structure?
-	 */
-	for (i = rinfo->done; i < rinfo->count; i++)
-		if (rinfo->oids[i] == oid)
+		if (rinfo->oids[rinfo->done] == oid) {
+			if (rinfo->suspended)
+				break;
+			/*
+			 * When recovery is not suspended,
+			 * rinfo->oids[rinfo->done] is currently being recovered
+			 * and no need to call prepare_schedule_oid().
+			 */
+			return true;
+		}
+
+		/*
+		 * Check if oid is in the list that to be recovered later
+		 *
+		 * FIXME: do we need more efficient yet complex data structure?
+		 */
+		if (xlfind(&oid, rinfo->oids + rinfo->done + 1, rinfo->count,
+			   oid_cmp))
 			break;
 
-	/*
-	 * Newly created object after prepare_object_list() might not be
-	 * in the list
-	 */
-	if (i == rinfo->count) {
-		sd_err("%"PRIx64" is not in the recovery list", oid);
+		/*
+		 * Newly created object after prepare_object_list() might not be
+		 * in the list
+		 */
+		sd_debug("%"PRIx64" is not in the recovery list", oid);
+		return false;
+	case RW_NOTIFY_COMPLETION:
+		sd_debug("the object %" PRIx64 " is already recoverd", oid);
 		return false;
 	}
 
-- 
1.7.9.5