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_WINSOCK2_H
  33 #include <winsock2.h>
  34 #endif
  35 #include "cathreadpool.h"
  36 #include "logger.h"
  37 #include "oic_malloc.h"
  38 #include "uarraylist.h"
  39 #include "octhread.h"
  40 #include "ocrandom.h"
  41 #include "platform_features.h"
  42
  43 #define TAG PCF("UTHREADPOOL")
  44
  45 /**
  46  * empty struct to represent the details.  This implementation has no data
  47  * that it needs to keep track of, so it only uses NULL for the internal value.
  48  */
  49 typedef struct ca_thread_pool_details_t
  50 {
  51     u_arraylist_t* threads_list;
  52     oc_mutex list_lock;
  53 } ca_thread_pool_details_t;
  54
  55 /**
  56  * struct to wrap the pthreads callback properly.  The function pointer for
  57  * pthreads requires a void* return value, however u_thread_func is a void.
  58  */
  59 typedef struct ca_thread_pool_callback_info_t
  60 {
  61     ca_thread_func func;
  62     void* data;
  63 } ca_thread_pool_callback_info_t;
  64
  65 typedef struct ca_thread_pool_thread_info_t
  66 {
  67     oc_thread thread;
  68     uint32_t taskId;
  69 } ca_thread_pool_thread_info_t;
  70
  71 // passthrough function to convert the pthreads call to a u_thread_func call
  72 void* ca_thread_pool_pthreads_delegate(void* data)
  73 {
  74     ca_thread_pool_callback_info_t* info = (ca_thread_pool_callback_info_t*)data;
  75     info->func(info->data);
  76     OICFree(info);
  77     return NULL;
  78 }
  79
  80 // this implementation doesn't do a thread pool, so this function is essentially
  81 // a no-op besides creating a valid ca_thread_pool_t object.  It was determined after
  82 // reading through the existing implementation that the thread-pooling was unnecessary
  83 // for the posix platforms.  Behavior shouldn't be changed since previously num_of_threads
  84 // was greater than the number of requested threads.
  85 CAResult_t ca_thread_pool_init(int32_t num_of_threads, ca_thread_pool_t *thread_pool)
  86 {
  87     OIC_LOG(DEBUG, TAG, "IN");
  88
  89     if(!thread_pool)
  90     {
  91         OIC_LOG(ERROR, TAG, "Parameter thread_pool was null!");
  92         return CA_STATUS_INVALID_PARAM;
  93     }
  94
  95     if(num_of_threads <= 0)
  96     {
  97         OIC_LOG(ERROR, TAG, "num_of_threads must be positive and non-zero");
  98         return CA_STATUS_INVALID_PARAM;
  99     }
 100
 101     *thread_pool = OICMalloc(sizeof(struct ca_thread_pool));
 102
 103     if(!*thread_pool)
 104     {
 105         OIC_LOG(ERROR, TAG, "Failed to allocate for thread-pool");
 106         return CA_MEMORY_ALLOC_FAILED;
 107     }
 108
 109     (*thread_pool)->details = OICMalloc(sizeof(struct ca_thread_pool_details_t));
 110     if(!(*thread_pool)->details)
 111     {
 112         OIC_LOG(ERROR, TAG, "Failed to allocate for thread-pool details");
 113         OICFree(*thread_pool);
 114         *thread_pool=NULL;
 115         return CA_MEMORY_ALLOC_FAILED;
 116     }
 117
 118     (*thread_pool)->details->list_lock = oc_mutex_new();
 119
 120     if(!(*thread_pool)->details->list_lock)
 121     {
 122         OIC_LOG(ERROR, TAG, "Failed to create thread-pool mutex");
 123         goto exit;
 124     }
 125
 126     (*thread_pool)->details->threads_list = u_arraylist_create();
 127
 128     if(!(*thread_pool)->details->threads_list)
 129     {
 130         OIC_LOG(ERROR, TAG, "Failed to create thread-pool list");
 131         if(!oc_mutex_free((*thread_pool)->details->list_lock))
 132         {
 133             OIC_LOG(ERROR, TAG, "Failed to free thread-pool mutex");
 134         }
 135         goto exit;
 136     }
 137
 138     OIC_LOG(DEBUG, TAG, "OUT");
 139     return CA_STATUS_OK;
 140
 141 exit:
 142     OICFree((*thread_pool)->details);
 143     OICFree(*thread_pool);
 144     *thread_pool = NULL;
 145     return CA_STATUS_FAILED;
 146 }
 147
 148 #ifndef __TIZENRT__
 149 CAResult_t ca_thread_pool_add_task(ca_thread_pool_t thread_pool, ca_thread_func method, void *data,
 150                                    uint32_t *taskId)
 151 #else
 152 CAResult_t ca_thread_pool_add_task(ca_thread_pool_t thread_pool, ca_thread_func method, void *data,
 153                                    uint32_t *taskId, const char *task_name, int stack_size)
 154 #endif
 155 {
 156     OIC_LOG_V(DEBUG, TAG, "In %s", __func__);
 157
 158     if(NULL == thread_pool || NULL == method)
 159     {
 160         OIC_LOG(ERROR, TAG, "thread_pool or method was NULL");
 161         return CA_STATUS_INVALID_PARAM;
 162     }
 163
 164     ca_thread_pool_callback_info_t* info = OICMalloc(sizeof(ca_thread_pool_callback_info_t));
 165     if(!info)
 166     {
 167         OIC_LOG(ERROR, TAG, "Failed to allocate for memory wrapper");
 168         return CA_MEMORY_ALLOC_FAILED;
 169     }
 170
 171     info->func = method;
 172     info->data = data;
 173
 174     ca_thread_pool_thread_info_t *threadInfo =
 175             (ca_thread_pool_thread_info_t *) OICCalloc(1, sizeof(ca_thread_pool_thread_info_t));
 176     if (!threadInfo)
 177     {
 178         OIC_LOG(ERROR, TAG, "Memory allocation failed");
 179         OICFree(info);
 180         return CA_STATUS_FAILED;
 181     }
 182     threadInfo->taskId = OCGetRandom();
 183     if (taskId)
 184     {
 185         *taskId = threadInfo->taskId;
 186     }
 187
 188     oc_mutex_lock(thread_pool->details->list_lock);
 189     bool addResult = u_arraylist_add(thread_pool->details->threads_list, (void*) threadInfo);
 190     if (!addResult)
 191     {
 192         // Note that this is considered non-fatal.
 193         oc_mutex_unlock(thread_pool->details->list_lock);
 194         OIC_LOG(ERROR, TAG, "Arraylist add failed");
 195         OICFree(info);
 196         OICFree(threadInfo);
 197         return CA_STATUS_FAILED;
 198     }
 199
 200 #ifndef __TIZENRT__
 201     int thrRet = oc_thread_new(&threadInfo->thread, ca_thread_pool_pthreads_delegate, info);
 202 #else
 203     int thrRet = oc_thread_new(&threadInfo->thread, ca_thread_pool_pthreads_delegate, info,
 204                                task_name, stack_size);
 205 #endif
 206     if (thrRet != 0)
 207     {
 208         uint32_t index = 0;
 209         if (u_arraylist_get_index(thread_pool->details->threads_list, threadInfo, &index))
 210         {
 211             u_arraylist_remove(thread_pool->details->threads_list, index);
 212         }
 213         oc_mutex_unlock(thread_pool->details->list_lock);
 214         OIC_LOG_V(ERROR, TAG, "Thread start failed with error %d", thrRet);
 215         OICFree(info);
 216         return CA_STATUS_FAILED;
 217     }
 218     OIC_LOG_V(INFO, TAG, "created thread: %p, taskId: %u", threadInfo->thread, threadInfo->taskId);
 219     oc_mutex_unlock(thread_pool->details->list_lock);
 220
 221     OIC_LOG_V(DEBUG, TAG, "Out %s", __func__);
 222     return CA_STATUS_OK;
 223 }
 224
 225 CAResult_t ca_thread_pool_remove_task(ca_thread_pool_t thread_pool, uint32_t taskId)
 226 {
 227     OIC_LOG_V(DEBUG, TAG, "In %s", __func__);
 228
 229     if (!thread_pool)
 230     {
 231         OIC_LOG(ERROR, TAG, "Invalid parameter thread_pool was NULL");
 232         return CA_STATUS_FAILED;
 233     }
 234
 235     oc_mutex_lock(thread_pool->details->list_lock);
 236     for (uint32_t i = 0; i < u_arraylist_length(thread_pool->details->threads_list); ++i)
 237     {
 238         ca_thread_pool_thread_info_t *threadInfo = (ca_thread_pool_thread_info_t *)
 239                 u_arraylist_get(thread_pool->details->threads_list, i);
 240         if (threadInfo)
 241         {
 242             if (threadInfo->taskId == taskId)
 243             {
 244                 OIC_LOG_V(INFO, TAG, "waiting.. thread: %p, taskId: %u", threadInfo->thread,
 245                           threadInfo->taskId);
 246                 oc_thread_wait(threadInfo->thread);
 247
 248                 OIC_LOG_V(INFO, TAG, "removed.. thread: %p, taskId: %u", threadInfo->thread,
 249                           threadInfo->taskId);
 250                 u_arraylist_remove(thread_pool->details->threads_list, i);
 251                 oc_thread_free(threadInfo->thread);
 252                 OICFree(threadInfo);
 253                 break;
 254             }
 255         }
 256     }
 257     oc_mutex_unlock(thread_pool->details->list_lock);
 258
 259     OIC_LOG_V(DEBUG, TAG, "Out %s", __func__);
 260     return CA_STATUS_OK;
 261 }
 262
 263 void ca_thread_pool_free(ca_thread_pool_t thread_pool)
 264 {
 265     OIC_LOG_V(DEBUG, TAG, "In %s", __func__);
 266
 267     if (!thread_pool)
 268     {
 269         OIC_LOG(ERROR, TAG, "Invalid parameter thread_pool was NULL");
 270         return;
 271     }
 272
 273     oc_mutex_lock(thread_pool->details->list_lock);
 274
 275     for (uint32_t i = 0; i < u_arraylist_length(thread_pool->details->threads_list); ++i)
 276     {
 277         ca_thread_pool_thread_info_t *threadInfo = (ca_thread_pool_thread_info_t *)
 278                 u_arraylist_get(thread_pool->details->threads_list, i);
 279         if (threadInfo)
 280         {
 281             if (threadInfo->thread)
 282             {
 283 #ifdef __TIZEN__
 284                 OIC_LOG_V(INFO, TAG, "canceling.. thread: %p, taskId: %u", threadInfo->thread,
 285                           threadInfo->taskId);
 286                 oc_thread_cancel(threadInfo->thread);
 287 #endif
 288                 OIC_LOG_V(INFO, TAG, "waiting.. thread: %p, taskId: %u", threadInfo->thread,
 289                           threadInfo->taskId);
 290                 oc_thread_wait(threadInfo->thread);
 291                 oc_thread_free(threadInfo->thread);
 292             }
 293             OICFree(threadInfo);
 294         }
 295     }
 296
 297     u_arraylist_free(&(thread_pool->details->threads_list));
 298
 299     oc_mutex_unlock(thread_pool->details->list_lock);
 300     oc_mutex_free(thread_pool->details->list_lock);
 301
 302     OICFree(thread_pool->details);
 303     OICFree(thread_pool);
 304
 305     OIC_LOG_V(DEBUG, TAG, "Out %s", __func__);
 306 }