Cleanup: Remove unused threading C-API functions

BLI_task.hh has newer/better equivalents now. Pull Request: https://projects.blender.org/blender/blender/pulls/115539
2023-12-01 15:29:36 +01:00
parent 5e370ee643
commit 4c79b87d9a
5 changed files with 0 additions and 622 deletions
@@ -208,73 +208,6 @@ void BLI_task_parallel_range(int start,
                             TaskParallelRangeFunc func,
                             const TaskParallelSettings *settings);
 /**
 * This data is shared between all tasks, its access needs thread lock or similar protection.
 */
 typedef struct TaskParallelIteratorStateShared {
  /* Maximum amount of items to acquire at once. */
  int chunk_size;
  /* Next item to be acquired. */
  void *next_item;
  /* Index of the next item to be acquired. */
  int next_index;
  /* Indicates that end of iteration has been reached. */
  bool is_finished;
  /* Helper lock to protect access to this data in iterator getter callback,
   * can be ignored (if the callback implements its own protection system, using atomics e.g.).
   * Will be NULL when iterator is actually processed in a single thread. */
  SpinLock *spin_lock;
 } TaskParallelIteratorStateShared;
 typedef void (*TaskParallelIteratorIterFunc)(void *__restrict userdata,
                                             const TaskParallelTLS *__restrict tls,
                                             void **r_next_item,
                                             int *r_next_index,
                                             bool *r_do_abort);
 typedef void (*TaskParallelIteratorFunc)(void *__restrict userdata,
                                         void *item,
                                         int index,
                                         const TaskParallelTLS *__restrict tls);
 /**
 * This function allows to parallelize for loops using a generic iterator.
 *
 * \param userdata: Common userdata passed to all instances of \a func.
 * \param iter_func: Callback function used to generate chunks of items.
 * \param init_item: The initial item, if necessary (may be NULL if unused).
 * \param init_index: The initial index.
 * \param items_num: The total amount of items to iterate over
 *                   (if unknown, set it to a negative number).
 * \param func: Callback function.
 * \param settings: See public API doc of TaskParallelSettings for description of all settings.
 *
 * \note Static scheduling is only available when \a items_num is >= 0.
 */
 void BLI_task_parallel_iterator(void *userdata,
                                TaskParallelIteratorIterFunc iter_func,
                                void *init_item,
                                int init_index,
                                int items_num,
                                TaskParallelIteratorFunc func,
                                const TaskParallelSettings *settings);
 /**
 * This function allows to parallelize for loops over ListBase items.
 *
 * \param listbase: The double linked list to loop over.
 * \param userdata: Common userdata passed to all instances of \a func.
 * \param func: Callback function.
 * \param settings: See public API doc of ParallelRangeSettings for description of all settings.
 *
 * \note There is no static scheduling here,
 * since it would need another full loop over items to count them.
 */
 void BLI_task_parallel_listbase(struct ListBase *listbase,
                                void *userdata,
                                TaskParallelIteratorFunc func,
                                const TaskParallelSettings *settings);
 typedef struct MempoolIterData MempoolIterData;
 typedef void (*TaskParallelMempoolFunc)(void *userdata,
@@ -37,298 +37,6 @@
 /** \} */
 /* -------------------------------------------------------------------- */
 /** \name Generic Iteration
 * \{ */
 BLI_INLINE void task_parallel_calc_chunk_size(const TaskParallelSettings *settings,
                                              const int items_num,
                                              int tasks_num,
                                              int *r_chunk_size)
 {
  int chunk_size = 0;
  if (!settings->use_threading) {
    /* Some users of this helper will still need a valid chunk size in case processing is not
     * threaded. We can use a bigger one than in default threaded case then. */
    chunk_size = 1024;
    tasks_num = 1;
  }
  else if (settings->min_iter_per_thread > 0) {
    /* Already set by user, no need to do anything here. */
    chunk_size = settings->min_iter_per_thread;
  }
  else {
    /* Multiplier used in heuristics below to define "optimal" chunk size.
     * The idea here is to increase the chunk size to compensate for a rather measurable threading
     * overhead caused by fetching tasks. With too many CPU threads we are starting
     * to spend too much time in those overheads.
     * First values are: 1 if tasks_num < 16;
     *              else 2 if tasks_num < 32;
     *              else 3 if tasks_num < 48;
     *              else 4 if tasks_num < 64;
     *                   etc.
     * NOTE: If we wanted to keep the 'power of two' multiplier, we'd need something like:
     *     1 << max_ii(0, (int)(sizeof(int) * 8) - 1 - bitscan_reverse_i(tasks_num) - 3)
     */
    const int tasks_num_factor = max_ii(1, tasks_num >> 3);
    /* We could make that 'base' 32 number configurable in TaskParallelSettings too, or maybe just
     * always use that heuristic using TaskParallelSettings.min_iter_per_thread as basis? */
    chunk_size = 32 * tasks_num_factor;
    /* Basic heuristic to avoid threading on low amount of items.
     * We could make that limit configurable in settings too. */
    if (items_num > 0 && items_num < max_ii(256, chunk_size * 2)) {
      chunk_size = items_num;
    }
  }
  BLI_assert(chunk_size > 0);
  *r_chunk_size = chunk_size;
 }
 typedef struct TaskParallelIteratorState {
  void *userdata;
  TaskParallelIteratorIterFunc iter_func;
  TaskParallelIteratorFunc func;
  /* *** Data used to 'acquire' chunks of items from the iterator. *** */
  /* Common data also passed to the generator callback. */
  TaskParallelIteratorStateShared iter_shared;
  /* Total number of items. If unknown, set it to a negative number. */
  int items_num;
 } TaskParallelIteratorState;
 static void parallel_iterator_func_do(TaskParallelIteratorState *__restrict state,
                                      void *userdata_chunk)
 {
  TaskParallelTLS tls = {
      .userdata_chunk = userdata_chunk,
  };
  void **current_chunk_items;
  int *current_chunk_indices;
  int current_chunk_size;
  const size_t items_size = sizeof(*current_chunk_items) * (size_t)state->iter_shared.chunk_size;
  const size_t indices_size = sizeof(*current_chunk_indices) *
                              (size_t)state->iter_shared.chunk_size;
  current_chunk_items = MALLOCA(items_size);
  current_chunk_indices = MALLOCA(indices_size);
  current_chunk_size = 0;
  for (bool do_abort = false; !do_abort;) {
    if (state->iter_shared.spin_lock != NULL) {
      BLI_spin_lock(state->iter_shared.spin_lock);
    }
    /* Get current status. */
    int index = state->iter_shared.next_index;
    void *item = state->iter_shared.next_item;
    int i;
    /* 'Acquire' a chunk of items from the iterator function. */
    for (i = 0; i < state->iter_shared.chunk_size && !state->iter_shared.is_finished; i++) {
      current_chunk_indices[i] = index;
      current_chunk_items[i] = item;
      state->iter_func(state->userdata, &tls, &item, &index, &state->iter_shared.is_finished);
    }
    /* Update current status. */
    state->iter_shared.next_index = index;
    state->iter_shared.next_item = item;
    current_chunk_size = i;
    do_abort = state->iter_shared.is_finished;
    if (state->iter_shared.spin_lock != NULL) {
      BLI_spin_unlock(state->iter_shared.spin_lock);
    }
    for (i = 0; i < current_chunk_size; ++i) {
      state->func(state->userdata, current_chunk_items[i], current_chunk_indices[i], &tls);
    }
  }
  MALLOCA_FREE(current_chunk_items, items_size);
  MALLOCA_FREE(current_chunk_indices, indices_size);
 }
 static void parallel_iterator_func(TaskPool *__restrict pool, void *userdata_chunk)
 {
  TaskParallelIteratorState *__restrict state = BLI_task_pool_user_data(pool);
  parallel_iterator_func_do(state, userdata_chunk);
 }
 static void task_parallel_iterator_no_threads(const TaskParallelSettings *settings,
                                              TaskParallelIteratorState *state)
 {
  /* Prepare user's TLS data. */
  void *userdata_chunk = settings->userdata_chunk;
  if (userdata_chunk) {
    if (settings->func_init != NULL) {
      settings->func_init(state->userdata, userdata_chunk);
    }
  }
  /* Also marking it as non-threaded for the iterator callback. */
  state->iter_shared.spin_lock = NULL;
  parallel_iterator_func_do(state, userdata_chunk);
  if (userdata_chunk) {
    if (settings->func_free != NULL) {
      /* `func_free` should only free data that was created during execution of `func`. */
      settings->func_free(state->userdata, userdata_chunk);
    }
  }
 }
 static void task_parallel_iterator_do(const TaskParallelSettings *settings,
                                      TaskParallelIteratorState *state)
 {
  const int threads_num = BLI_task_scheduler_num_threads();
  task_parallel_calc_chunk_size(
      settings, state->items_num, threads_num, &state->iter_shared.chunk_size);
  if (!settings->use_threading) {
    task_parallel_iterator_no_threads(settings, state);
    return;
  }
  const int chunk_size = state->iter_shared.chunk_size;
  const int items_num = state->items_num;
  const size_t tasks_num = items_num >= 0 ?
                               (size_t)min_ii(threads_num, state->items_num / chunk_size) :
                               (size_t)threads_num;
  BLI_assert(tasks_num > 0);
  if (tasks_num == 1) {
    task_parallel_iterator_no_threads(settings, state);
    return;
  }
  SpinLock spin_lock;
  BLI_spin_init(&spin_lock);
  state->iter_shared.spin_lock = &spin_lock;
  void *userdata_chunk = settings->userdata_chunk;
  const size_t userdata_chunk_size = settings->userdata_chunk_size;
  void *userdata_chunk_local = NULL;
  void *userdata_chunk_array = NULL;
  const bool use_userdata_chunk = (userdata_chunk_size != 0) && (userdata_chunk != NULL);
  TaskPool *task_pool = BLI_task_pool_create(state, TASK_PRIORITY_HIGH);
  if (use_userdata_chunk) {
    userdata_chunk_array = MALLOCA(userdata_chunk_size * tasks_num);
  }
  for (size_t i = 0; i < tasks_num; i++) {
    if (use_userdata_chunk) {
      userdata_chunk_local = (char *)userdata_chunk_array + (userdata_chunk_size * i);
      memcpy(userdata_chunk_local, userdata_chunk, userdata_chunk_size);
      if (settings->func_init != NULL) {
        settings->func_init(state->userdata, userdata_chunk_local);
      }
    }
    /* Use this pool's pre-allocated tasks. */
    BLI_task_pool_push(task_pool, parallel_iterator_func, userdata_chunk_local, false, NULL);
  }
  BLI_task_pool_work_and_wait(task_pool);
  BLI_task_pool_free(task_pool);
  if (use_userdata_chunk) {
    if (settings->func_reduce != NULL || settings->func_free != NULL) {
      for (size_t i = 0; i < tasks_num; i++) {
        userdata_chunk_local = (char *)userdata_chunk_array + (userdata_chunk_size * i);
        if (settings->func_reduce != NULL) {
          settings->func_reduce(state->userdata, userdata_chunk, userdata_chunk_local);
        }
        if (settings->func_free != NULL) {
          settings->func_free(state->userdata, userdata_chunk_local);
        }
      }
    }
    MALLOCA_FREE(userdata_chunk_array, userdata_chunk_size * tasks_num);
  }
  BLI_spin_end(&spin_lock);
  state->iter_shared.spin_lock = NULL;
 }
 void BLI_task_parallel_iterator(void *userdata,
                                TaskParallelIteratorIterFunc iter_func,
                                void *init_item,
                                const int init_index,
                                const int items_num,
                                TaskParallelIteratorFunc func,
                                const TaskParallelSettings *settings)
 {
  TaskParallelIteratorState state = {0};
  state.items_num = items_num;
  state.iter_shared.next_index = init_index;
  state.iter_shared.next_item = init_item;
  state.iter_shared.is_finished = false;
  state.userdata = userdata;
  state.iter_func = iter_func;
  state.func = func;
  task_parallel_iterator_do(settings, &state);
 }
 /** \} */
 /* -------------------------------------------------------------------- */
 /** \name ListBase Iteration
 * \{ */
 static void task_parallel_listbase_get(void *__restrict UNUSED(userdata),
                                       const TaskParallelTLS *__restrict UNUSED(tls),
                                       void **r_next_item,
                                       int *r_next_index,
                                       bool *r_do_abort)
 {
  /* Get current status. */
  Link *link = *r_next_item;
  if (link->next == NULL) {
    *r_do_abort = true;
  }
  *r_next_item = link->next;
  (*r_next_index)++;
 }
 void BLI_task_parallel_listbase(ListBase *listbase,
                                void *userdata,
                                TaskParallelIteratorFunc func,
                                const TaskParallelSettings *settings)
 {
  if (BLI_listbase_is_empty(listbase)) {
    return;
  }
  TaskParallelIteratorState state = {0};
  state.items_num = BLI_listbase_count(listbase);
  state.iter_shared.next_index = 0;
  state.iter_shared.next_item = listbase->first;
  state.iter_shared.is_finished = false;
  state.userdata = userdata;
  state.iter_func = task_parallel_listbase_get;
  state.func = func;
  task_parallel_iterator_do(settings, &state);
 }
 /** \} */
 /* -------------------------------------------------------------------- */
 /** \name MemPool Iteration
 * \{ */
@@ -234,55 +234,6 @@ TEST(task, MempoolIterTLS)
  BLI_threadapi_exit();
 }
 /* *** Parallel iterations over double-linked list items. *** */
 static void task_listbase_iter_func(void *userdata,
                                    void *item,
                                    int index,
                                    const TaskParallelTLS *__restrict /*tls*/)
 {
  LinkData *data = (LinkData *)item;
  int *count = (int *)userdata;
  data->data = POINTER_FROM_INT(POINTER_AS_INT(data->data) + index);
  atomic_sub_and_fetch_uint32((uint32_t *)count, 1);
 }
 TEST(task, ListBaseIter)
 {
  ListBase list = {nullptr, nullptr};
  LinkData *items_buffer = (LinkData *)MEM_calloc_arrayN(
      ITEMS_NUM, sizeof(*items_buffer), __func__);
  BLI_threadapi_init();
  int i;
  int items_num = 0;
  for (i = 0; i < ITEMS_NUM; i++) {
    BLI_addtail(&list, &items_buffer[i]);
    items_num++;
  }
  TaskParallelSettings settings;
  BLI_parallel_range_settings_defaults(&settings);
  BLI_task_parallel_listbase(&list, &items_num, task_listbase_iter_func, &settings);
  /* Those checks should ensure us all items of the listbase were processed once, and only once -
   * as expected. */
  EXPECT_EQ(items_num, 0);
  LinkData *item;
  for (i = 0, item = (LinkData *)list.first; i < ITEMS_NUM && item != nullptr;
       i++, item = item->next)
  {
    EXPECT_EQ(POINTER_AS_INT(item->data), i);
  }
  EXPECT_EQ(ITEMS_NUM, i);
  MEM_freeN(items_buffer);
  BLI_threadapi_exit();
 }
 TEST(task, ParallelInvoke)
 {
  std::atomic<int> counter = 0;
@@ -1,213 +0,0 @@
 /* SPDX-FileCopyrightText: 2023 Blender Authors
 *
 * SPDX-License-Identifier: Apache-2.0 */
 #include "BLI_ressource_strings.h"
 #include "testing/testing.h"
 #include "atomic_ops.h"
 #define GHASH_INTERNAL_API
 #include "MEM_guardedalloc.h"
 #include "BLI_utildefines.h"
 #include "BLI_listbase.h"
 #include "BLI_mempool.h"
 #include "BLI_task.h"
 #include "PIL_time.h"
 #define NUM_RUN_AVERAGED 100
 static uint gen_pseudo_random_number(uint num)
 {
  /* NOTE: this is taken from BLI_ghashutil_uinthash(), don't want to depend on external code that
   * might change here... */
  num += ~(num << 16);
  num ^= (num >> 5);
  num += (num << 3);
  num ^= (num >> 13);
  num += ~(num << 9);
  num ^= (num >> 17);
  /* Make final number in [65 - 16385] range. */
  return ((num & 255) << 6) + 1;
 }
 /* *** Parallel iterations over double-linked list items. *** */
 static void task_listbase_light_iter_func(void * /*userdata*/,
                                          void *item,
                                          int index,
                                          const TaskParallelTLS *__restrict /*tls*/)
 {
  LinkData *data = (LinkData *)item;
  data->data = POINTER_FROM_INT(POINTER_AS_INT(data->data) + index);
 }
 static void task_listbase_light_membarrier_iter_func(void *userdata,
                                                     void *item,
                                                     int index,
                                                     const TaskParallelTLS *__restrict /*tls*/)
 {
  LinkData *data = (LinkData *)item;
  int *count = (int *)userdata;
  data->data = POINTER_FROM_INT(POINTER_AS_INT(data->data) + index);
  atomic_sub_and_fetch_uint32((uint32_t *)count, 1);
 }
 static void task_listbase_heavy_iter_func(void * /*userdata*/,
                                          void *item,
                                          int index,
                                          const TaskParallelTLS *__restrict /*tls*/)
 {
  LinkData *data = (LinkData *)item;
  /* 'Random' number of iterations. */
  const uint num = gen_pseudo_random_number(uint(index));
  for (uint i = 0; i < num; i++) {
    data->data = POINTER_FROM_INT(POINTER_AS_INT(data->data) + ((i % 2) ? -index : index));
  }
 }
 static void task_listbase_heavy_membarrier_iter_func(void *userdata,
                                                     void *item,
                                                     int index,
                                                     const TaskParallelTLS *__restrict /*tls*/)
 {
  LinkData *data = (LinkData *)item;
  int *count = (int *)userdata;
  /* 'Random' number of iterations. */
  const uint num = gen_pseudo_random_number(uint(index));
  for (uint i = 0; i < num; i++) {
    data->data = POINTER_FROM_INT(POINTER_AS_INT(data->data) + ((i % 2) ? -index : index));
  }
  atomic_sub_and_fetch_uint32((uint32_t *)count, 1);
 }
 static void task_listbase_test_do(ListBase *list,
                                  const int items_num,
                                  int *items_tmp_num,
                                  const char *id,
                                  TaskParallelIteratorFunc func,
                                  const bool use_threads,
                                  const bool check_items_tmp_num)
 {
  TaskParallelSettings settings;
  BLI_parallel_range_settings_defaults(&settings);
  settings.use_threading = use_threads;
  double averaged_timing = 0.0;
  for (int i = 0; i < NUM_RUN_AVERAGED; i++) {
    const double init_time = PIL_check_seconds_timer();
    BLI_task_parallel_listbase(list, items_tmp_num, func, &settings);
    averaged_timing += PIL_check_seconds_timer() - init_time;
    /* Those checks should ensure us all items of the listbase were processed once, and only once -
     * as expected. */
    if (check_items_tmp_num) {
      EXPECT_EQ(*items_tmp_num, 0);
    }
    LinkData *item;
    int j;
    for (j = 0, item = (LinkData *)list->first; j < items_num && item != nullptr;
         j++, item = item->next)
    {
      EXPECT_EQ(POINTER_AS_INT(item->data), j);
      item->data = POINTER_FROM_INT(0);
    }
    EXPECT_EQ(items_num, j);
    *items_tmp_num = items_num;
  }
  printf("\t%s: done in %fs on average over %d runs\n",
         id,
         averaged_timing / NUM_RUN_AVERAGED,
         NUM_RUN_AVERAGED);
 }
 static void task_listbase_test(const char *id, const int count, const bool use_threads)
 {
  printf("\n========== STARTING %s ==========\n", id);
  ListBase list = {nullptr, nullptr};
  LinkData *items_buffer = (LinkData *)MEM_calloc_arrayN(count, sizeof(*items_buffer), __func__);
  BLI_threadapi_init();
  int items_num = 0;
  for (int i = 0; i < count; i++) {
    BLI_addtail(&list, &items_buffer[i]);
    items_num++;
  }
  int items_tmp_num = items_num;
  task_listbase_test_do(&list,
                        items_num,
                        &items_tmp_num,
                        "Light iter",
                        task_listbase_light_iter_func,
                        use_threads,
                        false);
  task_listbase_test_do(&list,
                        items_num,
                        &items_tmp_num,
                        "Light iter with mem barrier",
                        task_listbase_light_membarrier_iter_func,
                        use_threads,
                        true);
  task_listbase_test_do(&list,
                        items_num,
                        &items_tmp_num,
                        "Heavy iter",
                        task_listbase_heavy_iter_func,
                        use_threads,
                        false);
  task_listbase_test_do(&list,
                        items_num,
                        &items_tmp_num,
                        "Heavy iter with mem barrier",
                        task_listbase_heavy_membarrier_iter_func,
                        use_threads,
                        true);
  MEM_freeN(items_buffer);
  BLI_threadapi_exit();
  printf("========== ENDED %s ==========\n\n", id);
 }
 TEST(task, ListBaseIterNoThread10k)
 {
  task_listbase_test("ListBase parallel iteration - Single thread - 10000 items", 10000, false);
 }
 TEST(task, ListBaseIter10k)
 {
  task_listbase_test("ListBase parallel iteration - Threaded - 10000 items", 10000, true);
 }
 TEST(task, ListBaseIterNoThread100k)
 {
  task_listbase_test("ListBase parallel iteration - Single thread - 100000 items", 100000, false);
 }
 TEST(task, ListBaseIter100k)
 {
  task_listbase_test("ListBase parallel iteration - Threaded - 100000 items", 100000, true);
 }
@@ -19,4 +19,3 @@ set(LIB
 )
 blender_add_performancetest_executable(BLI_ghash_performance "BLI_ghash_performance_test.cc" "${INC}" "${INC_SYS}" "${LIB}")
 blender_add_performancetest_executable(BLI_task_performance "BLI_task_performance_test.cc" "${INC}" "${INC_SYS}" "${LIB}")