RIOT-OS/RIOT
RIOT-OS/RIOT
1
0
Fork 0
mirror of https://github.com/RIOT-OS/RIOT.git synced 2025-01-18 12:52:44 +01:00
Code Releases Activity

tests/sys_sema_inv: add test for inverse semaphore

Browse Source
This commit is contained in:
Benjamin Valentin 2021-01-13 17:38:44 +01:00
parent 5602587dd7
commit b0d476c890
4 changed files with 166 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
tests/sys_sema_inv/Makefile Normal file
View File

@ -0,0 +1,6 @@
include ../Makefile.tests_common
USEMODULE += sema_inv
USEMODULE += xtimer
include $(RIOTBASE)/Makefile.include

7
tests/sys_sema_inv/Makefile.ci Normal file
View File

@ -0,0 +1,7 @@
BOARD_INSUFFICIENT_MEMORY := \
arduino-duemilanove \
arduino-nano \
arduino-uno \
atmega328p \
nucleo-l011k4 \
#

127
tests/sys_sema_inv/main.c Normal file
View File

@ -0,0 +1,127 @@
/*
* Copyright (C) 2021 ML!PA Consulting GmbH
*
* This file is subject to the terms and conditions of the GNU Lesser
* General Public License v2.1. See the file LICENSE in the top level
* directory for more details.
*/
/**
* @ingroup tests
* @{
*
* @file
* @brief Inverse Semaphore Test Application
*
* @author Benjamin Valentin <benjamin.valentin@ml-pa.com>
* @}
*/
#include <assert.h>
#include <stdio.h>
#include "sema_inv.h"
#include "xtimer.h"
char t1_stack[THREAD_STACKSIZE_SMALL];
char t2_stack[THREAD_STACKSIZE_SMALL];
char t3_stack[THREAD_STACKSIZE_SMALL];
struct thread_ctx {
sema_inv_t *sync;
unsigned id;
};
static void *thread_count(void *arg)
{
struct thread_ctx *ctx = arg;
printf("THREAD %u start\n", ctx->id);
xtimer_msleep(5);
if (sema_inv_post(ctx->sync)) {
printf("THREAD %u woke main thread\n", ctx->id);
}
return arg;
}
static void *thread_bit(void *arg)
{
struct thread_ctx *ctx = arg;
printf("THREAD %u start\n", ctx->id);
xtimer_msleep(5);
if (sema_inv_post_mask(ctx->sync, 1 << ctx->id)) {
printf("THREAD %u woke main thread\n", ctx->id);
}
return arg;
}
static void test_counter_mode(void)
{
sema_inv_t sync;
struct thread_ctx ctx[3] = {
{ .sync = &sync, .id = 1 },
{ .sync = &sync, .id = 2 },
{ .sync = &sync, .id = 3 },
};
puts("counter mode");
sema_inv_init(&sync, 3);
thread_create(t1_stack, sizeof(t1_stack), THREAD_PRIORITY_MAIN - 1,
THREAD_CREATE_STACKTEST, thread_count, &ctx[0], "nr1");
thread_create(t2_stack, sizeof(t2_stack), THREAD_PRIORITY_MAIN + 1,
THREAD_CREATE_STACKTEST, thread_count, &ctx[1], "nr2");
thread_create(t3_stack, sizeof(t3_stack), THREAD_PRIORITY_MAIN + 1,
THREAD_CREATE_STACKTEST, thread_count, &ctx[2], "nr3");
sema_inv_wait(&sync);
puts("thread synced");
/* wait for all threads to terminate, we are going to re-use the stack */
xtimer_msleep(10);
}
static void test_mask_mode(void)
{
sema_inv_t sync;
struct thread_ctx ctx[3] = {
{ .sync = &sync, .id = 1 },
{ .sync = &sync, .id = 2 },
{ .sync = &sync, .id = 3 },
};
puts("mask mode");
sema_inv_init(&sync, 0xE);
thread_create(t1_stack, sizeof(t1_stack), THREAD_PRIORITY_MAIN - 1,
THREAD_CREATE_STACKTEST, thread_bit, &ctx[0], "nr1");
thread_create(t2_stack, sizeof(t2_stack), THREAD_PRIORITY_MAIN + 1,
THREAD_CREATE_STACKTEST, thread_bit, &ctx[1], "nr2");
thread_create(t3_stack, sizeof(t3_stack), THREAD_PRIORITY_MAIN + 1,
THREAD_CREATE_STACKTEST, thread_bit, &ctx[2], "nr3");
sema_inv_wait(&sync);
puts("thread synced");
/* wait for all threads to terminate, we are going to re-use the stack */
xtimer_msleep(10);
}
int main(void)
{
test_counter_mode();
test_mask_mode();
puts("SUCCESS");
return 0;
}

26
tests/sys_sema_inv/tests/01-run.py Executable file
View File

@ -0,0 +1,26 @@
#!/usr/bin/env python3
import sys
from testrunner import run
def testfunc(child):
child.expect_exact('counter mode')
child.expect_exact('THREAD 1 start')
child.expect_exact('THREAD 2 start')
child.expect_exact('THREAD 3 start')
child.expect_exact('thread synced')
child.expect_exact('THREAD 3 woke main thread')
child.expect_exact('mask mode')
child.expect_exact('THREAD 1 start')
child.expect_exact('THREAD 2 start')
child.expect_exact('THREAD 3 start')
child.expect_exact('thread synced')
child.expect_exact('THREAD 3 woke main thread')
child.expect_exact('SUCCESS')
if __name__ == "__main__":
sys.exit(run(testfunc))