[Win32] Add win32 threading & synchronization.

* Modified camutex_pthreads
* Modified cathreadpool_pthreads
* Modified camutex_t; use win32 GetSystemTime() in getAbsTime()

Change-Id: I9793a784398c29dab4ed838ff43b041c62120cc3
Signed-off-by: Daniel Ferguson <daniel.j.ferguson@intel.com>
Reviewed-on: https://gerrit.iotivity.org/gerrit/5517
Tested-by: jenkins-iotivity <jenkins-iotivity@opendaylight.org>
Reviewed-by: David Antler <david.a.antler@intel.com>

diff --git a/resource/csdk/connectivity/common/src/camutex_pthreads.c b/resource/csdk/connectivity/common/src/camutex_pthreads.c
index 95ee2eb..f0e73b6 100644 (file)
--- a/resource/csdk/connectivity/common/src/camutex_pthreads.c
+++ b/resource/csdk/connectivity/common/src/camutex_pthreads.c
@@ -36,6 +36,7 @@
 #define _POSIX_C_SOURCE 200809L
 #endif
 
+#if !defined(_WIN32)
 #include <string.h>
 #include <pthread.h>
 #include <errno.h>
@@ -44,9 +45,20 @@
 #include <sys/time.h>
 #include <assert.h>
 #include <oic_malloc.h>
+#include "camutex.h"
+#include "logger.h"
+#endif
 
+#if defined(_WIN32)
+#include <string.h>
+#include <time.h>
+#include <assert.h>
+#include <oic_malloc.h>
 #include "camutex.h"
 #include "logger.h"
+#include <winsock2.h>
+#include <stdio.h>
+#endif
 
 /**
  * TAG
@@ -76,13 +88,21 @@ static const uint64_t NANOSECS_PER_SEC      = 1000000000L;
 
 typedef struct _tagMutexInfo_t
 {
+#if defined(_WIN32)
+    CRITICAL_SECTION mutex;
+#else
     pthread_mutex_t mutex;
+#endif
 } ca_mutex_internal;
 
 typedef struct _tagEventInfo_t
 {
+#if defined(_WIN32)
+    CONDITION_VARIABLE cond;
+#else
     pthread_cond_t cond;
     pthread_condattr_t condattr;
+#endif
 } ca_cond_internal;
 
 ca_mutex ca_mutex_new(void)
@@ -91,6 +111,10 @@ ca_mutex ca_mutex_new(void)
     ca_mutex_internal *mutexInfo = (ca_mutex_internal*) OICMalloc(sizeof(ca_mutex_internal));
     if (NULL != mutexInfo)
     {
+#if defined(_WIN32)
+        InitializeCriticalSection(&mutexInfo->mutex);
+        retVal = (ca_mutex)mutexInfo;
+#else
         // create the mutex with the attributes set
         int ret=pthread_mutex_init(&(mutexInfo->mutex), PTHREAD_MUTEX_DEFAULT);
         if (0 == ret)
@@ -102,6 +126,11 @@ ca_mutex ca_mutex_new(void)
             OIC_LOG_V(ERROR, TAG, "%s Failed to initialize mutex !", __func__);
             OICFree(mutexInfo);
         }
+#endif
+    }
+    else
+    {
+        OIC_LOG_V(ERROR, TAG, "%s Failed to allocate mutex!", __func__);
     }
 
     return retVal;
@@ -114,6 +143,11 @@ bool ca_mutex_free(ca_mutex mutex)
     ca_mutex_internal *mutexInfo = (ca_mutex_internal*) mutex;
     if (mutexInfo)
     {
+#if defined(_WIN32)
+        DeleteCriticalSection(&mutexInfo->mutex);
+        OICFree(mutexInfo);
+        bRet=true;
+#else
         int ret = pthread_mutex_destroy(&mutexInfo->mutex);
         if (0 == ret)
         {
@@ -124,6 +158,7 @@ bool ca_mutex_free(ca_mutex mutex)
         {
             OIC_LOG_V(ERROR, TAG, "%s Failed to free mutex !", __func__);
         }
+#endif
     }
     else
     {
@@ -138,12 +173,16 @@ void ca_mutex_lock(ca_mutex mutex)
     ca_mutex_internal *mutexInfo = (ca_mutex_internal*) mutex;
     if (mutexInfo)
     {
+#if defined(_WIN32)
+        EnterCriticalSection(&mutexInfo->mutex);
+#else
         int ret = pthread_mutex_lock(&mutexInfo->mutex);
         if(ret != 0)
         {
             OIC_LOG_V(ERROR, TAG, "Pthread Mutex lock failed: %d", ret);
             exit(ret);
         }
+#endif
     }
     else
     {
@@ -157,6 +196,9 @@ void ca_mutex_unlock(ca_mutex mutex)
     ca_mutex_internal *mutexInfo = (ca_mutex_internal*) mutex;
     if (mutexInfo)
     {
+#if defined(_WIN32)
+        LeaveCriticalSection(&mutexInfo->mutex);
+#else
         int ret = pthread_mutex_unlock(&mutexInfo->mutex);
         if(ret != 0)
         {
@@ -164,11 +206,12 @@ void ca_mutex_unlock(ca_mutex mutex)
             exit(ret);
         }
         (void)ret;
+#endif
     }
     else
     {
-          OIC_LOG_V(ERROR, TAG, "%s: Invalid mutex !", __func__);
-          return;
+        OIC_LOG_V(ERROR, TAG, "%s: Invalid mutex !", __func__);
+        return;
     }
 }
 
@@ -178,6 +221,10 @@ ca_cond ca_cond_new(void)
     ca_cond_internal *eventInfo = (ca_cond_internal*) OICMalloc(sizeof(ca_cond_internal));
     if (NULL != eventInfo)
     {
+#if defined(_WIN32)
+        InitializeConditionVariable(&eventInfo->cond);
+        retVal = (ca_cond) eventInfo;
+#else
         int ret = pthread_condattr_init(&(eventInfo->condattr));
         if(0 != ret)
         {
@@ -187,14 +234,14 @@ ca_cond ca_cond_new(void)
             return retVal;
         }
 
-#if (defined(__ANDROID__) || _POSIX_TIMERS > 0) && !defined(__msys_nt__)
-#ifdef __ANDROID__
+ #if defined(__ANDROID__) || _POSIX_TIMERS > 0
+ #ifdef __ANDROID__
         if (camutex_condattr_setclock) {
             ret = camutex_condattr_setclock(&(eventInfo->condattr), CLOCK_MONOTONIC);
-#else
+ #else
         {
             ret = pthread_condattr_setclock(&(eventInfo->condattr), CLOCK_MONOTONIC);
-#endif /*  __ANDROID__ */
+ #endif /*  __ANDROID__ */
             if(0 != ret)
             {
                 OIC_LOG_V(ERROR, TAG, "%s: Failed to set condition variable clock %d!",
@@ -204,7 +251,7 @@ ca_cond ca_cond_new(void)
                 return retVal;
             }
         }
-#endif /* (defined(__ANDROID__) || _POSIX_TIMERS > 0) && !defined(__msys_nt__)*/
+ #endif /* defined(__ANDROID__) || _POSIX_TIMERS > 0 */
         ret = pthread_cond_init(&(eventInfo->cond), &(eventInfo->condattr));
         if (0 == ret)
         {
@@ -216,6 +263,11 @@ ca_cond ca_cond_new(void)
             pthread_condattr_destroy(&(eventInfo->condattr));
             OICFree(eventInfo);
         }
+#endif
+    }
+    else
+    {
+        OIC_LOG_V(ERROR, TAG, "%s: Failed to allocate condition variable!", __func__);
     }
 
     return retVal;
@@ -226,6 +278,9 @@ void ca_cond_free(ca_cond cond)
     ca_cond_internal *eventInfo = (ca_cond_internal*) cond;
     if (eventInfo != NULL)
     {
+#if defined(_WIN32)
+        OICFree(cond);
+#else
         int ret = pthread_cond_destroy(&(eventInfo->cond));
         int ret2 = pthread_condattr_destroy(&(eventInfo->condattr));
         if (0 == ret && 0 == ret2)
@@ -237,6 +292,7 @@ void ca_cond_free(ca_cond cond)
             OIC_LOG_V(ERROR, TAG, "%s: Failed to destroy condition variable %d, %d",
                     __func__, ret, ret2);
         }
+#endif
     }
     else
     {
@@ -249,11 +305,15 @@ void ca_cond_signal(ca_cond cond)
     ca_cond_internal *eventInfo = (ca_cond_internal*) cond;
     if (eventInfo != NULL)
     {
+#if defined(_WIN32)
+        WakeConditionVariable(&eventInfo->cond);
+#else
         int ret = pthread_cond_signal(&(eventInfo->cond));
         if (0 != ret)
         {
             OIC_LOG_V(ERROR, TAG, "%s: Failed to signal condition variable", __func__);
         }
+#endif
     }
     else
     {
@@ -266,11 +326,15 @@ void ca_cond_broadcast(ca_cond cond)
     ca_cond_internal* eventInfo = (ca_cond_internal*) cond;
     if (eventInfo != NULL)
     {
+#if defined(_WIN32)
+        WakeAllConditionVariable(&eventInfo->cond);
+#else
         int ret = pthread_cond_broadcast(&(eventInfo->cond));
         if (0 != ret)
         {
             OIC_LOG_V(ERROR, TAG, "%s: failed to signal condition variable", __func__);
         }
+#endif
     }
     else
     {
@@ -283,6 +347,14 @@ void ca_cond_wait(ca_cond cond, ca_mutex mutex)
     ca_cond_wait_for(cond, mutex, 0L);
 }
 
+#ifndef TIMEVAL_TO_TIMESPEC
+#define TIMEVAL_TO_TIMESPEC(tv, ts) {               \
+    (ts)->tv_sec = (tv)->tv_sec;                    \
+    (ts)->tv_nsec = (tv)->tv_usec * 1000;           \
+}
+#endif
+
+#if !defined(_WIN32)
 struct timespec ca_get_current_time()
 {
 #if defined(__ANDROID__) || _POSIX_TIMERS > 0
@@ -308,6 +380,7 @@ void ca_add_microseconds_to_timespec(struct timespec* ts, uint64_t microseconds)
     ts->tv_nsec = (totalNs)% NANOSECS_PER_SEC;
     ts->tv_sec += secPart + secOfNs;
 }
+#endif
 
 CAWaitResult_t ca_cond_wait_for(ca_cond cond, ca_mutex mutex, uint64_t microseconds)
 {
@@ -330,6 +403,25 @@ CAWaitResult_t ca_cond_wait_for(ca_cond cond, ca_mutex mutex, uint64_t microseco
 
     if (microseconds > 0)
     {
+#if defined(_WIN32)
+        // Wait for the given time
+        DWORD milli = (DWORD)(microseconds / 1000);
+        if (!SleepConditionVariableCS(&eventInfo->cond, &mutexInfo->mutex, milli))
+        {
+            if (GetLastError() == ERROR_TIMEOUT)
+            {
+                retVal = CA_WAIT_TIMEDOUT;
+            }
+            else
+            {
+                OIC_LOG_V(ERROR, TAG, "SleepConditionVariableCS() with Timeout failed %i", GetLastError());
+                retVal = CA_WAIT_INVAL;
+            }
+        }else
+        {
+            retVal = CA_WAIT_SUCCESS;
+        }
+#else
         int ret;
         struct timespec abstime;
 
@@ -367,14 +459,26 @@ CAWaitResult_t ca_cond_wait_for(ca_cond cond, ca_mutex mutex, uint64_t microseco
                 retVal = CA_WAIT_INVAL;
                 break;
         }
+#endif
     }
     else
     {
+#if defined(_WIN32)
+        // Wait forever
+        if (!SleepConditionVariableCS(&eventInfo->cond, &mutexInfo->mutex, INFINITE))
+        {
+            OIC_LOG_V(ERROR, TAG, "SleepConditionVariableCS() w/o Timeout failed %i", GetLastError());
+            retVal = CA_WAIT_INVAL;
+        }else
+        {
+            retVal = CA_WAIT_SUCCESS;
+        }
+#else
         // Wait forever
         int ret = pthread_cond_wait(&eventInfo->cond, &mutexInfo->mutex);
         retVal = ret == 0 ? CA_WAIT_SUCCESS : CA_WAIT_INVAL;
+#endif
     }
-
     return retVal;
 }
 

diff --git a/resource/csdk/connectivity/common/src/cathreadpool_pthreads.c b/resource/csdk/connectivity/common/src/cathreadpool_pthreads.c
index 0fb57de..3ca8cd6 100644 (file)
--- a/resource/csdk/connectivity/common/src/cathreadpool_pthreads.c
+++ b/resource/csdk/connectivity/common/src/cathreadpool_pthreads.c
@@ -28,7 +28,13 @@
 #define _GNU_SOURCE
 #endif
 #include <errno.h>
+#if !defined(__msys_nt__)
 #include <pthread.h>
+#endif
+
+#if defined(__msys_nt__)
+#include <windows.h>
+#endif
 #include "cathreadpool.h"
 #include "logger.h"
 #include "oic_malloc.h"
@@ -155,8 +161,17 @@ CAResult_t ca_thread_pool_add_task(ca_thread_pool_t thread_pool, ca_thread_func
     info->func = method;
     info->data = data;
 
+#if defined(__msys_nt__)
+    HANDLE threadHandle;
+    DWORD threadId;
+    threadHandle = CreateThread(NULL, 0, ca_thread_pool_pthreads_delegate, info, 0, &threadId);
+    if (threadHandle == NULL)
+    {
+        OIC_LOG_V(ERROR, TAG, "CreateThread failed with error %i", GetLastError());
+        return CA_STATUS_FAILED;
+    }
+#else
     pthread_t threadHandle;
-
     int result = pthread_create(&threadHandle, NULL, ca_thread_pool_pthreads_delegate, info);
 
     if(result != 0)
@@ -164,7 +179,7 @@ CAResult_t ca_thread_pool_add_task(ca_thread_pool_t thread_pool, ca_thread_func
         OIC_LOG_V(ERROR, TAG, "Thread start failed with error %d", result);
         return CA_STATUS_FAILED;
     }
-
+#endif
     ca_mutex_lock(thread_pool->details->list_lock);
     bool addResult = u_arraylist_add(thread_pool->details->threads_list, (void*)threadHandle);
     ca_mutex_unlock(thread_pool->details->list_lock);
@@ -193,12 +208,21 @@ void ca_thread_pool_free(ca_thread_pool_t thread_pool)
 
     for(uint32_t i = 0; i<u_arraylist_length(thread_pool->details->threads_list); ++i)
     {
+#if defined(__msys_nt__)
+        HANDLE tid = (HANDLE)u_arraylist_get(thread_pool->details->threads_list, i);
+        DWORD joinres = WaitForSingleObject(tid, INFINITE);
+        if(WAIT_OBJECT_0 != joinres)
+        {
+            OIC_LOG_V(ERROR, TAG, "Failed to join thread at index %u with error %d", i, joinres);
+        }
+#else
         pthread_t tid = (pthread_t)u_arraylist_get(thread_pool->details->threads_list, i);
         int joinres = pthread_join(tid, NULL);
         if(0 != joinres)
         {
             OIC_LOG_V(ERROR, TAG, "Failed to join thread at index %u with error %d", i, joinres);
         }
+#endif
     }
 
     u_arraylist_free(&(thread_pool->details->threads_list));

diff --git a/resource/csdk/connectivity/test/camutex_tests.cpp b/resource/csdk/connectivity/test/camutex_tests.cpp
index 1c3f34e..66e6116 100644 (file)
--- a/resource/csdk/connectivity/test/camutex_tests.cpp
+++ b/resource/csdk/connectivity/test/camutex_tests.cpp
@@ -68,6 +68,10 @@ uint64_t getAbsTime()
     struct timespec ts;
     clock_gettime(CLOCK_MONOTONIC, &ts);
     currentTime = ts.tv_sec * USECS_PER_SEC + ts.tv_nsec / 1000;
+#elif defined(_WIN32)
+    SYSTEMTIME time;
+    GetSystemTime(&time);
+    currentTime = time.wSecond * 100000 + time.wMilliseconds * 1000;
 #else
     struct timeval tv;
     gettimeofday(&tv, NULL);