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         fd_type_t              fd_type;
  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->fd_type == FD_TYPE_EVENT) {
 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->fd_type == FD_TYPE_EVENT ||
 249                 gsource->fd_type == FD_TYPE_FENCE)
 250                 close(gsource->fd);
 251
 252         gsource->fd = -1;
 253         gsource->thread = NULL;
 254         gsource->gsource_funcs = NULL;
 255         gsource->data = NULL;
 256         gsource->finalizer = NULL;
 257 }
 258
 259 static GSourceFuncs _thread_source_funcs = {
 260         .prepare = _thread_source_prepare,
 261         .check = _thread_source_check,
 262         .dispatch = _thread_source_dispatch,
 263         .finalize = _thread_source_finalize,
 264 };
 265
 266 tpl_gsource *
 267 tpl_gsource_create(tpl_gthread *thread, void *data, int fd, fd_type_t fd_type,
 268                                    tpl_gsource_functions *funcs, tpl_gsource_type_t type)
 269 {
 270         tpl_gsource *new_gsource = NULL;
 271
 272         new_gsource = (tpl_gsource *)g_source_new(&_thread_source_funcs,
 273                                   sizeof(tpl_gsource));
 274         if (!new_gsource) {
 275                 TPL_ERR("Failed to create new tpl_gsource");
 276                 return NULL;
 277         }
 278
 279         if (fd < 0) {
 280                 new_gsource->fd = eventfd(0, EFD_CLOEXEC);
 281                 if (new_gsource->fd < 0) {
 282                         TPL_ERR("Failed to create eventfd. errno(%d)", errno);
 283                         g_source_unref(&new_gsource->gsource);
 284                         return NULL;
 285                 }
 286
 287                 new_gsource->fd_type = FD_TYPE_EVENT;
 288         } else {
 289                 new_gsource->fd = fd;
 290                 new_gsource->fd_type = fd_type;
 291         }
 292
 293         new_gsource->thread        = thread;
 294         new_gsource->gsource_funcs = funcs;
 295         new_gsource->data          = data;
 296         new_gsource->type          = type;
 297         new_gsource->intended_destroy = TPL_FALSE;
 298
 299         if (new_gsource->type == SOURCE_TYPE_NORMAL) {
 300                 tpl_gsource *finalizer = tpl_gsource_create(thread, new_gsource, -1, FD_TYPE_NONE,
 301                                                                                                         NULL, SOURCE_TYPE_FINALIZER);
 302                 new_gsource->finalizer = finalizer;
 303         } else
 304                 new_gsource->finalizer = NULL;
 305
 306         new_gsource->tag = g_source_add_unix_fd(&new_gsource->gsource,
 307                                                                                         new_gsource->fd,
 308                                                                                         G_IO_IN | G_IO_ERR);
 309         g_source_attach(&new_gsource->gsource,
 310                                         g_main_loop_get_context(thread->loop));
 311
 312         TPL_LOG_D("[GSOURCE][CREATE]", "tpl_gsource(%p) thread(%p) data(%p) fd(%d) type(%d)",
 313                           new_gsource, thread, data, new_gsource->fd, type);
 314
 315         return new_gsource;
 316 }
 317
 318 static void
 319 __gsource_remove_and_destroy(tpl_gsource *source)
 320 {
 321         if (g_source_is_destroyed(&source->gsource))
 322                 return;
 323
 324         TPL_LOG_D("[GSOURCE][DESTROY]", "tpl_gsource(%p) type(%d)",
 325                           source, source->type);
 326
 327         g_source_remove_unix_fd(&source->gsource, source->tag);
 328         g_source_destroy(&source->gsource);
 329         g_source_unref(&source->gsource);
 330 }
 331
 332 void
 333 tpl_gsource_destroy(tpl_gsource *source, tpl_bool_t destroy_in_thread)
 334 {
 335         tpl_gthread *thread = source->thread;
 336
 337         if (g_source_is_destroyed(&source->gsource)) {
 338                 TPL_WARN("gsource(%p) already has been destroyed.",
 339                                  source);
 340                 return;
 341         }
 342
 343         g_mutex_lock(&thread->thread_mutex);
 344         if (source->type == SOURCE_TYPE_NORMAL &&
 345                 source->finalizer) {
 346                 tpl_gsource *finalizer = source->finalizer;
 347
 348                 if (destroy_in_thread) {
 349                         finalizer->intended_destroy = TPL_TRUE;
 350                         tpl_gsource_send_message(finalizer, 1);
 351                 } else {
 352                         __gsource_remove_and_destroy(finalizer);
 353                         source->finalizer = NULL;
 354                 }
 355         }
 356
 357         if (!destroy_in_thread) {
 358                 if (source->fd_type == FD_TYPE_FENCE &&
 359                         source->type == SOURCE_TYPE_DISPOSABLE) {
 360                         TPL_LOG_D("[GSOURCE][DESTROY]", "tpl_gsource(%p) type(%d)",
 361                                           source, source->type);
 362
 363                         g_source_remove_unix_fd(&source->gsource, source->tag);
 364                         source->data = NULL;
 365                         g_source_destroy(&source->gsource);
 366                         g_source_unref(&source->gsource);
 367                 }
 368                 else
 369                         __gsource_remove_and_destroy(source);
 370         }
 371
 372         g_mutex_unlock(&thread->thread_mutex);
 373 }
 374
 375 void
 376 tpl_gsource_send_message(tpl_gsource *source, uint64_t message)
 377 {
 378         uint64_t value = message;
 379         int ret;
 380
 381         if (source->fd_type != FD_TYPE_EVENT) {
 382                 TPL_ERR("source is not using eventfd. source(%p) fd(%d)",
 383                                 source, source->fd);
 384                 return;
 385         }
 386
 387         ret = write(source->fd, &value, sizeof(uint64_t));
 388         if (ret == -1) {
 389                 TPL_ERR("failed to send devent. tpl_gsource(%p)",
 390                                 source);
 391         }
 392 }
 393
 394 void *
 395 tpl_gsource_get_data(tpl_gsource *source)
 396 {
 397         void *data = NULL;
 398
 399         if (source)
 400                 data = source->data;
 401
 402         return data;
 403 }
 404
 405 tpl_bool_t
 406 tpl_gsource_check_io_condition(tpl_gsource *source)
 407 {
 408         GIOCondition cond;
 409
 410         if (!source) {
 411                 TPL_ERR("Invalid parameter tpl_gsource is null");
 412                 return TPL_FALSE;
 413         }
 414
 415         cond = g_source_query_unix_fd(&source->gsource, source->tag);
 416         if (cond & G_IO_IN)
 417                 return TPL_TRUE;
 418
 419         return TPL_FALSE;
 420 }
 421
 422 void
 423 tpl_gmutex_init(tpl_gmutex *gmutex)
 424 {
 425         g_mutex_init(gmutex);
 426 }
 427
 428 void
 429 tpl_gmutex_clear(tpl_gmutex *gmutex)
 430 {
 431         g_mutex_clear(gmutex);
 432 }
 433
 434 void
 435 tpl_gmutex_lock(tpl_gmutex *gmutex)
 436 {
 437         g_mutex_lock(gmutex);
 438 }
 439
 440 void
 441 tpl_gmutex_unlock(tpl_gmutex *gmutex)
 442 {
 443         g_mutex_unlock(gmutex);
 444 }
 445
 446 void
 447 tpl_gcond_init(tpl_gcond *gcond)
 448 {
 449         g_cond_init(gcond);
 450 }
 451
 452 void
 453 tpl_gcond_clear(tpl_gcond *gcond)
 454 {
 455         g_cond_clear(gcond);
 456 }
 457
 458 void
 459 tpl_gcond_wait(tpl_gcond *gcond, tpl_gmutex *gmutex)
 460 {
 461         g_cond_wait(gcond, gmutex);
 462 }
 463
 464 tpl_result_t
 465 tpl_gcond_timed_wait(tpl_gcond *gcond, tpl_gmutex *gmutex,
 466                                         int64_t timeout_ms)
 467 {
 468         gint64 end_time = g_get_monotonic_time() +
 469                                                 (timeout_ms * G_TIME_SPAN_MILLISECOND);
 470         if (!g_cond_wait_until(gcond, gmutex, end_time))
 471                 return TPL_ERROR_TIME_OUT;
 472
 473         return TPL_ERROR_NONE;
 474 }
 475
 476 void
 477 tpl_gcond_signal(tpl_gcond *gcond)
 478 {
 479         g_cond_signal(gcond);
 480 }
 481
 482 static gboolean
 483 _thread_idle_cb(gpointer data)
 484 {
 485         tpl_gthread *gthread = (tpl_gthread *)data;
 486
 487         TPL_LOG_D("[WAIT_IDLE]", "THREAD IDLE CALLBACK");
 488
 489         g_mutex_lock(&gthread->idle_mutex);
 490         gthread->is_idle = TPL_TRUE;
 491         g_cond_broadcast(&gthread->idle_cond);
 492         g_mutex_unlock(&gthread->idle_mutex);
 493
 494         /* If the caller thread of tpl_gthread_wait_idle locked the pause_mutex,
 495          * thread will be paused here until unlock */
 496         TPL_LOG_D("[THREAD_PAUSE]", "try to lock pause_mutex");
 497         g_mutex_lock(&gthread->pause_mutex);
 498         gthread->paused = TPL_FALSE;
 499         g_mutex_unlock(&gthread->pause_mutex);
 500         TPL_LOG_D("[THREAD_RESUME]", "thread resumes");
 501
 502         return G_SOURCE_REMOVE;
 503 }
 504
 505 #define WAIT_IDLE_TIMEOUT 500
 506
 507 tpl_result_t
 508 tpl_gthread_wait_idle(tpl_gthread *gthread)
 509 {
 510         GSource *idle_source = NULL;
 511         gint64 end_time;
 512         gboolean ret = TRUE;
 513         tpl_result_t res = TPL_ERROR_NONE;
 514
 515         TPL_LOG_D("[WAIT_IDLE]", "BEGIN");
 516
 517         g_mutex_lock(&gthread->idle_mutex);
 518
 519         idle_source = g_idle_source_new();
 520         if (idle_source == NULL) {
 521                 TPL_WARN("Failed to create and attach idle source");
 522                 res = TPL_ERROR_INVALID_OPERATION;
 523                 g_mutex_unlock(&gthread->idle_mutex);
 524                 return res;
 525         }
 526
 527         g_source_set_priority(idle_source, G_PRIORITY_LOW);
 528         g_source_set_callback(idle_source,
 529                                                   _thread_idle_cb, (gpointer)gthread,
 530                                                   NULL);
 531         g_source_attach(idle_source, g_main_loop_get_context(gthread->loop));
 532         g_source_unref(idle_source);
 533
 534         /* 200ms timeout */
 535         end_time = g_get_monotonic_time() +
 536                                 (WAIT_IDLE_TIMEOUT * G_TIME_SPAN_MILLISECOND);
 537         do {
 538                 ret = g_cond_wait_until(&gthread->idle_cond,
 539                                                                 &gthread->idle_mutex,
 540                                                                 end_time);
 541                 if (!ret) {
 542                         TPL_ERR("wait_idle timeout!");
 543                         res = TPL_ERROR_TIME_OUT;
 544                         break;
 545                 }
 546         } while (!gthread->is_idle);
 547
 548         gthread->is_idle = TPL_FALSE;
 549
 550         g_mutex_unlock(&gthread->idle_mutex);
 551
 552         TPL_LOG_D("[WAIT_IDLE]", "END");
 553
 554         return res;
 555 }
 556
 557 tpl_bool_t
 558 tpl_gthread_pause_in_idle(tpl_gthread *gthread)
 559 {
 560         TPL_CHECK_ON_NULL_RETURN_VAL(gthread, TPL_FALSE);
 561
 562         tpl_result_t res;
 563         int cnt = 0;
 564
 565         /* Assume three threads. (M, C, wl-egl-thread)
 566          * C thread : already locked pause_mutex and doing their own job.
 567          * M thread : call pause_in_idle and trying to lock pause_mutex.
 568          * wl-egl-thread : trying to lock pause_mutex in _thread_idle_cb.
 569          *
 570          * When C thread calls tpl_gthread_continue and unlock pause_mutex,
 571          * M thread may receive schedule and lock pause_mutex.
 572          * In that case, M thread should yield to wl-egl-thread, which is
 573          * paused in thread_idle_cb(). */
 574         do {
 575                 g_mutex_lock(&gthread->pause_mutex);
 576                 if (gthread->paused) {
 577                         g_mutex_unlock(&gthread->pause_mutex);
 578                         sched_yield();
 579                 } else {
 580                         break;
 581                 }
 582         } while (++cnt <= 100);
 583
 584         res = tpl_gthread_wait_idle(gthread);
 585         if (res != TPL_ERROR_NONE) {
 586                 TPL_ERR("Failed to wait idle. | res(%d)", res);
 587                 gthread->paused = TPL_FALSE;
 588                 g_mutex_unlock(&gthread->pause_mutex);
 589         } else {
 590                 gthread->paused = TPL_TRUE;
 591         }
 592
 593         return gthread->paused;
 594 }
 595
 596 void
 597 tpl_gthread_continue(tpl_gthread *gthread)
 598 {
 599         TPL_CHECK_ON_NULL_RETURN(gthread);
 600
 601         if (!gthread->paused) return;
 602         g_mutex_unlock(&gthread->pause_mutex);
 603 }