ports/openmpi
ports/openmpi
1
1
Форкнуть 0
Код Релизы Активность

Use select() for Windows events by default.

Просмотр исходного кода 
For some historical reasons, we didn't use select() for the Windows events. Now it could be merged back to have a better performance on Windows.

This commit was SVN r21899.
Этот коммит содержится в:
Shiqing Fan 2009-08-27 08:11:56 +00:00
родитель 6e57772ebd
Коммит 4119497c5a
3 изменённых файлов: 336 добавлений и 397 удалений
Показать статистику Скачать Patch файл Скачать Diff файл Показать все файлы Свернуть все файлы

569
opal/event/WIN32-Code/win32.c
Просмотреть файл

@ -46,18 +46,17 @@
#define RB_AUGMENT(x) (void)(x)
#include "./tree.h"
#include "opal/event/event_rename.h"
#include "opal/event/log.h"
#include "opal/event/event.h"
#include "opal/event/event-internal.h"
#include "opal/event/WIN32-Code/misc.h"
#define XFREE(ptr) do { if (ptr) free(ptr); } while(0)
extern struct event_list timequeue;
extern struct event_list addqueue;
#if 0
extern struct event_list signalqueue;
#endif
struct win_fd_set {
u_int fd_count;
@ -69,9 +68,12 @@ volatile sig_atomic_t signal_caught = 0;
/* MSDN says this is required to handle SIGFPE */
volatile double SIGFPE_REQ = 0.0f;
#if 0
static void signal_handler(int sig);
void signal_process(struct event_base *base);
int signal_recalc(struct event_base *base);
void signal_process(void);
int signal_recalc(void);
#endif
struct event_entry {
RB_ENTRY(event_entry) node;
@ -100,11 +102,11 @@ struct win32op {
struct win_fd_set *readset_out;
struct win_fd_set *writeset_out;
struct win_fd_set *exset_out;
int n_events;
int n_events_alloc;
opal_event_t **events;
RB_HEAD(event_map, event_entry) event_root;
};
RB_PROTOTYPE(event_map, event_entry, node, compare);
RB_GENERATE(event_map, event_entry, node, compare);
void *win32_init (struct event_base *);
int win32_insert (void *, struct event *);
@ -122,355 +124,289 @@ struct eventop win32ops = {
0
};
#define FD_SET_ALLOC_SIZE(n) ((sizeof(struct win_fd_set) + ((n)-1)*sizeof(SOCKET)))
static int
realloc_fd_sets(struct win32op *op, size_t new_size)
{
size_t size;
assert(new_size >= op->readset_in->fd_count &&
new_size >= op->writeset_in->fd_count);
assert(new_size >= 1);
size = FD_SET_ALLOC_SIZE(new_size);
if (!(op->readset_in = realloc(op->readset_in, size)))
return (-1);
if (!(op->writeset_in = realloc(op->writeset_in, size)))
return (-1);
if (!(op->readset_out = realloc(op->readset_out, size)))
return (-1);
if (!(op->exset_out = realloc(op->exset_out, size)))
return (-1);
if (!(op->writeset_out = realloc(op->writeset_out, size)))
return (-1);
op->fd_setsz = new_size;
return (0);
}
static int
timeval_to_ms(struct timeval *tv)
{
return ((tv->tv_sec * 1000) + (tv->tv_usec / 1000));
}
static struct event_entry*
get_event_entry(struct win32op *op, SOCKET s, int create)
{
struct event_entry key, *val;
key.sock = s;
val = RB_FIND(event_map, &op->event_root, &key);
if (val || !create)
return val;
if (!(val = calloc(1, sizeof(struct event_entry)))) {
event_warn("%s: calloc", __func__);
return NULL;
}
val->sock = s;
val->read_pos = val->write_pos = -1;
RB_INSERT(event_map, &op->event_root, val);
return val;
}
static int
do_fd_set(struct win32op *op, struct event_entry *ent, int read)
{
SOCKET s = ent->sock;
struct win_fd_set *set = read ? op->readset_in : op->writeset_in;
if (read) {
if (ent->read_pos >= 0)
return (0);
} else {
if (ent->write_pos >= 0)
return (0);
}
if (set->fd_count == op->fd_setsz) {
if (realloc_fd_sets(op, op->fd_setsz*2))
return (-1);
/* set pointer will have changed and needs reiniting! */
set = read ? op->readset_in : op->writeset_in;
}
set->fd_array[set->fd_count] = s;
if (read)
ent->read_pos = set->fd_count;
else
ent->write_pos = set->fd_count;
return (set->fd_count++);
}
static int
do_fd_clear(struct win32op *op, struct event_entry *ent, int read)
{
int i;
struct win_fd_set *set = read ? op->readset_in : op->writeset_in;
if (read) {
i = ent->read_pos;
ent->read_pos = -1;
} else {
i = ent->write_pos;
ent->write_pos = -1;
}
if (i < 0)
return (0);
if (--set->fd_count != i) {
struct event_entry *ent2;
SOCKET s2;
s2 = set->fd_array[i] = set->fd_array[set->fd_count];
ent2 = get_event_entry(op, s2, 0);
if (!ent) /* This indicates a bug. */
return (0);
if (read)
ent2->read_pos = i;
else
ent2->write_pos = i;
}
return (0);
}
#define NEVENT 64
void *
win32_init(void)
win32_init(struct event_base *_base)
{
struct win32op *winop;
size_t size;
if (!(winop = (struct win32op*)calloc(1, sizeof(struct win32op))))
if (!(winop = calloc(1, sizeof(struct win32op))))
return NULL;
winop->fd_setsz = NEVENT;
size = FD_SET_ALLOC_SIZE(NEVENT);
if (!(winop->readset_in = (struct win_fd_set*)malloc(size)))
if (!(winop->readset_in = malloc(size)))
goto err;
if (!(winop->writeset_in = (struct win_fd_set*)malloc(size)))
if (!(winop->writeset_in = malloc(size)))
goto err;
if (!(winop->readset_out = (struct win_fd_set*)malloc(size)))
if (!(winop->readset_out = malloc(size)))
goto err;
if (!(winop->writeset_out = (struct win_fd_set*)malloc(size)))
if (!(winop->writeset_out = malloc(size)))
goto err;
if (!(winop->exset_out = (struct win_fd_set*)malloc(size)))
goto err;
winop->n_events = 0;
winop->n_events_alloc = NEVENT;
if (!(winop->events = (opal_event_t**)malloc(NEVENT*sizeof(opal_event_t*))))
if (!(winop->exset_out = malloc(size)))
goto err;
RB_INIT(&winop->event_root);
winop->readset_in->fd_count = winop->writeset_in->fd_count = 0;
winop->readset_out->fd_count = winop->writeset_out->fd_count
= winop->exset_out->fd_count = 0;
return (winop);
evsignal_init(_base);
return (winop);
err:
XFREE(winop->readset_in);
XFREE(winop->writeset_in);
XFREE(winop->readset_out);
XFREE(winop->writeset_out);
XFREE(winop->exset_out);
XFREE(winop->events);
XFREE(winop);
return (NULL);
}
int
win32_recalc(struct event_base *base, void *arg, int max)
win32_insert(void *op, struct event *ev)
{
return 0/*(signal_recalc())*/;
}
struct win32op *win32op = op;
struct event_entry *ent;
static bool win32_is_fd_a_socket( int fd )
{
int error;
u_long value = 123456;
if( SOCKET_ERROR == WSAHtonl( fd, value, &value ) ) {
error = WSAGetLastError();
return false;
if (ev->ev_events & EV_SIGNAL) {
return (evsignal_add(ev));
}
return true;
}
void CALLBACK win32_socket_event_callback( void* lpParameter, BOOLEAN TimerOrWaitFired )
{
opal_event_t* master = (opal_event_t*)lpParameter;
WSANETWORKEVENTS network_events;
int got, error;
opal_event_t* next;
assert( FALSE == TimerOrWaitFired );
/* The handle will be automatically reset */
if( SOCKET_ERROR == WSAEnumNetworkEvents( master->ev_fd, master->base_handle, &network_events ) ) {
error = WSAGetLastError();
return;
}
do {
got = 0;
next = master->ev_similar;
if( network_events.lNetworkEvents & (FD_READ | FD_ACCEPT) ) {
if( master->ev_events & OPAL_EV_READ ) {
network_events.lNetworkEvents &= ~(FD_READ | FD_ACCEPT);
got |= OPAL_EV_READ;
}
}
if( master->ev_events & OPAL_EV_WRITE ) {
got |= OPAL_EV_WRITE;
if( 0 == WSASetEvent(master->base_handle) ) {
int error = WSAGetLastError();
}
}
if( got ) {
if (!(master->ev_events & OPAL_EV_PERSIST)) {
opal_event_del(master);
}
SleepEx(1, true);
(*master->ev_callback)((int)master->ev_fd, got, master->ev_arg);
/*opal_event_active( ev, got, 1 );*/
}
master = next;
} while( master != ((opal_event_t*)lpParameter) );
}
void CALLBACK win32_file_event_callback( void* lpParameter, BOOLEAN TimerOrWaitFired )
{
opal_event_t* ev = (opal_event_t*)lpParameter;
int got = 0;
assert( FALSE == TimerOrWaitFired );
if( ev->ev_events & OPAL_EV_READ ) {
got |= OPAL_EV_READ;
}
if( ev->ev_events & OPAL_EV_WRITE ) {
got |= OPAL_EV_WRITE;
}
if (!(ev->ev_events & OPAL_EV_PERSIST)) {
opal_event_del(ev);
}
if( got ) {
/*(*ev->ev_callback)((int)ev->ev_fd, got, ev->ev_arg);*/
opal_event_active( ev, got, 1 );
}
}
static int win32_recompute_event( opal_event_t* master )
{
long flags = FD_CLOSE;
opal_event_t* temp;
int error;
if( INVALID_HANDLE_VALUE == master->base_handle ) {
master->base_handle = WSACreateEvent();
if( INVALID_HANDLE_VALUE == master->base_handle ) {
return 0;
}
}
/* Compute the flags we're looking at */
temp = master;
do {
if( temp->ev_events & OPAL_EV_READ ) flags |= FD_READ | FD_ACCEPT;
if( temp->ev_events & OPAL_EV_WRITE ) flags |= FD_WRITE | FD_CONNECT;
temp = temp->ev_similar;
} while( temp != master );
if( SOCKET_ERROR == WSAEventSelect( master->ev_fd, master->base_handle, flags ) ) {
error = WSAGetLastError();
WSACloseEvent( master->base_handle );
master->base_handle = INVALID_HANDLE_VALUE;
return 0;
}
if( INVALID_HANDLE_VALUE == master->registered_handle ) {
if( 0 == RegisterWaitForSingleObject( &master->registered_handle, master->base_handle, win32_socket_event_callback,
(void*)master, INFINITE, WT_EXECUTEINWAITTHREAD ) ) {
error = GetLastError();
WSACloseEvent( master->base_handle );
master->base_handle = INVALID_HANDLE_VALUE;
master->registered_handle = INVALID_HANDLE_VALUE;
return 0;
}
}
if( flags & FD_WRITE ) {
if( 0 == WSASetEvent(master->base_handle) ) {
error = WSAGetLastError();
}
}
return 1;
}
int
win32_insert(struct win32op *win32op, opal_event_t *ev)
{
int i;
opal_event_t* master = NULL;
if (ev->ev_events & OPAL_EV_SIGNAL) {
if (ev->ev_events & (OPAL_EV_READ|OPAL_EV_WRITE))
event_errx(1, "%s: EV_SIGNAL incompatible use",
__func__);
if( signal(OPAL_EVENT_SIGNAL(ev), signal_handler) == SIG_ERR)
return (-1);
if (!(ev->ev_events & (EV_READ|EV_WRITE)))
return (0);
}
ent = get_event_entry(win32op, ev->ev_fd, 1);
if (!ent)
return (-1); /* out of memory */
/**
* Find a place for the current event.
*/
for( i = 0; i < win32op->n_events; ++i ) {
if( win32op->events[i]->ev_fd != ev->ev_fd ) continue;
master = win32op->events[i];
if( master == ev ) {
event_debug( ("%s: Event for %d already inserted.",
__func__, (int)ev->ev_fd));
return (0);
}
if( master->ev_events & ev->ev_events ) {
event_debug( ("%d Event for %d already have a similar event posted.",
__func__, (int)ev->ev_fd) );
}
ev->ev_similar = master->ev_similar;
master->ev_similar = ev;
break;
}
if( NULL == master ) {
event_debug(("%s: adding event for %d", __func__, (int)ev->ev_fd));
if (win32op->n_events_alloc == win32op->n_events) {
size_t sz;
win32op->n_events_alloc *= 2;
sz = sizeof(opal_event_t*) * win32op->n_events_alloc;
if (!(win32op->events = (opal_event_t**)realloc(win32op->events, sz)))
if (ev->ev_events & EV_READ) {
if (do_fd_set(win32op, ent, 1)<0)
return (-1);
ent->read_event = ev;
}
win32op->events[win32op->n_events++] = ev;
ev->ev_similar = ev;
master = ev;
if (ev->ev_events & EV_WRITE) {
if (do_fd_set(win32op, ent, 0)<0)
return (-1);
ent->write_event = ev;
}
ev->base_handle = INVALID_HANDLE_VALUE;
ev->registered_handle = INVALID_HANDLE_VALUE;
/**
* Decide if we have a socket or a normal file descriptor. If it's a socket
* create a WSA event otherwise a normal file event will be what we need.
*/
if( win32_is_fd_a_socket(ev->ev_fd) ) {
win32_recompute_event( master );
} else {
/*
if( ev->ev_events & OPAL_EV_READ ) flags |= FD_READ;
if( ev->ev_events & OPAL_EV_WRITE ) flags |= FD_WRITE;
ev->base_handle = (HANDLE)_get_osfhandle( ev->ev_fd );
if( INVALID_HANDLE_VALUE == ev->base_handle ) {
int error = errno;
return (0);
}
int
win32_del(void *op, struct event *ev)
{
struct win32op *win32op = op;
struct event_entry *ent;
if (ev->ev_events & EV_SIGNAL)
return (evsignal_del(ev));
if (!(ent = get_event_entry(win32op, ev->ev_fd, 0)))
return (-1);
event_debug(("%s: Removing event for %d", __func__, ev->ev_fd));
if (ev == ent->read_event) {
do_fd_clear(win32op, ent, 1);
ent->read_event = NULL;
}
if( 0 == RegisterWaitForSingleObject( &ev->registered_handle, ev->base_handle, win32_file_event_callback,
(void*)ev, INFINITE, WT_EXECUTEINWAITTHREAD ) ) {
error = GetLastError();
}*/
if (ev == ent->write_event) {
do_fd_clear(win32op, ent, 0);
ent->write_event = NULL;
}
if (!ent->read_event && !ent->write_event) {
RB_REMOVE(event_map, &win32op->event_root, ent);
free(ent);
}
return 0;
}
static void
fd_set_copy(struct win_fd_set *out, const struct win_fd_set *in)
{
out->fd_count = in->fd_count;
memcpy(out->fd_array, in->fd_array, in->fd_count * (sizeof(SOCKET)));
}
/*
static void dump_fd_set(struct win_fd_set *s)
{
unsigned int i;
printf("[ ");
for(i=0;i<s->fd_count;++i)
printf("%d ",(int)s->fd_array[i]);
printf("]\n");
}
*/
int
win32_dispatch(struct event_base *base, void *op,
struct timeval *tv)
{
struct win32op *win32op = op;
int res = 0;
int i;
int fd_count;
fd_set_copy(win32op->readset_out, win32op->readset_in);
fd_set_copy(win32op->exset_out, win32op->readset_in);
fd_set_copy(win32op->writeset_out, win32op->writeset_in);
fd_count =
(win32op->readset_out->fd_count > win32op->writeset_out->fd_count) ?
win32op->readset_out->fd_count : win32op->writeset_out->fd_count;
if (!fd_count) {
/* Windows doesn't like you to call select() with no sockets */
Sleep(timeval_to_ms(tv));
evsignal_process(base);
return (0);
}
res = select(fd_count,
(struct fd_set*)win32op->readset_out,
(struct fd_set*)win32op->writeset_out,
(struct fd_set*)win32op->exset_out, tv);
event_debug(("%s: select returned %d", __func__, res));
if(res <= 0) {
evsignal_process(base);
return res;
} else if (base->sig.evsignal_caught) {
evsignal_process(base);
}
for (i=0; i<win32op->readset_out->fd_count; ++i) {
struct event_entry *ent;
SOCKET s = win32op->readset_out->fd_array[i];
if ((ent = get_event_entry(win32op, s, 0)) && ent->read_event)
event_active(ent->read_event, EV_READ, 1);
}
for (i=0; i<win32op->exset_out->fd_count; ++i) {
struct event_entry *ent;
SOCKET s = win32op->exset_out->fd_array[i];
if ((ent = get_event_entry(win32op, s, 0)) && ent->read_event)
event_active(ent->read_event, EV_READ, 1);
}
for (i=0; i<win32op->writeset_out->fd_count; ++i) {
struct event_entry *ent;
SOCKET s = win32op->writeset_out->fd_array[i];
if ((ent = get_event_entry(win32op, s, 0)) && ent->write_event)
event_active(ent->write_event, EV_WRITE, 1);
}
#if 0
if (signal_recalc() == -1)
return (-1);
#endif
return (0);
}
int
win32_del(struct win32op *win32op, opal_event_t *ev)
{
int i, error;
opal_event_t *master = NULL, *temp;
if (ev->ev_events & OPAL_EV_SIGNAL)
return (signal(OPAL_EVENT_SIGNAL(ev), SIG_IGN) != SIG_ERR);
for( i = 0; i < win32op->n_events; ++i ) {
if( win32op->events[i]->ev_fd != ev->ev_fd ) continue;
master = win32op->events[i];
break;
}
if( NULL == master ) {
event_debug(("%s: Unable to remove non-inserted event for %d",
__func__, ev->ev_fd));
return (-1);
}
event_debug(("%s: Removing event for %d", __func__, ev->ev_fd));
/**
* Remove the current event and recompute the registered event
* based on the opal events pending on the same master.
*/
if( master == ev ) {
/* Disable all pending events */
if( INVALID_HANDLE_VALUE != ev->registered_handle ) {
if( 0 == UnregisterWait(ev->registered_handle) ) {
error = GetLastError();
}
ev->registered_handle = INVALID_HANDLE_VALUE;
}
if( ev->ev_similar == ev ) {
/* Only one event in the queue. Remove everything. */
if( INVALID_HANDLE_VALUE != ev->base_handle ) { /* socket */
/* Now detach the base event from the socket. */
if( SOCKET_ERROR == WSAEventSelect( ev->ev_fd, ev->base_handle, 0) ) {
error = WSAGetLastError();
}
/* Finally, destroy the event handle. */
if( 0 == WSACloseEvent(ev->base_handle) ) {
error = WSAGetLastError();
}
ev->base_handle = INVALID_HANDLE_VALUE;
}
if (i != --win32op->n_events) {
win32op->events[i] = win32op->events[win32op->n_events];
}
return 0;
}
master = ev->ev_similar;
master->base_handle = ev->base_handle;
master->registered_handle = ev->registered_handle;
win32op->events[i] = master;
temp = master;
while( temp->ev_similar != ev ) {
temp = temp->ev_similar;
}
temp->ev_similar = master;
} else {
temp = master;
while( temp->ev_similar != ev ) {
temp = temp->ev_similar;
}
temp->ev_similar = ev->ev_similar;
}
win32_recompute_event( master );
return 0;
}
int
win32_dispatch( struct event_base *base, struct win32op *win32op,
struct timeval *tv )
{
DWORD milisec;
/*milisec = tv->tv_sec * 1000;
if( tv->tv_usec > 1000 ) {
milisec += tv->tv_usec / 1000;
}*/
milisec = tv->tv_sec; /* BLAH BLAH REMOVE ME */
SleepEx( milisec, TRUE );
if( 0 != signal_caught ) {
signal_process(base);
signal_recalc(base);
}
return 0;
}
void
win32_dealloc(struct event_base *_base, void *arg)
{
@ -493,6 +429,7 @@ win32_dealloc(struct event_base *_base, void *arg)
free(win32op);
}
#if 0
static void
signal_handler(int sig)
{
@ -501,12 +438,12 @@ signal_handler(int sig)
}
int
signal_recalc(struct event_base base)
signal_recalc(void)
{
struct event *ev;
/* Reinstall our signal handler. */
TAILQ_FOREACH(ev, &base.sig.signalqueue, ev_signal_next) {
TAILQ_FOREACH(ev, &signalqueue, ev_signal_next) {
if((int)signal(EVENT_SIGNAL(ev), signal_handler) == -1)
return (-1);
}
@ -514,12 +451,12 @@ signal_recalc(struct event_base base)
}
void
signal_process(struct event_base base)
signal_process(void)
{
struct event *ev;
short ncalls;
TAILQ_FOREACH(ev, &base.sig.signalqueue, ev_signal_next) {
TAILQ_FOREACH(ev, &signalqueue, ev_signal_next) {
ncalls = evsigcaught[EVENT_SIGNAL(ev)];
if (ncalls) {
if (!(ev->ev_events & EV_PERSIST))
@ -531,4 +468,4 @@ signal_process(struct event_base base)
memset(evsigcaught, 0, sizeof(evsigcaught));
signal_caught = 0;
}
#endif

2
orte/mca/iof/hnp/iof_hnp_read.c
Просмотреть файл

@ -58,6 +58,8 @@ bool orte_iof_hnp_stdin_check(int fd)
if( isatty(fd) && (getpgrp() != tcgetpgrp(fd)) ) {
return false;
}
#elif defined(__WINDOWS__)
return false;
#endif /* !defined(__WINDOWS__) */
return true;
}

2
orte/tools/orterun/orterun.c
Просмотреть файл

@ -635,6 +635,7 @@ int orterun(int argc, char *argv[])
goto DONE;
}
#ifndef __WINDOWS__
/** setup callbacks for abort signals - from this point
* forward, we need to abort in a manner that allows us
* to cleanup
@ -646,7 +647,6 @@ int orterun(int argc, char *argv[])
abort_signal_callback, &int_handler);
opal_signal_add(&int_handler, NULL);
#ifndef __WINDOWS__
/** setup callbacks for signals we should foward */
opal_signal_set(&sigusr1_handler, SIGUSR1,
signal_forward_callback, &sigusr1_handler);