[PATCH] ctdb: Adding memory pool for queue callback

[Swen Schillig]

Please review and push if happy.

Thanks for your support in advance.

Cheers Swen
-------------- next part --------------
From 41cac803946abcdf028e8cf4cbaf045665019fed Mon Sep 17 00:00:00 2001
From: Swen Schillig <swen at vnet.ibm.com>
Date: Mon, 12 Mar 2018 17:56:21 +0100
Subject: [PATCH] ctdb: Adding memory pool for queue callback

The received packet is copied into a newly allocated memory chunk for further
processing by the assigned callback. Once this is done, the memory is free'd.
This is repeated for each received packet making the memory allocation / free
an expensive task. To optimize this process, a memory pool is defined which
is sized identically to the queue's buffer and is therefore adjustable via
tunable.queue_buffer_size.

Signed-off-by: Swen Schillig <swen at vnet.ibm.com>
---
 ctdb/common/ctdb_io.c | 14 +++++++++++---
 1 file changed, 11 insertions(+), 3 deletions(-)

diff --git a/ctdb/common/ctdb_io.c b/ctdb/common/ctdb_io.c
index 657b51e7715..9c828605fd0 100644
--- a/ctdb/common/ctdb_io.c
+++ b/ctdb/common/ctdb_io.c
@@ -65,6 +65,7 @@ struct ctdb_queue {
 	size_t alignment;
 	void *private_data;
 	ctdb_queue_cb_fn_t callback;
+	char *data_pool;
 	const char *name;
 	uint32_t buffer_size;
 };
@@ -115,7 +116,7 @@ static void queue_process(struct ctdb_queue *queue)
 	}
 
 	/* Extract complete packet */
-	data = talloc_memdup(queue,
+	data = talloc_memdup(queue->data_pool,
 			     queue->buffer.data + queue->buffer.offset,
 			     pkt_size);
 
@@ -368,8 +369,9 @@ int ctdb_queue_send(struct ctdb_queue *queue, uint8_t *data, uint32_t length)
 		if (length2 == 0) return 0;
 	}
 
-	pkt = talloc_size(
-		queue, offsetof(struct ctdb_queue_pkt, buf) + length2);
+	pkt = talloc_size(queue->data_pool,
+			  offsetof(struct ctdb_queue_pkt, buf) + length2);
+
 	CTDB_NO_MEMORY(queue->ctdb, pkt);
 	talloc_set_name_const(pkt, "struct ctdb_queue_pkt");
 
@@ -480,5 +482,11 @@ struct ctdb_queue *ctdb_queue_setup(struct ctdb_context *ctdb,
 		queue->buffer_size = 1024;
 	}
 
+	queue->data_pool = talloc_pool(queue, queue->buffer_size);
+	if (queue->data_pool == NULL) {
+		TALLOC_FREE(queue);
+		return NULL;
+	}
+
 	return queue;
 }
-- 
2.14.4