Allocate packet buffers in the transport drivers

This change paves the way for eliminating a copy within the Infiniband
driver in the future by transferring responsibility for allocating and
freeing message buffers to the transport driver layer.

Tested under valgrind on a single-node cluster.

Signed-off-by: Zane Bitter <zane.bitter@gmail.com>
Reviewed-by: Steven Dake <sdake@redhat.com>

exec/totemiba.c

@@ -1317,6 +1317,16 @@ int totemiba_initialize (
 	return (res);
 }
 
+void *totemiba_buffer_alloc (void)
+{
+	return malloc (MAX_MTU_SIZE);
+}
+
+void totemiba_buffer_release (void *ptr)
+{
+	return free (ptr);
+}
+
 int totemiba_processor_count_set (
 	void *iba_context,
 	int processor_count)

exec/totemiba.h

@@ -62,6 +62,10 @@ extern int totemiba_initialize (
 	void (*target_set_completed) (
 		void *context));
 
+extern void *totemiba_buffer_alloc (void);
+
+extern void totemiba_buffer_release (void *ptr);
+
 extern int totemiba_processor_count_set (
 	void *iba_context,
 	int processor_count);

exec/totemnet.c

@@ -35,6 +35,8 @@
 
 #include <config.h>
 
+#include <assert.h>
+
 #ifdef HAVE_RDMA
 #include <totemiba.h>
 #endif
@@ -67,6 +69,10 @@ struct transport {
 		void (*target_set_completed) (
 			void *context));
 
+	void *(*buffer_alloc) (void);
+
+	void (*buffer_release) (void *ptr);
+
 	int (*processor_count_set) (
 		void *transport_context,
 		int processor_count);
@@ -127,6 +133,8 @@ struct transport transport_entries[] = {
 	{
 		.name = "UDP/IP Multicast",
 		.initialize = totemudp_initialize,
+		.buffer_alloc = totemudp_buffer_alloc,
+		.buffer_release = totemudp_buffer_release,
 		.processor_count_set = totemudp_processor_count_set,
 		.token_send = totemudp_token_send,
 		.mcast_flush_send = totemudp_mcast_flush_send,
@@ -145,6 +153,8 @@ struct transport transport_entries[] = {
 	{
 		.name = "UDP/IP Unicast",
 		.initialize = totemudpu_initialize,
+		.buffer_alloc = totemudpu_buffer_alloc,
+		.buffer_release = totemudpu_buffer_release,
 		.processor_count_set = totemudpu_processor_count_set,
 		.token_send = totemudpu_token_send,
 		.mcast_flush_send = totemudpu_mcast_flush_send,
@@ -166,6 +176,8 @@ struct transport transport_entries[] = {
 	{
 		.name = "Infiniband/IP",
 		.initialize = totemiba_initialize,
+		.buffer_alloc = totemiba_buffer_alloc,
+		.buffer_release = totemiba_buffer_release,
 		.processor_count_set = totemiba_processor_count_set,
 		.token_send = totemiba_token_send,
 		.mcast_flush_send = totemiba_mcast_flush_send,
@@ -296,6 +308,22 @@ error_destroy:
 	return (-1);
 }
 
+void *totemnet_buffer_alloc (void *net_context)
+{
+	struct totemnet_instance *instance = net_context;
+	assert (instance != NULL);
+	assert (instance->transport != NULL);
+	return instance->transport->buffer_alloc();
+}
+
+void totemnet_buffer_release (void *net_context, void *ptr)
+{
+	struct totemnet_instance *instance = net_context;
+	assert (instance != NULL);
+	assert (instance->transport != NULL);
+	instance->transport->buffer_release (ptr);
+}
+
 int totemnet_processor_count_set (
 	void *net_context,
 	int processor_count)

exec/totemnet.h

@@ -69,6 +69,10 @@ extern int totemnet_initialize (
 	void (*target_set_completed) (
 		void *context));
 
+extern void *totemnet_buffer_alloc (void *net_context);
+
+extern void totemnet_buffer_release (void *net_context, void *ptr);
+
 extern int totemnet_processor_count_set (
 	void *net_context,
 	int processor_count);

exec/totemrrp.c

@@ -1665,6 +1665,20 @@ error_destroy:
 	return (res);
 }
 
+void *totemrrp_buffer_alloc (void *rrp_context)
+{
+	struct totemrrp_instance *instance = rrp_context;
+	assert (instance != NULL);
+	return totemnet_buffer_alloc (instance->net_handles[0]);
+}
+
+void totemrrp_buffer_release (void *rrp_context, void *ptr)
+{
+	struct totemrrp_instance *instance = rrp_context;
+	assert (instance != NULL);
+	totemnet_buffer_release (instance->net_handles[0], ptr);
+}
+
 int totemrrp_processor_count_set (
 	void *rrp_context,
 	unsigned int processor_count)

exec/totemrrp.h

@@ -78,6 +78,12 @@ extern int totemrrp_initialize (
 		void *context)
 	);
 
+extern void *totemrrp_buffer_alloc (
+	void *rrp_context);
+
+extern void totemrrp_buffer_release (
+	void *rrp_context,
+	void *ptr);
 
 extern int totemrrp_processor_count_set (
 	void *rrp_context,

exec/totemsrp.c

@@ -606,6 +606,8 @@ static void timer_function_heartbeat_timeout (void *data);
 static void timer_function_token_retransmit_timeout (void *data);
 static void timer_function_token_hold_retransmit_timeout (void *data);
 static void timer_function_merge_detect_timeout (void *data);
+static void *totemsrp_buffer_alloc (struct totemsrp_instance *instance);
+static void totemsrp_buffer_release (struct totemsrp_instance *instance, void *ptr);
 
 void main_deliver_fn (
 	void *context,
@@ -1356,6 +1358,18 @@ static void memb_set_print (
 }
 #endif
 
+static void *totemsrp_buffer_alloc (struct totemsrp_instance *instance)
+{
+	assert (instance != NULL);
+	return totemrrp_buffer_alloc (instance->totemrrp_context);
+}
+
+static void totemsrp_buffer_release (struct totemsrp_instance *instance, void *ptr)
+{
+	assert (instance != NULL);
+	totemrrp_buffer_release (instance->totemrrp_context, ptr);
+}
+
 static void reset_token_retransmit_timeout (struct totemsrp_instance *instance)
 {
 	poll_timer_delete (instance->totemsrp_poll_handle,
@@ -2067,7 +2081,7 @@ static void memb_state_recovery_enter (
 		messages_originated++;
 		memset (&message_item, 0, sizeof (struct message_item));
 	// TODO	 LEAK
-		message_item.mcast = malloc (FRAME_SIZE_MAX);
+		message_item.mcast = totemsrp_buffer_alloc (instance);
 		assert (message_item.mcast);
 		message_item.mcast->header.type = MESSAGE_TYPE_MCAST;
 		srp_addr_copy (&message_item.mcast->system_from, &instance->my_id);
@@ -2140,7 +2154,7 @@ int totemsrp_mcast (
 	/*
 	 * Allocate pending item
 	 */
-	message_item.mcast = malloc (FRAME_SIZE_MAX);
+	message_item.mcast = totemsrp_buffer_alloc (instance);
 	if (message_item.mcast == 0) {
 		goto error_mcast;
 	}
@@ -2278,7 +2292,7 @@ static void messages_free (
 			instance->last_released + i, &ptr);
 		if (res == 0) {
 			regular_message = ptr;
-			free (regular_message->mcast);
+			totemsrp_buffer_release (instance, regular_message->mcast);
 		}
 		sq_items_release (&instance->regular_sort_queue,
 			instance->last_released + i);
@@ -3823,7 +3837,7 @@ static int message_handler_mcast (
 		 * Allocate new multicast memory block
 		 */
 // TODO LEAK
-		sort_queue_item.mcast = malloc (msg_len);
+		sort_queue_item.mcast = totemsrp_buffer_alloc (instance);
 		if (sort_queue_item.mcast == NULL) {
 			return (-1); /* error here is corrected by the algorithm */
 		}

exec/totemudp.c

@@ -1832,6 +1832,16 @@ int totemudp_initialize (
 	return (0);
 }
 
+void *totemudp_buffer_alloc (void)
+{
+	return malloc (FRAME_SIZE_MAX);
+}
+
+void totemudp_buffer_release (void *ptr)
+{
+	return free (ptr);
+}
+
 int totemudp_processor_count_set (
 	void *udp_context,
 	int processor_count)

exec/totemudp.h

@@ -63,6 +63,10 @@ extern int totemudp_initialize (
 	void (*target_set_completed) (
 		void *context));
 
+extern void *totemudp_buffer_alloc (void);
+
+extern void totemudp_buffer_release (void *ptr);
+
 extern int totemudp_processor_count_set (
 	void *udp_context,
 	int processor_count);

exec/totemudpu.c

@@ -1497,6 +1497,16 @@ int totemudpu_initialize (
 	return (0);
 }
 
+void *totemudpu_buffer_alloc (void)
+{
+	return malloc (FRAME_SIZE_MAX);
+}
+
+void totemudpu_buffer_release (void *ptr)
+{
+	return free (ptr);
+}
+
 int totemudpu_processor_count_set (
 	void *udpu_context,
 	int processor_count)

exec/totemudpu.h

@@ -63,6 +63,10 @@ extern int totemudpu_initialize (
 	void (*target_set_completed) (
 		void *context));
 
+extern void *totemudpu_buffer_alloc (void);
+
+extern void totemudpu_buffer_release (void *ptr);
+
 extern int totemudpu_processor_count_set (
 	void *udpu_context,
 	int processor_count);