src/examples/ecore_thread_example.c

   1 /*
   2  * gcc -o ecore_thread_example ecore_thread_example.c `pkg-config --cflags --libs ecore`
   3  */
   4 #include <stdio.h>
   5 #include <stdlib.h>
   6 #include <dirent.h>
   7 #include <Ecore.h>
   8 #include <Ecore_Getopt.h>
   9
  10 typedef struct
  11 {
  12    Ecore_Thread  *thread_3;
  13    int            msgs_received;
  14    int            max_msgs;
  15    Eina_Lock      mutex;
  16    Eina_Condition condition;
  17 } App_Data;
  18
  19 typedef struct
  20 {
  21    Eina_List *list;
  22 } Thread_Data;
  23
  24 typedef struct
  25 {
  26    char     *name;
  27    char     *base;
  28    Eina_Lock mutex;
  29 } Feedback_Thread_Data;
  30
  31 typedef struct
  32 {
  33    int        all_done;
  34    Eina_List *list;
  35 } App_Msg;
  36
  37 static void
  38 _local_data_free(void *data)
  39 {
  40    Thread_Data *td = data;
  41    char *str;
  42
  43    EINA_LIST_FREE(td->list, str)
  44      {
  45         printf("Freeing string: %s\n", str);
  46         free(str);
  47      }
  48    free(td);
  49 }
  50
  51 static void
  52 _short_job(void *data, Ecore_Thread *th)
  53 {
  54    Thread_Data *td;
  55    int i;
  56
  57    td = ecore_thread_local_data_find(th, "data");
  58    if (!td)
  59      {
  60         td = calloc(1, sizeof(Thread_Data));
  61         if (!td)
  62           {
  63              ecore_thread_cancel(th);
  64              return;
  65           }
  66         ecore_thread_local_data_add(th, "data", td, _local_data_free,
  67                                     EINA_FALSE);
  68      }
  69
  70    for (i = 0; i < 10; i++)
  71      {
  72         char buf[200];
  73
  74         if (ecore_thread_check(th))
  75           {
  76              ecore_thread_local_data_del(th, "data");
  77              break;
  78           }
  79
  80         snprintf(buf, sizeof(buf), "Thread %p: String number %d", th, i);
  81         td->list = eina_list_append(td->list, strdup(buf));
  82         sleep(1);
  83      }
  84 }
  85
  86 static void
  87 _feedback_job(void *data, Ecore_Thread *th)
  88 {
  89    time_t t;
  90    int i, count;
  91    Feedback_Thread_Data *ftd = NULL;
  92    DIR *dir;
  93    App_Msg *msg;
  94
  95    count = (int)ecore_thread_global_data_find("count");
  96    for (i = 0; i < count; i++)
  97      {
  98         char buf[32];
  99         snprintf(buf, sizeof(buf), "data%d", i);
 100         ftd = ecore_thread_global_data_find(buf);
 101         if (!ftd)
 102           continue;
 103         if (eina_lock_take_try(&ftd->mutex))
 104           break;
 105         else
 106           ftd = NULL;
 107      }
 108    if (!ftd)
 109      return;
 110
 111    dir = opendir(ftd->base);
 112    if (!dir)
 113      goto the_end;
 114
 115    msg = calloc(1, sizeof(App_Msg));
 116
 117    t = time(NULL);
 118    while (time(NULL) < t + 2)
 119      {
 120         struct dirent entry, *result;
 121
 122         if (readdir_r(dir, &entry, &result))
 123           break;
 124         if (!result)
 125           break;
 126
 127         if (strlen(result->d_name) >= 10)
 128           msg->list = eina_list_append(msg->list,
 129                                        strdup(result->d_name));
 130      }
 131
 132    closedir(dir);
 133    ecore_thread_feedback(th, msg);
 134
 135 the_end:
 136    ecore_thread_global_data_del(ftd->name);
 137    free(ftd->name);
 138    free(ftd->base);
 139    eina_lock_release(&ftd->mutex);
 140    eina_lock_free(&ftd->mutex);
 141    free(ftd);
 142    ecore_thread_reschedule(th);
 143 }
 144
 145 static void
 146 _out_of_pool_job(void *data, Ecore_Thread *th)
 147 {
 148    App_Data *ad = data;
 149    App_Msg *msg;
 150
 151    while (1)
 152      {
 153         int msgs;
 154         eina_condition_wait(&ad->condition);
 155         msgs = ad->msgs_received;
 156         eina_lock_release(&ad->mutex);
 157         if (msgs == ad->max_msgs)
 158           {
 159              msg = calloc(1, sizeof(App_Msg));
 160              msg->all_done = 1;
 161              ecore_thread_feedback(th, msg);
 162              return;
 163           }
 164      }
 165 }
 166
 167 static void
 168 _print_status(void)
 169 {
 170    int active, pending_total, pending_feedback, pending_short, available;
 171
 172    active = ecore_thread_active_get();
 173    pending_total = ecore_thread_pending_total_get();
 174    pending_feedback = ecore_thread_pending_feedback_get();
 175    pending_short = ecore_thread_pending_get();
 176    available = ecore_thread_available_get();
 177
 178    printf("Status:\n\t* Active threads: %d\n"
 179           "\t* Available threads: %d\n"
 180           "\t* Pending short jobs: %d\n"
 181           "\t* Pending feedback jobs: %d\n"
 182           "\t* Pending total: %d\n", active, available, pending_short,
 183           pending_feedback, pending_total);
 184 }
 185
 186 static void
 187 _feedback_job_msg_cb(void *data, Ecore_Thread *th, void *msg_data)
 188 {
 189    App_Data *ad = data;
 190    App_Msg *msg = msg_data;
 191    char *str;
 192
 193    if (msg->all_done)
 194      {
 195         ecore_main_loop_quit();
 196         free(msg);
 197         return;
 198      }
 199
 200    _print_status();
 201
 202    if (!msg->list)
 203      printf("Received an empty list from thread %p\n", th);
 204    else
 205      {
 206         int i = 0;
 207         printf("Received %d elements from threads %p (printing first 5):\n",
 208                eina_list_count(msg->list), th);
 209         EINA_LIST_FREE(msg->list, str)
 210           {
 211              if (i <= 5)
 212                printf("\t%s\n", str);
 213              free(str);
 214              i++;
 215           }
 216      }
 217
 218    eina_lock_take(&ad->mutex);
 219    ad->msgs_received++;
 220    eina_condition_signal(&ad->condition);
 221    eina_lock_release(&ad->mutex);
 222
 223    free(msg);
 224 }
 225
 226 static void
 227 _thread_end_cb(void *data, Ecore_Thread *th)
 228 {
 229    App_Data *ad = data;
 230
 231    printf("Normal termination for thread %p.\n", th);
 232    if (th == ad->thread_3)
 233      ad->thread_3 = NULL;
 234 }
 235
 236 static void
 237 _thread_cancel_cb(void *data, Ecore_Thread *th)
 238 {
 239    App_Data *ad = data;
 240
 241    printf("Thread %p got cancelled.\n", th);
 242    if (th == ad->thread_3)
 243      ad->thread_3 = NULL;
 244 }
 245
 246 static Eina_Bool
 247 _cancel_timer_cb(void *data)
 248 {
 249    App_Data *ad = data;
 250
 251    if (ad->thread_3 && !ecore_thread_check(ad->thread_3))
 252      ecore_thread_cancel(ad->thread_3);
 253
 254    return EINA_FALSE;
 255 }
 256
 257 static Eina_Bool
 258 _status_timer_cb(void *data)
 259 {
 260    _print_status();
 261
 262    return EINA_TRUE;
 263 }
 264
 265 static const Ecore_Getopt optdesc = {
 266    "ecore_thread_example",
 267    NULL,
 268    "0.0",
 269    "(C) 2011 Enlightenment",
 270    "Public domain?",
 271    "Example program for Ecore_Thread",
 272    0,
 273    {
 274       ECORE_GETOPT_STORE_INT('t', "threads", "Max number of threads to run"),
 275       ECORE_GETOPT_STORE_INT('m', "msgs", "Max number of messages to receive"),
 276       ECORE_GETOPT_APPEND_METAVAR('p', "path", "Add path for feedback job",
 277                                   "STRING", ECORE_GETOPT_TYPE_STR),
 278       ECORE_GETOPT_HELP('h', "help"),
 279       ECORE_GETOPT_SENTINEL
 280    }
 281 };
 282
 283 int
 284 main(int argc, char *argv[])
 285 {
 286    int i, max_threads = 0, max_msgs = 0;
 287    Eina_Bool opt_quit = EINA_FALSE;
 288    Eina_List *path_list = NULL;
 289    App_Data appdata;
 290    Ecore_Getopt_Value values[] = {
 291       ECORE_GETOPT_VALUE_INT(max_threads),
 292       ECORE_GETOPT_VALUE_INT(max_msgs),
 293       ECORE_GETOPT_VALUE_LIST(path_list),
 294       ECORE_GETOPT_VALUE_BOOL(opt_quit),
 295       ECORE_GETOPT_VALUE_NONE
 296    };
 297
 298    ecore_init();
 299
 300    i = ecore_thread_max_get();
 301    printf("Initial max threads: %d\n", i);
 302
 303    memset(&appdata, 0, sizeof(App_Data));
 304    appdata.max_msgs = 1;
 305
 306    if (ecore_getopt_parse(&optdesc, values, argc, argv) < 0)
 307      {
 308         printf("Argument parsing failed\n");
 309         return 1;
 310      }
 311
 312    if (opt_quit)
 313      return 0;
 314
 315    if (max_threads)
 316      {
 317         ecore_thread_max_set(max_threads);
 318         printf("Max threads: %d\n", ecore_thread_max_get());
 319      }
 320    if (max_msgs)
 321      appdata.max_msgs = max_msgs;
 322
 323    if (!path_list)
 324      {
 325         Feedback_Thread_Data *ftd;
 326         ecore_thread_global_data_add("count", (void *)3, NULL, EINA_FALSE);
 327         ftd = calloc(1, sizeof(Feedback_Thread_Data));
 328         ftd->name = strdup("data0");
 329         ftd->base = strdup("/usr/bin");
 330         eina_lock_new(&ftd->mutex);
 331         ecore_thread_global_data_add(ftd->name, ftd, NULL, EINA_TRUE);
 332         ftd = calloc(1, sizeof(Feedback_Thread_Data));
 333         ftd->name = strdup("data1");
 334         ftd->base = strdup("/usr/lib");
 335         eina_lock_new(&ftd->mutex);
 336         ecore_thread_global_data_add(ftd->name, ftd, NULL, EINA_TRUE);
 337         ftd = calloc(1, sizeof(Feedback_Thread_Data));
 338         ftd->name = strdup("data2");
 339         ftd->base = strdup("/usr/share");
 340         eina_lock_new(&ftd->mutex);
 341         ecore_thread_global_data_add(ftd->name, ftd, NULL, EINA_TRUE);
 342      }
 343    else
 344      {
 345         Feedback_Thread_Data *ftd;
 346         char *str;
 347         ecore_thread_global_data_add("count",
 348                                      (void *)eina_list_count(path_list), NULL,
 349                                      EINA_FALSE);
 350         i = 0;
 351         EINA_LIST_FREE(path_list, str)
 352           {
 353              char buf[32];
 354              snprintf(buf, sizeof(buf), "data%d", i);
 355              ftd = calloc(1, sizeof(Feedback_Thread_Data));
 356              ftd->name = strdup(buf);
 357              ftd->base = strdup(str);
 358              eina_lock_new(&ftd->mutex);
 359              ecore_thread_global_data_add(ftd->name, ftd, NULL, EINA_TRUE);
 360              free(str);
 361              i++;
 362           }
 363      }
 364
 365    eina_lock_new(&appdata.mutex);
 366    eina_condition_new(&appdata.condition, &appdata.mutex);
 367
 368    ecore_thread_feedback_run(_out_of_pool_job, _feedback_job_msg_cb, NULL,
 369                              NULL, &appdata, EINA_TRUE);
 370
 371    ecore_thread_run(_short_job, _thread_end_cb, _thread_cancel_cb, &appdata);
 372    ecore_thread_feedback_run(_feedback_job, _feedback_job_msg_cb,
 373                              _thread_end_cb, _thread_cancel_cb, &appdata,
 374                              EINA_FALSE);
 375    appdata.thread_3 = ecore_thread_run(_short_job, _thread_end_cb,
 376                                        _thread_cancel_cb, &appdata);
 377    ecore_thread_feedback_run(_feedback_job, _feedback_job_msg_cb,
 378                              _thread_end_cb, _thread_cancel_cb, &appdata,
 379                              EINA_FALSE);
 380
 381    ecore_timer_add(1.0, _cancel_timer_cb, &appdata);
 382    ecore_timer_add(2.0, _status_timer_cb, NULL);
 383
 384    _print_status();
 385
 386    ecore_main_loop_begin();
 387
 388    eina_condition_free(&appdata.condition);
 389    eina_lock_free(&appdata.mutex);
 390
 391    ecore_shutdown();
 392
 393    return 0;
 394 }
 395