/*

* This application is Copyright 2012 Red Hat, Inc.

* Doug Ledford <dledford@redhat.com>

*

* mq_perf_tests is free software: you can redistribute it and/or modify

* it under the terms of the GNU General Public License as published by

* the Free Software Foundation, version 3.

*

* mq_perf_tests is distributed in the hope that it will be useful,

* but WITHOUT ANY WARRANTY; without even the implied warranty of

* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the

* GNU General Public License for more details.

*

* For the full text of the license, see <http://www.gnu.org/licenses/>.

*

* mq_perf_tests.c

* Tests various types of message queue workloads, concentrating on those

* situations that invole large message sizes, large message queue depths,

* or both, and reports back useful metrics about kernel message queue

* performance.

*

*/

#define _GNU_SOURCE

#include <stdio.h>

#include <stdlib.h>

#include <unistd.h>

#include <fcntl.h>

#include <string.h>

#include <limits.h>

#include <errno.h>

#include <signal.h>

#include <pthread.h>

#include <sched.h>

#include <sys/types.h>

#include <sys/time.h>

#include <sys/resource.h>

#include <sys/stat.h>

#include <sys/param.h>

#include <mqueue.h>

#include <popt.h>

#include <error.h>

#include "kselftest.h"

static char *usage =

"Usage:\n"

" %s [-c #[,#..] -f] path\n"

"\n"

" -c # Skip most tests and go straight to a high queue depth test\n"

" and then run that test continuously (useful for running at\n"

" the same time as some other workload to see how much the\n"

" cache thrashing caused by adding messages to a very deep\n"

" queue impacts the performance of other programs). The number\n"

" indicates which CPU core we should bind the process to during\n"

" the run. If you have more than one physical CPU, then you\n"

" will need one copy per physical CPU package, and you should\n"

" specify the CPU cores to pin ourself to via a comma separated\n"

" list of CPU values.\n"

" -f Only usable with continuous mode. Pin ourself to the CPUs\n"

" as requested, then instead of looping doing a high mq\n"

" workload, just busy loop. This will allow us to lock up a\n"

" single CPU just like we normally would, but without actually\n"

" thrashing the CPU cache. This is to make it easier to get\n"

" comparable numbers from some other workload running on the\n"

" other CPUs. One set of numbers with # CPUs locked up running\n"

" an mq workload, and another set of numbers with those same\n"

" CPUs locked away from the test workload, but not doing\n"

" anything to trash the cache like the mq workload might.\n"

" path Path name of the message queue to create\n"

"\n"

" Note: this program must be run as root in order to enable all tests\n"

"\n";

char *MAX_MSGS = "/proc/sys/fs/mqueue/msg_max";

char *MAX_MSGSIZE = "/proc/sys/fs/mqueue/msgsize_max";

#define MAX_CPUS 64

char *cpu_option_string;

int cpus_to_pin[MAX_CPUS];

int num_cpus_to_pin;

pthread_t cpu_threads[MAX_CPUS];

pthread_t main_thread;

cpu_set_t *cpu_set;

int cpu_set_size;

int cpus_online;

#define MSG_SIZE 16

#define TEST1_LOOPS 10000000

#define TEST2_LOOPS 100000

int continuous_mode;

int continuous_mode_fake;

struct rlimit saved_limits, cur_limits;

int saved_max_msgs, saved_max_msgsize;

int cur_max_msgs, cur_max_msgsize;

FILE *max_msgs, *max_msgsize;

int cur_nice;

char *queue_path = "/mq_perf_tests";

mqd_t queue = -1;

struct mq_attr result;

int mq_prio_max;

const struct poptOption options[] = {

{

.longName = "continuous",

.shortName = 'c',

.argInfo = POPT_ARG_STRING,

.arg = &cpu_option_string,

.val = 'c',

.descrip = "Run continuous tests at a high queue depth in "

"order to test the effects of cache thrashing on "

"other tasks on the system. This test is intended "

"to be run on one core of each physical CPU while "

"some other CPU intensive task is run on all the other "

"cores of that same physical CPU and the other task "

"is timed. It is assumed that the process of adding "

"messages to the message queue in a tight loop will "

"impact that other task to some degree. Once the "

"tests are performed in this way, you should then "

"re-run the tests using fake mode in order to check "

"the difference in time required to perform the CPU "

"intensive task",

.argDescrip = "cpu[,cpu]",

},

{

.longName = "fake",

.shortName = 'f',

.argInfo = POPT_ARG_NONE,

.arg = &continuous_mode_fake,

.val = 0,

.descrip = "Tie up the CPUs that we would normally tie up in"

"continuous mode, but don't actually do any mq stuff, "

"just keep the CPU busy so it can't be used to process "

"system level tasks as this would free up resources on "

"the other CPU cores and skew the comparison between "

"the no-mqueue work and mqueue work tests",

.argDescrip = NULL,

},

{

.longName = "path",

.shortName = 'p',

.argInfo = POPT_ARG_STRING | POPT_ARGFLAG_SHOW_DEFAULT,

.arg = &queue_path,

.val = 'p',

.descrip = "The name of the path to use in the mqueue "

"filesystem for our tests",

.argDescrip = "pathname",

},

POPT_AUTOHELP

POPT_TABLEEND

};

static inline void __set(FILE *stream, int value, char *err_msg);

void shutdown(int exit_val, char *err_cause, int line_no);

void sig_action_SIGUSR1(int signum, siginfo_t *info, void *context);

void sig_action(int signum, siginfo_t *info, void *context);

static inline int get(FILE *stream);

static inline void set(FILE *stream, int value);

static inline int try_set(FILE *stream, int value);

static inline void getr(int type, struct rlimit *rlim);

static inline void setr(int type, struct rlimit *rlim);

static inline void open_queue(struct mq_attr *attr);

void increase_limits(void);

static inline void __set(FILE *stream, int value, char *err_msg)

{

rewind(stream);

if (fprintf(stream, "%d", value) < 0)

perror(err_msg);

}

void shutdown(int exit_val, char *err_cause, int line_no)

{

static int in_shutdown = 0;

int errno_at_shutdown = errno;

int i;

/* In case we get called by multiple threads or from an sighandler */

if (in_shutdown++)

return;

/* Free the cpu_set allocated using CPU_ALLOC in main function */

CPU_FREE(cpu_set);

for (i = 0; i < num_cpus_to_pin; i++)

if (cpu_threads[i]) {

pthread_kill(cpu_threads[i], SIGUSR1);

pthread_join(cpu_threads[i], NULL);

}

if (queue != -1)

if (mq_close(queue))

perror("mq_close() during shutdown");

if (queue_path)

/*

* Be silent if this fails, if we cleaned up already it's

* expected to fail

*/

mq_unlink(queue_path);

if (saved_max_msgs)

__set(max_msgs, saved_max_msgs,

"failed to restore saved_max_msgs");

if (saved_max_msgsize)

__set(max_msgsize, saved_max_msgsize,

"failed to restore saved_max_msgsize");

if (exit_val)

error(exit_val, errno_at_shutdown, "%s at %d",

err_cause, line_no);

exit(0);

}

void sig_action_SIGUSR1(int signum, siginfo_t *info, void *context)

{

if (pthread_self() != main_thread)

pthread_exit(0);

else {

fprintf(stderr, "Caught signal %d in SIGUSR1 handler, "

"exiting\n", signum);

shutdown(0, "", 0);

fprintf(stderr, "\n\nReturned from shutdown?!?!\n\n");

exit(0);

}

}

void sig_action(int signum, siginfo_t *info, void *context)

{

if (pthread_self() != main_thread)

pthread_kill(main_thread, signum);

else {

fprintf(stderr, "Caught signal %d, exiting\n", signum);

shutdown(0, "", 0);

fprintf(stderr, "\n\nReturned from shutdown?!?!\n\n");

exit(0);

}

}

static inline int get(FILE *stream)

{

int value;

rewind(stream);

if (fscanf(stream, "%d", &value) != 1)

shutdown(4, "Error reading /proc entry", __LINE__);

return value;

}

static inline void set(FILE *stream, int value)

{

int new_value;

rewind(stream);

if (fprintf(stream, "%d", value) < 0)

return shutdown(5, "Failed writing to /proc file", __LINE__);

new_value = get(stream);

if (new_value != value)

return shutdown(5, "We didn't get what we wrote to /proc back",

__LINE__);

}

static inline int try_set(FILE *stream, int value)

{

int new_value;

rewind(stream);

fprintf(stream, "%d", value);

new_value = get(stream);

return new_value == value;

}

static inline void getr(int type, struct rlimit *rlim)

{

if (getrlimit(type, rlim))

shutdown(6, "getrlimit()", __LINE__);

}

static inline void setr(int type, struct rlimit *rlim)

{

if (setrlimit(type, rlim))

shutdown(7, "setrlimit()", __LINE__);

}

/**

* open_queue - open the global queue for testing

* @attr - An attr struct specifying the desired queue traits

* @result - An attr struct that lists the actual traits the queue has

*

* This open is not allowed to fail, failure will result in an orderly

* shutdown of the program. The global queue_path is used to set what

* queue to open, the queue descriptor is saved in the global queue

* variable.

*/

static inline void open_queue(struct mq_attr *attr)

{

int flags = O_RDWR | O_EXCL | O_CREAT | O_NONBLOCK;

int perms = DEFFILEMODE;

queue = mq_open(queue_path, flags, perms, attr);

if (queue == -1)

shutdown(1, "mq_open()", __LINE__);

if (mq_getattr(queue, &result))

shutdown(1, "mq_getattr()", __LINE__);

printf("\n\tQueue %s created:\n", queue_path);

printf("\t\tmq_flags:\t\t\t%s\n", result.mq_flags & O_NONBLOCK ?

"O_NONBLOCK" : "(null)");

printf("\t\tmq_maxmsg:\t\t\t%lu\n", result.mq_maxmsg);

printf("\t\tmq_msgsize:\t\t\t%lu\n", result.mq_msgsize);

printf("\t\tmq_curmsgs:\t\t\t%lu\n", result.mq_curmsgs);

}

void *fake_cont_thread(void *arg)

{

int i;

for (i = 0; i < num_cpus_to_pin; i++)

if (cpu_threads[i] == pthread_self())

break;

printf("\tStarted fake continuous mode thread %d on CPU %d\n", i,

cpus_to_pin[i]);

while (1)

;

}

void *cont_thread(void *arg)

{

char buff[MSG_SIZE];

int i;

unsigned int priority;

for (i = 0; i < num_cpus_to_pin; i++)

if (cpu_threads[i] == pthread_self())

break;

printf("\tStarted continuous mode thread %d on CPU %d\n", i,

cpus_to_pin[i]);

while (1) {

while (mq_send(queue, buff, sizeof(buff), 0) == 0)

;

mq_receive(queue, buff, sizeof(buff), &priority);

}

}

#define drain_queue() \

while (mq_receive(queue, buff, MSG_SIZE, &prio_in) == MSG_SIZE)

#define do_untimed_send() \

do { \

if (mq_send(queue, buff, MSG_SIZE, prio_out)) \

shutdown(3, "Test send failure", __LINE__); \

} while (0)

#define do_send_recv() \

do { \

clock_gettime(clock, &start); \

if (mq_send(queue, buff, MSG_SIZE, prio_out)) \

shutdown(3, "Test send failure", __LINE__); \

clock_gettime(clock, &middle); \

if (mq_receive(queue, buff, MSG_SIZE, &prio_in) != MSG_SIZE) \

shutdown(3, "Test receive failure", __LINE__); \

clock_gettime(clock, &end); \

nsec = ((middle.tv_sec - start.tv_sec) * 1000000000) + \

(middle.tv_nsec - start.tv_nsec); \

send_total.tv_nsec += nsec; \

if (send_total.tv_nsec >= 1000000000) { \

send_total.tv_sec++; \

send_total.tv_nsec -= 1000000000; \

} \

nsec = ((end.tv_sec - middle.tv_sec) * 1000000000) + \

(end.tv_nsec - middle.tv_nsec); \

recv_total.tv_nsec += nsec; \

if (recv_total.tv_nsec >= 1000000000) { \

recv_total.tv_sec++; \

recv_total.tv_nsec -= 1000000000; \

} \

} while (0)

struct test {

char *desc;

void (*func)(int *);

};

void const_prio(int *prio)

{

return;

}

void inc_prio(int *prio)

{

if (++*prio == mq_prio_max)

*prio = 0;

}

void dec_prio(int *prio)

{

if (--*prio < 0)

*prio = mq_prio_max - 1;

}

void random_prio(int *prio)

{

*prio = random() % mq_prio_max;

}

struct test test2[] = {

{"\n\tTest #2a: Time send/recv message, queue full, constant prio\n",

const_prio},

{"\n\tTest #2b: Time send/recv message, queue full, increasing prio\n",

inc_prio},

{"\n\tTest #2c: Time send/recv message, queue full, decreasing prio\n",

dec_prio},

{"\n\tTest #2d: Time send/recv message, queue full, random prio\n",

random_prio},

{NULL, NULL}

};

/**

* Tests to perform (all done with MSG_SIZE messages):

*

* 1) Time to add/remove message with 0 messages on queue

* 1a) with constant prio

* 2) Time to add/remove message when queue close to capacity:

* 2a) with constant prio

* 2b) with increasing prio

* 2c) with decreasing prio

* 2d) with random prio

* 3) Test limits of priorities honored (double check _SC_MQ_PRIO_MAX)

*/

void *perf_test_thread(void *arg)

{

char buff[MSG_SIZE];

int prio_out;

unsigned int prio_in;

int i;

clockid_t clock;

pthread_t *t;

struct timespec res, start, middle, end, send_total, recv_total;

unsigned long long nsec;

struct test *cur_test;

t = &cpu_threads[0];

printf("\n\tStarted mqueue performance test thread on CPU %d\n",

cpus_to_pin[0]);

mq_prio_max = sysconf(_SC_MQ_PRIO_MAX);

if (mq_prio_max == -1)

shutdown(2, "sysconf(_SC_MQ_PRIO_MAX)", __LINE__);

if (pthread_getcpuclockid(cpu_threads[0], &clock) != 0)

shutdown(2, "pthread_getcpuclockid", __LINE__);

if (clock_getres(clock, &res))

shutdown(2, "clock_getres()", __LINE__);

printf("\t\tMax priorities:\t\t\t%d\n", mq_prio_max);

printf("\t\tClock resolution:\t\t%lu nsec%s\n", res.tv_nsec,

res.tv_nsec > 1 ? "s" : "");

printf("\n\tTest #1: Time send/recv message, queue empty\n");

printf("\t\t(%d iterations)\n", TEST1_LOOPS);

prio_out = 0;

send_total.tv_sec = 0;

send_total.tv_nsec = 0;

recv_total.tv_sec = 0;

recv_total.tv_nsec = 0;

for (i = 0; i < TEST1_LOOPS; i++)

do_send_recv();

printf("\t\tSend msg:\t\t\t%ld.%lus total time\n",

send_total.tv_sec, send_total.tv_nsec);

nsec = ((unsigned long long)send_total.tv_sec * 1000000000 +

send_total.tv_nsec) / TEST1_LOOPS;

printf("\t\t\t\t\t\t%lld nsec/msg\n", nsec);

printf("\t\tRecv msg:\t\t\t%ld.%lus total time\n",

recv_total.tv_sec, recv_total.tv_nsec);

nsec = ((unsigned long long)recv_total.tv_sec * 1000000000 +

recv_total.tv_nsec) / TEST1_LOOPS;

printf("\t\t\t\t\t\t%lld nsec/msg\n", nsec);

for (cur_test = test2; cur_test->desc != NULL; cur_test++) {

printf("%s:\n", cur_test->desc);

printf("\t\t(%d iterations)\n", TEST2_LOOPS);

prio_out = 0;

send_total.tv_sec = 0;

send_total.tv_nsec = 0;

recv_total.tv_sec = 0;

recv_total.tv_nsec = 0;

printf("\t\tFilling queue...");

fflush(stdout);

clock_gettime(clock, &start);

for (i = 0; i < result.mq_maxmsg - 1; i++) {

do_untimed_send();

cur_test->func(&prio_out);

}

clock_gettime(clock, &end);

nsec = ((unsigned long long)(end.tv_sec - start.tv_sec) *

1000000000) + (end.tv_nsec - start.tv_nsec);

printf("done.\t\t%lld.%llds\n", nsec / 1000000000,

nsec % 1000000000);

printf("\t\tTesting...");

fflush(stdout);

for (i = 0; i < TEST2_LOOPS; i++) {

do_send_recv();

cur_test->func(&prio_out);

}

printf("done.\n");

printf("\t\tSend msg:\t\t\t%ld.%lus total time\n",

send_total.tv_sec, send_total.tv_nsec);

nsec = ((unsigned long long)send_total.tv_sec * 1000000000 +

send_total.tv_nsec) / TEST2_LOOPS;

printf("\t\t\t\t\t\t%lld nsec/msg\n", nsec);

printf("\t\tRecv msg:\t\t\t%ld.%lus total time\n",

recv_total.tv_sec, recv_total.tv_nsec);

nsec = ((unsigned long long)recv_total.tv_sec * 1000000000 +

recv_total.tv_nsec) / TEST2_LOOPS;

printf("\t\t\t\t\t\t%lld nsec/msg\n", nsec);

printf("\t\tDraining queue...");

fflush(stdout);

clock_gettime(clock, &start);

drain_queue();

clock_gettime(clock, &end);

nsec = ((unsigned long long)(end.tv_sec - start.tv_sec) *

1000000000) + (end.tv_nsec - start.tv_nsec);

printf("done.\t\t%lld.%llds\n", nsec / 1000000000,

nsec % 1000000000);

}

return 0;

}

void increase_limits(void)

{

cur_limits.rlim_cur = RLIM_INFINITY;

cur_limits.rlim_max = RLIM_INFINITY;

setr(RLIMIT_MSGQUEUE, &cur_limits);

while (try_set(max_msgs, cur_max_msgs += 10))

;

cur_max_msgs = get(max_msgs);

while (try_set(max_msgsize, cur_max_msgsize += 1024))

;

cur_max_msgsize = get(max_msgsize);

if (setpriority(PRIO_PROCESS, 0, -20) != 0)

shutdown(2, "setpriority()", __LINE__);

cur_nice = -20;

}

int main(int argc, char *argv[])

{

struct mq_attr attr;

char *option, *next_option;

int i, cpu, rc;

struct sigaction sa;

poptContext popt_context;

void *retval;

main_thread = pthread_self();

num_cpus_to_pin = 0;

if (sysconf(_SC_NPROCESSORS_ONLN) == -1) {

perror("sysconf(_SC_NPROCESSORS_ONLN)");

exit(1);

}

if (getuid() != 0)

ksft_exit_skip("Not running as root, but almost all tests "

"require root in order to modify\nsystem settings. "

"Exiting.\n");

cpus_online = MIN(MAX_CPUS, sysconf(_SC_NPROCESSORS_ONLN));

cpu_set = CPU_ALLOC(cpus_online);

if (cpu_set == NULL) {

perror("CPU_ALLOC()");

exit(1);

}

cpu_set_size = CPU_ALLOC_SIZE(cpus_online);

CPU_ZERO_S(cpu_set_size, cpu_set);

popt_context = poptGetContext(NULL, argc, (const char **)argv,

options, 0);

while ((rc = poptGetNextOpt(popt_context)) > 0) {

switch (rc) {

case 'c':

continuous_mode = 1;

option = cpu_option_string;

do {

next_option = strchr(option, ',');

if (next_option)

*next_option = '\0';

cpu = atoi(option);

if (cpu >= cpus_online)

fprintf(stderr, "CPU %d exceeds "

"cpus online, ignoring.\n",

cpu);

else

cpus_to_pin[num_cpus_to_pin++] = cpu;

if (next_option)

option = ++next_option;

} while (next_option && num_cpus_to_pin < MAX_CPUS);

/* Double check that they didn't give us the same CPU

* more than once */

for (cpu = 0; cpu < num_cpus_to_pin; cpu++) {

if (CPU_ISSET_S(cpus_to_pin[cpu], cpu_set_size,

cpu_set)) {

fprintf(stderr, "Any given CPU may "

"only be given once.\n");

goto err_code;

} else

CPU_SET_S(cpus_to_pin[cpu],

cpu_set_size, cpu_set);

}

break;

case 'p':

/*

* Although we can create a msg queue with a

* non-absolute path name, unlink will fail. So,

* if the name doesn't start with a /, add one

* when we save it.

*/

option = queue_path;

if (*option != '/') {

queue_path = malloc(strlen(option) + 2);

if (!queue_path) {

perror("malloc()");

goto err_code;

}

queue_path[0] = '/';

queue_path[1] = 0;

strcat(queue_path, option);

free(option);

}

break;

}

}

if (continuous_mode && num_cpus_to_pin == 0) {

fprintf(stderr, "Must pass at least one CPU to continuous "

"mode.\n");

poptPrintUsage(popt_context, stderr, 0);

goto err_code;

} else if (!continuous_mode) {

num_cpus_to_pin = 1;

cpus_to_pin[0] = cpus_online - 1;

}

max_msgs = fopen(MAX_MSGS, "r+");

max_msgsize = fopen(MAX_MSGSIZE, "r+");

if (!max_msgs)

shutdown(2, "Failed to open msg_max", __LINE__);

if (!max_msgsize)

shutdown(2, "Failed to open msgsize_max", __LINE__);

/* Load up the current system values for everything we can */

getr(RLIMIT_MSGQUEUE, &saved_limits);

cur_limits = saved_limits;

saved_max_msgs = cur_max_msgs = get(max_msgs);

saved_max_msgsize = cur_max_msgsize = get(max_msgsize);

errno = 0;

cur_nice = getpriority(PRIO_PROCESS, 0);

if (errno)

shutdown(2, "getpriority()", __LINE__);

/* Tell the user our initial state */

printf("\nInitial system state:\n");

printf("\tUsing queue path:\t\t\t%s\n", queue_path);

printf("\tRLIMIT_MSGQUEUE(soft):\t\t\t%ld\n",

(long) saved_limits.rlim_cur);

printf("\tRLIMIT_MSGQUEUE(hard):\t\t\t%ld\n",

(long) saved_limits.rlim_max);

printf("\tMaximum Message Size:\t\t\t%d\n", saved_max_msgsize);

printf("\tMaximum Queue Size:\t\t\t%d\n", saved_max_msgs);

printf("\tNice value:\t\t\t\t%d\n", cur_nice);

printf("\n");

increase_limits();

printf("Adjusted system state for testing:\n");

if (cur_limits.rlim_cur == RLIM_INFINITY) {

printf("\tRLIMIT_MSGQUEUE(soft):\t\t\t(unlimited)\n");

printf("\tRLIMIT_MSGQUEUE(hard):\t\t\t(unlimited)\n");

} else {

printf("\tRLIMIT_MSGQUEUE(soft):\t\t\t%ld\n",

(long) cur_limits.rlim_cur);

printf("\tRLIMIT_MSGQUEUE(hard):\t\t\t%ld\n",

(long) cur_limits.rlim_max);

}

printf("\tMaximum Message Size:\t\t\t%d\n", cur_max_msgsize);

printf("\tMaximum Queue Size:\t\t\t%d\n", cur_max_msgs);

printf("\tNice value:\t\t\t\t%d\n", cur_nice);

printf("\tContinuous mode:\t\t\t(%s)\n", continuous_mode ?

(continuous_mode_fake ? "fake mode" : "enabled") :

"disabled");

printf("\tCPUs to pin:\t\t\t\t%d", cpus_to_pin[0]);

for (cpu = 1; cpu < num_cpus_to_pin; cpu++)

printf(",%d", cpus_to_pin[cpu]);

printf("\n");

sa.sa_sigaction = sig_action_SIGUSR1;

sigemptyset(&sa.sa_mask);

sigaddset(&sa.sa_mask, SIGHUP);

sigaddset(&sa.sa_mask, SIGINT);

sigaddset(&sa.sa_mask, SIGQUIT);

sigaddset(&sa.sa_mask, SIGTERM);

sa.sa_flags = SA_SIGINFO;

if (sigaction(SIGUSR1, &sa, NULL) == -1)

shutdown(1, "sigaction(SIGUSR1)", __LINE__);

sa.sa_sigaction = sig_action;

if (sigaction(SIGHUP, &sa, NULL) == -1)

shutdown(1, "sigaction(SIGHUP)", __LINE__);

if (sigaction(SIGINT, &sa, NULL) == -1)

shutdown(1, "sigaction(SIGINT)", __LINE__);

if (sigaction(SIGQUIT, &sa, NULL) == -1)

shutdown(1, "sigaction(SIGQUIT)", __LINE__);

if (sigaction(SIGTERM, &sa, NULL) == -1)

shutdown(1, "sigaction(SIGTERM)", __LINE__);

if (!continuous_mode_fake) {

attr.mq_flags = O_NONBLOCK;

attr.mq_maxmsg = cur_max_msgs;

attr.mq_msgsize = MSG_SIZE;

open_queue(&attr);

}

for (i = 0; i < num_cpus_to_pin; i++) {

pthread_attr_t thread_attr;

void *thread_func;

if (continuous_mode_fake)

thread_func = &fake_cont_thread;

else if (continuous_mode)

thread_func = &cont_thread;

else

thread_func = &perf_test_thread;

CPU_ZERO_S(cpu_set_size, cpu_set);

CPU_SET_S(cpus_to_pin[i], cpu_set_size, cpu_set);

pthread_attr_init(&thread_attr);

pthread_attr_setaffinity_np(&thread_attr, cpu_set_size,

cpu_set);

if (pthread_create(&cpu_threads[i], &thread_attr, thread_func,

NULL))

shutdown(1, "pthread_create()", __LINE__);

pthread_attr_destroy(&thread_attr);

}

if (!continuous_mode) {

pthread_join(cpu_threads[0], &retval);

shutdown((long)retval, "perf_test_thread()", __LINE__);

} else {

while (1)

sleep(1);

}

shutdown(0, "", 0);

err_code:

CPU_FREE(cpu_set);

exit(1);

}