The R13B03 release.OTP_R13B03

1 files changed, 1361 insertions, 0 deletions

diff --git a/erts/emulator/sys/win32/erl_poll.c b/erts/emulator/sys/win32/erl_poll.c
new file mode 100644
index 0000000000..d816cc2c07
--- /dev/null
+++ b/erts/emulator/sys/win32/erl_poll.c
@@ -0,0 +1,1361 @@
+/*
+ * %CopyrightBegin%
+ * 
+ * Copyright Ericsson AB 2007-2009. All Rights Reserved.
+ * 
+ * The contents of this file are subject to the Erlang Public License,
+ * Version 1.1, (the "License"); you may not use this file except in
+ * compliance with the License. You should have received a copy of the
+ * Erlang Public License along with this software. If not, it can be
+ * retrieved online at http://www.erlang.org/.
+ * 
+ * Software distributed under the License is distributed on an "AS IS"
+ * basis, WITHOUT WARRANTY OF ANY KIND, either express or implied. See
+ * the License for the specific language governing rights and limitations
+ * under the License.
+ * 
+ * %CopyrightEnd%
+ */
+#ifdef HAVE_CONFIG_H
+#  include "config.h"
+#endif
+
+#define WANT_NONBLOCKING
+
+#include "sys.h"
+#include "erl_alloc.h"
+#include "erl_poll.h"
+
+/*
+ * Some debug macros 
+ */
+
+/*#define HARDDEBUG */
+#ifdef HARDDEBUG
+#ifdef HARDTRACE
+#define HARDTRACEF(X) my_debug_printf##X
+#else
+#define HARDTRACEF(X)
+#endif
+
+#define HARDDEBUGF(X) my_debug_printf##X
+static void my_debug_printf(char *fmt, ...)
+{
+    char buffer[1024];
+    va_list args;
+
+    va_start(args, fmt);
+    erts_vsnprintf(buffer,1024,fmt,args);
+    va_end(args);
+    erts_fprintf(stderr,"%s\r\n",buffer);
+}
+#else
+#define HARDTRACEF(X)
+#define HARDDEBUGF(X)
+#endif
+
+#ifdef DEBUG
+#define NoMansLandFill 0xFD	/* fill no-man's land with this */
+#define DeadLandFill   0xDD	/* fill free objects with this */
+#define CleanLandFill  0xCD	/* fill new objects with this */
+
+static void consistency_check(struct _Waiter* w);
+static void* debug_alloc(ErtsAlcType_t, Uint);
+static void* debug_realloc(ErtsAlcType_t, void *, Uint, Uint);
+
+#  define SEL_ALLOC	debug_alloc
+#  define SEL_REALLOC	debug_realloc
+#  define SEL_FREE	erts_free
+
+static void *debug_alloc(ErtsAlcType_t type, Uint size)
+{
+    void* p = erts_alloc(type, size);
+    memset(p, CleanLandFill, size);
+    return p;
+}
+
+static void *debug_realloc(ErtsAlcType_t type, void *ptr, Uint prev_size,
+			   Uint size)
+{
+    void *p;
+    size_t fill_size;
+    void *fill_ptr;
+
+    if (prev_size > size) {
+	size_t fill_size = (size_t) (prev_size - size);
+	void *fill_ptr = (void *) (((char *) ptr) + size);
+	memset(fill_ptr, NoMansLandFill, fill_size);
+    }
+
+    p = erts_realloc(type, ptr, size);
+
+    if (size > prev_size) {
+	size_t fill_size = (size_t) (size - prev_size);
+	void *fill_ptr = (void *) (((char *) p) + prev_size);
+	memset(fill_ptr, CleanLandFill, fill_size);
+    }
+
+    return p;
+}
+#else
+#  define SEL_ALLOC	erts_alloc
+#  define SEL_REALLOC	realloc_wrap
+#  define SEL_FREE	erts_free
+
+static ERTS_INLINE void *
+realloc_wrap(ErtsAlcType_t t, void *p, Uint ps, Uint s)
+{
+    return erts_realloc(t, p, s);
+}
+#endif
+
+
+#ifdef HARD_POLL_DEBUG
+#define OP_SELECT 1
+#define OP_DESELECT 2
+#define OP_FIRED 3
+#define OP_READ_BEGIN 4
+#define OP_READ_DONE 5
+#define OP_WRITE_BEGIN 6
+#define OP_WRITE_DONE 7
+#define OP_REPORTED 8
+#define OP_DIED 9
+#define OP_ASYNC_INIT 10
+#define OP_ASYNC_IMMED 11
+#define OP_FD_MOVED 12
+
+static struct {
+    int op;
+    ErtsSysFdType active;
+    int xdata;
+} debug_save_ops[1024];
+
+static int num_debug_save_ops = 0;
+
+static ErtsSysFdType active_debug_fd;
+static int active_debug_fd_set = 0;
+
+static erts_mtx_t save_ops_mtx;
+
+static void poll_debug_init(void)
+{
+    erts_mtx_init(&save_ops_mtx, "save_ops_lock");
+}
+
+void poll_debug_set_active_fd(ErtsSysFdType fd)
+{
+    erts_mtx_lock(&save_ops_mtx);
+    active_debug_fd_set = 1;
+    active_debug_fd = fd;
+    erts_mtx_unlock(&save_ops_mtx);
+}
+
+static void do_save_op(ErtsSysFdType fd, int op, int xdata)
+{
+    erts_mtx_lock(&save_ops_mtx);
+    if (fd == active_debug_fd && num_debug_save_ops < 1024) {
+	int x = num_debug_save_ops++;
+	debug_save_ops[x].op = op;
+	debug_save_ops[x].active = fd;
+	debug_save_ops[x].xdata = xdata;
+    }
+    erts_mtx_unlock(&save_ops_mtx);
+}
+
+void poll_debug_moved(ErtsSysFdType fd, int s1, int s2)
+{
+    do_save_op(fd,OP_FD_MOVED,s1 | (s2 << 16));
+}
+
+void poll_debug_select(ErtsSysFdType fd, int mode)
+{
+    do_save_op(fd,OP_SELECT,mode);
+}
+
+void poll_debug_deselect(ErtsSysFdType fd)
+{
+    do_save_op(fd,OP_DESELECT,0);
+}
+
+void poll_debug_fired(ErtsSysFdType fd)
+{
+    do_save_op(fd,OP_FIRED,0);
+}
+
+void poll_debug_read_begin(ErtsSysFdType fd)
+{
+    do_save_op(fd,OP_READ_BEGIN,0);
+}
+
+void poll_debug_read_done(ErtsSysFdType fd, int bytes)
+{
+    do_save_op(fd,OP_READ_DONE,bytes);
+}
+
+void poll_debug_async_initialized(ErtsSysFdType fd)
+{
+    do_save_op(fd,OP_ASYNC_INIT,0);
+}
+
+void poll_debug_async_immediate(ErtsSysFdType fd, int bytes)
+{
+    do_save_op(fd,OP_ASYNC_IMMED,bytes);
+}
+
+void poll_debug_write_begin(ErtsSysFdType fd)
+{
+    do_save_op(fd,OP_WRITE_BEGIN,0);
+}
+
+void poll_debug_write_done(ErtsSysFdType fd, int bytes)
+{
+    do_save_op(fd,OP_WRITE_DONE,bytes);
+}
+
+void poll_debug_reported(ErtsSysFdType fd, int mode)
+{
+    do_save_op(fd,OP_REPORTED,mode);
+}
+
+void poll_debug_died(ErtsSysFdType fd)
+{
+    do_save_op(fd,OP_DIED,0);
+}
+
+#endif /* DEBUG */
+
+/*
+ * End of debug macros
+ */
+
+
+
+/*
+ * Handles that we poll, but that are actually signalled from outside 
+ * this module
+ */
+
+extern HANDLE erts_service_event;
+extern HANDLE erts_sys_break_event;
+
+
+/*
+ * The structure we hold for each event (i.e. fd)
+ */
+typedef struct _EventData {
+    HANDLE event;		/* For convenience. */
+    ErtsPollEvents mode;	/* The current select mode. */
+    struct _EventData *next;	/* Next in free or delete lists. */
+} EventData;
+
+/*
+ * The structure to represent a waiter thread
+ */
+typedef struct _Waiter {
+    HANDLE events[MAXIMUM_WAIT_OBJECTS];     /* The events. */
+    EventData* evdata[MAXIMUM_WAIT_OBJECTS]; /* Pointers to associated data. */
+    int active_events;		/* Number of events to wait for */
+    int total_events;		/* Total number of events in the arrays. */
+    int highwater;              /* Events processed up to here */
+    EventData evdata_heap[MAXIMUM_WAIT_OBJECTS]; /* Pre-allocated EventDatas */
+    EventData* first_free_evdata; /* Index of first free EventData object. */
+    HANDLE go_ahead;		/* The waiter may continue. (Auto-reset) */
+    void *xdata;                /* used when thread parameter */
+    erts_tid_t this;            /* Thread "handle" of this waiter */
+    erts_mtx_t mtx;             /* Mutex for updating/reading pollset, but the
+				   currently used set require thread stopping 
+				   to be updated */
+} Waiter;
+
+/*
+ * The structure for a pollset. There can currently be only one...
+ */
+struct ErtsPollSet_ {
+    Waiter** waiter;
+    int allocated_waiters;  /* Size ow waiter array */ 
+    int num_waiters;	    /* Number of waiter threads. */
+    erts_atomic_t sys_io_ready; /* Tells us there is I/O ready (already). */
+    int restore_events;        /* Tells us to restore waiters events 
+				  next time around */
+    HANDLE event_io_ready;     /* To be used when waiting for io */
+    /* These are used to wait for workers to enter standby */
+    volatile int standby_wait_counter; /* Number of threads to wait for */
+    CRITICAL_SECTION standby_crit;     /* CS to guard the counter */
+    HANDLE standby_wait_event;         /* Event signalled when counte == 0 */
+#ifdef ERTS_SMP
+    erts_smp_atomic_t woken;
+    erts_smp_mtx_t mtx;
+    erts_smp_atomic_t interrupt;
+#endif
+    erts_smp_atomic_t timeout;
+};
+
+#ifdef ERTS_SMP
+
+#define ERTS_POLLSET_LOCK(PS) \
+  erts_smp_mtx_lock(&(PS)->mtx)
+#define ERTS_POLLSET_UNLOCK(PS) \
+  erts_smp_mtx_unlock(&(PS)->mtx)
+#define ERTS_POLLSET_SET_POLLED_CHK(PS) \
+  ((int) erts_smp_atomic_xchg(&(PS)->polled, (long) 1))
+#define ERTS_POLLSET_SET_POLLED(PS) \
+  erts_smp_atomic_set(&(PS)->polled, (long) 1)
+#define ERTS_POLLSET_UNSET_POLLED(PS) \
+  erts_smp_atomic_set(&(PS)->polled, (long) 0)
+#define ERTS_POLLSET_IS_POLLED(PS) \
+  ((int) erts_smp_atomic_read(&(PS)->polled))
+#define ERTS_POLLSET_SET_POLLER_WOKEN_CHK(PS) \
+  ((int) erts_smp_atomic_xchg(&(PS)->woken, (long) 1))
+#define ERTS_POLLSET_SET_POLLER_WOKEN(PS) \
+  erts_smp_atomic_set(&(PS)->woken, (long) 1)
+#define ERTS_POLLSET_UNSET_POLLER_WOKEN(PS) \
+  erts_smp_atomic_set(&(PS)->woken, (long) 0)
+#define ERTS_POLLSET_IS_POLLER_WOKEN(PS) \
+  ((int) erts_smp_atomic_read(&(PS)->woken))
+
+#define ERTS_POLLSET_UNSET_INTERRUPTED_CHK(PS) \
+  ((int) erts_smp_atomic_xchg(&(PS)->interrupt, (long) 0))
+#define ERTS_POLLSET_UNSET_INTERRUPTED(PS) \
+  erts_smp_atomic_set(&(PS)->interrupt, (long) 0)
+#define ERTS_POLLSET_SET_INTERRUPTED(PS) \
+  erts_smp_atomic_set(&(PS)->interrupt, (long) 1)
+#define ERTS_POLLSET_IS_INTERRUPTED(PS) \
+  ((int) erts_smp_atomic_read(&(PS)->interrupt))
+
+#else
+
+#define ERTS_POLLSET_LOCK(PS)
+#define ERTS_POLLSET_UNLOCK(PS)
+#define ERTS_POLLSET_SET_POLLED_CHK(PS) 0
+#define ERTS_POLLSET_UNSET_POLLED(PS)
+#define ERTS_POLLSET_IS_POLLED(PS) 0
+#define ERTS_POLLSET_SET_POLLER_WOKEN_CHK(PS) 1
+#define ERTS_POLLSET_SET_POLLER_WOKEN(PS)
+#define ERTS_POLLSET_UNSET_POLLER_WOKEN(PS)
+#define ERTS_POLLSET_IS_POLLER_WOKEN(PS) 1
+
+
+#endif
+
+/*
+ * While atomics are not yet implemented for windows in the common library...
+ *
+ * MSDN doc states that SMP machines and old compilers require
+ * InterLockedExchange to properly read and write interlocked
+ * variables, otherwise the processors might reschedule
+ * the access and order of atomics access is destroyed... 
+ * While they only mention it in white-papers, the problem
+ * in VS2003 is due to the IA64 arch, so we can still count
+ * on the CPU not rescheduling the access to volatile in X86 arch using 
+ * even the slightly older compiler...
+ *
+ * So here's (hopefully) a subset of the generally working atomic 
+ * variable access...
+ */
+
+#if defined(__GNUC__)
+#  if defined(__i386__) || defined(__x86_64__)
+#    define VOLATILE_IN_SEQUENCE 1
+#  else
+#    define VOLATILE_IN_SEQUENCE 0
+#  endif 
+#elif defined(_MSC_VER)
+#  if _MSC_VER < 1300
+#    define VOLATILE_IN_SEQUENCE 0 /* Dont trust really old compilers */
+#  else
+#    if defined(_M_IX86)
+#      define VOLATILE_IN_SEQUENCE 1
+#    else /* I.e. IA64 */
+#      if _MSC_VER >= 1400 
+#        define VOLATILE_IN_SEQUENCE 1
+#      else
+#        define VOLATILE_IN_SEQUENCE 0
+#      endif
+#    endif
+#  endif
+#else
+# define VOLATILE_IN_SEQUENCE 0
+#endif
+ 
+
+
+/*
+ * Communication with sys_interrupt
+ */
+
+#ifdef ERTS_SMP
+extern erts_smp_atomic_t erts_break_requested;
+#define ERTS_SET_BREAK_REQUESTED \
+  erts_smp_atomic_set(&erts_break_requested, (long) 1)
+#define ERTS_UNSET_BREAK_REQUESTED \
+  erts_smp_atomic_set(&erts_break_requested, (long) 0)
+#else
+extern volatile int erts_break_requested;
+#define ERTS_SET_BREAK_REQUESTED (erts_break_requested = 1)
+#define ERTS_UNSET_BREAK_REQUESTED (erts_break_requested = 0)
+#endif
+
+static erts_mtx_t break_waiter_lock;
+static HANDLE break_happened_event;
+static erts_atomic_t break_waiter_state;
+#define BREAK_WAITER_GOT_BREAK 1
+#define BREAK_WAITER_GOT_HALT 2
+
+
+/* 
+ * Forward declarations
+ */
+
+static void *threaded_waiter(void *param);
+static void *break_waiter(void *param);
+
+/*
+ * Sychronization macros and functions
+ */
+#define START_WAITER(PS, w) \
+    SetEvent((w)->go_ahead)
+
+#define STOP_WAITER(PS,w) \
+do { \
+    setup_standby_wait((PS),1); \
+    SetEvent((w)->events[0]); \
+    wait_standby(PS); \
+} while(0)
+
+#define START_WAITERS(PS) \
+do { \
+    int i; \
+    for (i = 0; i < (PS)->num_waiters; i++) { \
+	SetEvent((PS)->waiter[i]->go_ahead); \
+    } \
+ } while(0)
+
+#define STOP_WAITERS(PS) \
+do { \
+    int i; \
+    setup_standby_wait((PS),(PS)->num_waiters); \
+    for (i = 0; i < (PS)->num_waiters; i++) { \
+	SetEvent((PS)->waiter[i]->events[0]); \
+    } \
+    wait_standby(PS); \
+ } while(0)
+
+#if ERTS_POLL_ASYNC_INTERRUPT_SUPPORT && !defined(ERTS_SMP)
+
+static ERTS_INLINE int
+unset_interrupted_chk(ErtsPollSet ps)
+{
+    /* This operation isn't atomic, but we have no need at all for an
+       atomic operation here... */
+    int res = ps->interrupt;
+    ps->interrupt = 0;
+    return res;
+}
+
+#endif
+
+#ifdef ERTS_SMP
+static ERTS_INLINE void
+wake_poller(ErtsPollSet ps)
+{
+    if (!ERTS_POLLSET_SET_POLLER_WOKEN_CHK(ps)) {
+	SetEvent(ps->event_io_ready);
+    }
+}
+#endif
+
+static void setup_standby_wait(ErtsPollSet ps, int num_threads)
+{
+    EnterCriticalSection(&(ps->standby_crit));
+    ps->standby_wait_counter = num_threads;
+    ResetEvent(ps->standby_wait_event);
+    LeaveCriticalSection(&(ps->standby_crit));
+}
+
+static void signal_standby(ErtsPollSet ps) 
+{
+    EnterCriticalSection(&(ps->standby_crit));
+    --(ps->standby_wait_counter);
+    if (ps->standby_wait_counter < 0) {
+	LeaveCriticalSection(&(ps->standby_crit));
+	erl_exit(1,"Standby signalled by more threads than expected");
+    }
+    if (!(ps->standby_wait_counter)) {
+	SetEvent(ps->standby_wait_event);
+    }
+    LeaveCriticalSection(&(ps->standby_crit));
+}
+
+static void wait_standby(ErtsPollSet ps)
+{
+    WaitForSingleObject(ps->standby_wait_event,INFINITE);
+}
+
+static void remove_event_from_set(Waiter *w, int j)
+{
+    w->evdata[j]->event = INVALID_HANDLE_VALUE;
+    w->evdata[j]->mode = 0;
+    w->evdata[j]->next = w->first_free_evdata;
+    w->first_free_evdata = w->evdata[j];
+    
+    /*
+     * If the event is active, we will overwrite it
+     * with the last active event and make the hole
+     * the first non-active event.
+     */
+    
+    if (j < w->active_events) {
+	w->active_events--;
+	w->highwater--;
+	w->total_events--;
+	w->events[j] = w->events[w->active_events];
+	w->evdata[j] = w->evdata[w->active_events];
+	w->events[w->active_events] = w->events[w->highwater];
+	w->evdata[w->active_events] = w->evdata[w->highwater];
+	w->events[w->highwater] =  w->events[w->total_events];
+	w->evdata[w->highwater] =  w->evdata[w->total_events];
+    } else if (j < w->highwater) {
+	w->highwater--;
+	w->total_events--;
+	w->events[j] = w->events[w->highwater];
+	w->evdata[j] = w->evdata[w->highwater];
+	w->events[w->highwater] =  w->events[w->total_events];
+	w->evdata[w->highwater] =  w->evdata[w->total_events];
+    } else {
+	w->total_events--;
+	w->events[j] = w->events[w->total_events];
+	w->evdata[j] = w->evdata[w->total_events];
+    }
+    
+#ifdef DEBUG
+    w->events[w->total_events] = (HANDLE) CleanLandFill;
+    w->evdata[w->total_events] = (EventData *) CleanLandFill;
+    consistency_check(w);
+#endif
+}    
+
+/* 
+ * Thread handling
+ */
+
+#ifdef DEBUG
+static void consistency_check(Waiter* w)
+{
+    int i;
+    
+    ASSERT(w->active_events <= w->total_events);
+    ASSERT(w->evdata[0] == NULL);
+
+    for (i = 1; i < w->total_events; i++) {
+	ASSERT(w->events[i] == w->evdata[i]->event);
+	ASSERT(w->evdata[i]->mode != 0);
+    }
+}
+
+#endif
+
+static void new_waiter(ErtsPollSet ps)
+{
+    register Waiter* w;
+    DWORD tid;			/* Id for thread. */
+    erts_tid_t thread;
+    int i;
+    int tres;
+
+    if (ps->num_waiters == ps->allocated_waiters) {
+	Uint old_size = sizeof(Waiter *)*ps->allocated_waiters;
+	ps->allocated_waiters += 64;
+	ps->waiter = SEL_REALLOC(ERTS_ALC_T_WAITER_OBJ,
+				 (void *) ps->waiter,
+				 old_size,
+				 sizeof(Waiter *) * (ps->allocated_waiters));
+    }
+	
+    w = (Waiter *) SEL_ALLOC(ERTS_ALC_T_WAITER_OBJ, sizeof(Waiter));
+    ps->waiter[ps->num_waiters] = w;
+
+    w->events[0] = CreateAutoEvent(FALSE);
+    w->evdata[0] = NULL;	/* Should never be used. */
+    w->active_events = 1;
+    w->highwater = 1;
+    w->total_events = 1;
+    erts_mtx_init(&w->mtx, "pollwaiter");
+
+
+    /*
+     * Form the free list of EventData objects.
+     */
+
+    w->evdata_heap[0].next = 0;	/* Last in free list. */
+    for (i = 1; i < MAXIMUM_WAIT_OBJECTS; i++) {
+	w->evdata_heap[i].next = w->evdata_heap+i-1;
+    }
+    w->first_free_evdata = w->evdata_heap+MAXIMUM_WAIT_OBJECTS-1;
+
+    /*
+     * Create the other events.
+     */
+
+    w->go_ahead = CreateAutoEvent(FALSE);
+
+    /*
+     * Create the thread.
+     */
+    w->xdata = ps;
+    erts_thr_create(&thread, &threaded_waiter, w, NULL);
+    w->this = thread;
+
+    /*
+     * Finally, done.
+     */
+
+    (ps->num_waiters)++;
+}
+
+static void *break_waiter(void *param)
+{
+    HANDLE harr[2];
+    int i = 0;
+    harr[i++] = erts_sys_break_event;
+    if (erts_service_event != NULL) {
+	harr[i++] = erts_service_event;
+    }
+
+    for(;;) {
+	switch (WaitForMultipleObjects(i,harr,FALSE,INFINITE)) {
+	case WAIT_OBJECT_0:
+	    ResetEvent(harr[0]);
+	    erts_mtx_lock(&break_waiter_lock);
+	    erts_atomic_set(&break_waiter_state,BREAK_WAITER_GOT_BREAK);
+	    SetEvent(break_happened_event);
+	    erts_mtx_unlock(&break_waiter_lock);
+	    break;
+	case (WAIT_OBJECT_0+1):
+	    ResetEvent(harr[1]);
+	    erts_mtx_lock(&break_waiter_lock);
+	    erts_atomic_set(&break_waiter_state,BREAK_WAITER_GOT_HALT);
+	    SetEvent(break_happened_event);
+	    erts_mtx_unlock(&break_waiter_lock);
+	    break;
+	default:
+	    erl_exit(1,"Unexpected event in break_waiter");
+	}
+    }
+}
+		    
+static void *threaded_waiter(void *param)
+{
+    register Waiter* w = (Waiter *) param;
+    ErtsPollSet ps = (ErtsPollSet) w->xdata;
+#ifdef HARD_POLL_DEBUG2
+    HANDLE oold_fired[64];
+    int num_oold_fired;
+    HANDLE old_fired[64];
+    int num_old_fired = 0;
+    HANDLE fired[64];
+    int num_fired = 0;
+    HANDLE errors[1024];
+    int num_errors = 0;
+    HANDLE save_events[64];
+    int save_active_events;
+    int save_total_events;
+    int save_highwater;
+#endif
+
+ again:
+    WaitForSingleObject(w->go_ahead, INFINITE);
+    /* Atomic enough when just checking, skip lock */
+    if (w->total_events == 0) {
+	return NULL;
+    }
+    if (w->active_events == 0) {
+	goto again;
+    }
+    ASSERT(w->evdata[0] == NULL);
+#ifdef HARD_POLL_DEBUG2
+    num_oold_fired = num_old_fired;
+    memcpy(oold_fired,old_fired,num_old_fired*sizeof(HANDLE));
+    num_old_fired = num_fired;
+    memcpy(old_fired,fired,num_fired*sizeof(HANDLE));
+    num_fired = 0;
+#endif
+    for (;;) {
+	int i;
+	int j;
+#ifdef HARD_POLL_DEBUG2
+	erts_mtx_lock(&w->mtx);
+	memcpy(save_events,w->events,w->active_events*sizeof(HANDLE));
+	save_active_events = w->active_events;
+	save_total_events = w->total_events;
+	save_highwater = w->highwater;
+	erts_mtx_unlock(&w->mtx);	
+#endif
+	i = WaitForMultipleObjects(w->active_events, w->events, FALSE, INFINITE);
+	switch (i) {
+	case WAIT_FAILED:
+	    DEBUGF(("Wait failed: %s\n", last_error()));
+	    erts_mtx_lock(&w->mtx);
+	    /* Dont wait for our signal event */
+	    for (j = 1; j < w->active_events; j++) {
+		int tmp;
+		if ((tmp = WaitForSingleObject(w->events[j], 0)) 
+		    == WAIT_FAILED) {
+		    DEBUGF(("Invalid handle: i = %d, handle = 0x%0x\n",
+			    j, w->events[j]));
+#ifdef HARD_POLL_DEBUG2
+		    if (num_errors < 1024)
+			errors[num_errors++] = w->events[j];
+#endif
+#ifdef HARD_POLL_DEBUG
+		    poll_debug_died(w->events[j]);
+#endif
+		    remove_event_from_set(w,j);
+#ifdef DEBUG
+		    consistency_check(w);
+#endif
+		} else if (tmp == WAIT_OBJECT_0) {
+		    i = WAIT_OBJECT_0 + j;
+		    goto event_happened;
+		}
+	    }
+	    erts_mtx_unlock(&w->mtx);
+	    break;
+	case WAIT_OBJECT_0:
+	    signal_standby(ps);
+	    goto again;
+#ifdef DEBUG
+	case WAIT_TIMEOUT:
+	    ASSERT(0);
+#endif
+	default:
+	    erts_mtx_lock(&w->mtx);
+#ifdef HARD_POLL_DEBUG2
+	    {
+		int x = memcmp(save_events,w->events,w->active_events*sizeof(HANDLE));
+		ASSERT(x == 0 && save_active_events == w->active_events);
+	    }
+#endif
+event_happened:
+#ifdef DEBUG
+	    consistency_check(w);
+#endif
+	    ASSERT(WAIT_OBJECT_0 < i && i < WAIT_OBJECT_0+w->active_events);
+	    if (!erts_atomic_xchg(&ps->sys_io_ready,1)) { 
+		HARDDEBUGF(("SET EventIoReady (%d)",erts_atomic_read(&ps->sys_io_ready)));
+		SetEvent(ps->event_io_ready);
+	    } else {
+		HARDDEBUGF(("DONT SET EventIoReady"));
+	    }
+
+	    /* 
+	     * The main thread wont start working on our arrays untill we're
+	     * stopped, so we can work in peace although the main thread runs 
+	     */
+	    ASSERT(i >= WAIT_OBJECT_0+1);
+	    i -= WAIT_OBJECT_0;
+	    ASSERT(i >= 1);
+	    w->active_events--;
+	    HARDDEBUGF(("i = %d, a,h,t = %d,%d,%d",i,
+			w->active_events, w->highwater, w->total_events));
+#ifdef HARD_POLL_DEBUG2
+	    fired[num_fired++] = w->events[i];
+#endif
+#ifdef HARD_POLL_DEBUG
+	    poll_debug_fired(w->events[i]);
+#endif
+	    if (i < w->active_events) {
+		HANDLE te = w->events[i];
+		EventData* tp = w->evdata[i];
+		w->events[i] = w->events[w->active_events];
+		w->evdata[i] = w->evdata[w->active_events];
+		w->events[w->active_events] = te;
+		w->evdata[w->active_events] = tp;
+	    }
+	    HARDDEBUGF(("i = %d, a,h,t = %d,%d,%d",i,
+			w->active_events, w->highwater, w->total_events));
+#ifdef DEBUG
+	    consistency_check(w);
+#endif
+	    erts_mtx_unlock(&w->mtx);	    
+	    break;
+	}
+    }
+}
+
+/*
+ * The actual adding and removing from pollset utilities 
+ */
+
+static int set_driver_select(ErtsPollSet ps, HANDLE event, ErtsPollEvents mode)
+{
+    int i;
+    int best_waiter = -1;	/* The waiter with lowest number of events. */
+    int lowest = MAXIMUM_WAIT_OBJECTS; /* Lowest number of events
+					* in any waiter.
+					*/
+    EventData* ev;
+    Waiter* w;
+
+    /*
+     * Find the waiter which is least busy.
+     */
+
+#ifdef HARD_POLL_DEBUG
+    poll_debug_select(event, mode);
+#endif
+
+    /* total_events can no longer be read without the lock, it's changed in the waiter */
+    for (i = 0; i < ps->num_waiters; i++) {
+	erts_mtx_lock(&(ps->waiter[i]->mtx));
+	if (ps->waiter[i]->total_events < lowest) {
+	    lowest = ps->waiter[i]->total_events;
+	    best_waiter = i;
+	}
+	erts_mtx_unlock(&(ps->waiter[i]->mtx));
+    }
+
+    /*
+     * Stop the selected waiter, or start a new waiter if all were busy.
+     */
+
+    if (best_waiter >= 0) {
+	w = ps->waiter[best_waiter];
+	STOP_WAITER(ps,w);
+	erts_mtx_lock(&w->mtx);
+    } else {
+	new_waiter(ps);
+	w = ps->waiter[(ps->num_waiters)-1];
+	erts_mtx_lock(&w->mtx);
+    }
+
+#ifdef DEBUG
+    consistency_check(w);
+#endif
+
+    /*
+     * Allocate and initialize an EventData structure.
+     */
+
+    ev = w->first_free_evdata;
+    w->first_free_evdata = ev->next;
+    ev->event = event;
+    ev->mode = mode;
+    ev->next = NULL;
+
+    /*
+     * At this point, the selected waiter (newly-created or not) is
+     * standing by.  Put the new event into the active part of the array.
+     */
+
+    if (w->active_events < w->total_events) {
+	/*
+	 * Move the first event beyond the active part of the array to
+	 * the very end to make place for the new event.
+	 */
+
+#ifdef HARD_POLL_DEBUG
+	poll_debug_moved(w->events[w->highwater],w->highwater,w->total_events);
+#endif
+	w->events[w->total_events] = w->events[w->highwater];
+	w->evdata[w->total_events] = w->evdata[w->highwater];
+#ifdef HARD_POLL_DEBUG
+	poll_debug_moved(w->events[w->active_events],w->active_events,w->highwater);
+#endif
+	w->events[w->highwater] = w->events[w->active_events];
+	w->evdata[w->highwater] = w->evdata[w->active_events];
+
+    }
+    w->events[w->active_events] = event;
+    w->evdata[w->active_events] = ev;
+    w->active_events++;
+    w->highwater++;
+    w->total_events++;
+
+#ifdef DEBUG
+    consistency_check(w);
+#endif
+    erts_mtx_unlock(&w->mtx);
+    START_WAITER(ps,w);
+    HARDDEBUGF(("add select %d %d %d %d",best_waiter,
+		w->active_events,w->highwater,w->total_events));
+    return mode;
+}
+
+
+static int cancel_driver_select(ErtsPollSet ps, HANDLE event)
+{
+    int i;
+
+    ASSERT(event != INVALID_HANDLE_VALUE);
+ restart:
+    for (i = 0; i < ps->num_waiters; i++) {
+	Waiter* w = ps->waiter[i];
+	int j;
+
+	erts_mtx_lock(&w->mtx);
+#ifdef DEBUG
+	consistency_check(w);
+#endif
+	for (j = 0; j < w->total_events; j++) {
+	    if (w->events[j] == event) {
+		int stopped = 0;
+		/*
+		 * Free the event's EventData structure.
+		 */
+
+		if (j < w->active_events) {
+		    HARDDEBUGF(("Stopped in remove select"));
+		    stopped = 1;
+		    erts_mtx_unlock(&w->mtx);
+		    STOP_WAITER(ps,w);
+		    erts_mtx_lock(&w->mtx);
+		    if ( j >= w->active_events || w->events[j] != event) {
+			/* things happened while unlocked */
+			START_WAITER(ps,w);
+			erts_mtx_unlock(&w->mtx);
+			goto restart;
+		    }
+		}
+#ifdef HARD_POLL_DEBUG
+		poll_debug_deselect(w->events[j]);
+#endif
+		remove_event_from_set(w, j);
+		if (stopped) {
+		    START_WAITER(ps,w);
+		}
+		HARDDEBUGF(("removed select %d,%d %d %d %d",i,j,
+			    w->active_events,w->highwater,w->total_events));
+		break;
+	    }
+	}
+	erts_mtx_unlock(&w->mtx);
+    }
+    return 0;
+}
+
+/*
+ * Interface functions
+ */
+
+void  erts_poll_interrupt(ErtsPollSet ps, int set /* bool */)
+{
+    HARDTRACEF(("In erts_poll_interrupt(%d)",set));
+#ifdef ERTS_SMP
+    if (set) {
+	ERTS_POLLSET_SET_INTERRUPTED(ps);
+	wake_poller(ps);
+    }
+    else {
+	ERTS_POLLSET_UNSET_INTERRUPTED(ps);
+    }
+#endif
+    HARDTRACEF(("Out erts_poll_interrupt(%d)",set));
+}
+
+void erts_poll_interrupt_timed(ErtsPollSet ps,
+			       int set /* bool */,
+			       long msec)
+{
+    HARDTRACEF(("In erts_poll_interrupt_timed(%d,%ld)",set,msec));
+#ifdef ERTS_SMP
+    if (set) {
+	if (erts_smp_atomic_read(&ps->timeout) > msec) {
+	    ERTS_POLLSET_SET_INTERRUPTED(ps);
+	    wake_poller(ps);
+	}
+    }
+    else {
+	ERTS_POLLSET_UNSET_INTERRUPTED(ps);
+    }
+#endif
+    HARDTRACEF(("Out erts_poll_interrupt_timed"));
+}
+
+
+/*
+ * Windows is special, there is actually only one event type, and
+ * the only difference between ERTS_POLL_EV_IN and ERTS_POLL_EV_OUT
+ * is which driver callback will eventually be called.
+ */
+static ErtsPollEvents do_poll_control(ErtsPollSet ps,
+				      ErtsSysFdType fd,
+				      ErtsPollEvents pe,
+				      int on /* bool */)
+{
+    HANDLE event = (HANDLE) fd;
+    ErtsPollEvents mode;
+    ErtsPollEvents result;
+    ASSERT(event != INVALID_HANDLE_VALUE);
+
+    if (on) {
+	if (pe & ERTS_POLL_EV_IN || !(pe & ERTS_POLL_EV_OUT )) {
+	    mode = ERTS_POLL_EV_IN;
+	} else {
+	    mode = ERTS_POLL_EV_OUT; /* ready output only in this case */
+	}
+	result = set_driver_select(ps, event, mode);
+    } else {
+	result = cancel_driver_select(ps, event);
+    }
+    return result;
+}
+
+ErtsPollEvents erts_poll_control(ErtsPollSet ps,
+				 ErtsSysFdType fd,
+				 ErtsPollEvents pe,
+				 int on,
+				 int* do_wake) /* In: Wake up polling thread */
+				               /* Out: Poller is woken */
+{
+    ErtsPollEvents result;
+    HARDTRACEF(("In erts_poll_control(0x%08X, %u, %d)",(unsigned long) fd, (unsigned) pe, on));
+    ERTS_POLLSET_LOCK(ps);
+    result=do_poll_control(ps,fd,pe,on);
+    ERTS_POLLSET_UNLOCK(ps);
+    *do_wake = 0; /* Never any need to wake polling threads on windows */
+    HARDTRACEF(("Out erts_poll_control -> %u",(unsigned) result));
+    return result;
+}
+
+void erts_poll_controlv(ErtsPollSet ps,
+			ErtsPollControlEntry pcev[],
+			int len)
+{
+    int i;
+    int hshur = 0;
+    int do_wake = 0;
+
+    HARDTRACEF(("In erts_poll_controlv(%d)",len));
+    ERTS_POLLSET_LOCK(ps);
+
+    for (i = 0; i < len; i++) {
+	pcev[i].events = do_poll_control(ps,
+					 pcev[i].fd,
+					 pcev[i].events,
+					 pcev[i].on);
+    }
+    ERTS_POLLSET_LOCK(ps);
+    HARDTRACEF(("Out erts_poll_controlv"));
+}
+
+int erts_poll_wait(ErtsPollSet ps,
+		   ErtsPollResFd pr[],
+		   int *len,
+		   SysTimeval *utvp)
+{
+    SysTimeval *tvp = utvp;
+    SysTimeval itv;
+    int no_fds;
+    DWORD timeout;
+    EventData* ev;
+    int res = 0;
+    int num = 0;
+    int n; 
+    int i;
+    int break_state;
+
+    HARDTRACEF(("In erts_poll_wait"));
+    ERTS_POLLSET_LOCK(ps);
+
+    if (!erts_atomic_read(&ps->sys_io_ready) && ps->restore_events) {
+	HARDDEBUGF(("Restore events: %d",ps->num_waiters));
+	ps->restore_events = 0;
+	for (i = 0; i < ps->num_waiters; ++i) {
+	   Waiter* w = ps->waiter[i];
+	   erts_mtx_lock(&w->mtx);
+	   HARDDEBUGF(("Maybe reset %d %d %d %d",i,
+		       w->active_events,w->highwater,w->total_events));
+	   if (w->active_events < w->total_events) {
+	       erts_mtx_unlock(&w->mtx);
+	       STOP_WAITER(ps,w);
+	       HARDDEBUGF(("Need reset %d %d %d %d",i,
+			   w->active_events,w->highwater,w->total_events));
+	       erts_mtx_lock(&w->mtx);	       
+	       /* Need reset, just check that it doesn't have got more to tell */
+	       if (w->highwater != w->active_events) {
+		   HARDDEBUGF(("Oups!"));
+		   /* Oups, got signalled before we took the lock, can't reset */
+		   if(erts_atomic_read(&ps->sys_io_ready) == 0) {
+		       erl_exit(1,"Internal error: "
+				"Inconsistent io structures in erl_poll.\n");
+		   }
+		   START_WAITER(ps,w);
+		   erts_mtx_unlock(&w->mtx);
+		   ps->restore_events = 1; 
+		   continue;
+	       }
+	       w->active_events = w->highwater =  w->total_events;
+	       START_WAITER(ps,w);
+	       erts_mtx_unlock(&w->mtx);
+	   } else {
+	       erts_mtx_unlock(&w->mtx);
+	   }
+	}
+    }
+
+    no_fds = *len;
+
+#ifdef ERTS_POLL_MAX_RES
+    if (no_fds >= ERTS_POLL_MAX_RES)
+	no_fds = ERTS_POLL_MAX_RES;
+#endif
+
+
+    ResetEvent(ps->event_io_ready);
+    ERTS_POLLSET_UNSET_POLLER_WOKEN(ps);
+
+#ifdef ERTS_SMP
+    if (ERTS_POLLSET_IS_INTERRUPTED(ps)) {
+	/* Interrupt use zero timeout */
+	itv.tv_sec = 0;
+	itv.tv_usec = 0;
+	tvp = &itv;
+    }
+#endif
+
+    timeout = tvp->tv_sec * 1000 + tvp->tv_usec / 1000;
+    /*HARDDEBUGF(("timeout = %ld",(long) timeout));*/
+    erts_smp_atomic_set(&ps->timeout, timeout);
+
+    if (timeout > 0 && ! erts_atomic_read(&ps->sys_io_ready) && ! erts_atomic_read(&break_waiter_state)) {
+	HANDLE harr[2] = {ps->event_io_ready, break_happened_event};
+	int num_h = 2;
+
+	HARDDEBUGF(("Start waiting %d [%d]",num_h, (long) timeout));
+	ERTS_POLLSET_UNLOCK(ps);
+	WaitForMultipleObjects(num_h, harr, FALSE, timeout);
+	ERTS_POLLSET_LOCK(ps);
+	HARDDEBUGF(("Stop waiting %d [%d]",num_h, (long) timeout));
+    }
+
+    ERTS_UNSET_BREAK_REQUESTED;
+    if(erts_atomic_read(&break_waiter_state)) {
+	erts_mtx_lock(&break_waiter_lock);
+	break_state = erts_atomic_read(&break_waiter_state);
+	erts_atomic_set(&break_waiter_state,0);
+	ResetEvent(break_happened_event);
+	erts_mtx_unlock(&break_waiter_lock);
+	switch (break_state) {
+	case  BREAK_WAITER_GOT_BREAK:
+	    ERTS_SET_BREAK_REQUESTED;
+	    break;
+	case  BREAK_WAITER_GOT_HALT:
+	    erl_exit(0,"");
+	    break;
+	default:
+	    break;
+	}
+    }
+
+    ERTS_POLLSET_SET_POLLER_WOKEN(ps);
+
+    if (!erts_atomic_read(&ps->sys_io_ready)) {
+	res = EINTR;
+	HARDDEBUGF(("EINTR!"));
+	goto done; 
+    }
+
+    erts_atomic_set(&ps->sys_io_ready,0);
+
+    n = ps->num_waiters;	
+
+    for (i = 0; i < n; i++) {
+	Waiter* w = ps->waiter[i];
+	int j;
+	int first;
+	int last;
+	erts_mtx_lock(&w->mtx);
+#ifdef DEBUG
+	consistency_check(w);
+#endif
+
+	first = w->active_events;
+	last = w->highwater;
+	w->highwater = w->active_events;
+
+	for (j = last-1; j >= first; --j) {
+	    if (num >= no_fds) {
+		w->highwater=j+1;
+		erts_mtx_unlock(&w->mtx);
+		/* This might mean we still have data to report, set
+		   back the global flag! */
+		erts_atomic_set(&ps->sys_io_ready,1);
+		HARDDEBUGF(("To many FD's to report!"));
+		goto done;
+	    }
+	    HARDDEBUGF(("SET! Restore events"));
+	    ps->restore_events = 1;
+	    HARDDEBUGF(("Report %d,%d",i,j));
+	    pr[num].fd = (ErtsSysFdType) w->events[j];
+	    pr[num].events = w->evdata[j]->mode;
+#ifdef HARD_POLL_DEBUG
+	    poll_debug_reported(w->events[j],w->highwater | (j << 16));
+	    poll_debug_reported(w->events[j],first | (last << 16));
+#endif
+	    ++num;
+	}
+
+#ifdef DEBUG
+	consistency_check(w);
+#endif
+	erts_mtx_unlock(&w->mtx);
+    }
+ done:
+    erts_smp_atomic_set(&ps->timeout, LONG_MAX);
+    *len = num;
+    ERTS_POLLSET_UNLOCK(ps);
+    HARDTRACEF(("Out erts_poll_wait"));
+    return res;
+
+}
+
+int erts_poll_max_fds(void)
+{
+    int res = sys_max_files();
+    HARDTRACEF(("In/Out erts_poll_max_fds -> %d",res));
+    return res;
+}
+
+void erts_poll_info(ErtsPollSet ps,
+		    ErtsPollInfo *pip)
+{
+    Uint size = 0;
+    Uint num_events = 0;
+    int i;
+
+    HARDTRACEF(("In erts_poll_info"));
+    ERTS_POLLSET_LOCK(ps);
+
+    size += sizeof(struct ErtsPollSet_);
+    size += sizeof(Waiter *) * ps->allocated_waiters;
+    for (i = 0; i < ps->num_waiters; ++i) {
+	Waiter *w = ps->waiter[i];
+	if (w != NULL) {
+	    size += sizeof(Waiter);
+	    erts_mtx_lock(&w->mtx);
+	    size += sizeof(EventData) * w->total_events;
+	    num_events += (w->total_events - 1); /* First event is internal */
+	    erts_mtx_unlock(&w->mtx);
+	}
+    }
+
+    pip->primary = "WaitForMultipleObjects"; 
+
+    pip->fallback = NULL;
+
+    pip->kernel_poll = NULL;
+
+    pip->memory_size = size;
+
+    pip->poll_set_size = num_events;
+
+    pip->fallback_poll_set_size = 0;
+
+    pip->lazy_updates = 0;
+
+    pip->pending_updates = 0;
+
+    pip->batch_updates = 0;
+
+    pip->concurrent_updates = 0;
+    ERTS_POLLSET_UNLOCK(ps);
+
+    pip->max_fds = erts_poll_max_fds();
+    HARDTRACEF(("Out erts_poll_info"));
+
+}
+
+ErtsPollSet erts_poll_create_pollset(void)
+{
+    ErtsPollSet ps = SEL_ALLOC(ERTS_ALC_T_POLLSET,
+			       sizeof(struct ErtsPollSet_));
+    HARDTRACEF(("In erts_poll_create_pollset"));
+
+    ps->num_waiters = 0;
+    ps->allocated_waiters = 64;
+    ps->waiter = SEL_ALLOC(ERTS_ALC_T_WAITER_OBJ,
+			   sizeof(Waiter *)*ps->allocated_waiters);
+    InitializeCriticalSection(&(ps->standby_crit));
+    ps->standby_wait_counter = 0;
+    ps->event_io_ready = CreateManualEvent(FALSE);
+    ps->standby_wait_event = CreateManualEvent(FALSE); 
+    erts_atomic_init(&ps->sys_io_ready,0);
+    ps->restore_events = 0;
+
+#ifdef ERTS_SMP
+    erts_smp_atomic_init(&ps->woken, 0);
+    erts_smp_mtx_init(&ps->mtx, "pollset");
+    erts_smp_atomic_init(&ps->interrupt, 0);
+#endif
+    erts_smp_atomic_init(&ps->timeout, LONG_MAX);
+
+    HARDTRACEF(("Out erts_poll_create_pollset"));
+    return ps;
+}
+
+void erts_poll_destroy_pollset(ErtsPollSet ps)
+{
+    int i;
+    HARDTRACEF(("In erts_poll_destroy_pollset"));
+    ERTS_POLLSET_LOCK(ps);
+    STOP_WAITERS(ps);
+    for (i=0;i<ps->num_waiters;++i) {
+	Waiter *w = ps->waiter[i];
+	void *dummy;
+	erts_tid_t t = w->this;
+	/* Assume we're alone, no locking here... */
+	w->active_events = w->total_events = w->highwater = 0;
+	START_WAITER(ps,w);
+	erts_thr_join(t,&dummy);
+	CloseHandle(w->go_ahead);
+	CloseHandle(w->events[0]);
+	erts_mtx_destroy(&w->mtx);
+	SEL_FREE(ERTS_ALC_T_WAITER_OBJ, (void *) w);
+    }
+    SEL_FREE(ERTS_ALC_T_WAITER_OBJ,ps->waiter);
+    CloseHandle(ps->event_io_ready);
+    CloseHandle(ps->standby_wait_event);
+    ERTS_POLLSET_UNLOCK(ps);
+#ifdef ERTS_SMP
+    erts_smp_mtx_destroy(&ps->mtx);
+#endif
+    SEL_FREE(ERTS_ALC_T_POLLSET, (void *) ps);
+    HARDTRACEF(("Out erts_poll_destroy_pollset"));
+}
+
+/*
+ * Actually mostly initializes the friend module sys_interrupt...
+ */
+void  erts_poll_init(void)
+{
+    erts_tid_t thread;
+
+#ifdef HARD_POLL_DEBUG
+    poll_debug_init();
+#endif
+
+    HARDTRACEF(("In erts_poll_init"));
+    erts_sys_break_event = CreateManualEvent(FALSE);
+
+    erts_mtx_init(&break_waiter_lock,"break_waiter_lock");
+    break_happened_event = CreateManualEvent(FALSE);
+    erts_atomic_init(&break_waiter_state, 0); 
+
+    erts_thr_create(&thread, &break_waiter, NULL, NULL);
+    ERTS_UNSET_BREAK_REQUESTED;
+    HARDTRACEF(("Out erts_poll_init"));
+}
+
+/*
+ * Non windows friendly interface, not used when fd's are not continous
+ */
+void  erts_poll_get_selected_events(ErtsPollSet ps,
+				    ErtsPollEvents ev[],
+				    int len)
+{
+    int i;
+    HARDTRACEF(("In erts_poll_get_selected_events"));
+    for (i = 0; i < len; ++i)
+	ev[i] = 0;
+    HARDTRACEF(("Out erts_poll_get_selected_events"));
+}