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 typedef struct ca_thread_pool_thread_info_t
  65 {
  66     oc_thread thread;
  67 } ca_thread_pool_thread_info_t;
  68
  69 // passthrough function to convert the pthreads call to a u_thread_func call
  70 void* ca_thread_pool_pthreads_delegate(void* data)
  71 {
  72     ca_thread_pool_callback_info_t* info = (ca_thread_pool_callback_info_t*)data;
  73     info->func(info->data);
  74     OICFree(info);
  75     return NULL;
  76 }
  77
  78 // this implementation doesn't do a thread pool, so this function is essentially
  79 // a no-op besides creating a valid ca_thread_pool_t object.  It was determined after
  80 // reading through the existing implementation that the thread-pooling was unnecessary
  81 // for the posix platforms.  Behavior shouldn't be changed since previously num_of_threads
  82 // was greater than the number of requested threads.
  83 CAResult_t ca_thread_pool_init(int32_t num_of_threads, ca_thread_pool_t *thread_pool)
  84 {
  85     OIC_LOG(DEBUG, TAG, "IN");
  86
  87     if(!thread_pool)
  88     {
  89         OIC_LOG(ERROR, TAG, "Parameter thread_pool was null!");
  90         return CA_STATUS_INVALID_PARAM;
  91     }
  92
  93     if(num_of_threads <= 0)
  94     {
  95         OIC_LOG(ERROR, TAG, "num_of_threads must be positive and non-zero");
  96         return CA_STATUS_INVALID_PARAM;
  97     }
  98
  99     *thread_pool = OICMalloc(sizeof(struct ca_thread_pool));
 100
 101     if(!*thread_pool)
 102     {
 103         OIC_LOG(ERROR, TAG, "Failed to allocate for thread-pool");
 104         return CA_MEMORY_ALLOC_FAILED;
 105     }
 106
 107     (*thread_pool)->details = OICMalloc(sizeof(struct ca_thread_pool_details_t));
 108     if(!(*thread_pool)->details)
 109     {
 110         OIC_LOG(ERROR, TAG, "Failed to allocate for thread-pool details");
 111         OICFree(*thread_pool);
 112         *thread_pool=NULL;
 113         return CA_MEMORY_ALLOC_FAILED;
 114     }
 115
 116     (*thread_pool)->details->list_lock = oc_mutex_new();
 117
 118     if(!(*thread_pool)->details->list_lock)
 119     {
 120         OIC_LOG(ERROR, TAG, "Failed to create thread-pool mutex");
 121         goto exit;
 122     }
 123
 124     (*thread_pool)->details->threads_list = u_arraylist_create();
 125
 126     if(!(*thread_pool)->details->threads_list)
 127     {
 128         OIC_LOG(ERROR, TAG, "Failed to create thread-pool list");
 129         if(!oc_mutex_free((*thread_pool)->details->list_lock))
 130         {
 131             OIC_LOG(ERROR, TAG, "Failed to free thread-pool mutex");
 132         }
 133         goto exit;
 134     }
 135
 136     OIC_LOG(DEBUG, TAG, "OUT");
 137     return CA_STATUS_OK;
 138
 139 exit:
 140     OICFree((*thread_pool)->details);
 141     OICFree(*thread_pool);
 142     *thread_pool = NULL;
 143     return CA_STATUS_FAILED;
 144 }
 145
 146 CAResult_t ca_thread_pool_add_task(ca_thread_pool_t thread_pool, ca_thread_func method,
 147                                    void *data)
 148 {
 149     OIC_LOG(DEBUG, TAG, "IN");
 150
 151     if(NULL == thread_pool || NULL == method)
 152     {
 153         OIC_LOG(ERROR, TAG, "thread_pool or method was NULL");
 154         return CA_STATUS_INVALID_PARAM;
 155     }
 156
 157     ca_thread_pool_callback_info_t* info = OICMalloc(sizeof(ca_thread_pool_callback_info_t));
 158     if(!info)
 159     {
 160         OIC_LOG(ERROR, TAG, "Failed to allocate for memory wrapper");
 161         return CA_MEMORY_ALLOC_FAILED;
 162     }
 163
 164     info->func = method;
 165     info->data = data;
 166
 167     ca_thread_pool_thread_info_t *threadInfo =
 168             (ca_thread_pool_thread_info_t *) OICCalloc(1, sizeof(ca_thread_pool_thread_info_t));
 169     if (!threadInfo)
 170     {
 171         OIC_LOG(ERROR, TAG, "Memory allocation failed");
 172         OICFree(info);
 173         return CA_STATUS_FAILED;
 174     }
 175
 176     oc_mutex_lock(thread_pool->details->list_lock);
 177     bool addResult = u_arraylist_add(thread_pool->details->threads_list, (void*) threadInfo);
 178     if (!addResult)
 179     {
 180         // Note that this is considered non-fatal.
 181         oc_mutex_unlock(thread_pool->details->list_lock);
 182         OIC_LOG(ERROR, TAG, "Arraylist add failed");
 183         OICFree(info);
 184         OICFree(threadInfo);
 185         return CA_STATUS_FAILED;
 186     }
 187
 188     int thrRet = oc_thread_new(&threadInfo->thread, ca_thread_pool_pthreads_delegate, info);
 189     if (thrRet != 0)
 190     {
 191         uint32_t index = 0;
 192         if (u_arraylist_get_index(thread_pool->details->threads_list, threadInfo, &index))
 193         {
 194             u_arraylist_remove(thread_pool->details->threads_list, index);
 195         }
 196         oc_mutex_unlock(thread_pool->details->list_lock);
 197         OIC_LOG_V(ERROR, TAG, "Thread start failed with error %d", thrRet);
 198         OICFree(info);
 199         return CA_STATUS_FAILED;
 200     }
 201     oc_mutex_unlock(thread_pool->details->list_lock);
 202
 203     OIC_LOG(DEBUG, TAG, "OUT");
 204     return CA_STATUS_OK;
 205 }
 206
 207 void ca_thread_pool_free(ca_thread_pool_t thread_pool)
 208 {
 209     OIC_LOG(DEBUG, TAG, "IN");
 210
 211     if (!thread_pool)
 212     {
 213         OIC_LOG(ERROR, TAG, "Invalid parameter thread_pool was NULL");
 214         return;
 215     }
 216
 217     oc_mutex_lock(thread_pool->details->list_lock);
 218
 219     for (uint32_t i = 0; i < u_arraylist_length(thread_pool->details->threads_list); ++i)
 220     {
 221         ca_thread_pool_thread_info_t *threadInfo = (ca_thread_pool_thread_info_t *)
 222                 u_arraylist_get(thread_pool->details->threads_list, i);
 223         if (threadInfo)
 224         {
 225             if (threadInfo->thread)
 226             {
 227                 oc_thread_wait(threadInfo->thread);
 228                 oc_thread_free(threadInfo->thread);
 229             }
 230             OICFree(threadInfo);
 231         }
 232     }
 233
 234     u_arraylist_free(&(thread_pool->details->threads_list));
 235
 236     oc_mutex_unlock(thread_pool->details->list_lock);
 237     oc_mutex_free(thread_pool->details->list_lock);
 238
 239     OICFree(thread_pool->details);
 240     OICFree(thread_pool);
 241
 242     OIC_LOG(DEBUG, TAG, "OUT");
 243 }