ctdb: Adding memory pool for queue callback

[Swen Schillig]

This is an updated version of the patch I sent out a few weeks ago.

The reason for having an update was that further tests showed,
that it was not beneficial to have the queue_send routine being a
consumer of the pool as well.
In our tests we saw that it is not likely that additional memory needs
to be allocated, but if it is, the size is so big that all of the pool
memory is used and kept for a considerably long time.
This has the negative side effect on the other consumer
(receiver / queue_read) which can not benefit from the pool anymore
until this memory chunk is free'd again.

Please review and push if happy.

Thanks for your support in advance.

Cheers Swen.
-------------- next part --------------
From cb11fc3c4f69ff0e352a63883562c50b177a9cf7 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.
During tests it could be seen that more than 95% of all messages were sized
below the standard buffer_size of 1k. All received messages are processed
synchronously per queue, therefore, a larger memory pool is not requried.

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

diff --git a/ctdb/common/ctdb_io.c b/ctdb/common/ctdb_io.c
index 27caf6b527e..3f4d5c4bd90 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);
 
@@ -477,5 +478,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.17.1