server/mm_sound_thread_pool.c

   1 /*
   2  * libmm-sound
   3  *
   4  * Copyright (c) 2000 - 2011 Samsung Electronics Co., Ltd. All rights reserved.
   5  *
   6  * Contact: Seungbae Shin <seungbae.shin@samsung.com>
   7  *
   8  * Licensed under the Apache License, Version 2.0 (the "License");
   9  * you may not use this file except in compliance with the License.
  10  * You may obtain a copy of the License at
  11  *
  12  * http://www.apache.org/licenses/LICENSE-2.0
  13  *
  14  * Unless required by applicable law or agreed to in writing, software
  15  * distributed under the License is distributed on an "AS IS" BASIS,
  16  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  17  * See the License for the specific language governing permissions and
  18  * limitations under the License.
  19  *
  20  */
  21
  22 #include <stdio.h>
  23 #include <stdlib.h>
  24 #include <unistd.h>
  25
  26 #include <mm_error.h>
  27 #include <mm_debug.h>
  28 #include <mm_sound_thread_pool.h>
  29
  30 #define USE_G_THREAD_POOL
  31 #ifdef USE_G_THREAD_POOL
  32 #include <glib.h>
  33 GThreadPool* g_pool;
  34
  35 #define MAX_UNUSED_THREADS_IN_THREADPOOL        10
  36
  37 #define MIN_RUNNING_THREAD 0
  38
  39 typedef struct __THREAD_INFO
  40 {
  41         void (*func)(gpointer data);
  42         void *param;
  43 } THREAD_INFO;
  44
  45 static void __DummyWork (void* param)
  46 {
  47         debug_msg ("thread index = %d\n", (int)param);
  48         sleep (1);
  49 }
  50
  51 static void __ThreadWork(gpointer data, gpointer user_data)
  52 {
  53         THREAD_INFO* info = (THREAD_INFO*)data;
  54         if (info) {
  55                  if (info->func) {
  56                                 debug_log ("Calling [%p] with param [%p]\n", info->func, info->param);
  57                                 info->func (info->param);
  58                  } else {
  59                                 debug_warning ("No func to call....\n");
  60                  }
  61
  62                  /* Info was allocated by MMSoundThreadPoolRun().
  63                         The actual content of info should be  freed whether inside func or outside (if handle) */
  64                  debug_log ("free [%p]\n", info);
  65                  free (info);
  66                  info = NULL;
  67         } else {
  68                 debug_warning ("No valid thread info...Nothing to do...\n");
  69         }
  70 }
  71
  72 int MMSoundThreadPoolDump(int fulldump)
  73 {
  74         if (g_pool == NULL) {
  75                 debug_error ("No thread pool initialized....\n");
  76                 return MM_ERROR_SOUND_INTERNAL;
  77         }
  78
  79         if (fulldump) {
  80                 debug_log ("##### [ThreadPool] max threads=[%d], max unused=[%d], max idle time=[%d]\n",
  81                                 g_thread_pool_get_max_threads (g_pool),
  82                                 g_thread_pool_get_max_unused_threads(),
  83                                 g_thread_pool_get_max_idle_time()       );
  84         }
  85         debug_log ("***** [ThreadPool] running=[%d], unused=[%d]\n",
  86                         g_thread_pool_get_num_threads (g_pool),
  87                         g_thread_pool_get_num_unused_threads() );
  88
  89         return MM_ERROR_NONE;
  90 }
  91
  92 gboolean IsMMSoundThreadPoolRunning(void)
  93 {
  94         MMSoundThreadPoolDump(FALSE);
  95         return (g_thread_pool_get_num_threads(g_pool) > MIN_RUNNING_THREAD);
  96 }
  97
  98 int MMSoundThreadPoolInit()
  99 {
 100         int i=0;
 101         GError* error = NULL;
 102
 103         /* Create thread pool (non-exclude mode with infinite max threads) */
 104         g_pool = g_thread_pool_new (__ThreadWork, NULL, -1, FALSE, &error);
 105         if (g_pool == NULL && error != NULL) {
 106                 debug_error ("thread pool created failed : %s\n", error->message);
 107                 g_error_free (error);
 108                 return MM_ERROR_SOUND_INTERNAL;
 109         }
 110         debug_msg ("thread pool created successfully\n");
 111
 112         MMSoundThreadPoolDump(TRUE);
 113
 114         /* Thread pool setting : this will maintain at least 10 unused threads and this will be reused. */
 115         /* If no unused thread left, new thread will be created, but always maintain 10 unused thread */
 116         debug_msg ("thread pool set max unused threads to %d\n", MAX_UNUSED_THREADS_IN_THREADPOOL);
 117         g_thread_pool_set_max_unused_threads (MAX_UNUSED_THREADS_IN_THREADPOOL);
 118
 119         /* To reserve unused threads, let's start some threads for beggining
 120            This dummy thread will be remained unused as soon as it started */
 121         debug_msg ("run threads to reserve minimum thread\n");
 122         for (i=0; i<MAX_UNUSED_THREADS_IN_THREADPOOL; i++) {
 123                 MMSoundThreadPoolRun ((void *)i, __DummyWork);
 124         }
 125
 126         MMSoundThreadPoolDump(TRUE);
 127
 128      return MM_ERROR_NONE;
 129 }
 130
 131 int MMSoundThreadPoolRun(void *param, void (*func)(void*))
 132 {
 133         GError* error = NULL;
 134
 135         /* Dump current thread pool */
 136         MMSoundThreadPoolDump(FALSE);
 137
 138         /* Create thread info structure.
 139            This thread info data will be free in __ThreadWork(), after use. */
 140         THREAD_INFO* thread_info = (THREAD_INFO*)malloc (sizeof(THREAD_INFO));
 141         if (thread_info) {
 142                 thread_info->func = func;
 143                 thread_info->param = param;
 144                 debug_log ("alloc thread_info = %p\n", thread_info);
 145
 146                 /* Add thread to queue of thread pool */
 147                 g_thread_pool_push (g_pool, thread_info, &error);
 148                 if (error) {
 149                         debug_error ("g_thread_pool_push failed : %s\n", error->message);
 150                         g_error_free (error);
 151                         free(thread_info);
 152                         return MM_ERROR_SOUND_INTERNAL;
 153                 }
 154         } else {
 155                 debug_error("failed to alloc thread info\n");
 156                 return MM_ERROR_SOUND_INTERNAL;
 157         }
 158
 159         return MM_ERROR_NONE;
 160 }
 161
 162 int MMSoundThreadPoolFini(void)
 163 {
 164         /* If immediate is TRUE, no new task is processed for pool.
 165         Otherwise pool is not freed before the last task is processed.
 166         Note however, that no thread of this pool is interrupted, while processing a task.
 167         Instead at least all still running threads can finish their tasks before the pool is freed.
 168
 169         If wait_ is TRUE, the functions does not return before all tasks to be processed
 170         (dependent on immediate, whether all or only the currently running) are ready.
 171         Otherwise the function returns immediately.     */
 172         debug_msg ("thread pool will be free\n");
 173         g_thread_pool_free (g_pool, TRUE, FALSE);
 174
 175         return MM_ERROR_NONE;
 176 }
 177
 178 #else // USE_G_THREAD_POOL
 179
 180 #define THREAD_POOL_MAX 10
 181
 182 struct __control_t
 183 {
 184         pthread_t threadid;
 185         pthread_cond_t condition;
 186         int stopflag;
 187         void *param;
 188         void (*func)(void *);
 189 };
 190
 191 static int threadmap[THREAD_POOL_MAX];
 192 static struct __control_t control[THREAD_POOL_MAX];
 193
 194 static pthread_cond_t startcond = PTHREAD_COND_INITIALIZER;
 195 static pthread_mutex_t startsync = PTHREAD_MUTEX_INITIALIZER;
 196 static pthread_mutex_t controlsync = PTHREAD_MUTEX_INITIALIZER;
 197 static pthread_mutex_t funcsync = PTHREAD_MUTEX_INITIALIZER;
 198
 199 static int __GetEmptyPool(void)
 200 {
 201         int count = 0;
 202         while ((count < THREAD_POOL_MAX) && threadmap[count] == 1)
 203                 ++count;
 204         return count == THREAD_POOL_MAX ? -1 : count;
 205 }
 206
 207 static void __SetPool(int n)
 208 {
 209     threadmap[n] = 1;
 210 }
 211
 212 static void __ResetPool(int n)
 213 {
 214     threadmap[n] = 0;
 215 }
 216
 217 static void __InitPool(void)
 218 {
 219     int count = 0;
 220     for (count = 0; count < THREAD_POOL_MAX; count++)
 221     {
 222         threadmap[count] = 0;
 223         if (pthread_cond_init(&control[count].condition, NULL) != 0)
 224             perror("Make Thread Condition");
 225     }
 226 }
 227
 228 static void __DestroyPool(void)
 229 {
 230     int count = 0;
 231     for (count = 0; count < THREAD_POOL_MAX; count++)
 232     {
 233         if (pthread_cond_destroy(&control[count].condition) != 0)
 234         {
 235             perror("Remove Thread Condition");
 236             exit(0);
 237         }
 238     }
 239 }
 240
 241 static void* __ThreadWork(void *param)
 242 {
 243     int myid = -1;
 244
 245     myid = (int)param;
 246
 247     pthread_mutex_lock(&startsync);
 248     pthread_cond_signal(&startcond);
 249     pthread_mutex_unlock(&startsync);
 250
 251     while(1)
 252     {
 253         pthread_mutex_lock(&controlsync);
 254         pthread_cond_wait(&control[myid].condition, &controlsync);
 255         pthread_mutex_unlock(&controlsync);
 256
 257         if (control[myid].func != NULL)
 258             control[myid].func(control[myid].param);
 259 /*        if (control[myid].param != NULL)
 260             free(control[myid].param);*/
 261
 262         control[myid].func = NULL;
 263         control[myid].param = NULL;
 264         pthread_mutex_lock(&startsync);
 265         __ResetPool(myid);
 266         pthread_mutex_unlock(&startsync);
 267
 268         if (control[myid].stopflag) {
 269             pthread_exit(0);
 270         }
 271     }
 272 }
 273
 274 int MMSoundThreadPoolDump(void)
 275 {
 276         int count = 0;
 277         int ret = MM_ERROR_NONE;
 278
 279         fprintf(stdout, "================================================================================\n");
 280         fprintf(stdout, "                              Thread States                                     \n");
 281         fprintf(stdout, "--------------------------------------------------------------------------------\n");
 282         for (count = 0; count < THREAD_POOL_MAX; count ++)
 283         {
 284                 fprintf(stdout, "Thread %d\n", control[count].threadid);
 285                 fprintf(stdout, "Current State is \"%s\"\n", threadmap[count] ? "Running" : "Ready");
 286                 if (threadmap[count])
 287                 {
 288                         fprintf(stdout, "Running function address %p\n", control[count].func);
 289                 }
 290                 else
 291                 {
 292                         fprintf(stdout, "Threadmap is NULL\n");
 293                         ret = MM_ERROR_SOUND_INTERNAL;
 294                 }
 295         }
 296         fprintf(stdout, "================================================================================\n");
 297         return ret;
 298 }
 299
 300 int MMSoundThreadPoolInit(void)
 301 {
 302     volatile int count = 0;
 303     pthread_mutex_lock(&funcsync);
 304
 305     __InitPool();
 306
 307     for (count = 0; count < THREAD_POOL_MAX; count++)
 308     {
 309         control[count].stopflag = 0;
 310         pthread_mutex_lock(&startsync);
 311         if (pthread_create(&control[count].threadid, NULL, __ThreadWork, (void*)count) < 0)
 312         {
 313             perror("Make Thread Fail");
 314             exit(0);
 315         }
 316         pthread_cond_wait(&startcond, &startsync);
 317         pthread_mutex_unlock(&startsync);
 318         usleep(100); /* Delay for thread init */
 319     }
 320     pthread_mutex_unlock(&funcsync);
 321     return MM_ERROR_NONE;
 322 }
 323
 324 int MMSoundThreadPoolRun(void *param, void (*func)(void*))
 325 {
 326     int poolnum = -1;
 327
 328     pthread_mutex_lock(&funcsync);
 329     pthread_mutex_lock(&startsync);
 330     poolnum = __GetEmptyPool();
 331     if (poolnum < 0) {
 332         pthread_mutex_unlock(&startsync);
 333         pthread_mutex_unlock(&funcsync);
 334         return MM_ERROR_COMMON_NO_FREE_SPACE;
 335     }
 336     __SetPool(poolnum);
 337     pthread_mutex_unlock(&startsync);
 338     control[poolnum].param = param;
 339     control[poolnum].func = func;
 340     pthread_cond_signal(&control[poolnum].condition);
 341     pthread_mutex_unlock(&funcsync);
 342
 343     return MM_ERROR_NONE;
 344 }
 345
 346 int MMSoundThreadPoolFini(void)
 347 {
 348     int count = 0;
 349
 350     pthread_mutex_lock(&funcsync);
 351     for (count = 0; count < THREAD_POOL_MAX; count++)
 352     {
 353         control[count].stopflag = 1;
 354         pthread_cond_signal(&control[count].condition);
 355         if (pthread_join(control[count].threadid, NULL) < 0)
 356         {
 357             perror("Join Fail");
 358             exit(0);
 359         }
 360     }
 361     __DestroyPool();
 362     pthread_mutex_unlock(&funcsync);
 363
 364     return MM_ERROR_NONE;
 365 }
 366 #endif // USE_G_THREAD_POOL
 367