/* Unit tests for queue.

*

* Author: Steffen Vogel <[email protected]>

* SPDX-FileCopyrightText: 2014-2023 Institute for Automation of Complex Power Systems, RWTH Aachen University

* SPDX-License-Identifier: Apache-2.0

*/

#include <cstdint>

#include <cstdlib>

#include <ctime>

#include <criterion/criterion.h>

#include <criterion/parameterized.h>

#include <pthread.h>

#include <stdio.h>

#include <unistd.h>

#include <villas/log.hpp>

#include <villas/node/memory.hpp>

#include <villas/queue.h>

#include <villas/tsc.hpp>

#include <villas/utils.hpp>

using namespace villas;

using namespace villas::node;

extern void init_memory();

#define SIZE (1 << 10)

static struct CQueue q;

#if defined(_POSIX_BARRIERS) && _POSIX_BARRIERS > 0

static pthread_barrier_t barrier;

#endif

struct param {

int iter_count;

int queue_size;

int thread_count;

bool many;

int batch_size;

struct memory::Type *mt;

volatile int start;

struct CQueue queue;

};

/* Get thread id as integer

* In contrast to pthread_t which is an opaque type */

#ifdef __linux__

#include <sys/syscall.h>

#endif

uint64_t thread_get_id() {

#ifdef __MACH__

uint64_t id;

pthread_threadid_np(pthread_self(), &id);

return id;

#elif defined(SYS_gettid)

return (int)syscall(SYS_gettid);

#endif

return -1;

}

// Sleep, do nothing (Architecture-Specific)

__attribute__((always_inline)) static inline void nop() {

#if defined(__aarch64__)

__asm__ volatile("yield"); // ARM equivalent of rep nop.

#else

__asm__ volatile("rep nop"); // x86 and most other architectures use rep nop.

#endif

}

static void *producer(void *ctx) {

int ret;

struct param *p = (struct param *)ctx;

srand((unsigned)time(0) + thread_get_id());

size_t nops = rand() % 1000;

// Wait for global start signal

while (p->start == 0)

sched_yield();

// Wait for a random time

for (size_t i = 0; i != nops; i += 1)

nop();

// Enqueue

for (intptr_t count = 0; count < p->iter_count; count++) {

do {

ret = queue_push(&p->queue, (void *)count);

sched_yield();

} while (ret != 1);

}

return nullptr;

}

static void *consumer(void *ctx) {

int ret;

struct param *p = (struct param *)ctx;

srand((unsigned)time(0) + thread_get_id());

size_t nops = rand() % 1000;

// Wait for global start signal

while (p->start == 0)

sched_yield();

// Wait for a random time

for (size_t i = 0; i != nops; i += 1)

nop();

// Dequeue

for (intptr_t count = 0; count < p->iter_count; count++) {

intptr_t ptr;

do {

ret = queue_pull(&p->queue, (void **)&ptr);

} while (ret != 1);

//logger->info("consumer: {}", count);

//cr_assert_eq((intptr_t) ptr, count);

}

return nullptr;

}

#if defined(_POSIX_BARRIERS) && _POSIX_BARRIERS > 0

void *producer_consumer(void *ctx) {

struct param *p = (struct param *)ctx;

srand((unsigned)time(0) + thread_get_id());

size_t nops = rand() % 1000;

// Wait for global start signal

while (p->start == 0)

sched_yield();

// Wait for a random time

for (size_t i = 0; i != nops; i += 1)

nop();

for (int iter = 0; iter < p->iter_count; ++iter) {

pthread_barrier_wait(&barrier);

for (intptr_t i = 0; i < p->batch_size; i++) {

void *ptr = (void *)(iter * p->batch_size + i);

while (!queue_push(&p->queue, ptr))

sched_yield(); // queue full, let other threads proceed

}

for (intptr_t i = 0; i < p->batch_size; i++) {

void *ptr;

while (!queue_pull(&p->queue, &ptr))

sched_yield(); // queue empty, let other threads proceed

}

}

return 0;

}

void *producer_consumer_many(void *ctx) {

struct param *p = (struct param *)ctx;

srand((unsigned)time(0) + thread_get_id());

size_t nops = rand() % 1000;

// Wait for global start signal

while (p->start == 0)

sched_yield();

// Wait for a random time

for (size_t i = 0; i != nops; i += 1)

nop();

void *ptrs[p->batch_size];

for (int iter = 0; iter < p->iter_count; ++iter) {

for (intptr_t i = 0; i < p->batch_size; i++)

ptrs[i] = (void *)(iter * p->batch_size + i);

pthread_barrier_wait(&barrier);

int pushed = 0;

do {

pushed +=

queue_push_many(&p->queue, &ptrs[pushed], p->batch_size - pushed);

if (pushed != p->batch_size)

sched_yield(); // queue full, let other threads proceed

} while (pushed < p->batch_size);

int pulled = 0;

do {

pulled +=

queue_pull_many(&p->queue, &ptrs[pulled], p->batch_size - pulled);

if (pulled != p->batch_size)

sched_yield(); // queue empty, let other threads proceed

} while (pulled < p->batch_size);

}

return 0;

}

#endif // _POSIX_BARRIERS

// cppcheck-suppress unknownMacro

Test(queue, single_threaded, .init = init_memory) {

int ret;

struct param p;

p.iter_count = 1 << 8;

p.queue_size = 1 << 10;

p.start = 1; // we start immeadiatly

ret = queue_init(&p.queue, p.queue_size, &memory::heap);

cr_assert_eq(ret, 0, "Failed to create queue");

producer(&p);

consumer(&p);

cr_assert_eq(queue_available(&q), 0);

ret = queue_destroy(&p.queue);

cr_assert_eq(ret, 0, "Failed to create queue");

}

#if defined(_POSIX_BARRIERS) && _POSIX_BARRIERS > 0

// cppcheck-suppress unknownMacro

ParameterizedTestParameters(queue, multi_threaded) {

static struct param params[] = {{.iter_count = 1 << 12,

.queue_size = 1 << 9,

.thread_count = 32,

.many = true,

.batch_size = 10,

.mt = &memory::heap},

{.iter_count = 1 << 8,

.queue_size = 1 << 9,

.thread_count = 4,

.many = true,

.batch_size = 100,

.mt = &memory::heap},

{.iter_count = 1 << 16,

.queue_size = 1 << 14,

.thread_count = 16,

.many = true,

.batch_size = 100,

.mt = &memory::heap},

{.iter_count = 1 << 8,

.queue_size = 1 << 9,

.thread_count = 4,

.many = true,

.batch_size = 10,

.mt = &memory::heap},

{.iter_count = 1 << 16,

.queue_size = 1 << 9,

.thread_count = 16,

.many = false,

.batch_size = 10,

.mt = &memory::mmap_hugetlb}};

return cr_make_param_array(struct param, params, ARRAY_LEN(params));

}

// cppcheck-suppress unknownMacro

ParameterizedTest(struct param *p, queue, multi_threaded, .timeout = 20,

.init = init_memory) {

int ret, cycpop;

struct Tsc tsc;

Logger logger = Log::get("test:queue:multi_threaded");

if (!utils::isPrivileged() && p->mt == &memory::mmap_hugetlb)

cr_skip_test("Skipping memory_mmap_hugetlb tests allocatpr because we are "

"running in an unprivileged environment.");

pthread_t threads[p->thread_count];

p->start = 0;

ret = queue_init(&p->queue, p->queue_size, p->mt);

cr_assert_eq(ret, 0, "Failed to create queue");

uint64_t start_tsc_time, end_tsc_time;

pthread_barrier_init(&barrier, nullptr, p->thread_count);

for (int i = 0; i < p->thread_count; ++i)

pthread_create(&threads[i], nullptr,

p->many ? producer_consumer_many : producer_consumer, p);

sleep(0.2);

ret = tsc_init(&tsc);

cr_assert(!ret);

start_tsc_time = tsc_now(&tsc);

p->start = 1;

for (int i = 0; i < p->thread_count; ++i)

pthread_join(threads[i], nullptr);

end_tsc_time = tsc_now(&tsc);

cycpop = (end_tsc_time - start_tsc_time) / p->iter_count;

if (cycpop < 400)

logger->debug("Cycles/op: {}", cycpop);

else

logger->warn(

"Cycles/op are very high ({}). Are you running on a hypervisor?",

cycpop);

ret = queue_available(&q);

cr_assert_eq(ret, 0);

ret = queue_destroy(&p->queue);

cr_assert_eq(ret, 0, "Failed to destroy queue");

ret = pthread_barrier_destroy(&barrier);

cr_assert_eq(ret, 0, "Failed to destroy barrier");

}

#endif // _POSIX_BARRIERS

// cppcheck-suppress unknownMacro

Test(queue, init_destroy, .init = init_memory) {

int ret;

struct CQueue q;

ret = queue_init(&q, 1024, &memory::heap);

cr_assert_eq(ret, 0); // Should succeed

ret = queue_destroy(&q);

cr_assert_eq(ret, 0); // Should succeed

}