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                 if (gsource->type == SOURCE_TYPE_DISPOSABLE) {
 207                         if (gsource->gsource_funcs && gsource->gsource_funcs->dispatch)
 208                                 ret = gsource->gsource_funcs->dispatch(gsource, 0);
 209                 }
 210         }
 211
 212         if (gsource->type == SOURCE_TYPE_DISPOSABLE) {
 213                 g_mutex_lock(&thread->thread_mutex);
 214                 __gsource_remove_and_destroy(gsource);
 215                 ret = G_SOURCE_REMOVE;
 216                 g_mutex_unlock(&thread->thread_mutex);
 217         }
 218
 219         return ret;
 220 }
 221
 222 static void
 223 _thread_source_finalize(GSource *source)
 224 {
 225         tpl_gsource *gsource = (tpl_gsource *)source;
 226
 227         if (gsource->gsource_funcs && gsource->gsource_funcs->finalize)
 228                 gsource->gsource_funcs->finalize(gsource);
 229
 230         if (gsource->is_eventfd)
 231                 close(gsource->fd);
 232
 233         gsource->fd = -1;
 234         gsource->thread = NULL;
 235         gsource->gsource_funcs = NULL;
 236         gsource->data = NULL;
 237         gsource->finalizer = NULL;
 238 }
 239
 240 static GSourceFuncs _thread_source_funcs = {
 241         .prepare = _thread_source_prepare,
 242         .check = _thread_source_check,
 243         .dispatch = _thread_source_dispatch,
 244         .finalize = _thread_source_finalize,
 245 };
 246
 247 tpl_gsource *
 248 tpl_gsource_create(tpl_gthread *thread, void *data, int fd,
 249                                    tpl_gsource_functions *funcs, tpl_gsource_type_t type)
 250 {
 251         tpl_gsource *new_gsource = NULL;
 252
 253         new_gsource = (tpl_gsource *)g_source_new(&_thread_source_funcs,
 254                                   sizeof(tpl_gsource));
 255         if (!new_gsource) {
 256                 TPL_ERR("Failed to create new tpl_gsource");
 257                 return NULL;
 258         }
 259
 260         if (fd < 0) {
 261                 new_gsource->fd = eventfd(0, EFD_CLOEXEC);
 262                 if (new_gsource->fd < 0) {
 263                         TPL_ERR("Failed to create eventfd. errno(%d)", errno);
 264                         g_source_unref(&new_gsource->gsource);
 265                         return NULL;
 266                 }
 267
 268                 new_gsource->is_eventfd = TPL_TRUE;
 269         } else {
 270                 new_gsource->fd = fd;
 271                 new_gsource->is_eventfd = TPL_FALSE;
 272         }
 273
 274         new_gsource->thread        = thread;
 275         new_gsource->gsource_funcs = funcs;
 276         new_gsource->data          = data;
 277         new_gsource->type          = type;
 278         new_gsource->intended_destroy = TPL_FALSE;
 279
 280         if (new_gsource->type == SOURCE_TYPE_NORMAL) {
 281                 tpl_gsource *finalizer = tpl_gsource_create(thread, new_gsource, -1,
 282                                                                                                         NULL, SOURCE_TYPE_FINALIZER);
 283                 new_gsource->finalizer = finalizer;
 284         } else
 285                 new_gsource->finalizer = NULL;
 286
 287         new_gsource->tag = g_source_add_unix_fd(&new_gsource->gsource,
 288                                                                                         new_gsource->fd,
 289                                                                                         G_IO_IN | G_IO_ERR);
 290         g_source_attach(&new_gsource->gsource,
 291                                         g_main_loop_get_context(thread->loop));
 292
 293         TPL_DEBUG("[GSOURCE_CREATE] tpl_gsource(%p) thread(%p) data(%p) fd(%d) type(%d)",
 294                           new_gsource, thread, data, new_gsource->fd, type);
 295
 296         return new_gsource;
 297 }
 298
 299 static void
 300 __gsource_remove_and_destroy(tpl_gsource *source)
 301 {
 302         if (g_source_is_destroyed(&source->gsource))
 303                 return;
 304
 305         TPL_DEBUG("[GSOURCE_DESTROY] tpl_gsource(%p) type(%d)",
 306                           source, source->type);
 307
 308         g_source_remove_unix_fd(&source->gsource, source->tag);
 309         g_source_destroy(&source->gsource);
 310         g_source_unref(&source->gsource);
 311 }
 312
 313 void
 314 tpl_gsource_destroy(tpl_gsource *source, tpl_bool_t destroy_in_thread)
 315 {
 316         tpl_gthread *thread = source->thread;
 317
 318         if (g_source_is_destroyed(&source->gsource)) {
 319                 TPL_WARN("gsource(%p) already has been destroyed.",
 320                                  source);
 321                 return;
 322         }
 323
 324         g_mutex_lock(&thread->thread_mutex);
 325         if (source->type == SOURCE_TYPE_NORMAL &&
 326                 source->finalizer) {
 327                 tpl_gsource *finalizer = source->finalizer;
 328
 329                 if (destroy_in_thread) {
 330                         finalizer->intended_destroy = TPL_TRUE;
 331                         tpl_gsource_send_message(finalizer, 1);
 332
 333                         g_cond_wait(&thread->thread_cond, &thread->thread_mutex);
 334                 } else {
 335                         __gsource_remove_and_destroy(finalizer);
 336                         source->finalizer = NULL;
 337                 }
 338         }
 339
 340         if (!destroy_in_thread) {
 341                 __gsource_remove_and_destroy(source);
 342         }
 343         g_mutex_unlock(&thread->thread_mutex);
 344 }
 345
 346 void
 347 tpl_gsource_send_message(tpl_gsource *source, uint64_t message)
 348 {
 349         uint64_t value = message;
 350         int ret;
 351
 352         if (!source->is_eventfd) {
 353                 TPL_ERR("source is not using eventfd. source(%p) fd(%d)",
 354                                 source, source->fd);
 355                 return;
 356         }
 357
 358         ret = write(source->fd, &value, sizeof(uint64_t));
 359         if (ret == -1) {
 360                 TPL_ERR("failed to send devent. tpl_gsource(%p)",
 361                                 source);
 362         }
 363 }
 364
 365 void *
 366 tpl_gsource_get_data(tpl_gsource *source)
 367 {
 368         void *data = NULL;
 369
 370         if (source)
 371                 data = source->data;
 372
 373         return data;
 374 }
 375
 376 tpl_bool_t
 377 tpl_gsource_check_io_condition(tpl_gsource *source)
 378 {
 379         GIOCondition cond;
 380
 381         if (!source) {
 382                 TPL_ERR("Invalid parameter tpl_gsource is null");
 383                 return TPL_FALSE;
 384         }
 385
 386         cond = g_source_query_unix_fd(&source->gsource, source->tag);
 387         if (cond & G_IO_IN)
 388                 return TPL_TRUE;
 389
 390         return TPL_FALSE;
 391 }
 392
 393 void
 394 tpl_gmutex_init(tpl_gmutex *gmutex)
 395 {
 396         g_mutex_init(gmutex);
 397 }
 398
 399 void
 400 tpl_gmutex_clear(tpl_gmutex *gmutex)
 401 {
 402         g_mutex_clear(gmutex);
 403 }
 404
 405 void
 406 tpl_gmutex_lock(tpl_gmutex *gmutex)
 407 {
 408         g_mutex_lock(gmutex);
 409 }
 410
 411 void
 412 tpl_gmutex_unlock(tpl_gmutex *gmutex)
 413 {
 414         g_mutex_unlock(gmutex);
 415 }
 416
 417 void
 418 tpl_gcond_init(tpl_gcond *gcond)
 419 {
 420         g_cond_init(gcond);
 421 }
 422
 423 void
 424 tpl_gcond_clear(tpl_gcond *gcond)
 425 {
 426         g_cond_clear(gcond);
 427 }
 428
 429 void
 430 tpl_gcond_wait(tpl_gcond *gcond, tpl_gmutex *gmutex)
 431 {
 432         g_cond_wait(gcond, gmutex);
 433 }
 434
 435 tpl_result_t
 436 tpl_cond_timed_wait(tpl_gcond *gcond, tpl_gmutex *gmutex,
 437                                         int64_t timeout_ms)
 438 {
 439         gint64 end_time = g_get_monotonic_time() +
 440                                                 (timeout_ms * G_TIME_SPAN_MILLISECOND);
 441         if (!g_cond_wait_until(gcond, gmutex, end_time))
 442                 return TPL_ERROR_TIME_OUT;
 443
 444         return TPL_ERROR_NONE;
 445 }
 446
 447 void
 448 tpl_gcond_signal(tpl_gcond *gcond)
 449 {
 450         g_cond_signal(gcond);
 451 }