1
0
Fork 0
mirror of https://git.rwth-aachen.de/acs/public/villas/node/ synced 2025-03-16 00:00:02 +01:00
VILLASnode/tests/unit/queue_signalled.cpp
Steffen Vogel 0735eb0f89 Make project REUSE compliant
And various other cleanups and harmonizations

Signed-off-by: Steffen Vogel <steffen.vogel@opal-rt.com>
2023-09-07 11:16:04 +02:00

155 lines
3.5 KiB
C++

/* Unit tests for queue_signalled
*
* Author: Steffen Vogel <post@steffenvogel.de>
* SPDX-FileCopyrightText: 2014-2023 Institute for Automation of Complex Power Systems, RWTH Aachen University
* SPDX-License-Identifier: Apache-2.0
*/
#include <criterion/criterion.h>
#include <criterion/parameterized.h>
#include <pthread.h>
#include <poll.h>
#include <villas/utils.hpp>
#include <villas/node/memory.hpp>
#include <villas/queue_signalled.h>
using namespace villas::node;
extern
void init_memory();
#define NUM_ELEM 1000
struct param {
enum QueueSignalledMode mode;
int flags;
bool polled;
};
static
void * producer(void * ctx)
{
int ret;
struct CQueueSignalled *q = (struct CQueueSignalled *) ctx;
for (intptr_t i = 0; i < NUM_ELEM; i++) {
ret = queue_signalled_push(q, (void *) i);
if (ret != 1)
return (void *) 1; // Indicates an error to the parent thread
usleep(0.1e-3 * 1e6); // 1 ms
}
return nullptr;
}
static
void * consumer(void * ctx)
{
int ret;
struct CQueueSignalled *q = (struct CQueueSignalled *) ctx;
void *data[NUM_ELEM];
for (intptr_t i = 0; i < NUM_ELEM;) {
ret = queue_signalled_pull_many(q, data, ARRAY_LEN(data));
if (ret <= 0)
return (void *) 1; // Indicates an error to the parent thread
for (intptr_t j = 0; j < ret; j++, i++) {
if ((intptr_t) data[j] != i)
return (void *) 2; // Indicates an error to the parent thread
}
}
return nullptr;
}
void * polled_consumer(void *ctx)
{
int ret, fd;
struct CQueueSignalled *q = (struct CQueueSignalled *) ctx;
fd = queue_signalled_fd(q);
cr_assert_geq(fd, 0);
struct pollfd pfd = {
.fd = fd,
.events = POLLIN
};
for (intptr_t i = 0; i < NUM_ELEM; i++) {
again: ret = poll(&pfd, 1, -1);
if (ret < 0)
return (void *) 3;
else if (ret == 0)
goto again;
void *p;
ret = queue_signalled_pull(q, &p);
if (ret != 1)
return (void *) 1; // Indicates an error to the parent thread
if ((intptr_t) p != i)
return (void *) 2; // Indicates an error to the parent thread
}
return nullptr;
}
ParameterizedTestParameters(queue_signalled, simple)
{
static
struct param params[] = {
{ QueueSignalledMode::AUTO, 0, false },
{ QueueSignalledMode::PTHREAD, 0, false },
{ QueueSignalledMode::PTHREAD, 0, false },
{ QueueSignalledMode::PTHREAD, (int) QueueSignalledFlags::PROCESS_SHARED, false },
{ QueueSignalledMode::POLLING, 0, false },
#if defined(__linux__) && defined(HAS_EVENTFD)
{ QueueSignalledMode::EVENTFD, 0, false },
{ QueueSignalledMode::EVENTFD, 0, true }
#endif
};
return cr_make_param_array(struct param, params, ARRAY_LEN(params));
}
// cppcheck-suppress unknownMacro
ParameterizedTest(struct param *param, queue_signalled, simple, .timeout = 5, .init = init_memory)
{
int ret;
void *r1, *r2;
struct CQueueSignalled q;
pthread_t t1, t2;
ret = queue_signalled_init(&q, LOG2_CEIL(NUM_ELEM), &memory::heap, param->mode, param->flags);
cr_assert_eq(ret, 0, "Failed to initialize queue: mode=%d, flags=%#x, ret=%d", (int) param->mode, param->flags, ret);
ret = pthread_create(&t1, nullptr, producer, &q);
cr_assert_eq(ret, 0);
ret = pthread_create(&t2, nullptr, param->polled ? polled_consumer : consumer, &q);
cr_assert_eq(ret, 0);
ret = pthread_join(t1, &r1);
cr_assert_eq(ret, 0);
ret = pthread_join(t2, &r2);
cr_assert_eq(ret, 0);
cr_assert_null(r1, "Producer failed: %p", r1);
cr_assert_null(r2, "Consumer failed: %p", r2);
ret = queue_signalled_available(&q);
cr_assert_eq(ret, 0);
ret = queue_signalled_close(&q);
cr_assert_eq(ret, 0);
ret = queue_signalled_destroy(&q);
cr_assert_eq(ret, 0);
}