Update a test and add two new ones for testing event lib thread support
This commit was SVN r24051.
Этот коммит содержится в:
Ralph Castain
родитель
b43a4509ac
Коммит
58e711a412
@ -1,4 +1,4 @@
|
||||
PROGS = no_op sigusr_trap spin orte_nodename orte_spawn orte_loop_spawn orte_loop_child orte_abort get_limits orte_ring spawn_child orte_tool orte_no_op binom oob_stress iof_stress iof_delay radix orte_barrier orte_mcast opal_interface mcast mcast_recv orte_spin segfault sysinfo orte_exit orte_db orte_sensor
|
||||
PROGS = no_op sigusr_trap spin orte_nodename orte_spawn orte_loop_spawn orte_loop_child orte_abort get_limits orte_ring spawn_child orte_tool orte_no_op binom oob_stress iof_stress iof_delay radix orte_barrier orte_mcast opal_interface mcast mcast_recv orte_spin segfault sysinfo orte_exit orte_db orte_sensor test-time event-threads
|
||||
|
||||
all: $(PROGS)
|
||||
|
||||
|
||||
153
orte/test/system/event-threads.c
Обычный файл
153
orte/test/system/event-threads.c
Обычный файл
@ -0,0 +1,153 @@
|
||||
#include <sys/types.h>
|
||||
#include <sys/stat.h>
|
||||
#include <fcntl.h>
|
||||
#include <stdlib.h>
|
||||
#include <stdio.h>
|
||||
#include <string.h>
|
||||
#ifndef WIN32
|
||||
#include <unistd.h>
|
||||
#include <sys/time.h>
|
||||
#endif
|
||||
#include <errno.h>
|
||||
|
||||
#include "opal/util/fd.h"
|
||||
#include "opal/threads/threads.h"
|
||||
#include "opal/runtime/opal.h"
|
||||
#include "opal/mca/event/event.h"
|
||||
|
||||
static opal_event_base_t *my_base=NULL;
|
||||
static opal_thread_t progress_thread;
|
||||
static bool progress_thread_stop=false;
|
||||
static int progress_thread_pipe[2];
|
||||
static opal_mutex_t lock;
|
||||
static opal_condition_t cond;
|
||||
static bool active=false;
|
||||
static opal_event_t write_event;
|
||||
static int my_fd;
|
||||
static bool fd_written=false;
|
||||
|
||||
static void* progress_engine(opal_object_t *obj);
|
||||
static void send_handler(int sd, short flags, void *arg);
|
||||
|
||||
int main(int argc, char **argv)
|
||||
{
|
||||
char byte='a';
|
||||
struct timespec tp={0, 100};
|
||||
int count=0;
|
||||
|
||||
/* Initialize the event library */
|
||||
opal_init(&argc, &argv);
|
||||
|
||||
/* setup for threads */
|
||||
opal_event_use_threads();
|
||||
|
||||
/* create a new base */
|
||||
my_base = opal_event_base_create();
|
||||
|
||||
/* launch a progress thread on that base*/
|
||||
pipe(progress_thread_pipe);
|
||||
OBJ_CONSTRUCT(&lock, opal_mutex_t);
|
||||
OBJ_CONSTRUCT(&cond, opal_condition_t);
|
||||
OBJ_CONSTRUCT(&progress_thread, opal_thread_t);
|
||||
progress_thread.t_run = progress_engine;
|
||||
if (OPAL_SUCCESS != opal_thread_start(&progress_thread)) {
|
||||
fprintf(stderr, "Unable to start progress thread\n");
|
||||
opal_event_base_finalize(my_base);
|
||||
exit(1);
|
||||
}
|
||||
|
||||
/* wait a little while - reflects reality in an async system */
|
||||
while (count < 100) {
|
||||
nanosleep(&tp, NULL);
|
||||
count++;
|
||||
}
|
||||
count=0;
|
||||
|
||||
/* define a file descriptor event - looks like an incoming socket
|
||||
* connection being created
|
||||
*/
|
||||
if (0 > (my_fd = open("foo", O_CREAT | O_TRUNC | O_RDWR, 0644))) {
|
||||
fprintf(stderr, "Couldnt open file\n");
|
||||
exit(1);
|
||||
}
|
||||
opal_event_set(my_base,
|
||||
&write_event,
|
||||
my_fd,
|
||||
OPAL_EV_WRITE|OPAL_EV_PERSIST,
|
||||
send_handler,
|
||||
NULL);
|
||||
opal_event_add(&write_event, 0);
|
||||
/* opal_fd_write(progress_thread_pipe[1], 1, &byte); */
|
||||
|
||||
/* wait for it to trigger */
|
||||
while (count < 1000) {
|
||||
OPAL_ACQUIRE_THREAD(&lock, &cond, &active);
|
||||
if (fd_written) {
|
||||
OPAL_RELEASE_THREAD(&lock, &cond, &active);
|
||||
break;
|
||||
}
|
||||
OPAL_RELEASE_THREAD(&lock, &cond, &active);
|
||||
if (0 == (count % 100)) {
|
||||
fprintf(stderr, "Waiting...\n");
|
||||
}
|
||||
nanosleep(&tp, NULL);
|
||||
count++;
|
||||
}
|
||||
fprintf(stderr, "Done waiting\n");
|
||||
|
||||
/* stop the thread */
|
||||
OPAL_ACQUIRE_THREAD(&lock, &cond, &active);
|
||||
progress_thread_stop = true;
|
||||
OPAL_RELEASE_THREAD(&lock, &cond, &active);
|
||||
opal_fd_write(progress_thread_pipe[1], 1, &byte);
|
||||
opal_thread_join(&progress_thread, NULL);
|
||||
|
||||
opal_finalize();
|
||||
return 0;
|
||||
}
|
||||
|
||||
static opal_event_t stop_event;
|
||||
static void stop_handler(int sd, short flags, void* cbdata)
|
||||
{
|
||||
char byte;
|
||||
opal_fd_read(progress_thread_pipe[0], 1, &byte);
|
||||
fprintf(stderr, "Stop handler called\n");
|
||||
/* reset the event */
|
||||
opal_event_add(&stop_event, 0);
|
||||
return;
|
||||
}
|
||||
|
||||
static void* progress_engine(opal_object_t *obj)
|
||||
{
|
||||
/* define an event that will be used to kick us out of a blocking
|
||||
* situation when we want to exit
|
||||
*/
|
||||
opal_event_set(my_base, &stop_event,
|
||||
progress_thread_pipe[0], OPAL_EV_READ, stop_handler, NULL);
|
||||
opal_event_add(&stop_event, 0);
|
||||
|
||||
while (1) {
|
||||
OPAL_ACQUIRE_THREAD(&lock, &cond, &active);
|
||||
if (progress_thread_stop) {
|
||||
fprintf(stderr, "Thread stopping\n");
|
||||
OPAL_RELEASE_THREAD(&lock, &cond, &active);
|
||||
opal_event_del(&stop_event);
|
||||
return OPAL_THREAD_CANCELLED;
|
||||
}
|
||||
OPAL_RELEASE_THREAD(&lock, &cond, &active);
|
||||
fprintf(stderr, "Looping...\n");
|
||||
opal_event_loop(my_base, OPAL_EVLOOP_ONCE);
|
||||
}
|
||||
}
|
||||
|
||||
static void send_handler(int sd, short flags, void *arg)
|
||||
{
|
||||
char *bytes="This is an output string\n";
|
||||
|
||||
fprintf(stderr, "Write event fired\n");
|
||||
opal_fd_write(my_fd, strlen(bytes), bytes);
|
||||
opal_event_del(&write_event);
|
||||
OPAL_ACQUIRE_THREAD(&lock, &cond, &active);
|
||||
fd_written = true;
|
||||
OPAL_RELEASE_THREAD(&lock, &cond, &active);
|
||||
}
|
||||
179
orte/test/system/evthread-test.c
Обычный файл
179
orte/test/system/evthread-test.c
Обычный файл
@ -0,0 +1,179 @@
|
||||
#include <sys/types.h>
|
||||
#include <sys/stat.h>
|
||||
#include <fcntl.h>
|
||||
#include <stdlib.h>
|
||||
#include <stdio.h>
|
||||
#include <string.h>
|
||||
#include <stdbool.h>
|
||||
#ifndef WIN32
|
||||
#include <unistd.h>
|
||||
#include <sys/time.h>
|
||||
#endif
|
||||
#include <errno.h>
|
||||
|
||||
#include <pthread.h>
|
||||
|
||||
#include <event2/util.h>
|
||||
#include <event2/thread.h>
|
||||
#include <event2/event.h>
|
||||
#include <event2/event_struct.h>
|
||||
|
||||
static struct event_base *my_base=NULL;
|
||||
static pthread_t progress_thread;
|
||||
static bool progress_thread_stop=false;
|
||||
static int progress_thread_pipe[2];
|
||||
static pthread_mutex_t lock;
|
||||
static struct event write_event;
|
||||
static int my_fd;
|
||||
static bool fd_written=false;
|
||||
|
||||
static void* progress_engine(void *obj);
|
||||
static void send_handler(int sd, short flags, void *arg);
|
||||
|
||||
int main(int argc, char **argv)
|
||||
{
|
||||
char byte='a';
|
||||
struct timespec tp={0, 100};
|
||||
int count=0;
|
||||
|
||||
/* setup for threads */
|
||||
evthread_use_pthreads();
|
||||
|
||||
/* create a new base */
|
||||
my_base = event_base_new();
|
||||
|
||||
/* launch a progress thread on that base*/
|
||||
pipe(progress_thread_pipe);
|
||||
|
||||
if (pthread_mutex_init(&lock, NULL)) {
|
||||
fprintf(stderr, "pthread_mutex_init failed\n");
|
||||
exit(1);
|
||||
}
|
||||
if (pthread_create(&progress_thread, NULL, progress_engine,
|
||||
NULL)) {
|
||||
fprintf(stderr, "pthread_create failed\n");
|
||||
exit(1);
|
||||
}
|
||||
/*
|
||||
pthread starts the thread running itself; no need to do anything to
|
||||
launch it.
|
||||
*/
|
||||
|
||||
/* wait a little while - reflects reality in an async system */
|
||||
while (count < 100) {
|
||||
nanosleep(&tp, NULL);
|
||||
count++;
|
||||
}
|
||||
count=0;
|
||||
|
||||
#ifdef WAKE_WITH_EVENT
|
||||
/* make a dummy event */
|
||||
fprintf(stderr, "activating the write_event");
|
||||
event_assign(&write_event,
|
||||
my_base,
|
||||
-1,
|
||||
0,
|
||||
send_handler,
|
||||
NULL);
|
||||
/* activate it. */
|
||||
event_active(&write_event, EV_WRITE, 1);
|
||||
#else
|
||||
fprintf(stderr, "opening the file");
|
||||
/* define a file descriptor event - looks like an incoming socket
|
||||
* connection being created, if we're lucky.
|
||||
*/
|
||||
my_fd = open("foo", O_CREAT | O_TRUNC | O_RDWR, 0644);
|
||||
if (my_fd <0) {
|
||||
perror("open");
|
||||
exit(1);
|
||||
}
|
||||
event_assign(&write_event,
|
||||
my_base,
|
||||
my_fd,
|
||||
EV_WRITE|EV_PERSIST,
|
||||
send_handler,
|
||||
NULL);
|
||||
event_add(&write_event, NULL);
|
||||
if (write(progress_thread_pipe[1], &byte, 1) < 0) {
|
||||
perror("write");
|
||||
exit(1);
|
||||
}
|
||||
#endif
|
||||
|
||||
/* wait for it to trigger */
|
||||
while (!fd_written && count < 1000) {
|
||||
if (0 == (count % 100)) {
|
||||
fprintf(stderr, "Waiting...\n");
|
||||
}
|
||||
nanosleep(&tp, NULL);
|
||||
count++;
|
||||
}
|
||||
|
||||
/* stop the thread */
|
||||
pthread_mutex_lock(&lock);
|
||||
progress_thread_stop = true;
|
||||
pthread_mutex_unlock(&lock);
|
||||
|
||||
write(progress_thread_pipe[1], &byte, 1);
|
||||
pthread_join(progress_thread, NULL);
|
||||
|
||||
return 0;
|
||||
}
|
||||
|
||||
static struct event stop_event;
|
||||
static void stop_handler(int sd, short flags, void* cbdata)
|
||||
{
|
||||
char byte;
|
||||
int n;
|
||||
if ((n = read(progress_thread_pipe[0], &byte, 1)) <= 0) {
|
||||
if (n == 0)
|
||||
fprintf(stderr, "got a close\n");
|
||||
else
|
||||
perror("read");
|
||||
}
|
||||
|
||||
/* reset the event */
|
||||
event_add(&stop_event, NULL);
|
||||
return;
|
||||
}
|
||||
|
||||
static void* progress_engine(void *obj)
|
||||
{
|
||||
/* define an event that will be used to kick us out of a blocking
|
||||
* situation when we want to exit
|
||||
*/
|
||||
event_assign(&stop_event, my_base,
|
||||
progress_thread_pipe[0], EV_READ, stop_handler, NULL);
|
||||
event_add(&stop_event, NULL);
|
||||
|
||||
while (1) {
|
||||
pthread_mutex_lock(&lock);
|
||||
if (progress_thread_stop) {
|
||||
fprintf(stderr, "Thread stopping\n");
|
||||
pthread_mutex_unlock(&lock); /* moved this */
|
||||
event_del(&stop_event);
|
||||
return (void*)1;
|
||||
}
|
||||
pthread_mutex_unlock(&lock);
|
||||
fprintf(stderr, "Looping...\n");
|
||||
event_base_loop(my_base, EVLOOP_ONCE);
|
||||
}
|
||||
}
|
||||
|
||||
static void send_handler(int sd, short flags, void *arg)
|
||||
{
|
||||
#ifdef WAKE_WITH_EVENT
|
||||
fprintf(stderr, "Write event fired\n");
|
||||
#else
|
||||
char *bytes="This is an output string\n";
|
||||
|
||||
fprintf(stderr, "Write event fired\n");
|
||||
if (write(my_fd, bytes, strlen(bytes)) < 0) {
|
||||
perror("write");
|
||||
exit(1);
|
||||
}
|
||||
event_del(&write_event);
|
||||
#endif
|
||||
fd_written = true; /* This needs a lock around it if you are reading it
|
||||
* in the main thread and changing it here XXX */
|
||||
}
|
||||
@ -19,12 +19,11 @@
|
||||
|
||||
int called = 0;
|
||||
|
||||
#define NEVENT 20000
|
||||
#define NEVENT 2000
|
||||
|
||||
struct opal_event_t *ev[NEVENT];
|
||||
opal_event_t* ev[NEVENT];
|
||||
|
||||
static int
|
||||
rand_int(int n)
|
||||
static int rand_int(int n)
|
||||
{
|
||||
#ifdef WIN32
|
||||
return (int)(rand() % n);
|
||||
@ -33,56 +32,53 @@ rand_int(int n)
|
||||
#endif
|
||||
}
|
||||
|
||||
static void
|
||||
time_cb(evutil_socket_t fd, short event, void *arg)
|
||||
static void time_cb(int fd, short event, void *arg)
|
||||
{
|
||||
struct timeval tv;
|
||||
int i, j;
|
||||
struct timeval tv;
|
||||
opal_event_t *tmp = (opal_event_t*)arg;
|
||||
|
||||
called++;
|
||||
called++;
|
||||
|
||||
if (called < 10*NEVENT) {
|
||||
for (i = 0; i < 10; i++) {
|
||||
j = rand_int(NEVENT);
|
||||
tv.tv_sec = 0;
|
||||
tv.tv_usec = rand_int(50000);
|
||||
if (tv.tv_usec % 2)
|
||||
opal_evtimer_add(ev[j], &tv);
|
||||
else
|
||||
opal_evtimer_del(ev[j]);
|
||||
}
|
||||
}
|
||||
if (0 == (called % 1000)) {
|
||||
fprintf(stderr, "Fired event %d\n", called);
|
||||
}
|
||||
|
||||
if (called < 10*NEVENT) {
|
||||
tv.tv_sec = 0;
|
||||
tv.tv_usec = rand_int(50000);
|
||||
opal_event_evtimer_add(tmp, &tv);
|
||||
}
|
||||
}
|
||||
|
||||
int
|
||||
main(int argc, char **argv)
|
||||
int main(int argc, char **argv)
|
||||
{
|
||||
struct timeval tv;
|
||||
int i;
|
||||
struct timeval tv;
|
||||
int i;
|
||||
#ifdef WIN32
|
||||
WORD wVersionRequested;
|
||||
WSADATA wsaData;
|
||||
int err;
|
||||
WORD wVersionRequested;
|
||||
WSADATA wsaData;
|
||||
int err;
|
||||
|
||||
wVersionRequested = MAKEWORD(2, 2);
|
||||
wVersionRequested = MAKEWORD(2, 2);
|
||||
|
||||
err = WSAStartup(wVersionRequested, &wsaData);
|
||||
err = WSAStartup(wVersionRequested, &wsaData);
|
||||
#endif
|
||||
|
||||
/* Initalize the event library */
|
||||
opal_init(argc, argv);
|
||||
/* Initialize the event library */
|
||||
opal_init(&argc, &argv);
|
||||
|
||||
for (i = 0; i < NEVENT; i++) {
|
||||
/* Initalize one event */
|
||||
ev[i] = opal_evtimer_new(time_cb, ev[i]);
|
||||
tv.tv_sec = 0;
|
||||
tv.tv_usec = rand_int(50000);
|
||||
opal_evtimer_add(ev[i], &tv);
|
||||
}
|
||||
for (i = 0; i < NEVENT; i++) {
|
||||
/* Initalize one event */
|
||||
ev[i] = (opal_event_t*)malloc(sizeof(opal_event_t));
|
||||
opal_event_evtimer_set(opal_event_base, ev[i], time_cb, ev[i]);
|
||||
tv.tv_sec = 0;
|
||||
tv.tv_usec = rand_int(50000);
|
||||
opal_event_evtimer_add(ev[i], &tv);
|
||||
}
|
||||
|
||||
opal_event_dispatch();
|
||||
opal_event_dispatch(opal_event_base);
|
||||
|
||||
opal_finalize();
|
||||
return (called < NEVENT);
|
||||
opal_finalize();
|
||||
return (called < NEVENT);
|
||||
}
|
||||
|
||||
|
||||
Загрузка…
x
Ссылка в новой задаче
Block a user