resource/csdk/connectivity/src/caqueueingthread.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 #include <stdio.h>
  22 #include <stdlib.h>
  23 #include <string.h>
  24 #include <unistd.h>
  25 #include <sys/syscall.h>
  26 #include <sys/types.h>
  27
  28 #include "caqueueingthread.h"
  29 #include "oic_malloc.h"
  30 #include "logger.h"
  31
  32 #define TAG PCF("CA")
  33
  34 static void CAQueueingThreadBaseRoutine(void *threadValue)
  35 {
  36     OIC_LOG_V(DEBUG, TAG, "message handler main thread start..");
  37
  38     CAQueueingThread_t *thread = (CAQueueingThread_t *) threadValue;
  39
  40     if (thread == NULL)
  41     {
  42         OIC_LOG_V(DEBUG, TAG, "thread data passing error!!");
  43
  44         return;
  45     }
  46
  47     while (!thread->isStop)
  48     {
  49         // mutex lock
  50         u_mutex_lock(thread->threadMutex);
  51
  52         // if queue is empty, thread will wait
  53         if (u_queue_get_size(thread->dataQueue) <= 0)
  54         {
  55             OIC_LOG_V(DEBUG, TAG, "wait..");
  56
  57             // wait
  58             u_cond_wait(thread->threadCond, thread->threadMutex);
  59
  60             OIC_LOG_V(DEBUG, TAG, "wake up..");
  61         }
  62
  63         // mutex unlock
  64         u_mutex_unlock(thread->threadMutex);
  65
  66         // check stop flag
  67         if (thread->isStop)
  68             continue;
  69
  70         // get data
  71         u_queue_message_t *message = u_queue_get_element(thread->dataQueue);
  72         if (message == NULL)
  73         {
  74             continue;
  75         }
  76
  77         void *data = message->msg;
  78
  79         // process data
  80         thread->threadTask(data);
  81
  82         // free
  83         if (thread->destroy != NULL)
  84         {
  85             thread->destroy(message->msg, message->size);
  86         }
  87         else
  88         {
  89             OICFree(message->msg);
  90         }
  91
  92         OICFree(message);
  93     }
  94
  95     // remove all remained list data.
  96     while (u_queue_get_size(thread->dataQueue) > 0)
  97     {
  98         // get data
  99         u_queue_message_t *message = u_queue_get_element(thread->dataQueue);
 100
 101         // free
 102         if(message != NULL) {
 103             if (thread->destroy != NULL)
 104             {
 105                 thread->destroy(message->msg, message->size);
 106             }
 107             else
 108             {
 109                 OICFree(message->msg);
 110             }
 111
 112             OICFree(message);
 113         }
 114     }
 115
 116     u_cond_signal(thread->threadCond);
 117
 118     OIC_LOG_V(DEBUG, TAG, "message handler main thread end..");
 119 }
 120
 121 CAResult_t CAQueueingThreadInitialize(CAQueueingThread_t *thread, u_thread_pool_t handle,
 122                                       CAThreadTask task, CADataDestroyFunction destroy)
 123 {
 124     if (thread == NULL)
 125     {
 126         OIC_LOG_V(DEBUG, TAG, "thread instance is empty..");
 127         return CA_STATUS_FAILED;
 128     }
 129
 130     if (handle == NULL)
 131     {
 132         OIC_LOG_V(DEBUG, TAG, "thread pool handle is empty..");
 133         return CA_STATUS_FAILED;
 134     }
 135
 136     OIC_LOG_V(DEBUG, TAG, "thread initialize..");
 137
 138     memset(thread, 0, sizeof(CAQueueingThread_t));
 139
 140     // mutex init
 141     u_mutex_init();
 142
 143     // set send thread data
 144     thread->threadPool = handle;
 145     thread->dataQueue = u_queue_create();
 146     thread->threadMutex = u_mutex_new();
 147     thread->threadCond = u_cond_new();
 148     thread->isStop = CA_TRUE;
 149     thread->threadTask = task;
 150     thread->destroy = destroy;
 151
 152     return CA_STATUS_OK;
 153 }
 154
 155 CAResult_t CAQueueingThreadStart(CAQueueingThread_t *thread)
 156 {
 157     if (thread == NULL)
 158     {
 159         OIC_LOG_V(DEBUG, TAG, "thread instance is empty..");
 160         return CA_STATUS_FAILED;
 161     }
 162
 163     if (thread->threadPool == NULL)
 164     {
 165         OIC_LOG_V(DEBUG, TAG, "thread pool handle is empty..");
 166         return CA_STATUS_FAILED;
 167     }
 168
 169     if (CA_FALSE == thread->isStop) //Queueing thread already running
 170     {
 171         OIC_LOG_V(DEBUG, TAG, "queueing thread already running..");
 172         return CA_STATUS_OK;
 173     }
 174
 175     thread->isStop = CA_FALSE;
 176     CAResult_t res = u_thread_pool_add_task(thread->threadPool, CAQueueingThreadBaseRoutine,
 177                                             thread);
 178     if (res != CA_STATUS_OK)
 179     {
 180         OIC_LOG_V(DEBUG, TAG, "thread pool add task error(send thread).");
 181         thread->isStop = CA_TRUE;
 182         return res;
 183     }
 184
 185     return res;
 186 }
 187
 188 CAResult_t CAQueueingThreadAddData(CAQueueingThread_t *thread, void *data, uint32_t size)
 189 {
 190     if (thread == NULL)
 191     {
 192         OIC_LOG_V(DEBUG, TAG, "thread instance is empty..");
 193         return CA_STATUS_FAILED;
 194     }
 195
 196     if (data == NULL || size == 0)
 197     {
 198         OIC_LOG_V(DEBUG, TAG, "data is empty..");
 199
 200         return CA_STATUS_FAILED;
 201     }
 202
 203     // create thread data
 204     u_queue_message_t *message = (u_queue_message_t *) OICMalloc(sizeof(u_queue_message_t));
 205
 206     if (message == NULL)
 207     {
 208         OIC_LOG_V(DEBUG, TAG, "memory error!!");
 209         return CA_MEMORY_ALLOC_FAILED;
 210     }
 211     memset(message, 0, sizeof(u_queue_message_t));
 212
 213     message->msg = data;
 214     message->size = sizeof(size);
 215
 216     // mutex lock
 217     u_mutex_lock(thread->threadMutex);
 218
 219     // add thread data into list
 220     u_queue_add_element(thread->dataQueue, message);
 221
 222     // notity the thread
 223     u_cond_signal(thread->threadCond);
 224
 225     // mutex unlock
 226     u_mutex_unlock(thread->threadMutex);
 227
 228     return CA_STATUS_OK;
 229 }
 230
 231 CAResult_t CAQueueingThreadDestroy(CAQueueingThread_t *thread)
 232 {
 233     if (thread == NULL)
 234     {
 235         OIC_LOG_V(DEBUG, TAG, "thread instance is empty..");
 236         return CA_STATUS_FAILED;
 237     }
 238
 239     OIC_LOG_V(DEBUG, TAG, "thread destroy..");
 240
 241     u_mutex_free(thread->threadMutex);
 242     thread->threadMutex = NULL;
 243     u_cond_free(thread->threadCond);
 244     u_queue_delete(thread->dataQueue);
 245
 246     return CA_STATUS_OK;
 247 }
 248
 249 CAResult_t CAQueueingThreadStop(CAQueueingThread_t *thread)
 250 {
 251     if (thread == NULL)
 252     {
 253         OIC_LOG_V(DEBUG, TAG, "thread instance is empty..");
 254         return CA_STATUS_FAILED;
 255     }
 256
 257     OIC_LOG_V(DEBUG, TAG, "thread stop request!!");
 258
 259     if (!thread->isStop)
 260     {
 261         // mutex lock
 262         u_mutex_lock(thread->threadMutex);
 263
 264         // set stop flag
 265         thread->isStop = CA_TRUE;
 266
 267         // notity the thread
 268         u_cond_signal(thread->threadCond);
 269
 270         u_cond_wait(thread->threadCond, thread->threadMutex);
 271
 272         // mutex unlock
 273         u_mutex_unlock(thread->threadMutex);
 274     }
 275
 276     return CA_STATUS_OK;
 277 }
 278