Line data    Source code

       1              : /*
       2              :  * SPDX-License-Identifier: GPL-3.0-or-later
       3              :  *
       4              :  * This file is part of colopresso
       5              :  *
       6              :  * Copyright (C) 2025-2026 COLOPL, Inc.
       7              :  *
       8              :  * Author: Go Kudo <g-kudo@colopl.co.jp>
       9              :  * Developed with AI (LLM) code assistance. See `NOTICE` for details.
      10              :  */
      11              : 
      12              : #include <colopresso.h>
      13              : #include <colopresso/portable.h>
      14              : 
      15              : #include "internal/log.h"
      16              : #include "internal/threads.h"
      17              : 
      18              : #if COLOPRESSO_ENABLE_THREADS
      19              : 
      20              : #include <stdlib.h>
      21              : 
      22              : typedef struct {
      23              :     parallel_func_t func;
      24              :     void *context;
      25              :     uint32_t start_index;
      26              :     uint32_t end_index;
      27              :     bool created;
      28              : } thread_work_t;
      29              : 
      30           47 : static void *thread_worker(void *arg)
      31              : {
      32           47 :     thread_work_t *work = (thread_work_t *)arg;
      33              : 
      34           47 :     if (work && work->func) {
      35           47 :         work->func(work->context, work->start_index, work->end_index);
      36              :     }
      37              : 
      38           47 :     return NULL;
      39              : }
      40              : 
      41           20 : extern uint32_t cpres_get_default_thread_count(void)
      42              : {
      43           20 :     uint32_t cpu_count = colopresso_get_cpu_count(), half = (uint32_t)(cpu_count / 2);
      44              : 
      45           20 :     return half > 0 ? half : 1;
      46              : }
      47              : 
      48           94 : extern uint32_t cpres_get_max_thread_count(void)
      49              : {
      50           94 :     return colopresso_get_cpu_count();
      51              : }
      52              : 
      53            1 : extern bool cpres_is_threads_enabled(void)
      54              : {
      55            1 :     return true;
      56              : }
      57              : 
      58           50 : extern bool colopresso_parallel_for(uint32_t use_threads, uint32_t total_items,
      59              :                                     parallel_func_t func, void *context)
      60              : {
      61              :     colopresso_thread_t *threads;
      62              :     thread_work_t *works;
      63              :     uint32_t thread_count, max_threads, chunk_size, remainder, start, end, i;
      64              :     int rc;
      65              : 
      66           50 :     if (!func || total_items == 0) {
      67            2 :         return false;
      68              :     }
      69              : 
      70           48 :     thread_count = use_threads > 0 ? use_threads : cpres_get_default_thread_count();
      71           48 :     max_threads = cpres_get_max_thread_count();
      72           48 :     if (thread_count > max_threads) {
      73            1 :         thread_count = max_threads;
      74              :     }
      75              : 
      76           48 :     if (thread_count > total_items) {
      77            1 :         thread_count = total_items;
      78              :     }
      79              : 
      80           48 :     if (thread_count <= 1) {
      81           27 :         func(context, 0, total_items);
      82              : 
      83           27 :         return true;
      84              :     }
      85              : 
      86           21 :     threads = (colopresso_thread_t *)malloc(sizeof(colopresso_thread_t) * thread_count);
      87           21 :     works = (thread_work_t *)malloc(sizeof(thread_work_t) * thread_count);
      88           21 :     if (!threads || !works) {
      89            0 :         free(threads);
      90            0 :         free(works);
      91            0 :         func(context, 0, total_items);
      92              : 
      93            0 :         return true;
      94              :     }
      95              : 
      96           21 :     chunk_size = total_items / thread_count;
      97           21 :     remainder = total_items % thread_count;
      98           21 :     start = 0;
      99              : 
     100           68 :     for (i = 0; i < thread_count; ++i) {
     101           47 :         end = start + chunk_size;
     102              : 
     103           47 :         if (i < remainder) {
     104            0 :             ++end;
     105              :         }
     106              : 
     107           47 :         works[i].func = func;
     108           47 :         works[i].context = context;
     109           47 :         works[i].start_index = start;
     110           47 :         works[i].end_index = end;
     111              : 
     112           47 :         rc = colopresso_thread_create(&threads[i], NULL, thread_worker, &works[i]);
     113           47 :         works[i].created = rc == 0;
     114           47 :         if (!works[i].created) {
     115            0 :             colopresso_log(CPRES_LOG_LEVEL_WARNING,
     116              :                            "Threads: colopresso_thread_create failed (rc=%d) - falling back to "
     117              :                            "inline execution",
     118              :                            rc);
     119            0 :             works[i].func(works[i].context, works[i].start_index, works[i].end_index);
     120              :         }
     121              : 
     122           47 :         start = end;
     123              :     }
     124              : 
     125           68 :     for (i = 0; i < thread_count; ++i) {
     126           47 :         if (works[i].created) {
     127           47 :             colopresso_thread_join(threads[i], NULL);
     128              :         }
     129              :     }
     130              : 
     131           21 :     free(threads);
     132           21 :     free(works);
     133              : 
     134           21 :     return true;
     135              : }
     136              : 
     137              : #else
     138              : 
     139              : extern uint32_t cpres_get_default_thread_count(void)
     140              : {
     141              :     return 1;
     142              : }
     143              : 
     144              : extern uint32_t cpres_get_max_thread_count(void)
     145              : {
     146              :     return 1;
     147              : }
     148              : 
     149              : extern bool cpres_is_threads_enabled(void)
     150              : {
     151              :     return false;
     152              : }
     153              : 
     154              : extern bool colopresso_parallel_for(uint32_t use_threads, uint32_t total_items,
     155              :                                     parallel_func_t func, void *context)
     156              : {
     157              :     if (!func || total_items == 0) {
     158              :         return false;
     159              :     }
     160              : 
     161              :     func(context, 0, total_items);
     162              : 
     163              :     return true;
     164              : }
     165              : 
     166              : #endif