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