machinarium: allocate msg payload data on create

examples/benchmark_channel.c

@@ -28,7 +28,7 @@ benchmark_writer(void *arg)
 	machine_channel_t channel = arg;
 	while (machine_active()) {
 		machine_msg_t msg;
-		msg = machine_msg_create(0);
+		msg = machine_msg_create(0, 0);
 		machine_channel_write(channel, msg);
 		ops++;
 		machine_sleep(0);
@@ -56,7 +56,6 @@ benchmark_runner(void *arg)
 
 	printf("done.\n");
 	printf("channel operations %d in 1 sec.\n", ops);
-	printf(".\n");
 }
 
 int

src/mm_msg.c

@@ -9,12 +9,20 @@
 #include <machinarium_private.h>
 
 MACHINE_API machine_msg_t
-machine_msg_create(int type)
+machine_msg_create(int type, int data_size)
 {
 	mm_msg_t *msg = mm_msgpool_pop(&machinarium.msg_pool);
 	if (msg == NULL)
 		return NULL;
 	msg->type = type;
+	if (data_size > 0) {
+		int rc;
+		rc = mm_buf_ensure(&msg->data, data_size);
+		if (rc == -1) {
+			mm_msg_unref(&machinarium.msg_pool, msg);
+			return NULL;
+		}
+	}
 	return msg;
 }
 
@@ -25,18 +33,11 @@ machine_msg_free(machine_msg_t obj)
 	mm_msgpool_push(&machinarium.msg_pool, msg);
 }
 
-MACHINE_API void
-machine_msg_set_data(machine_msg_t obj, void *data)
-{
-	mm_msg_t *msg = obj;
-	msg->data = data;
-}
-
 MACHINE_API void*
 machine_msg_get_data(machine_msg_t obj)
 {
 	mm_msg_t *msg = obj;
-	return msg->data;
+	return msg->data.start;
 }
 
 MACHINE_API int

src/mm_msg.h

@@ -12,7 +12,7 @@ typedef struct mm_msg_t mm_msg_t;
 struct mm_msg_t {
 	uint16_t  refs;
 	int       type;
-	void     *data;
+	mm_buf_t  data;
 	mm_list_t link;
 };
 

src/mm_msg_pool.c

@@ -45,7 +45,7 @@ mm_msgpool_pop(mm_msgpool_t *pool)
 init:
 	msg->refs = 0;
 	msg->type = 0;
-	msg->data = NULL;
+	mm_buf_init(&msg->data);
 	mm_list_init(&msg->link);
 	return msg;
 }

tests/test_channel_rw0.c

@@ -16,7 +16,7 @@ test_fiber(void *arg)
 	test(channel != NULL);
 
 	machine_msg_t msg;
-	msg = machine_msg_create(123);
+	msg = machine_msg_create(123, 0);
 	test(msg != NULL);
 
 	machine_channel_write(channel, msg);

tests/test_channel_rw1.c

@@ -19,7 +19,7 @@ test_fiber2(void *arg)
 	test(machine_msg_get_type(msg) == 123);
 	machine_msg_free(msg);
 
-	msg = machine_msg_create(321);
+	msg = machine_msg_create(321, 0);
 	machine_channel_write(channel, msg);
 }
 
@@ -33,7 +33,7 @@ test_fiber(void *arg)
 	id = machine_fiber_create(test_fiber2, NULL);
 
 	machine_msg_t msg;
-	msg = machine_msg_create(123);
+	msg = machine_msg_create(123, 0);
 	test(msg != NULL);
 	machine_channel_write(channel, msg);
 

tests/test_channel_rw2.c

@@ -19,7 +19,7 @@ test_fiber2(void *arg)
 	test(machine_msg_get_type(msg) == 123);
 	machine_msg_free(msg);
 
-	msg = machine_msg_create(321);
+	msg = machine_msg_create(321, 0);
 	machine_channel_write(channel, msg);
 }
 
@@ -34,7 +34,7 @@ test_fiber(void *arg)
 	machine_sleep(0);
 
 	machine_msg_t msg;
-	msg = machine_msg_create(123);
+	msg = machine_msg_create(123, 0);
 	test(msg != NULL);
 	machine_channel_write(channel, msg);
 

tests/test_channel_rw3.c

@@ -62,15 +62,15 @@ test_fiber(void *arg)
 	machine_sleep(0);
 
 	machine_msg_t msg;
-	msg = machine_msg_create(1);
+	msg = machine_msg_create(1, 0);
 	test(msg != NULL);
 	machine_channel_write(channel, msg);
 
-	msg = machine_msg_create(2);
+	msg = machine_msg_create(2, 0);
 	test(msg != NULL);
 	machine_channel_write(channel, msg);
 
-	msg = machine_msg_create(3);
+	msg = machine_msg_create(3, 0);
 	test(msg != NULL);
 	machine_channel_write(channel, msg);
 

tests/test_channel_rw4.c

@@ -62,17 +62,17 @@ test_fiber(void *arg)
 	machine_sleep(0);
 
 	machine_msg_t msg;
-	msg = machine_msg_create(1);
+	msg = machine_msg_create(1, 0);
 	test(msg != NULL);
 	machine_channel_write(channel, msg);
 	machine_sleep(0);
 
-	msg = machine_msg_create(2);
+	msg = machine_msg_create(2, 0);
 	test(msg != NULL);
 	machine_channel_write(channel, msg);
 	machine_sleep(0);
 
-	msg = machine_msg_create(3);
+	msg = machine_msg_create(3, 0);
 	test(msg != NULL);
 	machine_channel_write(channel, msg);
 	machine_sleep(0);

tests/test_queue_producer_consumer0.c

@@ -30,7 +30,7 @@ test_producer(void *arg)
 	int i = 0;
 	for (; i < 100; i++) {
 		machine_msg_t msg;
-		msg = machine_msg_create(i);
+		msg = machine_msg_create(i, 0);
 		test(msg != NULL);
 		machine_queue_put(queue, msg);
 	}

tests/test_queue_producer_consumer1.c

@@ -33,14 +33,14 @@ test_producer(void *arg)
 	int i = 0;
 	for (; i < 100000; i++) {
 		machine_msg_t msg;
-		msg = machine_msg_create(i);
+		msg = machine_msg_create(i, 0);
 		test(msg != NULL);
 		machine_queue_put(queue, msg);
 	}
 	/* exit */
 	for (i = 0; i < consumers_count; i++ ){
 		machine_msg_t msg;
-		msg = machine_msg_create(UINT32_MAX);
+		msg = machine_msg_create(UINT32_MAX, 0);
 		test(msg != NULL);
 		machine_queue_put(queue, msg);
 	}

tests/test_queue_rw0.c

@@ -16,7 +16,7 @@ test_fiber(void *arg)
 	test(queue != NULL);
 
 	machine_msg_t msg;
-	msg = machine_msg_create(123);
+	msg = machine_msg_create(123, 0);
 	test(msg != NULL);
 
 	machine_queue_put(queue, msg);

tests/test_queue_rw1.c

@@ -19,7 +19,7 @@ test_fiber2(void *arg)
 	test(machine_msg_get_type(msg) == 123);
 	machine_msg_free(msg);
 
-	msg = machine_msg_create(321);
+	msg = machine_msg_create(321, 0);
 	machine_queue_put(queue, msg);
 }
 
@@ -33,7 +33,7 @@ test_fiber(void *arg)
 	id = machine_fiber_create(test_fiber2, NULL);
 
 	machine_msg_t msg;
-	msg = machine_msg_create(123);
+	msg = machine_msg_create(123, 0);
 	test(msg != NULL);
 	machine_queue_put(queue, msg);
 

tests/test_queue_rw2.c

@@ -19,7 +19,7 @@ test_fiber2(void *arg)
 	test(machine_msg_get_type(msg) == 123);
 	machine_msg_free(msg);
 
-	msg = machine_msg_create(321);
+	msg = machine_msg_create(321, 0);
 	machine_queue_put(queue, msg);
 }
 
@@ -34,7 +34,7 @@ test_fiber(void *arg)
 	machine_sleep(0);
 
 	machine_msg_t msg;
-	msg = machine_msg_create(123);
+	msg = machine_msg_create(123, 0);
 	test(msg != NULL);
 	machine_queue_put(queue, msg);