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 "iotivity_config.h"
  31 #include <errno.h>
  32 #if defined HAVE_PTHREAD_H
  33 #include <pthread.h>
  34 #endif
  35 #if defined HAVE_WINSOCK2_H
  36 #include <winsock2.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 #include "platform_features.h"
  44
  45 #define TAG PCF("UTHREADPOOL")
  46
  47 /**
  48  * empty struct to represent the details.  This implementation has no data
  49  * that it needs to keep track of, so it only uses NULL for the internal value.
  50  */
  51 typedef struct ca_thread_pool_details_t
  52 {
  53     u_arraylist_t* threads_list;
  54     ca_mutex list_lock;
  55 } ca_thread_pool_details_t;
  56
  57 /**
  58  * struct to wrap the pthreads callback properly.  The function pointer for
  59  * pthreads requires a void* return value, however u_thread_func is a void.
  60  */
  61 typedef struct ca_thread_pool_callback_info_t
  62 {
  63     ca_thread_func func;
  64     void* data;
  65 } ca_thread_pool_callback_info_t;
  66
  67 // passthrough function to convert the pthreads call to a u_thread_func call
  68 void* ca_thread_pool_pthreads_delegate(void* data)
  69 {
  70     ca_thread_pool_callback_info_t* info = (ca_thread_pool_callback_info_t*)data;
  71     info->func(info->data);
  72     OICFree(info);
  73     return NULL;
  74 }
  75
  76 // this implementation doesn't do a thread pool, so this function is essentially
  77 // a no-op besides creating a valid ca_thread_pool_t object.  It was determined after
  78 // reading through the existing implementation that the thread-pooling was unnecessary
  79 // for the posix platforms.  Behavior shouldn't be changed since previously num_of_threads
  80 // was greater than the number of requested threads.
  81 CAResult_t ca_thread_pool_init(int32_t num_of_threads, ca_thread_pool_t *thread_pool)
  82 {
  83     OIC_LOG(DEBUG, TAG, "IN");
  84
  85     if(!thread_pool)
  86     {
  87         OIC_LOG(ERROR, TAG, "Parameter thread_pool was null!");
  88         return CA_STATUS_INVALID_PARAM;
  89     }
  90
  91     if(num_of_threads <= 0)
  92     {
  93         OIC_LOG(ERROR, TAG, "num_of_threads must be positive and non-zero");
  94         return CA_STATUS_INVALID_PARAM;
  95     }
  96
  97     *thread_pool = OICMalloc(sizeof(struct ca_thread_pool));
  98
  99     if(!*thread_pool)
 100     {
 101         OIC_LOG(ERROR, TAG, "Failed to allocate for thread-pool");
 102         return CA_MEMORY_ALLOC_FAILED;
 103     }
 104
 105     (*thread_pool)->details = OICMalloc(sizeof(struct ca_thread_pool_details_t));
 106     if(!(*thread_pool)->details)
 107     {
 108         OIC_LOG(ERROR, TAG, "Failed to allocate for thread-pool details");
 109         OICFree(*thread_pool);
 110         *thread_pool=NULL;
 111         return CA_MEMORY_ALLOC_FAILED;
 112     }
 113
 114     (*thread_pool)->details->list_lock = ca_mutex_new();
 115
 116     if(!(*thread_pool)->details->list_lock)
 117     {
 118         OIC_LOG(ERROR, TAG, "Failed to create thread-pool mutex");
 119         goto exit;
 120     }
 121
 122     (*thread_pool)->details->threads_list = u_arraylist_create();
 123
 124     if(!(*thread_pool)->details->threads_list)
 125     {
 126         OIC_LOG(ERROR, TAG, "Failed to create thread-pool list");
 127         if(!ca_mutex_free((*thread_pool)->details->list_lock))
 128         {
 129             OIC_LOG(ERROR, TAG, "Failed to free thread-pool mutex");
 130         }
 131         goto exit;
 132     }
 133
 134     OIC_LOG(DEBUG, TAG, "OUT");
 135     return CA_STATUS_OK;
 136
 137 exit:
 138     OICFree((*thread_pool)->details);
 139     OICFree(*thread_pool);
 140     *thread_pool = NULL;
 141     return CA_STATUS_FAILED;
 142 }
 143
 144 CAResult_t ca_thread_pool_add_task(ca_thread_pool_t thread_pool, ca_thread_func method,
 145                                     void *data)
 146 {
 147     OIC_LOG(DEBUG, TAG, "IN");
 148
 149     if(NULL == thread_pool || NULL == method)
 150     {
 151         OIC_LOG(ERROR, TAG, "thread_pool or method was NULL");
 152         return CA_STATUS_INVALID_PARAM;
 153     }
 154
 155     ca_thread_pool_callback_info_t* info = OICMalloc(sizeof(ca_thread_pool_callback_info_t));
 156     if(!info)
 157     {
 158         OIC_LOG(ERROR, TAG, "Failed to allocate for memory wrapper");
 159         return CA_MEMORY_ALLOC_FAILED;
 160     }
 161
 162     info->func = method;
 163     info->data = data;
 164
 165     pthread_t threadHandle;
 166     int result = pthread_create(&threadHandle, NULL, ca_thread_pool_pthreads_delegate, info);
 167
 168     if(result != 0)
 169     {
 170         OIC_LOG_V(ERROR, TAG, "Thread start failed with error %d", result);
 171         return CA_STATUS_FAILED;
 172     }
 173
 174     ca_mutex_lock(thread_pool->details->list_lock);
 175     bool addResult = u_arraylist_add(thread_pool->details->threads_list, (void*)threadHandle);
 176     ca_mutex_unlock(thread_pool->details->list_lock);
 177
 178     if (!addResult)
 179     {
 180         OIC_LOG_V(ERROR, TAG, "Arraylist Add failed, may not be properly joined: %d", addResult);
 181 #if defined(_WIN32)
 182         DWORD joinres = WaitForSingleObject(threadHandle, INFINITE);
 183         if (WAIT_OBJECT_0 != joinres)
 184         {
 185             OIC_LOG_V(ERROR, TAG, "Failed to join thread with error %d", joinres);
 186         }
 187         CloseHandle(threadHandle);
 188 #else
 189         int joinres = pthread_join(threadHandle, NULL);
 190         if (0 != joinres)
 191         {
 192             OIC_LOG_V(ERROR, TAG, "Failed to join thread with error %d", joinres);
 193         }
 194 #endif
 195         return CA_STATUS_FAILED;
 196     }
 197
 198     OIC_LOG(DEBUG, TAG, "OUT");
 199     return CA_STATUS_OK;
 200 }
 201
 202 void ca_thread_pool_free(ca_thread_pool_t thread_pool)
 203 {
 204     OIC_LOG(DEBUG, TAG, "IN");
 205
 206     if(!thread_pool)
 207     {
 208         OIC_LOG(ERROR, TAG, "Invalid parameter thread_pool was NULL");
 209         return;
 210     }
 211
 212     ca_mutex_lock(thread_pool->details->list_lock);
 213
 214     for(uint32_t i = 0; i<u_arraylist_length(thread_pool->details->threads_list); ++i)
 215     {
 216         pthread_t tid = (pthread_t)u_arraylist_get(thread_pool->details->threads_list, i);
 217 #if defined(_WIN32)
 218         DWORD joinres = WaitForSingleObject(tid, INFINITE);
 219         if (WAIT_OBJECT_0 != joinres)
 220         {
 221             OIC_LOG_V(ERROR, TAG, "Failed to join thread at index %u with error %d", i, joinres);
 222         }
 223         CloseHandle(tid);
 224 #else
 225         int joinres = pthread_join(tid, NULL);
 226         if(0 != joinres)
 227         {
 228             OIC_LOG_V(ERROR, TAG, "Failed to join thread at index %u with error %d", i, joinres);
 229         }
 230 #endif
 231     }
 232
 233     u_arraylist_free(&(thread_pool->details->threads_list));
 234
 235     ca_mutex_unlock(thread_pool->details->list_lock);
 236     ca_mutex_free(thread_pool->details->list_lock);
 237
 238     OICFree(thread_pool->details);
 239     OICFree(thread_pool);
 240
 241     OIC_LOG(DEBUG, TAG, "OUT");
 242 }