resource/csdk/connectivity/common/src/cathreadpool_pthreads.c

   1 /* ****************************************************************
   2  *
   3  * Copyright 2014 Samsung Electronics All Rights Reserved.
   4  *
   5  *
   6  *
   7  * Licensed under the Apache License, Version 2.0 (the "License");
   8  * you may not use this file except in compliance with the License.
   9  * You may obtain a copy of the License at
  10  *
  11  *      http://www.apache.org/licenses/LICENSE-2.0
  12  *
  13  * Unless required by applicable law or agreed to in writing, software
  14  * distributed under the License is distributed on an "AS IS" BASIS,
  15  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  16  * See the License for the specific language governing permissions and
  17  * limitations under the License.
  18  *
  19  ******************************************************************/
  20
  21 /**
  22  * @file
  23  *
  24  * This file provides APIs related to thread pool.
  25  */
  26
  27 #ifndef _GNU_SOURCE
  28 #define _GNU_SOURCE
  29 #endif
  30 #include <errno.h>
  31 #if !defined(__msys_nt__)
  32 #include <pthread.h>
  33 #endif
  34
  35 #if defined(__msys_nt__)
  36 #include <windows.h>
  37 #endif
  38 #include "cathreadpool.h"
  39 #include "logger.h"
  40 #include "oic_malloc.h"
  41 #include "uarraylist.h"
  42 #include "camutex.h"
  43
  44 #define TAG PCF("UTHREADPOOL")
  45
  46 /**
  47  * empty struct to represent the details.  This implementation has no data
  48  * that it needs to keep track of, so it only uses NULL for the internal value.
  49  */
  50 typedef struct ca_thread_pool_details_t
  51 {
  52     u_arraylist_t* threads_list;
  53     ca_mutex list_lock;
  54 } ca_thread_pool_details_t;
  55
  56 /**
  57  * struct to wrap the pthreads callback properly.  The function pointer for
  58  * pthreads requires a void* return value, however u_thread_func is a void.
  59  */
  60 typedef struct ca_thread_pool_callback_info_t
  61 {
  62     ca_thread_func func;
  63     void* data;
  64 } ca_thread_pool_callback_info_t;
  65
  66 // passthrough function to convert the pthreads call to a u_thread_func call
  67 void* ca_thread_pool_pthreads_delegate(void* data)
  68 {
  69     ca_thread_pool_callback_info_t* info = (ca_thread_pool_callback_info_t*)data;
  70     info->func(info->data);
  71     OICFree(info);
  72     return NULL;
  73 }
  74
  75 // this implementation doesn't do a thread pool, so this function is essentially
  76 // a no-op besides creating a valid ca_thread_pool_t object.  It was determined after
  77 // reading through the existing implementation that the thread-pooling was unnecessary
  78 // for the posix platforms.  Behavior shouldn't be changed since previously num_of_threads
  79 // was greater than the number of requested threads.
  80 CAResult_t ca_thread_pool_init(int32_t num_of_threads, ca_thread_pool_t *thread_pool)
  81 {
  82     OIC_LOG(DEBUG, TAG, "IN");
  83
  84     if(!thread_pool)
  85     {
  86         OIC_LOG(ERROR, TAG, "Parameter thread_pool was null!");
  87         return CA_STATUS_INVALID_PARAM;
  88     }
  89
  90     if(num_of_threads <= 0)
  91     {
  92         OIC_LOG(ERROR, TAG, "num_of_threads must be positive and non-zero");
  93         return CA_STATUS_INVALID_PARAM;
  94     }
  95
  96     *thread_pool = OICMalloc(sizeof(struct ca_thread_pool));
  97
  98     if(!*thread_pool)
  99     {
 100         OIC_LOG(ERROR, TAG, "Failed to allocate for thread-pool");
 101         return CA_MEMORY_ALLOC_FAILED;
 102     }
 103
 104     (*thread_pool)->details = OICMalloc(sizeof(struct ca_thread_pool_details_t));
 105     if(!(*thread_pool)->details)
 106     {
 107         OIC_LOG(ERROR, TAG, "Failed to allocate for thread-pool details");
 108         OICFree(*thread_pool);
 109         *thread_pool=NULL;
 110         return CA_MEMORY_ALLOC_FAILED;
 111     }
 112
 113     (*thread_pool)->details->list_lock = ca_mutex_new();
 114
 115     if(!(*thread_pool)->details->list_lock)
 116     {
 117         OIC_LOG(ERROR, TAG, "Failed to create thread-pool mutex");
 118         goto exit;
 119     }
 120
 121     (*thread_pool)->details->threads_list = u_arraylist_create();
 122
 123     if(!(*thread_pool)->details->threads_list)
 124     {
 125         OIC_LOG(ERROR, TAG, "Failed to create thread-pool list");
 126         if(!ca_mutex_free((*thread_pool)->details->list_lock))
 127         {
 128             OIC_LOG(ERROR, TAG, "Failed to free thread-pool mutex");
 129         }
 130         goto exit;
 131     }
 132
 133     OIC_LOG(DEBUG, TAG, "OUT");
 134     return CA_STATUS_OK;
 135
 136 exit:
 137     OICFree((*thread_pool)->details);
 138     OICFree(*thread_pool);
 139     *thread_pool = NULL;
 140     return CA_STATUS_FAILED;
 141 }
 142
 143 CAResult_t ca_thread_pool_add_task(ca_thread_pool_t thread_pool, ca_thread_func method,
 144                                     void *data)
 145 {
 146     OIC_LOG(DEBUG, TAG, "IN");
 147
 148     if(NULL == thread_pool || NULL == method)
 149     {
 150         OIC_LOG(ERROR, TAG, "thread_pool or method was NULL");
 151         return CA_STATUS_INVALID_PARAM;
 152     }
 153
 154     ca_thread_pool_callback_info_t* info = OICMalloc(sizeof(ca_thread_pool_callback_info_t));
 155     if(!info)
 156     {
 157         OIC_LOG(ERROR, TAG, "Failed to allocate for memory wrapper");
 158         return CA_MEMORY_ALLOC_FAILED;
 159     }
 160
 161     info->func = method;
 162     info->data = data;
 163
 164 #if defined(__msys_nt__)
 165     HANDLE threadHandle;
 166     DWORD threadId;
 167     threadHandle = CreateThread(NULL, 0, ca_thread_pool_pthreads_delegate, info, 0, &threadId);
 168     if (threadHandle == NULL)
 169     {
 170         OIC_LOG_V(ERROR, TAG, "CreateThread failed with error %i", GetLastError());
 171         return CA_STATUS_FAILED;
 172     }
 173 #else
 174     pthread_t threadHandle;
 175     int result = pthread_create(&threadHandle, NULL, ca_thread_pool_pthreads_delegate, info);
 176
 177     if(result != 0)
 178     {
 179         OIC_LOG_V(ERROR, TAG, "Thread start failed with error %d", result);
 180         return CA_STATUS_FAILED;
 181     }
 182 #endif
 183     ca_mutex_lock(thread_pool->details->list_lock);
 184     bool addResult = u_arraylist_add(thread_pool->details->threads_list, (void*)threadHandle);
 185     ca_mutex_unlock(thread_pool->details->list_lock);
 186
 187     if(!addResult)
 188     {
 189         OIC_LOG_V(ERROR, TAG, "Arraylist Add failed, may not be properly joined: %d", addResult);
 190         return CA_STATUS_FAILED;
 191     }
 192
 193     OIC_LOG(DEBUG, TAG, "OUT");
 194     return CA_STATUS_OK;
 195 }
 196
 197 void ca_thread_pool_free(ca_thread_pool_t thread_pool)
 198 {
 199     OIC_LOG(DEBUG, TAG, "IN");
 200
 201     if(!thread_pool)
 202     {
 203         OIC_LOG(ERROR, TAG, "Invalid parameter thread_pool was NULL");
 204         return;
 205     }
 206
 207     ca_mutex_lock(thread_pool->details->list_lock);
 208
 209     for(uint32_t i = 0; i<u_arraylist_length(thread_pool->details->threads_list); ++i)
 210     {
 211 #if defined(__msys_nt__)
 212         HANDLE tid = (HANDLE)u_arraylist_get(thread_pool->details->threads_list, i);
 213         DWORD joinres = WaitForSingleObject(tid, INFINITE);
 214         if(WAIT_OBJECT_0 != joinres)
 215         {
 216             OIC_LOG_V(ERROR, TAG, "Failed to join thread at index %u with error %d", i, joinres);
 217         }
 218 #else
 219         pthread_t tid = (pthread_t)u_arraylist_get(thread_pool->details->threads_list, i);
 220         int joinres = pthread_join(tid, NULL);
 221         if(0 != joinres)
 222         {
 223             OIC_LOG_V(ERROR, TAG, "Failed to join thread at index %u with error %d", i, joinres);
 224         }
 225 #endif
 226     }
 227
 228     u_arraylist_free(&(thread_pool->details->threads_list));
 229
 230     ca_mutex_unlock(thread_pool->details->list_lock);
 231     ca_mutex_free(thread_pool->details->list_lock);
 232
 233     OICFree(thread_pool->details);
 234     OICFree(thread_pool);
 235
 236     OIC_LOG(DEBUG, TAG, "OUT");
 237 }