Mostly working on Windows

cm.c

@@ -232,7 +232,7 @@ static thr_thread_t
 thr_fork(void*(*func)(void*), void *arg)
 {
     thr_thread_t new_thread = 0;
-    int err = thr_thread_create(&new_thread, NULL, (void*(*)(void*))func, arg);
+    int err = thr_thread_create(&new_thread, NULL, (void*)func, arg);
     if (err != 0) {
 	return (thr_thread_t) (intptr_t)NULL;
     } else {
@@ -258,7 +258,7 @@ INT_CMfork_comm_thread(CManager cm)
 	    if (server_thread ==  (thr_thread_t)(intptr_t) NULL) {
 		return 0;
 	    }
-	    cm->control_list->server_thread = server_thread;
+	    cm->control_list->server_thread = thr_get_thread_id(server_thread);
 	    cm->control_list->has_thread = 1;
 	    cm->reference_count++;
 	    CMtrace_out(cm, CMFreeVerbose, "Forked - CManager %p ref count now %d\n", 
@@ -1053,7 +1053,7 @@ CManager_free(CManager cm)
      new_list->select_data = NULL;
      new_list->add_select = NULL;
      new_list->remove_select = NULL;
-     new_list->server_thread =  (thr_thread_t)(intptr_t) NULL;
+     new_list->server_thread =  (thr_thread_id)0;
      new_list->network_blocking_function.func = NULL;
      new_list->network_polling_function.func = NULL;
      new_list->polling_function_list = NULL;
@@ -2088,6 +2088,7 @@ timeout_conn(CManager cm, void *client_data)
      if (cm_preread_hook) {
 	 do_read = cm_preread_hook(buffer_full_point - buffer_data_end, tmp_message_buffer);
      }
+     CMtrace_out(cm, CMLowLevelVerbose, "P5\n");
      if (do_read) {
 	 if (trans->read_to_buffer_func) {
 	 /* 
@@ -3851,7 +3852,7 @@ CM_init_select(CMControlList cl, CManager cm)
 	}
 	CMtrace_out(cm, CMLowLevelVerbose,
 		    "CM - Forked comm thread %p\n", (void*)(intptr_t)server_thread);
-	cm->control_list->server_thread = server_thread;
+	cm->control_list->server_thread = thr_get_thread_id(server_thread);
 	cm->control_list->cl_reference_count++;
 	cm->control_list->free_reference_count++;
 	cl->has_thread = 1;

cm_control.c

@@ -242,7 +242,7 @@ INT_CMCondition_wait(CManager cm, int condition)
 	       (void*)(intptr_t)cl->server_thread);
     }
     if (!cl->has_thread) {
-	if ((cl->server_thread ==  (thr_thread_t) (intptr_t) NULL) || (cl->server_thread == thr_thread_self())) {
+	if ((cl->server_thread ==  (thr_thread_id) 0) || (cl->server_thread == thr_thread_self())) {
 	    cl->cond_polling = 1;
 	    while (!(cond->signaled || cond->failed)) {
 		if (cm_control_debug_flag) {
@@ -255,7 +255,7 @@ INT_CMCondition_wait(CManager cm, int condition)
 		fprintf(cm->CMTrace_file, "CMLowLevel  after Polling for CMcondition %d\n", condition);
 	    }
 	    /* the poll and handle will set cl->server_thread, restore it */
-	    cl->server_thread =  (thr_thread_t) (intptr_t)NULL;
+	    cl->server_thread =  (thr_thread_id) (intptr_t)NULL;
 	    if (cm_control_debug_flag) {
 		fprintf(cm->CMTrace_file, "CMLowLevel  In condition wait, reset server thread = %lx\n", 
 		       (long)cl->server_thread);

cm_internal.h

@@ -34,24 +34,35 @@
 #else
 //#include <mutex>
 #include <Windows.h>
-#define thr_mutex_t HANDLE
-#define thr_thread_t DWORD
-#define thr_condition_t HANDLE
-#define thr_thread_create(w,x,y,z) 0
+extern int win_thread_create(HANDLE* w, void* x, void* y, void* z);
+extern void win_mutex_init(SRWLOCK *m);
+extern void win_mutex_lock(SRWLOCK* m);
+extern void win_mutex_unlock(SRWLOCK* m);
+extern void win_mutex_free(SRWLOCK* m);
+extern void win_condition_init(CONDITION_VARIABLE *c);
+extern void win_condition_wait(CONDITION_VARIABLE *c, SRWLOCK *m);
+extern void win_condition_signal(CONDITION_VARIABLE *c);
+extern void win_condition_free(CONDITION_VARIABLE *c);
+#define thr_mutex_t SRWLOCK
+#define thr_thread_t HANDLE
+#define thr_thread_id DWORD
+#define thr_condition_t CONDITION_VARIABLE
+#define thr_thread_create(w,x,y,z) win_thread_create(w,x,y,z)
 #define thr_thread_self() GetCurrentThreadId()
-#define thr_thread_exit(status) 
+#define thr_thread_exit(status) ExitThread((DWORD)(intptr_t)status)
+#define thr_get_thread_id(t) GetThreadId(t)
 #define thr_thread_detach(thread) 
 #define thr_thread_yield() 
 #define thr_thread_join(t, s) (void)s
-#define thr_mutex_init(m) 
-#define thr_mutex_lock(m)
-#define thr_mutex_unlock(m)
-#define thr_mutex_free(m)
-#define thr_condition_init(c)
-#define thr_condition_wait(c, m)
-#define thr_condition_signal(c)
-#define thr_condition_broadcast(c) 
-#define thr_condition_free(c) 
+#define thr_mutex_init(m) win_mutex_init(&m)
+#define thr_mutex_lock(m) win_mutex_lock(&m)
+#define thr_mutex_unlock(m) win_mutex_unlock(&m)
+#define thr_mutex_free(m) win_mutex_free(&m)
+#define thr_condition_init(c) win_condition_init(&c)
+#define thr_condition_wait(c, m) win_condition_wait(&c, &m)
+#define thr_condition_signal(c) win_condition_signal(&c)
+#define thr_condition_broadcast(c) error
+#define thr_condition_free(c) win_condition_free(&c)
 #endif
 
 #include <ev_internal.h>
@@ -240,7 +251,7 @@ typedef struct _CMControlList {
     int closed;
     int has_thread;
     int cond_polling;
-    thr_thread_t server_thread;
+    thr_thread_id server_thread;
 } CMControlList_s;
 
 struct queued_data_rec {

cm_threadio.c

@@ -46,7 +46,7 @@ static thr_thread_t
 thr_fork(void*(*func)(void*), void *arg)
 {
     thr_thread_t new_thread = 0;
-    int err = thr_thread_create(&new_thread, NULL, (void*(*)(void*))func, arg);
+    int err = thr_thread_create(&new_thread, 0, (void*)func, arg);
     if (err != 0) {
 	return  (thr_thread_t) (intptr_t) NULL;
     } else {

cm_util.c

@@ -1,5 +1,4 @@
 #include "config.h"
-#ifndef MODULE
 
 #include <stdlib.h>
 #ifdef HAVE_UNISTD_H
@@ -12,11 +11,7 @@
 #include <varargs.h>
 #endif
 #include <errno.h>
-#else
-#include "kernel/kcm.h"
-#include "kernel/cm_kernel.h"
-#include "kernel/library.h"
-#endif
+
 #include "atl.h"
 #include "evpath.h"
 #include "chr_time.h"
@@ -297,3 +292,50 @@ INT_CMfree(void *ptr)
     free(ptr);
 }
 
+#ifdef _MSC_VER
+#include <windows.h>
+#include <tchar.h>
+#include <strsafe.h>
+int
+win_thread_create(HANDLE *threadp, void *attr, void *startfunc, void *arg)
+{
+    HANDLE h = CreateThread(NULL, 0, (LPTHREAD_START_ROUTINE)startfunc, arg, 0, attr);
+    (*threadp) = h;
+    return 0;
+}
+void win_mutex_init(SRWLOCK *m)
+{
+    InitializeSRWLock(m);
+}
+
+
+void win_mutex_lock(SRWLOCK *m)
+{
+    AcquireSRWLockExclusive(m);
+}
+void win_mutex_unlock(SRWLOCK *m)
+{
+    ReleaseSRWLockExclusive(m);
+}
+void win_mutex_free(SRWLOCK *m)
+{
+  // nothing necessary
+}
+extern void win_condition_init(CONDITION_VARIABLE *c)
+{
+    InitializeConditionVariable(c);
+}
+extern void win_condition_wait(CONDITION_VARIABLE *c, SRWLOCK* m)
+{
+    SleepConditionVariableSRW(c, m, INFINITE, 0);
+}
+
+extern void win_condition_signal(CONDITION_VARIABLE *c)
+{
+    WakeConditionVariable(c);
+}
+extern void win_condition_free(CONDITION_VARIABLE *c)
+{
+  // nothing necessary
+}
+#endif

cmselect.c

@@ -109,7 +109,7 @@ typedef struct func_list_item {
 } FunctionListElement;
 
 typedef struct select_data {
-    thr_thread_t server_thread;
+    thr_thread_id server_thread;
 
     void *fdset;		/* bitmap of the fds for read select */
     void *write_set;		/* bitmap of the fds for write select */
@@ -153,7 +153,7 @@ init_select_data(CMtrans_services svc, select_data_ptr *sdp, CManager cm)
     EVPATH_FD_ZERO((fd_set *) sd->fdset);
     sd->write_set = svc->malloc_func(sizeof(fd_set));
     EVPATH_FD_ZERO((fd_set *) sd->write_set);
-    sd->server_thread =  (thr_thread_t)(intptr_t) NULL;
+    sd->server_thread =  (thr_thread_id)(intptr_t) NULL;
     sd->closed = 0;
     sd->sel_item_max = 0;
     sd->select_items = (FunctionListElement *) svc->malloc_func(sizeof(FunctionListElement));
@@ -179,7 +179,7 @@ init_select_data(CMtrans_services svc, select_data_ptr *sdp, CManager cm)
 typedef struct _periodic_task {
     int period_sec;
     int period_usec;
-    thr_thread_t executing;
+    thr_thread_id executing;
     struct timeval next_time;
     select_list_func func;
     void *arg1;
@@ -221,7 +221,7 @@ set_soonest_timeout(struct timeval *timeout, periodic_task_handle task_list, str
     if (task_list == NULL) return;
     this_delay.tv_sec = task_list->next_time.tv_sec - now.tv_sec;
     this_delay.tv_usec = task_list->next_time.tv_usec - now.tv_usec;
-    if (task_list->executing == (thr_thread_t)-1) {
+    if (task_list->executing == (thr_thread_id)-1) {
 	/* this task not executing already, see when it needs to run  */
 	if (this_delay.tv_usec < 0) {
 	    this_delay.tv_sec--;
@@ -260,15 +260,15 @@ socket_select(CMtrans_services svc, select_data_ptr sd, int timeout_sec, int tim
     int tmp_select_consistency_number = sd->select_consistency_number;
 
     if (sd->closed) {
-	sd->server_thread =  (thr_thread_t)(intptr_t) NULL; 
+	sd->server_thread =  (thr_thread_id)(intptr_t) NULL; 
 	return;
     }
 
     if (sd->cm) {
 	/* assert CM is locked */
 	assert(CM_LOCKED(svc, sd->cm));
     }
-    if (sd->server_thread ==  (thr_thread_t)(intptr_t) NULL) {
+    if (sd->server_thread ==  (thr_thread_id)(intptr_t) NULL) {
 	/* no server thread set, must be this one */
 	sd->server_thread = thr_thread_self();
     }
@@ -321,7 +321,7 @@ socket_select(CMtrans_services svc, select_data_ptr sd, int timeout_sec, int tim
 	ACQUIRE_CM_LOCK(svc, sd->cm);
     }
     if (sd->closed) {
-	sd->server_thread =  (thr_thread_t)(intptr_t) NULL; 
+	sd->server_thread =  (thr_thread_id)(intptr_t) NULL; 
 	return;
     }
 #ifndef HAVE_WINDOWS_H
@@ -440,7 +440,7 @@ socket_select(CMtrans_services svc, select_data_ptr sd, int timeout_sec, int tim
     if (res != 0) {
 	for (i = 0; i <= sd->sel_item_max; i++) {
 	    if (sd->closed) {
-		sd->server_thread = (thr_thread_t)(intptr_t) NULL;
+		sd->server_thread = (thr_thread_id)(intptr_t) NULL;
 		return;
 	    }
 	    if (FD_ISSET(i, &wr_set)) {
@@ -496,14 +496,14 @@ socket_select(CMtrans_services svc, select_data_ptr sd, int timeout_sec, int tim
 		increment_time(&this_periodic_task->next_time,
 			       this_periodic_task->period_sec,
 			       this_periodic_task->period_usec);
-		if (this_periodic_task->executing == (thr_thread_t)-1) {
+		if (this_periodic_task->executing == (thr_thread_id)-1) {
 		    this_periodic_task->executing = thr_thread_self();
 		    DROP_CM_LOCK(svc, sd->cm);
 		    this_periodic_task->func(this_periodic_task->arg1,
 					     this_periodic_task->arg2);
 		    ACQUIRE_CM_LOCK(svc, sd->cm);
 		    next = this_periodic_task->next;
-		    this_periodic_task->executing = (thr_thread_t) -1;
+		    this_periodic_task->executing = (thr_thread_id) -1;
 		    if ((this_periodic_task->period_sec == 0) &&
 			(this_periodic_task->period_usec == 0)) {
 		        remove_periodic_task(sd, this_periodic_task);
@@ -652,7 +652,7 @@ libcmselect_LTX_add_periodic(CMtrans_services svc, select_data_ptr *sdp, int int
     }
     handle->period_sec = interval_sec;
     handle->period_usec = interval_usec;
-    handle->executing = (thr_thread_t) -1;
+    handle->executing = (thr_thread_id) -1;
 #ifndef HAVE_WINDOWS_H
     gettimeofday(&handle->next_time, NULL);
 #else
@@ -699,7 +699,7 @@ libcmselect_LTX_add_delayed_task(CMtrans_services svc, select_data_ptr *sdp, int
     }
     handle->period_sec = 0;
     handle->period_usec = 0;
-    handle->executing = (thr_thread_t) -1;
+    handle->executing = (thr_thread_id) -1;
 #ifndef HAVE_WINDOWS_H
     gettimeofday(&handle->next_time, NULL);
 #else
@@ -751,7 +751,7 @@ remove_periodic_task(select_data_ptr sd, periodic_task_handle handle)
     if (handle->executing != thr_thread_self()) {
 	/* someone besides us executing this ? */
         int i = 0;
-	while (handle->executing != (thr_thread_t)-1) {
+	while (handle->executing != (thr_thread_id)-1) {
 	    /* wait until they're done */
 	    thr_thread_yield();
 	    i++;

cmsockets.c

@@ -879,6 +879,26 @@ set_block_state(CMtrans_services svc, socket_conn_data_ptr scd,
 		       scd->fd);
     }
 #else
+    if ((needed_block_state == Block) && (scd->block_state == Non_Block)) {
+      u_long mode = 0;  // 0 to enable blocking socket
+      int ret = ioctlsocket(scd->fd, FIONBIO, &mode);
+      scd->block_state = Block;
+      if (ret != NO_ERROR)
+	printf("ioctlsocket failed with error: %ld\n", ret);
+
+      svc->trace_out(scd->sd->cm, "CMSocket switch fd %d to blocking WIN properly",
+		     scd->fd);
+    } else if ((needed_block_state == Non_Block) && 
+	       (scd->block_state == Block)) {
+      u_long mode = 1;  // 1 to enable non-blocking socket
+      int ret = ioctlsocket(scd->fd, FIONBIO, &mode);
+      if (ret != NO_ERROR)
+	printf("ioctlsocket failed with error: %ld\n", ret);
+
+      scd->block_state = Non_Block;
+      svc->trace_out(scd->sd->cm, "CMSocket switch fd %d to nonblocking WIN properly",
+		     scd->fd);
+    }
 #endif
 }
 
@@ -909,7 +929,8 @@ libcmsockets_LTX_read_to_buffer_func(CMtrans_services svc, socket_conn_data_ptr
     iget = read(scd->fd, (char *) buffer, (int)requested_len);
     if ((iget == -1) || (iget == 0)) {
 	int lerrno = errno;
-	if ((lerrno != EWOULDBLOCK) &&
+	if ((lerrno != 0) &&
+	    (lerrno != EWOULDBLOCK) &&
 	    (lerrno != EAGAIN) &&
 	    (lerrno != EINTR)) {
 	    /* serious error */

ev_internal.h

@@ -234,7 +234,7 @@ typedef struct _EVclient_sources {
 
 typedef struct _ev_handler_activation_rec {
     struct _ev_handler_activation_rec *prev;
-    thr_thread_t thread_id;
+    thr_thread_id thread_id;
     EVstone stone_id;
     struct _ev_handler_activation_rec *next;
 } ev_handler_activation_rec, *ev_handler_activation_ptr;

evp.c

@@ -1632,7 +1632,7 @@ pop_activation_record_from_stack(CManager cm, ev_handler_activation_ptr rec)
 {
     event_path_data evp = cm->evp;
     ev_handler_activation_ptr tmp;
-    thr_thread_t self = thr_thread_self();
+    thr_thread_id self = thr_thread_self();
     if (!evp->activation_stack) {
 	printf("Activation stack inconsistency!  No records!\n"); 
 	return;
@@ -1663,7 +1663,7 @@ find_activation_record_from_stack(CManager cm)
 {
     event_path_data evp = cm->evp;
     ev_handler_activation_ptr tmp;
-    thr_thread_t self = thr_thread_self();
+    thr_thread_id self = thr_thread_self();
     tmp = evp->activation_stack;
     while(tmp) {
 	if (tmp->thread_id == self) {

mtests/cmconn.c

@@ -26,13 +26,14 @@
 #include <arpa/inet.h>
 #endif
 #include "evpath.h"
-#include <sys/wait.h>
 
 #ifdef _MSC_VER
 #define drand48() (((double)rand())/((double)RAND_MAX))
 #define lrand48() rand()
 #define srand48(x)
 #define kill(x,y) TerminateProcess(OpenProcess(0,0,(DWORD)x),y)
+#else
+#include <sys/wait.h>
 #endif
 typedef struct _msg_rec {
     char *contact_list;