/*
* Copyright (C) NGINX, Inc.
*/
#include <nxt_main.h>
#include <math.h>
#include <inttypes.h>
#ifndef NXT_NCQ_TEST
#define NXT_NCQ_TEST 1
#endif
#define NXT_QTEST_USE_THREAD 0
#if NXT_NCQ_TEST
#include <nxt_nncq.h>
#else
#include <nxt_nvbcq.h>
#endif
#define MAX_ITER 20
#define STAT_ITER 5
#define MIN_COV 0.02
extern char **environ;
static uintptr_t nops = 10000000;
static uintptr_t nprocs_enq = 0;
static uintptr_t nprocs_deq = 0;
static uintptr_t nprocs_wenq = 0;
static uintptr_t nprocs_wdeq = 0;
static uintptr_t nprocs_enq_deq = 0;
static uintptr_t nprocs_cas = 0;
static uintptr_t nprocs_faa = 0;
static uintptr_t nprocs = 1;
static size_t
elapsed_time(size_t us)
{
struct timeval t;
gettimeofday(&t, NULL);
return t.tv_sec * 1000000 + t.tv_usec - us;
}
static double
mean(const double *times, int n)
{
int i;
double sum;
sum = 0;
for (i = 0; i < n; i++) {
sum += times[i];
}
return sum / n;
}
static double
cov(const double *times, double mean, int n)
{
int i;
double variance;
variance = 0;
for (i = 0; i < n; i++) {
variance += (times[i] - mean) * (times[i] - mean);
}
variance /= n;
return sqrt(variance) / mean;
}
typedef struct {
#if NXT_NCQ_TEST
nxt_nncq_t free_queue;
nxt_nncq_t active_queue;
#else
nxt_nvbcq_t free_queue;
nxt_nvbcq_t active_queue;
#endif
uint32_t counter;
} nxt_cq_t;
static nxt_cq_t *pgq;
#if NXT_NCQ_TEST
#define nxt_cq_enqueue nxt_nncq_enqueue
#define nxt_cq_dequeue nxt_nncq_dequeue
#define nxt_cq_empty nxt_nncq_empty
#define nxt_cq_init nxt_nncq_init
#define NXT_CQ_SIZE NXT_NNCQ_SIZE
#else
#define nxt_cq_enqueue nxt_nvbcq_enqueue
#define nxt_cq_dequeue nxt_nvbcq_dequeue
#define nxt_cq_empty nxt_nvbcq_empty
#define nxt_cq_init nxt_nvbcq_init
#define NXT_CQ_SIZE NXT_NVBCQ_SIZE
#endif
typedef struct {
int id;
uint64_t enq;
uint64_t deq;
uint64_t wait_enq;
uint64_t wait_deq;
uint64_t own_res;
uint64_t cas;
uint64_t faa;
#if NXT_QTEST_USE_THREAD
nxt_thread_handle_t handle;
#else
nxt_pid_t pid;
int status;
#endif
} nxt_worker_info_t;
static void
cas_worker(void *p)
{
nxt_cq_t *q;
uint32_t c;
uintptr_t i;
nxt_worker_info_t *wi;
q = pgq;
wi = p;
for (i = 0; i < nops / nprocs_cas; i++) {
c = q->counter;
if (nxt_atomic_cmp_set(&q->counter, c, c + 1)) {
++wi->cas;
}
}
}
static void
faa_worker(void *p)
{
nxt_cq_t *q;
uintptr_t i;
nxt_worker_info_t *wi;
q = pgq;
wi = p;
for (i = 0; i < nops / nprocs_faa; i++) {
nxt_atomic_fetch_add(&q->counter, 1);
wi->faa++;
}
}
static void
enq_deq_worker(void *p)
{
nxt_cq_t *q;
uintptr_t i, v;
nxt_worker_info_t *wi;
q = pgq;
wi = p;
for (i = 0; i < nops / nprocs_enq_deq; i++) {
v = nxt_cq_dequeue(&q->free_queue);
if (v != nxt_cq_empty(&q->free_queue)) {
nxt_cq_enqueue(&q->active_queue, wi->id);
wi->enq++;
}
v = nxt_cq_dequeue(&q->active_queue);
if (v != nxt_cq_empty(&q->active_queue)) {
nxt_cq_enqueue(&q->free_queue, v);
wi->deq++;
if ((int) v == wi->id) {
wi->own_res++;
}
}
}
}
static void
enq_worker(void *p)
{
nxt_cq_t *q;
uintptr_t i, v;
nxt_worker_info_t *wi;
q = pgq;
wi = p;
for (i = 0; i < nops / nprocs_enq; i++) {
v = nxt_cq_dequeue(&q->free_queue);
if (v != nxt_cq_empty(&q->free_queue)) {
nxt_cq_enqueue(&q->active_queue, v);
wi->enq++;
}
}
}
static void
deq_worker(void *p)
{
nxt_cq_t *q;
uintptr_t i, v;
nxt_worker_info_t *wi;
q = pgq;
wi = p;
for (i = 0; i < nops / nprocs_deq; i++) {
v = nxt_cq_dequeue(&q->active_queue);
if (v != nxt_cq_empty(&q->active_queue)) {
nxt_cq_enqueue(&q->free_queue, v);
++wi->deq;
}
}
}
static void
wenq_worker(void *p)
{
nxt_cq_t *q;
uintptr_t i, v;
nxt_worker_info_t *wi;
q = pgq;
wi = p;
for (i = 0; i < nops / nprocs_wenq; i++) {
do {
wi->wait_enq++;
v = nxt_cq_dequeue(&q->free_queue);
} while (v == nxt_cq_empty(&q->free_queue));
nxt_cq_enqueue(&q->active_queue, v);
wi->enq++;
wi->wait_enq--;
}
}
static void
wdeq_worker(void *p)
{
nxt_cq_t *q;
uintptr_t i, v;
nxt_worker_info_t *wi;
q = pgq;
wi = p;
for (i = 0; i < nops / nprocs_wdeq; i++) {
do {
wi->wait_deq++;
v = nxt_cq_dequeue(&q->active_queue);
} while (v == nxt_cq_empty(&q->active_queue));
nxt_cq_enqueue(&q->free_queue, v);
wi->deq++;
wi->wait_deq--;
}
}
static nxt_int_t
worker_create(nxt_worker_info_t *wi, int id, nxt_thread_start_t start)
{
wi->id = id;
#if NXT_QTEST_USE_THREAD
nxt_thread_link_t *link;
link = nxt_zalloc(sizeof(nxt_thread_link_t));
link->start = start;
link->work.data = wi;
return nxt_thread_create(&wi->handle, link);
#else
pid_t pid = fork();
if (pid == 0) {
start(wi);
exit(0);
} else {
wi->pid = pid;
}
return NXT_OK;
#endif
}
static void
worker_wait(nxt_worker_info_t *wi)
{
#if NXT_QTEST_USE_THREAD
pthread_join(wi->handle, NULL);
#else
waitpid(wi->pid, &wi->status, 0);
#endif
}
int nxt_cdecl
main(int argc, char **argv)
{
int i, k, id, verbose, objective, rk;
char *a;
size_t start, elapsed;
double *stats, m, c;
uint64_t total_ops;
uintptr_t j;
nxt_task_t task;
nxt_thread_t *thr;
nxt_worker_info_t *wi;
double times[MAX_ITER], mopsec[MAX_ITER];
verbose = 0;
objective = 0;
for (i = 1; i < argc; i++) {
a = argv[i];
if (strcmp(a, "-v") == 0) {
verbose++;
continue;
}
if (strcmp(a, "-n") == 0 && (i + 1) < argc) {
nops = atoi(argv[++i]);
continue;
}
if (strcmp(a, "--enq") == 0 && (i + 1) < argc) {
nprocs_enq = atoi(argv[++i]);
continue;
}
if (strcmp(a, "--deq") == 0 && (i + 1) < argc) {
nprocs_deq = atoi(argv[++i]);
continue;
}
if (strcmp(a, "--wenq") == 0 && (i + 1) < argc) {
nprocs_wenq = atoi(argv[++i]);
continue;
}
if (strcmp(a, "--wdeq") == 0 && (i + 1) < argc) {
nprocs_wdeq = atoi(argv[++i]);
continue;
}
if (strcmp(a, "--ed") == 0 && (i + 1) < argc) {
nprocs_enq_deq = atoi(argv[++i]);
continue;
}
if (strcmp(a, "--cas") == 0 && (i + 1) < argc) {
nprocs_cas = atoi(argv[++i]);
continue;
}
if (strcmp(a, "--faa") == 0 && (i + 1) < argc) {
nprocs_faa = atoi(argv[++i]);
continue;
}
if (strcmp(a, "--obj") == 0 && (i + 1) < argc) {
objective = atoi(argv[++i]);
continue;
}
printf("unknown option %s", a);
return 1;
}
if (nxt_lib_start("ncq_test", argv, &environ) != NXT_OK) {
return 1;
}
nprocs = nprocs_enq + nprocs_deq + nprocs_wenq + nprocs_wdeq
+ nprocs_enq_deq + nprocs_cas + nprocs_faa;
if (nprocs == 0) {
return 0;
}
nxt_main_log.level = NXT_LOG_INFO;
task.log = &nxt_main_log;
thr = nxt_thread();
thr->task = &task;
pgq = mmap(NULL, sizeof(nxt_cq_t), PROT_READ | PROT_WRITE,
MAP_ANON | MAP_SHARED, -1, 0);
if (pgq == MAP_FAILED) {
return 2;
}
nxt_cq_init(&pgq->free_queue);
nxt_cq_init(&pgq->active_queue);
for(i = 0; i < NXT_CQ_SIZE; i++) {
nxt_cq_enqueue(&pgq->free_queue, i);
}
if (verbose >= 1) {
printf("number of workers: %d\n", (int) nprocs);
printf("number of ops: %d\n", (int) nops);
}
wi = mmap(NULL, nprocs * sizeof(nxt_worker_info_t), PROT_READ | PROT_WRITE,
MAP_ANON | MAP_SHARED, -1, 0);
if (wi == MAP_FAILED) {
return 3;
}
for (k = 0; k < MAX_ITER; k++) {
nxt_memzero(wi, nprocs * sizeof(nxt_worker_info_t));
nxt_cq_init(&pgq->free_queue);
nxt_cq_init(&pgq->active_queue);
for(i = 0; i < NXT_CQ_SIZE; i++) {
nxt_cq_enqueue(&pgq->free_queue, i);
}
start = elapsed_time(0);
id = 0;
for (j = 0; j < nprocs_enq; j++, id++) {
worker_create(wi + id, id, enq_worker);
}
for (j = 0; j < nprocs_deq; j++, id++) {
worker_create(wi + id, id, deq_worker);
}
for (j = 0; j < nprocs_wenq; j++, id++) {
worker_create(wi + id, id, wenq_worker);
}
for (j = 0; j < nprocs_wdeq; j++, id++) {
worker_create(wi + id, id, wdeq_worker);
}
for (j = 0; j < nprocs_enq_deq; j++, id++) {
worker_create(wi + id, id, enq_deq_worker);
}
for (j = 0; j < nprocs_cas; j++, id++) {
worker_create(wi + id, id, cas_worker);
}
for (j = 0; j < nprocs_faa; j++, id++) {
worker_create(wi + id, id, faa_worker);
}
for (j = 0; j < nprocs; j++) {
worker_wait(wi + j);
}
elapsed = elapsed_time(start);
for (j = 1; j < nprocs; j++) {
wi[0].enq += wi[j].enq;
wi[0].deq += wi[j].deq;
wi[0].wait_enq += wi[j].wait_enq;
wi[0].wait_deq += wi[j].wait_deq;
wi[0].own_res += wi[j].own_res;
wi[0].cas += wi[j].cas;
wi[0].faa += wi[j].faa;
}
total_ops = wi[0].enq + wi[0].deq + wi[0].cas + wi[0].faa;
if (total_ops == 0) {
total_ops = nops;
}
times[k] = elapsed / 1000.0;
mopsec[k] = (double) total_ops / elapsed;
if (verbose >= 2) {
printf("enq %10"PRIu64"\n", wi[0].enq);
printf("deq %10"PRIu64"\n", wi[0].deq);
printf("wait_enq %10"PRIu64"\n", wi[0].wait_enq);
printf("wait_deq %10"PRIu64"\n", wi[0].wait_deq);
printf("own_res %10"PRIu64"\n", wi[0].own_res);
printf("cas %10"PRIu64"\n", wi[0].cas);
printf("faa %10"PRIu64"\n", wi[0].faa);
printf("total ops %10"PRIu64"\n", total_ops);
printf("Mops/sec %13.2f\n", mopsec[k]);
printf("elapsed %10d us\n", (int) elapsed);
printf("per op %10d ns\n", (int) ((1000 * elapsed) / total_ops));
}
if (k >= STAT_ITER) {
stats = (objective == 0) ? times : mopsec;
m = mean(stats + k - STAT_ITER, STAT_ITER);
c = cov(stats + k - STAT_ITER, m, STAT_ITER);
if (verbose >= 1) {
if (objective == 0) {
printf(" #%02d elapsed time: %.2f ms; Mops/sec %.2f; "
"mean time %.2f ms; cov %.4f\n",
(int) k + 1, times[k], mopsec[k], m, c);
} else {
printf(" #%02d elapsed time: %.2f ms; Mops/sec %.2f; "
"mean Mop/sec %.2f; cov %.4f\n",
(int) k + 1, times[k], mopsec[k], m, c);
}
}
if (c < MIN_COV) {
rk = k - STAT_ITER;
for (i = rk + 1; i <= k; i++) {
if (fabs(stats[i] - m) < fabs(stats[rk] - m)) {
rk = i;
}
}
printf("#%d %.2f ms; %.2f\n", rk, times[rk], mopsec[rk]);
return 0;
}
} else {
if (verbose >= 1) {
printf(" #%02d elapsed time: %.2f ms; Mops/sec %.2f\n",
(int) k + 1, times[k], mopsec[k]);
}
}
}
return 0;
}
