fix audisp/queue.c wrap resize

Fixed the wrapped-resize data loss in audisp/queue.c:367. The grow path now
allocates a fresh queue, copies live entries in FIFO order starting from
q_last, then resets q_last to 0 and q_next to the live count. That preserves
dequeue order even when the old ring had wrapped.

Added a regression test in audisp/test/test-queue.c:242. It fills a 100-slot
queue, dequeues 80 entries, enqueues 20 more to force wrap, resizes to 200,
and verifies that entries 80..119 are still dequeued in order.

Author: stevegrubb

audisp/queue.c

@@ -368,20 +368,44 @@ void increase_queue_depth(unsigned int size)
 {
 	pthread_mutex_lock(&queue_lock);
 	if (size > q_depth) {
-		unsigned int i;
-		void *tmp_q;
+		volatile event_t **tmp_q;
+		unsigned int count, i, old_depth, old_last;
 
-		tmp_q = realloc(q, size * sizeof(event_t *));
+		old_depth = q_depth;
+		count = AUDIT_ATOMIC_LOAD(currently_used);
+#ifdef HAVE_ATOMIC
+		old_last = atomic_load_explicit(&q_last,
+				memory_order_relaxed) % old_depth;
+#else
+		old_last = q_last % old_depth;
+#endif
+
+		tmp_q = calloc(size, sizeof(event_t *));
 		if (tmp_q == NULL) {
 			fprintf(stderr, "Out of Memory. Check %s file, %d line",
 				__FILE__, __LINE__);
 			pthread_mutex_unlock(&queue_lock);
 			return;
 		}
+
+		/*
+		 * Preserve FIFO order across a grow. Once the ring has wrapped,
+		 * simply extending the backing array strands the entries stored
+		 * at the start of the old allocation.
+		 */
+		for (i = 0; i < count; i++)
+			tmp_q[i] = q[(old_last + i) % old_depth];
+
+		free((void *)q);
 		q = tmp_q;
-		for (i=q_depth; i<size; i++)
-			q[i] = NULL;
 		q_depth = size;
+#ifdef HAVE_ATOMIC
+		atomic_store_explicit(&q_last, 0, memory_order_relaxed);
+		atomic_store_explicit(&q_next, count, memory_order_relaxed);
+#else
+		q_last = 0;
+		q_next = count;
+#endif
 		overflowed = 0;
 	}
 	pthread_mutex_unlock(&queue_lock);
@@ -456,4 +480,3 @@ int queue_overflowed_p(void)
 {
        return overflowed;
 }
-

audisp/test/test-queue.c

@@ -239,6 +239,93 @@ static int persist_test(const char *logfile)
 	return rc;
 }
 
+static int resize_wrap_test(void)
+{
+	struct disp_conf conf;
+	event_t *e = NULL;
+	char buf[32];
+	int i, rc = 1;
+
+	memset(&conf, 0, sizeof(conf));
+	conf.overflow_action = O_IGNORE;
+
+	if (init_queue(100)) {
+		fprintf(stderr, "resize_wrap_test: init_queue failed\n");
+		return rc;
+	}
+
+	/* Fill the ring so the next enqueue sequence can wrap it. */
+	for (i = 0; i < 100; i++) {
+		snprintf(buf, sizeof(buf), "event-%03d", i);
+		e = make_event(buf);
+		if (!e || enqueue(e, &conf)) {
+			fprintf(stderr, "resize_wrap_test: enqueue failed\n");
+			goto out_q;
+		}
+		e = NULL;
+	}
+
+	for (i = 0; i < 80; i++) {
+		snprintf(buf, sizeof(buf), "event-%03d", i);
+		e = dequeue();
+		if (!e) {
+			fprintf(stderr, "resize_wrap_test: queue underflow\n");
+			goto out_q;
+		}
+		if (strcmp(e->data, buf) != 0) {
+			fprintf(stderr, "resize_wrap_test: initial data mismatch\n");
+			goto out_free;
+		}
+		free(e);
+		e = NULL;
+	}
+
+	/* Add more entries so the live data spans the end and start. */
+	for (i = 100; i < 120; i++) {
+		snprintf(buf, sizeof(buf), "event-%03d", i);
+		e = make_event(buf);
+		if (!e || enqueue(e, &conf)) {
+			fprintf(stderr,
+				"resize_wrap_test: wrapped enqueue failed\n");
+			goto out_q;
+		}
+		e = NULL;
+	}
+
+	/* Growing a wrapped ring must not strand the entries at slot 0. */
+	increase_queue_depth(200);
+
+	/* The consumer should still see the wrapped entries in FIFO order. */
+	for (i = 80; i < 120; i++) {
+		snprintf(buf, sizeof(buf), "event-%03d", i);
+		e = dequeue();
+		if (!e) {
+			fprintf(stderr,
+				"resize_wrap_test: queue underflow after resize\n");
+			goto out_q;
+		}
+		if (strcmp(e->data, buf) != 0) {
+			fprintf(stderr,
+				"resize_wrap_test: data mismatch after resize\n");
+			goto out_free;
+		}
+		free(e);
+		e = NULL;
+	}
+
+	if (queue_current_depth() != 0) {
+		fprintf(stderr, "resize_wrap_test: depth not zero\n");
+		goto out_q;
+	}
+
+	rc = 0;
+out_free:
+	free(e);
+out_q:
+	destroy_queue();
+	return rc;
+}
+
 int main(void)
 {
 	const char *srcdir = getenv("srcdir") ? getenv("srcdir") : ".";
@@ -247,10 +334,11 @@ int main(void)
 
 	if (basic_test(path))
 		return 1;
+	if (resize_wrap_test())
+		return 1;
 	if (persist_test(path))
 		return 1;
 	if (concurrency_test(path))
 		return 1;
 	return 0;
 }
-