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         GMutex                 idle_mutex;
  11         GCond                  idle_cond;
  12         GMutex                 pause_mutex;
  13         tpl_bool_t             is_idle;
  14         tpl_bool_t             paused;
  15
  16         tpl_gthread_func       init_func;
  17         void                  *func_data;
  18 };
  19
  20 struct _tpl_gsource {
  21         GSource                gsource;
  22         gpointer               tag;
  23
  24         tpl_gthread           *thread;
  25
  26         int                    fd;
  27         tpl_bool_t             is_eventfd;
  28         tpl_gsource_functions *gsource_funcs;
  29
  30         tpl_gsource_type_t     type;
  31         tpl_gsource           *finalizer;
  32         tpl_bool_t             intended_destroy;
  33
  34         void                  *data;
  35 };
  36
  37 static void
  38 __gsource_remove_and_destroy(tpl_gsource *source);
  39
  40 static gpointer
  41 _tpl_gthread_init(gpointer data)
  42 {
  43         tpl_gthread *thread = (tpl_gthread *)data;
  44
  45         g_mutex_lock(&thread->thread_mutex);
  46
  47         if (thread->init_func)
  48                 thread->init_func(thread->func_data);
  49
  50         g_cond_signal(&thread->thread_cond);
  51         g_mutex_unlock(&thread->thread_mutex);
  52
  53         g_main_loop_run(thread->loop);
  54
  55         return thread;
  56 }
  57
  58 tpl_gthread *
  59 tpl_gthread_create(const char *thread_name,
  60                                    tpl_gthread_func init_func, void *func_data)
  61 {
  62         GMainContext *context    = NULL;
  63         GMainLoop    *loop       = NULL;
  64         tpl_gthread  *new_thread = NULL;
  65
  66         context = g_main_context_new();
  67         if (!context) {
  68                 TPL_ERR("Failed to create GMainContext");
  69                 return NULL;
  70         }
  71
  72         loop = g_main_loop_new(context, FALSE);
  73         if (!loop) {
  74                 TPL_ERR("Failed to create GMainLoop");
  75                 g_main_context_unref(context);
  76                 return NULL;
  77         }
  78
  79         // context's ref count was increased in g_main_loop_new
  80         g_main_context_unref(context);
  81
  82         new_thread = calloc(1, sizeof(tpl_gthread));
  83         if (!new_thread) {
  84                 TPL_ERR("Failed to allocate tpl_gthread");
  85
  86                 // context is also destroyed when loop is destroyed.
  87                 g_main_loop_unref(loop);
  88
  89                 return NULL;
  90         }
  91
  92         g_mutex_init(&new_thread->thread_mutex);
  93         g_cond_init(&new_thread->thread_cond);
  94
  95         g_mutex_init(&new_thread->pause_mutex);
  96         g_mutex_init(&new_thread->idle_mutex);
  97         g_cond_init(&new_thread->idle_cond);
  98
  99         new_thread->is_idle = TPL_FALSE;
 100         new_thread->paused = TPL_FALSE;
 101
 102         g_mutex_lock(&new_thread->thread_mutex);
 103
 104         new_thread->loop      = loop;
 105         new_thread->init_func = init_func;
 106         new_thread->func_data = func_data;
 107         new_thread->thread    = g_thread_new(thread_name,
 108                                                                              _tpl_gthread_init, new_thread);
 109         g_cond_wait(&new_thread->thread_cond,
 110                                 &new_thread->thread_mutex);
 111
 112         g_mutex_unlock(&new_thread->thread_mutex);
 113
 114         return new_thread;
 115 }
 116
 117 void
 118 tpl_gthread_destroy(tpl_gthread *thread)
 119 {
 120         g_mutex_lock(&thread->thread_mutex);
 121
 122         g_main_loop_quit(thread->loop);
 123         g_thread_join(thread->thread);
 124         g_main_loop_unref(thread->loop);
 125
 126         thread->loop = NULL;
 127
 128         g_mutex_unlock(&thread->thread_mutex);
 129
 130         g_mutex_clear(&thread->pause_mutex);
 131         g_mutex_clear(&thread->idle_mutex);
 132         g_cond_clear(&thread->idle_cond);
 133
 134         g_mutex_clear(&thread->thread_mutex);
 135         g_cond_clear(&thread->thread_cond);
 136
 137         thread->func_data = NULL;
 138         thread->thread = NULL;
 139
 140         free(thread);
 141         thread = NULL;
 142 }
 143
 144 static gboolean
 145 _thread_source_prepare(GSource *source, gint *time)
 146 {
 147         tpl_gsource *gsource = (tpl_gsource *)source;
 148         tpl_bool_t ret       = TPL_FALSE;
 149
 150         if (gsource->type != SOURCE_TYPE_NORMAL)
 151                 return ret;
 152
 153         if (gsource->gsource_funcs && gsource->gsource_funcs->prepare)
 154                 ret = gsource->gsource_funcs->prepare(gsource);
 155
 156         *time = -1;
 157
 158         return ret;
 159 }
 160
 161 static gboolean
 162 _thread_source_check(GSource *source)
 163 {
 164         tpl_gsource *gsource = (tpl_gsource *)source;
 165         tpl_bool_t ret       = TPL_FALSE;
 166
 167         if (gsource->type != SOURCE_TYPE_NORMAL)
 168                 return ret;
 169
 170         if (gsource->gsource_funcs && gsource->gsource_funcs->check)
 171                 ret = gsource->gsource_funcs->check(gsource);
 172
 173         return ret;
 174 }
 175
 176 static gboolean
 177 _thread_source_dispatch(GSource *source, GSourceFunc cb, gpointer data)
 178 {
 179         tpl_gsource *gsource = (tpl_gsource *)source;
 180         gboolean ret         = G_SOURCE_CONTINUE;
 181         GIOCondition cond    = g_source_query_unix_fd(source, gsource->tag);
 182         tpl_gthread *thread  = gsource->thread;
 183
 184         TPL_IGNORE(cb);
 185
 186         if (cond & G_IO_IN) {
 187                 ssize_t s;
 188                 uint64_t message = 0;
 189
 190                 if (gsource->is_eventfd) {
 191                         s = read(gsource->fd, &message, sizeof(uint64_t));
 192                         if (s != sizeof(uint64_t)) {
 193                                 TPL_ERR("Failed to read from event_fd(%d)",
 194                                                 gsource->fd);
 195                         }
 196                 }
 197
 198                 if (gsource->gsource_funcs && gsource->gsource_funcs->dispatch)
 199                         ret = gsource->gsource_funcs->dispatch(gsource, message);
 200
 201                 if (gsource->type == SOURCE_TYPE_FINALIZER &&
 202                         gsource->intended_destroy == TPL_TRUE) {
 203                         tpl_gsource *del_source = (tpl_gsource *)gsource->data;
 204                         if (!g_source_is_destroyed(&del_source->gsource)) {
 205                                 g_mutex_lock(&thread->thread_mutex);
 206
 207                                 __gsource_remove_and_destroy(del_source);
 208                                 __gsource_remove_and_destroy(gsource);
 209
 210                                 g_cond_signal(&thread->thread_cond);
 211                                 g_mutex_unlock(&thread->thread_mutex);
 212
 213                                 return G_SOURCE_REMOVE;
 214                         }
 215                 }
 216         }
 217
 218         if (cond && !(cond & G_IO_IN)) {
 219                 /* When some io errors occur, it is not considered as a critical error.
 220                  * There may be problems with the screen, but it does not affect the operation. */
 221                 TPL_WARN("Invalid GIOCondition occured. tpl_gsource(%p) fd(%d) cond(%d)",
 222                                  gsource, gsource->fd, cond);
 223
 224                 if (gsource->type == SOURCE_TYPE_DISPOSABLE) {
 225                         if (gsource->gsource_funcs && gsource->gsource_funcs->dispatch)
 226                                 ret = gsource->gsource_funcs->dispatch(gsource, 0);
 227                 }
 228         }
 229
 230         if (gsource->type == SOURCE_TYPE_DISPOSABLE) {
 231                 g_mutex_lock(&thread->thread_mutex);
 232                 __gsource_remove_and_destroy(gsource);
 233                 ret = G_SOURCE_REMOVE;
 234                 g_mutex_unlock(&thread->thread_mutex);
 235         }
 236
 237         return ret;
 238 }
 239
 240 static void
 241 _thread_source_finalize(GSource *source)
 242 {
 243         tpl_gsource *gsource = (tpl_gsource *)source;
 244
 245         if (gsource->gsource_funcs && gsource->gsource_funcs->finalize)
 246                 gsource->gsource_funcs->finalize(gsource);
 247
 248         if (gsource->is_eventfd)
 249                 close(gsource->fd);
 250
 251         gsource->fd = -1;
 252         gsource->thread = NULL;
 253         gsource->gsource_funcs = NULL;
 254         gsource->data = NULL;
 255         gsource->finalizer = NULL;
 256 }
 257
 258 static GSourceFuncs _thread_source_funcs = {
 259         .prepare = _thread_source_prepare,
 260         .check = _thread_source_check,
 261         .dispatch = _thread_source_dispatch,
 262         .finalize = _thread_source_finalize,
 263 };
 264
 265 tpl_gsource *
 266 tpl_gsource_create(tpl_gthread *thread, void *data, int fd,
 267                                    tpl_gsource_functions *funcs, tpl_gsource_type_t type)
 268 {
 269         tpl_gsource *new_gsource = NULL;
 270
 271         new_gsource = (tpl_gsource *)g_source_new(&_thread_source_funcs,
 272                                   sizeof(tpl_gsource));
 273         if (!new_gsource) {
 274                 TPL_ERR("Failed to create new tpl_gsource");
 275                 return NULL;
 276         }
 277
 278         if (fd < 0) {
 279                 new_gsource->fd = eventfd(0, EFD_CLOEXEC);
 280                 if (new_gsource->fd < 0) {
 281                         TPL_ERR("Failed to create eventfd. errno(%d)", errno);
 282                         g_source_unref(&new_gsource->gsource);
 283                         return NULL;
 284                 }
 285
 286                 new_gsource->is_eventfd = TPL_TRUE;
 287         } else {
 288                 new_gsource->fd = fd;
 289                 new_gsource->is_eventfd = TPL_FALSE;
 290         }
 291
 292         new_gsource->thread        = thread;
 293         new_gsource->gsource_funcs = funcs;
 294         new_gsource->data          = data;
 295         new_gsource->type          = type;
 296         new_gsource->intended_destroy = TPL_FALSE;
 297
 298         if (new_gsource->type == SOURCE_TYPE_NORMAL) {
 299                 tpl_gsource *finalizer = tpl_gsource_create(thread, new_gsource, -1,
 300                                                                                                         NULL, SOURCE_TYPE_FINALIZER);
 301                 new_gsource->finalizer = finalizer;
 302         } else
 303                 new_gsource->finalizer = NULL;
 304
 305         new_gsource->tag = g_source_add_unix_fd(&new_gsource->gsource,
 306                                                                                         new_gsource->fd,
 307                                                                                         G_IO_IN | G_IO_ERR);
 308         g_source_attach(&new_gsource->gsource,
 309                                         g_main_loop_get_context(thread->loop));
 310
 311         TPL_DEBUG("[GSOURCE_CREATE] tpl_gsource(%p) thread(%p) data(%p) fd(%d) type(%d)",
 312                           new_gsource, thread, data, new_gsource->fd, type);
 313
 314         return new_gsource;
 315 }
 316
 317 static void
 318 __gsource_remove_and_destroy(tpl_gsource *source)
 319 {
 320         if (g_source_is_destroyed(&source->gsource))
 321                 return;
 322
 323         TPL_DEBUG("[GSOURCE_DESTROY] tpl_gsource(%p) type(%d)",
 324                           source, source->type);
 325
 326         g_source_remove_unix_fd(&source->gsource, source->tag);
 327         g_source_destroy(&source->gsource);
 328         g_source_unref(&source->gsource);
 329 }
 330
 331 void
 332 tpl_gsource_destroy(tpl_gsource *source, tpl_bool_t destroy_in_thread)
 333 {
 334         tpl_gthread *thread = source->thread;
 335
 336         if (g_source_is_destroyed(&source->gsource)) {
 337                 TPL_WARN("gsource(%p) already has been destroyed.",
 338                                  source);
 339                 return;
 340         }
 341
 342         g_mutex_lock(&thread->thread_mutex);
 343         if (source->type == SOURCE_TYPE_NORMAL &&
 344                 source->finalizer) {
 345                 tpl_gsource *finalizer = source->finalizer;
 346
 347                 if (destroy_in_thread) {
 348                         finalizer->intended_destroy = TPL_TRUE;
 349                         tpl_gsource_send_message(finalizer, 1);
 350                 } else {
 351                         __gsource_remove_and_destroy(finalizer);
 352                         source->finalizer = NULL;
 353                 }
 354         }
 355
 356         if (!destroy_in_thread) {
 357                 __gsource_remove_and_destroy(source);
 358         }
 359         g_mutex_unlock(&thread->thread_mutex);
 360 }
 361
 362 void
 363 tpl_gsource_send_message(tpl_gsource *source, uint64_t message)
 364 {
 365         uint64_t value = message;
 366         int ret;
 367
 368         if (!source->is_eventfd) {
 369                 TPL_ERR("source is not using eventfd. source(%p) fd(%d)",
 370                                 source, source->fd);
 371                 return;
 372         }
 373
 374         ret = write(source->fd, &value, sizeof(uint64_t));
 375         if (ret == -1) {
 376                 TPL_ERR("failed to send devent. tpl_gsource(%p)",
 377                                 source);
 378         }
 379 }
 380
 381 void *
 382 tpl_gsource_get_data(tpl_gsource *source)
 383 {
 384         void *data = NULL;
 385
 386         if (source)
 387                 data = source->data;
 388
 389         return data;
 390 }
 391
 392 tpl_bool_t
 393 tpl_gsource_check_io_condition(tpl_gsource *source)
 394 {
 395         GIOCondition cond;
 396
 397         if (!source) {
 398                 TPL_ERR("Invalid parameter tpl_gsource is null");
 399                 return TPL_FALSE;
 400         }
 401
 402         cond = g_source_query_unix_fd(&source->gsource, source->tag);
 403         if (cond & G_IO_IN)
 404                 return TPL_TRUE;
 405
 406         return TPL_FALSE;
 407 }
 408
 409 void
 410 tpl_gmutex_init(tpl_gmutex *gmutex)
 411 {
 412         g_mutex_init(gmutex);
 413 }
 414
 415 void
 416 tpl_gmutex_clear(tpl_gmutex *gmutex)
 417 {
 418         g_mutex_clear(gmutex);
 419 }
 420
 421 void
 422 tpl_gmutex_lock(tpl_gmutex *gmutex)
 423 {
 424         g_mutex_lock(gmutex);
 425 }
 426
 427 void
 428 tpl_gmutex_unlock(tpl_gmutex *gmutex)
 429 {
 430         g_mutex_unlock(gmutex);
 431 }
 432
 433 void
 434 tpl_gcond_init(tpl_gcond *gcond)
 435 {
 436         g_cond_init(gcond);
 437 }
 438
 439 void
 440 tpl_gcond_clear(tpl_gcond *gcond)
 441 {
 442         g_cond_clear(gcond);
 443 }
 444
 445 void
 446 tpl_gcond_wait(tpl_gcond *gcond, tpl_gmutex *gmutex)
 447 {
 448         g_cond_wait(gcond, gmutex);
 449 }
 450
 451 tpl_result_t
 452 tpl_gcond_timed_wait(tpl_gcond *gcond, tpl_gmutex *gmutex,
 453                                         int64_t timeout_ms)
 454 {
 455         gint64 end_time = g_get_monotonic_time() +
 456                                                 (timeout_ms * G_TIME_SPAN_MILLISECOND);
 457         if (!g_cond_wait_until(gcond, gmutex, end_time))
 458                 return TPL_ERROR_TIME_OUT;
 459
 460         return TPL_ERROR_NONE;
 461 }
 462
 463 void
 464 tpl_gcond_signal(tpl_gcond *gcond)
 465 {
 466         g_cond_signal(gcond);
 467 }
 468
 469 static gboolean
 470 _thread_idle_cb(gpointer data)
 471 {
 472         tpl_gthread *gthread = (tpl_gthread *)data;
 473
 474         TPL_DEBUG("THREAD IDLE CALLBACK");
 475
 476         g_mutex_lock(&gthread->idle_mutex);
 477         gthread->is_idle = TPL_TRUE;
 478         g_cond_signal(&gthread->idle_cond);
 479         g_mutex_unlock(&gthread->idle_mutex);
 480
 481         /* If the caller thread of tpl_gthread_wait_idle locked the pause_mutex,
 482          * thread will be paused here until unlock */
 483         g_mutex_lock(&gthread->pause_mutex);
 484         g_mutex_unlock(&gthread->pause_mutex);
 485
 486         return G_SOURCE_REMOVE;
 487 }
 488
 489 void
 490 tpl_gthread_wait_idle(tpl_gthread *gthread)
 491 {
 492         TPL_CHECK_ON_NULL_RETURN(gthread);
 493
 494         GSource *idle_source = NULL;
 495         gint64 end_time;
 496         gboolean ret = TRUE;
 497
 498         TPL_DEBUG("[WAIT IDLE] BEGIN");
 499
 500         g_mutex_lock(&gthread->idle_mutex);
 501         gthread->is_idle = TPL_FALSE;
 502
 503         idle_source = g_idle_source_new();
 504         if (idle_source == NULL) {
 505                 TPL_WARN("Failed to create and attach idle source");
 506                 g_mutex_unlock(&gthread->idle_mutex);
 507                 return;
 508         }
 509
 510         g_source_set_priority(idle_source, G_PRIORITY_LOW);
 511         g_source_set_callback(idle_source,
 512                                                   _thread_idle_cb, (gpointer)gthread,
 513                                                   NULL);
 514         g_source_attach(idle_source, g_main_loop_get_context(gthread->loop));
 515         g_source_unref(idle_source);
 516
 517         /* 200ms timeout */
 518         end_time = g_get_monotonic_time() +
 519                                 (200 * G_TIME_SPAN_MILLISECOND);
 520
 521         while (!gthread->is_idle) {
 522                 ret = g_cond_wait_until(&gthread->idle_cond,
 523                                                                 &gthread->idle_mutex,
 524                                                                 end_time);
 525                 if (!ret) {
 526                         TPL_ERR("wait_idle timeout!");
 527                         break;
 528                 }
 529         }
 530         g_mutex_unlock(&gthread->idle_mutex);
 531
 532         TPL_DEBUG("[WAIT IDLE] END");
 533 }
 534
 535 void
 536 tpl_gthread_pause_in_idle(tpl_gthread *gthread)
 537 {
 538         TPL_CHECK_ON_NULL_RETURN(gthread);
 539
 540         g_mutex_lock(&gthread->pause_mutex);
 541         tpl_gthread_wait_idle(gthread);
 542         gthread->paused = TPL_TRUE;
 543 }
 544
 545 void
 546 tpl_gthread_continue(tpl_gthread *gthread)
 547 {
 548         TPL_CHECK_ON_NULL_RETURN(gthread);
 549
 550         if (!gthread->paused) return;
 551         gthread->paused = TPL_FALSE;
 552         g_mutex_unlock(&gthread->pause_mutex);
 553 }