src/tpl_utils_gthread.c

   1 #include "tpl_utils_gthread.h"
   2
   3 struct _tpl_gthread {
   4         GThread               *thread;
   5         GMainLoop             *loop;
   6
   7         GMutex                 thread_mutex;
   8         GCond                  thread_cond;
   9
  10         tpl_gthread_func       init_func;
  11         void                  *func_data;
  12 };
  13
  14 struct _tpl_gsource {
  15         GSource                gsource;
  16         gpointer               tag;
  17
  18         tpl_gthread           *thread;
  19
  20         int                    fd;
  21         tpl_bool_t             is_eventfd;
  22         tpl_gsource_functions *gsource_funcs;
  23
  24         tpl_gsource_type_t     type;
  25         tpl_gsource           *finalizer;
  26         tpl_bool_t             intended_destroy;
  27
  28         void                  *data;
  29 };
  30
  31 static void
  32 __gsource_remove_and_destroy(tpl_gsource *source);
  33
  34 static gpointer
  35 _tpl_gthread_init(gpointer data)
  36 {
  37         tpl_gthread *thread = (tpl_gthread *)data;
  38
  39         g_mutex_lock(&thread->thread_mutex);
  40
  41         if (thread->init_func)
  42                 thread->init_func(thread->func_data);
  43
  44         g_cond_signal(&thread->thread_cond);
  45         g_mutex_unlock(&thread->thread_mutex);
  46
  47         g_main_loop_run(thread->loop);
  48
  49         return thread;
  50 }
  51
  52 tpl_gthread *
  53 tpl_gthread_create(const char *thread_name,
  54                                    tpl_gthread_func init_func, void *func_data)
  55 {
  56         GMainContext *context    = NULL;
  57         GMainLoop    *loop       = NULL;
  58         tpl_gthread  *new_thread = NULL;
  59
  60         context = g_main_context_new();
  61         if (!context) {
  62                 TPL_ERR("Failed to create GMainContext");
  63                 return NULL;
  64         }
  65
  66         loop = g_main_loop_new(context, FALSE);
  67         if (!loop) {
  68                 TPL_ERR("Failed to create GMainLoop");
  69                 g_main_context_unref(context);
  70                 return NULL;
  71         }
  72
  73         g_main_context_unref(context);
  74
  75         new_thread = calloc(1, sizeof(tpl_gthread));
  76         if (!new_thread) {
  77                 TPL_ERR("Failed to allocate tpl_gthread");
  78                 g_main_context_unref(context);
  79                 g_main_loop_unref(loop);
  80
  81                 return NULL;
  82         }
  83
  84         g_mutex_init(&new_thread->thread_mutex);
  85         g_cond_init(&new_thread->thread_cond);
  86
  87
  88         g_mutex_lock(&new_thread->thread_mutex);
  89
  90         new_thread->loop      = loop;
  91         new_thread->init_func = init_func;
  92         new_thread->func_data = func_data;
  93         new_thread->thread    = g_thread_new(thread_name,
  94                                                                              _tpl_gthread_init, new_thread);
  95         g_cond_wait(&new_thread->thread_cond,
  96                                 &new_thread->thread_mutex);
  97
  98         g_mutex_unlock(&new_thread->thread_mutex);
  99
 100         return new_thread;
 101 }
 102
 103 void
 104 tpl_gthread_destroy(tpl_gthread *thread)
 105 {
 106         g_mutex_lock(&thread->thread_mutex);
 107
 108         g_main_loop_quit(thread->loop);
 109         g_thread_join(thread->thread);
 110         g_main_loop_unref(thread->loop);
 111
 112         thread->loop = NULL;
 113
 114         g_mutex_unlock(&thread->thread_mutex);
 115
 116         g_mutex_clear(&thread->thread_mutex);
 117         g_cond_clear(&thread->thread_cond);
 118
 119         thread->func_data = NULL;
 120         thread->thread = NULL;
 121
 122         free(thread);
 123         thread = NULL;
 124 }
 125
 126 static gboolean
 127 _thread_source_prepare(GSource *source, gint *time)
 128 {
 129         tpl_gsource *gsource = (tpl_gsource *)source;
 130         tpl_bool_t ret       = TPL_FALSE;
 131
 132         if (gsource->type != SOURCE_TYPE_NORMAL)
 133                 return ret;
 134
 135         if (gsource->gsource_funcs && gsource->gsource_funcs->prepare)
 136                 ret = gsource->gsource_funcs->prepare(gsource);
 137
 138         *time = -1;
 139
 140         return ret;
 141 }
 142
 143 static gboolean
 144 _thread_source_check(GSource *source)
 145 {
 146         tpl_gsource *gsource = (tpl_gsource *)source;
 147         tpl_bool_t ret       = TPL_FALSE;
 148
 149         if (gsource->type != SOURCE_TYPE_NORMAL)
 150                 return ret;
 151
 152         if (gsource->gsource_funcs && gsource->gsource_funcs->check)
 153                 ret = gsource->gsource_funcs->check(gsource);
 154
 155         return ret;
 156 }
 157
 158 static gboolean
 159 _thread_source_dispatch(GSource *source, GSourceFunc cb, gpointer data)
 160 {
 161         tpl_gsource *gsource = (tpl_gsource *)source;
 162         gboolean ret         = G_SOURCE_CONTINUE;
 163         GIOCondition cond    = g_source_query_unix_fd(source, gsource->tag);
 164         tpl_gthread *thread  = gsource->thread;
 165
 166         TPL_IGNORE(cb);
 167
 168         if (cond & G_IO_IN) {
 169                 ssize_t s;
 170                 uint64_t message = 0;
 171
 172                 if (gsource->is_eventfd) {
 173                         s = read(gsource->fd, &message, sizeof(uint64_t));
 174                         if (s != sizeof(uint64_t)) {
 175                                 TPL_ERR("Failed to read from event_fd(%d)",
 176                                                 gsource->fd);
 177                         }
 178                 }
 179
 180                 if (gsource->gsource_funcs && gsource->gsource_funcs->dispatch)
 181                         ret = gsource->gsource_funcs->dispatch(gsource, message);
 182
 183                 if (gsource->type == SOURCE_TYPE_FINALIZER &&
 184                         gsource->intended_destroy == TPL_TRUE) {
 185                         tpl_gsource *del_source = (tpl_gsource *)gsource->data;
 186                         if (!g_source_is_destroyed(&del_source->gsource)) {
 187                                 g_mutex_lock(&thread->thread_mutex);
 188
 189                                 __gsource_remove_and_destroy(del_source);
 190                                 __gsource_remove_and_destroy(gsource);
 191
 192                                 g_cond_signal(&thread->thread_cond);
 193                                 g_mutex_unlock(&thread->thread_mutex);
 194
 195                                 return G_SOURCE_REMOVE;
 196                         }
 197                 }
 198         }
 199
 200         if (cond && !(cond & G_IO_IN)) {
 201                 /* When some io errors occur, it is not considered as a critical error.
 202                  * There may be problems with the screen, but it does not affect the operation. */
 203                 TPL_WARN("Invalid GIOCondition occured. tpl_gsource(%p) fd(%d) cond(%d)",
 204                                  gsource, gsource->fd, cond);
 205         }
 206
 207         if (gsource->type == SOURCE_TYPE_DISPOSABLE) {
 208                 g_mutex_lock(&thread->thread_mutex);
 209                 __gsource_remove_and_destroy(gsource);
 210                 ret = G_SOURCE_REMOVE;
 211                 g_mutex_unlock(&thread->thread_mutex);
 212         }
 213
 214         return ret;
 215 }
 216
 217 static void
 218 _thread_source_finalize(GSource *source)
 219 {
 220         tpl_gsource *gsource = (tpl_gsource *)source;
 221
 222         if (gsource->gsource_funcs && gsource->gsource_funcs->finalize)
 223                 gsource->gsource_funcs->finalize(gsource);
 224
 225         if (gsource->is_eventfd)
 226                 close(gsource->fd);
 227
 228         gsource->fd = -1;
 229         gsource->thread = NULL;
 230         gsource->gsource_funcs = NULL;
 231         gsource->data = NULL;
 232         gsource->finalizer = NULL;
 233 }
 234
 235 static GSourceFuncs _thread_source_funcs = {
 236         .prepare = _thread_source_prepare,
 237         .check = _thread_source_check,
 238         .dispatch = _thread_source_dispatch,
 239         .finalize = _thread_source_finalize,
 240 };
 241
 242 tpl_gsource *
 243 tpl_gsource_create(tpl_gthread *thread, void *data, int fd,
 244                                    tpl_gsource_functions *funcs, tpl_gsource_type_t type)
 245 {
 246         tpl_gsource *new_gsource = NULL;
 247
 248         new_gsource = (tpl_gsource *)g_source_new(&_thread_source_funcs,
 249                                   sizeof(tpl_gsource));
 250         if (!new_gsource) {
 251                 TPL_ERR("Failed to create new tpl_gsource");
 252                 return NULL;
 253         }
 254
 255         if (fd < 0) {
 256                 new_gsource->fd = eventfd(0, EFD_CLOEXEC);
 257                 if (new_gsource->fd < 0) {
 258                         TPL_ERR("Failed to create eventfd. errno(%d)", errno);
 259                         g_source_unref(&new_gsource->gsource);
 260                         return NULL;
 261                 }
 262
 263                 new_gsource->is_eventfd = TPL_TRUE;
 264         } else {
 265                 new_gsource->fd = fd;
 266                 new_gsource->is_eventfd = TPL_FALSE;
 267         }
 268
 269         new_gsource->thread        = thread;
 270         new_gsource->gsource_funcs = funcs;
 271         new_gsource->data          = data;
 272         new_gsource->type          = type;
 273         new_gsource->intended_destroy = TPL_FALSE;
 274
 275         if (new_gsource->type == SOURCE_TYPE_NORMAL) {
 276                 tpl_gsource *finalizer = tpl_gsource_create(thread, new_gsource, -1,
 277                                                                                                         NULL, SOURCE_TYPE_FINALIZER);
 278                 new_gsource->finalizer = finalizer;
 279         } else
 280                 new_gsource->finalizer = NULL;
 281
 282         new_gsource->tag = g_source_add_unix_fd(&new_gsource->gsource,
 283                                                                                         new_gsource->fd,
 284                                                                                         G_IO_IN | G_IO_ERR);
 285         g_source_attach(&new_gsource->gsource,
 286                                         g_main_loop_get_context(thread->loop));
 287
 288         TPL_DEBUG("[GSOURCE_CREATE] tpl_gsource(%p) thread(%p) data(%p) fd(%d) type(%d)",
 289                           new_gsource, thread, data, new_gsource->fd, type);
 290
 291         return new_gsource;
 292 }
 293
 294 static void
 295 __gsource_remove_and_destroy(tpl_gsource *source)
 296 {
 297         if (g_source_is_destroyed(&source->gsource))
 298                 return;
 299
 300         TPL_DEBUG("[GSOURCE_DESTROY] tpl_gsource(%p) type(%d)",
 301                           source, source->type);
 302
 303         g_source_remove_unix_fd(&source->gsource, source->tag);
 304         g_source_destroy(&source->gsource);
 305         g_source_unref(&source->gsource);
 306 }
 307
 308 void
 309 tpl_gsource_destroy(tpl_gsource *source, tpl_bool_t destroy_in_thread)
 310 {
 311         tpl_gthread *thread = source->thread;
 312
 313         if (g_source_is_destroyed(&source->gsource)) {
 314                 TPL_WARN("gsource(%p) already has been destroyed.",
 315                                  source);
 316                 return;
 317         }
 318
 319         g_mutex_lock(&thread->thread_mutex);
 320         if (source->type == SOURCE_TYPE_NORMAL &&
 321                 source->finalizer) {
 322                 tpl_gsource *finalizer = source->finalizer;
 323
 324                 if (destroy_in_thread) {
 325                         finalizer->intended_destroy = TPL_TRUE;
 326                         tpl_gsource_send_message(finalizer, 1);
 327
 328                         g_cond_wait(&thread->thread_cond, &thread->thread_mutex);
 329                 } else {
 330                         __gsource_remove_and_destroy(finalizer);
 331                         source->finalizer = NULL;
 332                 }
 333         }
 334
 335         if (!destroy_in_thread) {
 336                 __gsource_remove_and_destroy(source);
 337         }
 338         g_mutex_unlock(&thread->thread_mutex);
 339 }
 340
 341 void
 342 tpl_gsource_send_message(tpl_gsource *source, uint64_t message)
 343 {
 344         uint64_t value = message;
 345         int ret;
 346
 347         if (!source->is_eventfd) {
 348                 TPL_ERR("source is not using eventfd. source(%p) fd(%d)",
 349                                 source, source->fd);
 350                 return;
 351         }
 352
 353         ret = write(source->fd, &value, sizeof(uint64_t));
 354         if (ret == -1) {
 355                 TPL_ERR("failed to send devent. tpl_gsource(%p)",
 356                                 source);
 357         }
 358 }
 359
 360 void *
 361 tpl_gsource_get_data(tpl_gsource *source)
 362 {
 363         void *data = NULL;
 364
 365         if (source)
 366                 data = source->data;
 367
 368         return data;
 369 }
 370
 371 tpl_bool_t
 372 tpl_gsource_check_io_condition(tpl_gsource *source)
 373 {
 374         GIOCondition cond;
 375
 376         if (!source) {
 377                 TPL_ERR("Invalid parameter tpl_gsource is null");
 378                 return TPL_FALSE;
 379         }
 380
 381         cond = g_source_query_unix_fd(&source->gsource, source->tag);
 382         if (cond & G_IO_IN)
 383                 return TPL_TRUE;
 384
 385         return TPL_FALSE;
 386 }
 387
 388 void
 389 tpl_gmutex_init(tpl_gmutex *gmutex)
 390 {
 391         g_mutex_init(gmutex);
 392 }
 393
 394 void
 395 tpl_gmutex_clear(tpl_gmutex *gmutex)
 396 {
 397         g_mutex_clear(gmutex);
 398 }
 399
 400 void
 401 tpl_gmutex_lock(tpl_gmutex *gmutex)
 402 {
 403         g_mutex_lock(gmutex);
 404 }
 405
 406 void
 407 tpl_gmutex_unlock(tpl_gmutex *gmutex)
 408 {
 409         g_mutex_unlock(gmutex);
 410 }
 411
 412 void
 413 tpl_gcond_init(tpl_gcond *gcond)
 414 {
 415         g_cond_init(gcond);
 416 }
 417
 418 void
 419 tpl_gcond_clear(tpl_gcond *gcond)
 420 {
 421         g_cond_clear(gcond);
 422 }
 423
 424 void
 425 tpl_gcond_wait(tpl_gcond *gcond, tpl_gmutex *gmutex)
 426 {
 427         g_cond_wait(gcond, gmutex);
 428 }
 429
 430 tpl_result_t
 431 tpl_cond_timed_wait(tpl_gcond *gcond, tpl_gmutex *gmutex,
 432                                         int64_t timeout_ms)
 433 {
 434         gint64 end_time = g_get_monotonic_time() +
 435                                                 (timeout_ms * G_TIME_SPAN_MILLISECOND);
 436         if (!g_cond_wait_until(gcond, gmutex, end_time))
 437                 return TPL_ERROR_TIME_OUT;
 438
 439         return TPL_ERROR_NONE;
 440 }
 441
 442 void
 443 tpl_gcond_signal(tpl_gcond *gcond)
 444 {
 445         g_cond_signal(gcond);
 446 }