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