RHEL53 - clogd: Fix for bug 468438 - list corruption

[Jonathan Brassow <jbrassow at fedoraproject dot org>]

Gitweb:        http://git.fedorahosted.org/git/cluster.git?p=cluster.git;a=commitdiff;h=6bf04b872a87caadbd83f4e3b673323399085692
Commit:        6bf04b872a87caadbd83f4e3b673323399085692
Parent:        e07369b28d7a569e742d80152ef10c9d42bc2650
Author:        Jonathan Brassow <jbrassow@redhat.com>
AuthorDate:    Fri Oct 24 13:42:06 2008 -0500
Committer:     Jonathan Brassow <jbrassow@redhat.com>
CommitterDate: Mon Oct 27 10:54:28 2008 -0500

clogd: Fix for bug 468438 - list corruption

'commit e07369b28d7a569e742d80152ef10c9d42bc2650' introduced the
concept of a delay queue to hold requests while membership changes
occurred.  Sometimes, a request would be added to the delay_queue
/and/ the cluster_queue, resulting in list corruption.  Depending
on how the list was corrupted, infinite loops could occur, or
requests could simply be lost.
---
 cmirror/src/cluster.c   |   47 ++++++++++++++++++++++++++++-------------------
 cmirror/src/functions.c |   11 +++++++++++
 cmirror/src/link_mon.c  |    1 +
 cmirror/src/local.c     |   13 +++++++------
 cmirror/src/queues.c    |   38 ++------------------------------------
 5 files changed, 49 insertions(+), 61 deletions(-)

diff --git a/cmirror/src/cluster.c b/cmirror/src/cluster.c
index 24c2540..59aa1d6 100644
--- a/cmirror/src/cluster.c
+++ b/cmirror/src/cluster.c
@@ -116,7 +116,7 @@ static struct list_head clog_cpg_list;
  * cluster_send
  * @tfr
  *
- * Returns: 0 on success, -Exxx on error
+ * Returns: 0 on success, 1 if delayed, -Exxx on error
  */
 int cluster_send(struct clog_tfr *tfr)
 {
@@ -155,7 +155,7 @@ int cluster_send(struct clog_tfr *tfr)
 			 SHORT_UUID(tfr->uuid), queue_empty(entry->delay_queue) ? "" : "*",
 			 _RQ_TYPE(tfr->request_type), tfr->seq);
 		queue_add_tail(tfr, entry->delay_queue);
-		return 0;
+		return 1;
 	}
 
 	do {
@@ -208,15 +208,14 @@ static int handle_cluster_request(struct clog_cpg *entry,
 		 * Errors from previous functions are in the tfr struct.
 		 */
 		r = cluster_send(tfr);
-		if (r)
+		if (r < 0)
 			LOG_ERROR("cluster_send failed: %s", strerror(-r));
 	}
 
 	return r;
 }
 
-static int handle_cluster_response_poop(struct clog_cpg *entry,
-					struct clog_tfr *tfr, uint32_t who)
+static int handle_cluster_response(struct clog_cpg *entry, struct clog_tfr *tfr)
 {
 	int r = 0;
 	struct clog_tfr *orig_tfr;
@@ -236,8 +235,8 @@ static int handle_cluster_response_poop(struct clog_cpg *entry,
 
 		/* Unable to find match for response */
 
-		LOG_ERROR("[%s] No match for cluster response from %u: %s:%u",
-			  SHORT_UUID(tfr->uuid), who,
+		LOG_ERROR("[%s] No match for cluster response: %s:%u",
+			  SHORT_UUID(tfr->uuid),
 			  _RQ_TYPE(tfr->request_type), tfr->seq);
 
 		INIT_LIST_HEAD(&l);
@@ -252,7 +251,7 @@ static int handle_cluster_response_poop(struct clog_cpg *entry,
 			LOG_ERROR("   [%s]  %s:%u", SHORT_UUID(t->uuid),
 				  _RQ_TYPE(t->request_type),
 				  t->seq);
-			queue_add(t, cluster_queue);
+			queue_add_tail(t, cluster_queue);
 		}
 
 		INIT_LIST_HEAD(&l);
@@ -275,14 +274,15 @@ static int handle_cluster_response_poop(struct clog_cpg *entry,
 
 	if (log_resp_rec > 0) {
 		LOG_COND(log_resend_requests,
-			 "[%s] Response received to %s/#%u from %u",
+			 "[%s] Response received to %s/#%u",
 			 SHORT_UUID(tfr->uuid), _RQ_TYPE(tfr->request_type),
-			 tfr->seq, who);
+			 tfr->seq);
 		log_resp_rec--;
 	}
 
 	/* FIXME: Ensure memcpy cannot explode */
 	memcpy(orig_tfr, tfr, sizeof(*tfr) + tfr->data_size);
+	INIT_LIST_HEAD((struct list_head *)&orig_tfr->private);
 	r = kernel_send(orig_tfr);
 	if (r)
 		LOG_ERROR("Failed to send response to kernel");
@@ -541,17 +541,19 @@ rr_create_retry:
 		}
 	}
 	memset(tfr, 0, sizeof(*tfr));
+	INIT_LIST_HEAD((struct list_head *)&tfr->private);
 	tfr->request_type = DM_CLOG_CHECKPOINT_READY;
 	tfr->originator = cp->requester;  /* FIXME: hack to overload meaning of originator */
 	strncpy(tfr->uuid, cp->uuid, CPG_MAX_NAME_LENGTH);
 
 	/* FIXME: Clean-up and use better variable for rtn */
-	if ((len = cluster_send(tfr))) {
+	len = cluster_send(tfr);
+	if (len < 0) {
 		LOG_ERROR("Failed to send checkpoint ready notice");
 		queue_add(tfr, free_queue);
 		return len;
-	}
-	queue_add(tfr, free_queue);
+	} else if (!len)
+		queue_add(tfr, free_queue);
 
 	LOG_DBG("[%s] Checkpoint ready, notification sent to %u",
 		SHORT_UUID(cp->uuid), cp->requester);
@@ -755,6 +757,7 @@ static int resend_requests(struct clog_cpg *entry)
 
 	INIT_LIST_HEAD(&resend);
 	INIT_LIST_HEAD(&delay);
+
 	queue_remove_all(&delay, entry->delay_queue);
 
 	if (entry->resend_requests) {
@@ -806,7 +809,7 @@ static int resend_requests(struct clog_cpg *entry)
 						 _RQ_TYPE(tfr->request_type),
 						 tfr->seq, entry->lowest_id);
 					queue_add_tail(tfr, cluster_queue);
-					if (cluster_send(tfr))
+					if (cluster_send(tfr) < 0)
 						LOG_ERROR("Failed resend");
 				}
 			}
@@ -825,7 +828,7 @@ static int resend_requests(struct clog_cpg *entry)
 		queue_add_tail(tfr, cluster_queue);
 
 		if ((tfr->request_type != DM_CLOG_POSTSUSPEND) &&
-		    cluster_send(tfr))
+		    (cluster_send(tfr) < 0))
 			LOG_ERROR("Failed resend");
 	}
 
@@ -1042,7 +1045,7 @@ static void cpg_message_callback(cpg_handle_t handle, struct cpg_name *gname,
 
 	if (tfr->request_type & DM_CLOG_RESPONSE) {
 		response = 1;
-		r = handle_cluster_response_poop(match, tfr, nodeid);
+		r = handle_cluster_response(match, tfr);
 	} else {
 		tfr->originator = nodeid;
 
@@ -1069,6 +1072,7 @@ static void cpg_message_callback(cpg_handle_t handle, struct cpg_name *gname,
 			}
 
 			memcpy(startup_tfr, tfr, sizeof(*tfr) + tfr->data_size);
+			INIT_LIST_HEAD((struct list_head *)&startup_tfr->private);
 			startup_tfr->error = match->lowest_id;
 			queue_add_tail(startup_tfr, match->startup_queue);
 			goto out;
@@ -1232,7 +1236,7 @@ static void cpg_leave_callback(struct clog_cpg *match,
 
 			/* Leave in the cluster_queue if not of this log */
 			if (strcmp(match->name.value, tfr->uuid)) {
-				queue_add(tfr, cluster_queue);
+				queue_add_tail(tfr, cluster_queue);
 				continue;
 			}
 
@@ -1256,6 +1260,7 @@ static void cpg_leave_callback(struct clog_cpg *match,
 
 		cpg_finalize(match->handle);
 
+		/* FIXME: redundant */
 		if (match->startup_queue->count) {
 			LOG_ERROR("%d startup items remain in cluster log",
 				  match->startup_queue->count);
@@ -1507,8 +1512,12 @@ static int _destroy_cluster_cpg(struct clog_cpg *del)
 		tfr->originator = my_cluster_id;
 		tfr->seq = 0;
 		strncpy(tfr->uuid, del->name.value, CPG_MAX_NAME_LENGTH);
-		cluster_send(tfr);
-		queue_add(tfr, free_queue);
+		r = cluster_send(tfr);
+		if (r < 0)
+			LOG_ERROR("[%s] Failed to send request to cluster: %s",
+				  SHORT_UUID(tfr->uuid), _RQ_TYPE(tfr->request_type));
+		else if (!r)
+			queue_add(tfr, free_queue);
 	}
 
 	del->cpg_state = INVALID;
diff --git a/cmirror/src/functions.c b/cmirror/src/functions.c
index 10e2158..c341d7f 100644
--- a/cmirror/src/functions.c
+++ b/cmirror/src/functions.c
@@ -47,6 +47,7 @@ struct log_c {
 
 	uint32_t *clean_bits;
 	uint32_t *sync_bits;
+	uint32_t recoverer;
 	uint64_t recovering_region; /* -1 means not recovering */
 	int sync_search;
 
@@ -481,6 +482,7 @@ static int _clog_ctr(int argc, char **argv, uint64_t device_size)
 			LOG_ERROR("Unable to allocate memory for disk_buffer");
 			goto fail;
 		}
+		memset(lc->disk_buffer, 0, lc->disk_size);
 		LOG_DBG("Disk log ready");
 	}
 
@@ -1125,6 +1127,13 @@ static int clog_get_resync_work(struct clog_tfr *tfr)
 		LOG_DBG("[%s] Someone is already recovering region %llu",
 			SHORT_UUID(lc->uuid),
 			(unsigned long long)lc->recovering_region);
+		if (lc->recoverer == tfr->originator) {
+			LOG_PRINT("[%s] %u is re-requesting resync work",
+				  SHORT_UUID(lc->uuid), tfr->originator);
+			pkg->r = lc->recovering_region;
+			pkg->i = 1;
+		}
+
 		return 0;
 	}
 
@@ -1168,6 +1177,8 @@ static int clog_get_resync_work(struct clog_tfr *tfr)
 		(unsigned long long)pkg->r);
 	pkg->i = 1;
 	lc->recovering_region = pkg->r;
+	lc->recoverer = tfr->originator;
+
 	return 0;
 }
 
diff --git a/cmirror/src/link_mon.c b/cmirror/src/link_mon.c
index 8799c9b..978c47f 100644
--- a/cmirror/src/link_mon.c
+++ b/cmirror/src/link_mon.c
@@ -89,6 +89,7 @@ int links_unregister(int fd)
 			else
 				callbacks = c->next;
 			free(c);
+			break;
 		}
 
 	return 0;
diff --git a/cmirror/src/local.c b/cmirror/src/local.c
index 8e7212a..3e9c312 100644
--- a/cmirror/src/local.c
+++ b/cmirror/src/local.c
@@ -123,6 +123,7 @@ static int kernel_recv(struct clog_tfr **tfr)
 			LOG_ERROR("len = %d, msg->len = %d", len, msg->len);
 
 		memcpy(*tfr, msg->data, msg->len);
+		INIT_LIST_HEAD((struct list_head *)&((*tfr)->private));
 
 		if (!(*tfr)->request_type) {
 			LOG_DBG("Bad transmission, requesting resend [%u]", msg->seq);
@@ -262,14 +263,14 @@ static int do_local_work(void *data)
 	case DM_CLOG_SET_REGION_SYNC:
 	case DM_CLOG_IS_REMOTE_RECOVERING:
 		r = cluster_send(tfr);
-		if (r) {
+		if (r < 0) {
 			LOG_ERROR("[%s] Unable to send %s to cluster: %s",
 				  SHORT_UUID(tfr->uuid),
 				  RQ_TYPE(tfr->request_type), strerror(-r));
 			tfr->data_size = 0;
 			tfr->error = r;
 			kernel_send(tfr);
-		} else {
+		} else if (!r) {
 			/*
 			 * If this was multi-threaded, we would have to
 			 * add the 'tfr' to the queue before doing
@@ -283,7 +284,7 @@ static int do_local_work(void *data)
 		r = kernel_ack(tfr->seq, 0);
 
 		r = cluster_send(tfr);
-		if (r) {
+		if (r < 0) {
 			LOG_ERROR("[%s] Unable to send %s to cluster: %s",
 				  SHORT_UUID(tfr->uuid),
 				  RQ_TYPE(tfr->request_type), strerror(-r));
@@ -292,8 +293,8 @@ static int do_local_work(void *data)
 			 *        This would allow us to optimize MARK_REGION
 			 *        too.
 			 */
-		}
-		queue_add_tail(tfr, free_queue);
+		} else if (!r)
+			queue_add_tail(tfr, free_queue);
 
 		break;
 	case DM_CLOG_GET_REGION_SIZE:
@@ -303,7 +304,7 @@ static int do_local_work(void *data)
 		return 0;
 	}
 
-	if (r && !tfr->error) {
+	if ((r < 0) && !tfr->error) {
 		LOG_ERROR("Programmer error: tfr->error not set.");
 		tfr->error = r;
 	}
diff --git a/cmirror/src/queues.c b/cmirror/src/queues.c
index 8d8e308..5668bdc 100644
--- a/cmirror/src/queues.c
+++ b/cmirror/src/queues.c
@@ -33,9 +33,6 @@ static void queue_dtr(struct queue *q)
 			  WHICH_QUEUE(q));
 
 	list_for_each_safe(p, n, &q->list) {
-		/* FIXME: Use proper macros
-		tfr = list_entry(p, struct clog_tfr, private);
-		*/
 		tfr = (struct clog_tfr *)p;
 		list_del_init(p);
 		free(tfr);
@@ -120,7 +117,7 @@ int init_queues(void)
 	    (r = queue_ctr(&free_queue, 100))) {
 		if (cluster_queue)
 			queue_dtr(cluster_queue);
-		/* Don't need to check last queue */
+
 		return EXIT_QUEUE_NOMEM;
 	}
 
@@ -180,18 +177,10 @@ struct clog_tfr *queue_remove(struct queue *q)
 	if (list_empty(&q->list))
 		return NULL;
 
-	/* FIXME: Use proper macros
-	tfr = list_entry(q->list.next, struct clog_tfr, private);
-	*/
 	tfr = (struct clog_tfr *)q->list.next;
 	list_del_init((struct list_head *)&tfr->private);
 	q->count--;
-/*
-	if (q == cluster_queue)
-		LOG_ERROR("[%s] remove %s %u",
-			  SHORT_UUID(tfr->uuid), RQ_TYPE(tfr->request_type),
-			  tfr->seq);
-*/
+
 	return tfr;
 }
 
@@ -218,20 +207,11 @@ struct clog_tfr *queue_remove_match(struct queue *q,
 		return NULL;
 
 	list_for_each_safe(p, n, &q->list) {
-		/* FIXME: Use proper macros
-		tfr = list_entry(p, struct clog_tfr, private);
-		*/
 		tfr = (struct clog_tfr *)p;
 		if (f(tfr, tfr_cmp)) {
 			list_del_init(p);
 			q->count--;
 
-/*
-			if (q == cluster_queue)
-				LOG_ERROR("[%s] remove match %s %u",
-					  SHORT_UUID(tfr->uuid), RQ_TYPE(tfr->request_type),
-					  tfr->seq);
-*/
 			return tfr;
 		}
 	}
@@ -249,20 +229,6 @@ struct clog_tfr *queue_remove_match(struct queue *q,
  */
 void queue_remove_all(struct list_head *l, struct queue *q)
 {
-/*
-	struct clog_tfr *tfr;
-	struct list_head *p, *n;
-
-	if (q == cluster_queue) {
-		LOG_ERROR("[--] remove all");
-		list_for_each_safe(p, n, &q->list) {
-			tfr = (struct clog_tfr *)p;
-			LOG_ERROR("    [%s] %s %u",
-				  SHORT_UUID(tfr->uuid), RQ_TYPE(tfr->request_type),
-				  tfr->seq);
-		}
-	}		
-*/
 	list_splice_init(&q->list, l);
 	q->count = 0;
 }