third_party/machinarium/channel_api: add ready count fn

This function will be used to obtain information about workers
task_channel size in future commits.

Signed-off-by: rkhapov <[email protected]>
(cherry picked from commit fe27077)
Signed-off-by: rkhapov <[email protected]>

test/CMakeLists.txt

@@ -23,6 +23,7 @@ set(od_test_src
     machinarium/test_channel_rw2.c
     machinarium/test_channel_rw3.c
     machinarium/test_channel_rw4.c
+    machinarium/test_channel_ready_count.c
     machinarium/test_channel_timeout.c
     machinarium/test_channel_cancel.c
     machinarium/test_channel_shared_create.c

test/machinarium/test_channel_ready_count.c

@@ -0,0 +1,55 @@
+
+#include <machinarium.h>
+#include <odyssey_test.h>
+
+static void test_coroutine(void *arg)
+{
+	(void)arg;
+	machine_channel_t *channel;
+	channel = machine_channel_create(0);
+	test(channel != NULL);
+
+	machine_msg_t *msg;
+	msg = machine_msg_create(0);
+	test(msg != NULL);
+
+	test(machine_channel_ready_count(channel) == 0);
+
+	machine_msg_set_type(msg, 1);
+	machine_channel_write(channel, msg);
+	test(machine_channel_ready_count(channel) == 1);
+
+	machine_channel_write(channel, msg);
+	test(machine_channel_ready_count(channel) == 2);
+
+	machine_channel_write(channel, msg);
+	test(machine_channel_ready_count(channel) == 3);
+
+	(void)machine_channel_read_back(channel, UINT32_MAX);
+	test(machine_channel_ready_count(channel) == 2);
+
+	(void)machine_channel_read_back(channel, UINT32_MAX);
+	test(machine_channel_ready_count(channel) == 1);
+
+	(void)machine_channel_read_back(channel, UINT32_MAX);
+	test(machine_channel_ready_count(channel) == 0);
+
+	machine_msg_free(msg);
+
+	machine_channel_free(channel);
+}
+
+void machinarium_test_channel_ready_count(void)
+{
+	machinarium_init();
+
+	int id;
+	id = machine_create("test", test_coroutine, NULL);
+	test(id != -1);
+
+	int rc;
+	rc = machine_wait(id);
+	test(rc != -1);
+
+	machinarium_free();
+}

test/odyssey_test.c

@@ -38,6 +38,7 @@ extern void machinarium_test_channel_rw1(void);
 extern void machinarium_test_channel_rw2(void);
 extern void machinarium_test_channel_rw3(void);
 extern void machinarium_test_channel_rw4(void);
+extern void machinarium_test_channel_ready_count(void);
 extern void machinarium_test_channel_timeout(void);
 extern void machinarium_test_channel_cancel(void);
 extern void machinarium_test_channel_shared_create(void);
@@ -118,6 +119,7 @@ int main(int argc, char *argv[])
 	odyssey_test(machinarium_test_channel_rw2);
 	odyssey_test(machinarium_test_channel_rw3);
 	odyssey_test(machinarium_test_channel_rw4);
+	odyssey_test(machinarium_test_channel_ready_count);
 	odyssey_test(machinarium_test_channel_timeout);
 	odyssey_test(machinarium_test_channel_cancel);
 	odyssey_test(machinarium_test_channel_shared_create);

third_party/machinarium/sources/channel.c

@@ -184,3 +184,11 @@ mm_msg_t *mm_channel_read_back(mm_channel_t *channel, uint32_t time_ms)
 
 	return reader.result;
 }
+
+int mm_channel_ready_count(mm_channel_t *chan)
+{
+	MM_SLEEPLOCK_UNLOCK_ON_EXIT mm_sleeplock_t *lock = &chan->lock;
+	mm_sleeplock_lock(lock);
+
+	return chan->msg_list_count;
+}

third_party/machinarium/sources/channel.h

@@ -34,4 +34,6 @@ mm_retcode_t mm_channel_write(mm_channel_t *, mm_msg_t *);
 mm_msg_t *mm_channel_read(mm_channel_t *, uint32_t);
 mm_msg_t *mm_channel_read_back(mm_channel_t *, uint32_t);
 
+int mm_channel_ready_count(mm_channel_t *);
+
 #endif /* MM_CHANNEL_H */

third_party/machinarium/sources/channel_api.c

@@ -112,3 +112,20 @@ MACHINE_API machine_msg_t *machine_channel_read_back(machine_channel_t *obj,
 	msg = mm_channelfast_read(channel, time_ms);
 	return (machine_msg_t *)msg;
 }
+
+MACHINE_API int machine_channel_ready_count(machine_channel_t *obj)
+{
+	mm_channeltype_t *type;
+	type = mm_cast(mm_channeltype_t *, obj);
+
+	if (type->is_shared) {
+		mm_channel_t *channel;
+		channel = mm_cast(mm_channel_t *, obj);
+		return mm_channel_ready_count(channel);
+	}
+
+	mm_channelfast_t *channel;
+	channel = mm_cast(mm_channelfast_t *, obj);
+
+	return mm_channelfast_ready_count(channel);
+}

third_party/machinarium/sources/channel_fast.c

@@ -82,3 +82,8 @@ mm_msg_t *mm_channelfast_read(mm_channelfast_t *channel, uint32_t time_ms)
 	channel->incoming_count--;
 	return mm_container_of(first, mm_msg_t, link);
 }
+
+int mm_channelfast_ready_count(mm_channelfast_t *chan)
+{
+	return chan->incoming_count;
+}

third_party/machinarium/sources/channel_fast.h

@@ -35,4 +35,6 @@ mm_retcode_t mm_channelfast_write(mm_channelfast_t *, mm_msg_t *);
 
 mm_msg_t *mm_channelfast_read(mm_channelfast_t *, uint32_t);
 
+int mm_channelfast_ready_count(mm_channelfast_t *);
+
 #endif /* MM_CHANNEL_FAST_H */

third_party/machinarium/sources/machinarium.h

@@ -161,6 +161,8 @@ MACHINE_API machine_msg_t *machine_channel_read(machine_channel_t *,
 MACHINE_API machine_msg_t *machine_channel_read_back(machine_channel_t *,
 						     uint32_t time_ms);
 
+MACHINE_API int machine_channel_ready_count(machine_channel_t *);
+
 /* tls */
 
 MACHINE_API machine_tls_t *machine_tls_create(void);

third_party/machinarium/sources/sleep_lock.h

@@ -39,4 +39,12 @@ static inline void mm_sleeplock_unlock(mm_sleeplock_t *lock)
 	__sync_lock_release(lock);
 }
 
+static inline void mm_sleeplock_cleanup(mm_sleeplock_t **lock)
+{
+	mm_sleeplock_unlock(*lock);
+}
+
+#define MM_SLEEPLOCK_UNLOCK_ON_EXIT \
+	__attribute__((cleanup(mm_sleeplock_cleanup)))
+
 #endif /* MM_SLEEP_LOCK_H */