Changeset View
Changeset View
Standalone View
Standalone View
source/blender/blenlib/intern/task_iterator.c
| Show All 11 Lines | |||||
| * You should have received a copy of the GNU General Public License | * You should have received a copy of the GNU General Public License | ||||
| * along with this program; if not, write to the Free Software Foundation, | * along with this program; if not, write to the Free Software Foundation, | ||||
| * Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA. | * Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA. | ||||
| */ | */ | ||||
| /** \file | /** \file | ||||
| * \ingroup bli | * \ingroup bli | ||||
| * | * | ||||
| * A generic task system which can be used for any task based subsystem. | * Parallel tasks over all elements in a container. | ||||
| */ | */ | ||||
| #include <stdlib.h> | #include <stdlib.h> | ||||
| #include "MEM_guardedalloc.h" | #include "MEM_guardedalloc.h" | ||||
| #include "DNA_listBase.h" | #include "DNA_listBase.h" | ||||
| #include "BLI_listbase.h" | #include "BLI_listbase.h" | ||||
| #include "BLI_math.h" | #include "BLI_math.h" | ||||
| #include "BLI_mempool.h" | #include "BLI_mempool.h" | ||||
| #include "BLI_task.h" | #include "BLI_task.h" | ||||
| #include "BLI_threads.h" | #include "BLI_threads.h" | ||||
| #include "atomic_ops.h" | #include "atomic_ops.h" | ||||
| /* Parallel range routines */ | |||||
| /** | |||||
| * | |||||
| * Main functions: | |||||
| * - #BLI_task_parallel_range | |||||
| * - #BLI_task_parallel_listbase (#ListBase - double linked list) | |||||
| * | |||||
| * TODO: | |||||
| * - #BLI_task_parallel_foreach_link (#Link - single linked list) | |||||
| * - #BLI_task_parallel_foreach_ghash/gset (#GHash/#GSet - hash & set) | |||||
| * - #BLI_task_parallel_foreach_mempool (#BLI_mempool - iterate over mempools) | |||||
| */ | |||||
| /* Allows to avoid using malloc for userdata_chunk in tasks, when small enough. */ | /* Allows to avoid using malloc for userdata_chunk in tasks, when small enough. */ | ||||
| #define MALLOCA(_size) ((_size) <= 8192) ? alloca((_size)) : MEM_mallocN((_size), __func__) | #define MALLOCA(_size) ((_size) <= 8192) ? alloca((_size)) : MEM_mallocN((_size), __func__) | ||||
| #define MALLOCA_FREE(_mem, _size) \ | #define MALLOCA_FREE(_mem, _size) \ | ||||
| if (((_mem) != NULL) && ((_size) > 8192)) \ | if (((_mem) != NULL) && ((_size) > 8192)) \ | ||||
| MEM_freeN((_mem)) | MEM_freeN((_mem)) | ||||
| #if 0 | |||||
| /* Stores all needed data to perform a parallelized iteration, | /* Stores all needed data to perform a parallelized iteration, | ||||
| * with a same operation (callback function). | * with a same operation (callback function). | ||||
| * It can be chained with other tasks in a single-linked list way. */ | * It can be chained with other tasks in a single-linked list way. */ | ||||
| typedef struct TaskParallelRangeState { | typedef struct TaskParallelRangeState { | ||||
| struct TaskParallelRangeState *next; | struct TaskParallelRangeState *next; | ||||
| /* Start and end point of integer value iteration. */ | /* Start and end point of integer value iteration. */ | ||||
| int start, stop; | int start, stop; | ||||
| Show All 40 Lines | typedef struct TaskParallelRangePool { | ||||
| /* Linked list of range tasks to process. */ | /* Linked list of range tasks to process. */ | ||||
| TaskParallelRangeState *parallel_range_states; | TaskParallelRangeState *parallel_range_states; | ||||
| /* Current range task beeing processed, swapped atomically. */ | /* Current range task beeing processed, swapped atomically. */ | ||||
| TaskParallelRangeState *current_state; | TaskParallelRangeState *current_state; | ||||
| /* Scheduling settings common to all tasks. */ | /* Scheduling settings common to all tasks. */ | ||||
| TaskParallelSettings *settings; | TaskParallelSettings *settings; | ||||
| } TaskParallelRangePool; | } TaskParallelRangePool; | ||||
| #endif | |||||
| BLI_INLINE void task_parallel_calc_chunk_size(const TaskParallelSettings *settings, | BLI_INLINE void task_parallel_calc_chunk_size(const TaskParallelSettings *settings, | ||||
| const int tot_items, | const int tot_items, | ||||
| int num_tasks, | int num_tasks, | ||||
| int *r_chunk_size) | int *r_chunk_size) | ||||
| { | { | ||||
| int chunk_size = 0; | int chunk_size = 0; | ||||
| if (!settings->use_threading) { | if (!settings->use_threading) { | ||||
| Show All 28 Lines | else { | ||||
| /* Basic heuristic to avoid threading on low amount of items. | /* Basic heuristic to avoid threading on low amount of items. | ||||
| * We could make that limit configurable in settings too. */ | * We could make that limit configurable in settings too. */ | ||||
| if (tot_items > 0 && tot_items < max_ii(256, chunk_size * 2)) { | if (tot_items > 0 && tot_items < max_ii(256, chunk_size * 2)) { | ||||
| chunk_size = tot_items; | chunk_size = tot_items; | ||||
| } | } | ||||
| } | } | ||||
| BLI_assert(chunk_size > 0); | BLI_assert(chunk_size > 0); | ||||
| #if 0 | |||||
| if (tot_items > 0) { | if (tot_items > 0) { | ||||
| switch (settings->scheduling_mode) { | switch (settings->scheduling_mode) { | ||||
| case TASK_SCHEDULING_STATIC: | case TASK_SCHEDULING_STATIC: | ||||
| *r_chunk_size = max_ii(chunk_size, tot_items / num_tasks); | *r_chunk_size = max_ii(chunk_size, tot_items / num_tasks); | ||||
| break; | break; | ||||
| case TASK_SCHEDULING_DYNAMIC: | case TASK_SCHEDULING_DYNAMIC: | ||||
| *r_chunk_size = chunk_size; | *r_chunk_size = chunk_size; | ||||
| ▲ Show 20 Lines • Show All 173 Lines • ▼ Show 20 Lines | void BLI_task_parallel_range(const int start, | ||||
| range_pool.num_tasks = num_tasks = min_ii(num_tasks, | range_pool.num_tasks = num_tasks = min_ii(num_tasks, | ||||
| max_ii(1, (stop - start) / range_pool.chunk_size)); | max_ii(1, (stop - start) / range_pool.chunk_size)); | ||||
| if (num_tasks == 1) { | if (num_tasks == 1) { | ||||
| parallel_range_single_thread(&range_pool); | parallel_range_single_thread(&range_pool); | ||||
| return; | return; | ||||
| } | } | ||||
| TaskPool *task_pool = range_pool.pool = BLI_task_pool_create_suspended( | TaskPool *task_pool = range_pool.pool = BLI_task_pool_create( | ||||
| task_scheduler, &range_pool, TASK_PRIORITY_HIGH); | task_scheduler, &range_pool, TASK_PRIORITY_HIGH); | ||||
| range_pool.current_state = &state; | range_pool.current_state = &state; | ||||
| if (use_tls_data) { | if (use_tls_data) { | ||||
| state.flatten_tls_storage = flatten_tls_storage = MALLOCA(tls_data_size * (size_t)num_tasks); | state.flatten_tls_storage = flatten_tls_storage = MALLOCA(tls_data_size * (size_t)num_tasks); | ||||
| state.tls_data_size = tls_data_size; | state.tls_data_size = tls_data_size; | ||||
| } | } | ||||
| Show All 18 Lines | for (i = 0; i < num_tasks; i++) { | ||||
| if (settings->func_reduce) { | if (settings->func_reduce) { | ||||
| settings->func_reduce(userdata, tls_data, userdata_chunk_local); | settings->func_reduce(userdata, tls_data, userdata_chunk_local); | ||||
| } | } | ||||
| if (settings->func_free) { | if (settings->func_free) { | ||||
| /* `func_free` should only free data that was created during execution of `func`. */ | /* `func_free` should only free data that was created during execution of `func`. */ | ||||
| settings->func_free(userdata, userdata_chunk_local); | settings->func_free(userdata, userdata_chunk_local); | ||||
| } | } | ||||
| } | } | ||||
| MALLOCA_FREE(flatten_tls_storage, tls_data_size * (size_t)num_tasks); | |||||
| } | } | ||||
| MALLOCA_FREE(flatten_tls_storage, tls_data_size * (size_t)num_tasks); | |||||
| } | } | ||||
| /** | /** | ||||
| * Initialize a task pool to parallelize several for loops at the same time. | * Initialize a task pool to parallelize several for loops at the same time. | ||||
| * | * | ||||
| * See public API doc of ParallelRangeSettings for description of all settings. | * See public API doc of ParallelRangeSettings for description of all settings. | ||||
| * Note that loop-specific settings (like 'tls' data or reduce/free functions) must be left NULL | * Note that loop-specific settings (like 'tls' data or reduce/free functions) must be left NULL | ||||
| * here. Only settings controlling how iteration is parallelized must be defined, as those will | * here. Only settings controlling how iteration is parallelized must be defined, as those will | ||||
| ▲ Show 20 Lines • Show All 46 Lines • ▼ Show 20 Lines | void BLI_task_parallel_range_pool_push(TaskParallelRangePool *range_pool, | ||||
| state->tls_data_size = settings->userdata_chunk_size; | state->tls_data_size = settings->userdata_chunk_size; | ||||
| state->func_reduce = settings->func_reduce; | state->func_reduce = settings->func_reduce; | ||||
| state->func_free = settings->func_free; | state->func_free = settings->func_free; | ||||
| state->next = range_pool->parallel_range_states; | state->next = range_pool->parallel_range_states; | ||||
| range_pool->parallel_range_states = state; | range_pool->parallel_range_states = state; | ||||
| } | } | ||||
| static void parallel_range_func_finalize(TaskPool *__restrict pool, | static void parallel_range_func_reduce(TaskPool *__restrict pool, | ||||
| void *v_state, | void *v_state, | ||||
| int UNUSED(thread_id)) | int UNUSED(thread_id)) | ||||
| { | { | ||||
| TaskParallelRangePool *__restrict range_pool = BLI_task_pool_userdata(pool); | TaskParallelRangePool *__restrict range_pool = BLI_task_pool_userdata(pool); | ||||
| TaskParallelRangeState *state = v_state; | TaskParallelRangeState *state = v_state; | ||||
| for (int i = 0; i < range_pool->num_tasks; i++) { | for (int i = 0; i < range_pool->num_tasks; i++) { | ||||
| void *tls_data = (char *)state->flatten_tls_storage + (state->tls_data_size * (size_t)i); | void *tls_data = (char *)state->flatten_tls_storage + (state->tls_data_size * (size_t)i); | ||||
| if (state->func_reduce != NULL) { | if (state->func_reduce != NULL) { | ||||
| state->func_reduce(state->userdata_shared, state->initial_tls_memory, tls_data); | state->func_reduce(state->userdata_shared, state->initial_tls_memory, tls_data); | ||||
| ▲ Show 20 Lines • Show All 57 Lines • ▼ Show 20 Lines | state->flatten_tls_storage = userdata_chunk_array = MALLOCA(userdata_chunk_size * | ||||
| (size_t)num_tasks); | (size_t)num_tasks); | ||||
| for (int i = 0; i < num_tasks; i++) { | for (int i = 0; i < num_tasks; i++) { | ||||
| void *userdata_chunk_local = (char *)userdata_chunk_array + | void *userdata_chunk_local = (char *)userdata_chunk_array + | ||||
| (userdata_chunk_size * (size_t)i); | (userdata_chunk_size * (size_t)i); | ||||
| memcpy(userdata_chunk_local, userdata_chunk, userdata_chunk_size); | memcpy(userdata_chunk_local, userdata_chunk, userdata_chunk_size); | ||||
| } | } | ||||
| } | } | ||||
| TaskPool *task_pool = range_pool->pool = BLI_task_pool_create_suspended( | TaskPool *task_pool = range_pool->pool = BLI_task_pool_create( | ||||
| task_scheduler, range_pool, TASK_PRIORITY_HIGH); | task_scheduler, range_pool, TASK_PRIORITY_HIGH); | ||||
| range_pool->current_state = range_pool->parallel_range_states; | range_pool->current_state = range_pool->parallel_range_states; | ||||
| const int thread_id = BLI_task_pool_creator_thread_id(task_pool); | const int thread_id = BLI_task_pool_creator_thread_id(task_pool); | ||||
| for (int i = 0; i < num_tasks; i++) { | for (int i = 0; i < num_tasks; i++) { | ||||
| BLI_task_pool_push_from_thread( | BLI_task_pool_push_from_thread( | ||||
| task_pool, parallel_range_func, POINTER_FROM_INT(i), false, NULL, thread_id); | task_pool, parallel_range_func, POINTER_FROM_INT(i), false, NULL, thread_id); | ||||
| } | } | ||||
| Show All 10 Lines | for (TaskParallelRangeState *state = range_pool->parallel_range_states; state != NULL; | ||||
| UNUSED_VARS_NDEBUG(userdata_chunk_array); | UNUSED_VARS_NDEBUG(userdata_chunk_array); | ||||
| if (userdata_chunk_size == 0) { | if (userdata_chunk_size == 0) { | ||||
| BLI_assert(userdata_chunk_array == NULL); | BLI_assert(userdata_chunk_array == NULL); | ||||
| continue; | continue; | ||||
| } | } | ||||
| if (state->func_reduce != NULL || state->func_free != NULL) { | if (state->func_reduce != NULL || state->func_free != NULL) { | ||||
| BLI_task_pool_push_from_thread( | BLI_task_pool_push_from_thread( | ||||
| task_pool, parallel_range_func_finalize, state, false, NULL, thread_id); | task_pool, parallel_range_func_reduce, state, false, NULL, thread_id); | ||||
| } | } | ||||
| } | } | ||||
| BLI_task_pool_work_and_wait(task_pool); | BLI_task_pool_work_and_wait(task_pool); | ||||
| BLI_task_pool_free(task_pool); | BLI_task_pool_free(task_pool); | ||||
| range_pool->pool = NULL; | range_pool->pool = NULL; | ||||
| /* Cleanup all tasks. */ | /* Cleanup all tasks. */ | ||||
| Show All 22 Lines | void BLI_task_parallel_range_pool_free(TaskParallelRangePool *range_pool) | ||||
| TaskParallelRangeState *state_next = NULL; | TaskParallelRangeState *state_next = NULL; | ||||
| for (TaskParallelRangeState *state = range_pool->parallel_range_states; state != NULL; | for (TaskParallelRangeState *state = range_pool->parallel_range_states; state != NULL; | ||||
| state = state_next) { | state = state_next) { | ||||
| state_next = state->next; | state_next = state->next; | ||||
| MEM_freeN(state); | MEM_freeN(state); | ||||
| } | } | ||||
| MEM_freeN(range_pool->settings); | MEM_freeN(range_pool->settings); | ||||
| MEM_freeN(range_pool); | MEM_freeN(range_pool); | ||||
| #else | |||||
| *r_chunk_size = chunk_size; | |||||
| #endif | |||||
| } | } | ||||
| typedef struct TaskParallelIteratorState { | typedef struct TaskParallelIteratorState { | ||||
| void *userdata; | void *userdata; | ||||
| TaskParallelIteratorIterFunc iter_func; | TaskParallelIteratorIterFunc iter_func; | ||||
| TaskParallelIteratorFunc func; | TaskParallelIteratorFunc func; | ||||
| /* *** Data used to 'acquire' chunks of items from the iterator. *** */ | /* *** Data used to 'acquire' chunks of items from the iterator. *** */ | ||||
| /* Common data also passed to the generator callback. */ | /* Common data also passed to the generator callback. */ | ||||
| TaskParallelIteratorStateShared iter_shared; | TaskParallelIteratorStateShared iter_shared; | ||||
| /* Total number of items. If unknown, set it to a negative number. */ | /* Total number of items. If unknown, set it to a negative number. */ | ||||
| int tot_items; | int tot_items; | ||||
| } TaskParallelIteratorState; | } TaskParallelIteratorState; | ||||
| BLI_INLINE void task_parallel_iterator_calc_chunk_size(const TaskParallelSettings *settings, | |||||
| const int num_tasks, | |||||
| TaskParallelIteratorState *state) | |||||
| { | |||||
| task_parallel_calc_chunk_size( | |||||
| settings, state->tot_items, num_tasks, &state->iter_shared.chunk_size); | |||||
| } | |||||
| static void parallel_iterator_func_do(TaskParallelIteratorState *__restrict state, | static void parallel_iterator_func_do(TaskParallelIteratorState *__restrict state, | ||||
| void *userdata_chunk, | void *userdata_chunk) | ||||
| int threadid) | |||||
| { | { | ||||
| TaskParallelTLS tls = { | TaskParallelTLS tls = { | ||||
| .thread_id = threadid, | |||||
| .userdata_chunk = userdata_chunk, | .userdata_chunk = userdata_chunk, | ||||
| }; | }; | ||||
| void **current_chunk_items; | void **current_chunk_items; | ||||
| int *current_chunk_indices; | int *current_chunk_indices; | ||||
| int current_chunk_size; | int current_chunk_size; | ||||
| const size_t items_size = sizeof(*current_chunk_items) * (size_t)state->iter_shared.chunk_size; | const size_t items_size = sizeof(*current_chunk_items) * (size_t)state->iter_shared.chunk_size; | ||||
| Show All 36 Lines | for (i = 0; i < current_chunk_size; ++i) { | ||||
| state->func(state->userdata, current_chunk_items[i], current_chunk_indices[i], &tls); | state->func(state->userdata, current_chunk_items[i], current_chunk_indices[i], &tls); | ||||
| } | } | ||||
| } | } | ||||
| MALLOCA_FREE(current_chunk_items, items_size); | MALLOCA_FREE(current_chunk_items, items_size); | ||||
| MALLOCA_FREE(current_chunk_indices, indices_size); | MALLOCA_FREE(current_chunk_indices, indices_size); | ||||
| } | } | ||||
| static void parallel_iterator_func(TaskPool *__restrict pool, void *userdata_chunk, int threadid) | static void parallel_iterator_func(TaskPool *__restrict pool, void *userdata_chunk) | ||||
| { | { | ||||
| TaskParallelIteratorState *__restrict state = BLI_task_pool_userdata(pool); | TaskParallelIteratorState *__restrict state = BLI_task_pool_user_data(pool); | ||||
| parallel_iterator_func_do(state, userdata_chunk, threadid); | parallel_iterator_func_do(state, userdata_chunk); | ||||
| } | } | ||||
| static void task_parallel_iterator_no_threads(const TaskParallelSettings *settings, | static void task_parallel_iterator_no_threads(const TaskParallelSettings *settings, | ||||
| TaskParallelIteratorState *state) | TaskParallelIteratorState *state) | ||||
| { | { | ||||
| /* Prepare user's TLS data. */ | /* Prepare user's TLS data. */ | ||||
| void *userdata_chunk = settings->userdata_chunk; | void *userdata_chunk = settings->userdata_chunk; | ||||
| const size_t userdata_chunk_size = settings->userdata_chunk_size; | const size_t userdata_chunk_size = settings->userdata_chunk_size; | ||||
| void *userdata_chunk_local = NULL; | void *userdata_chunk_local = NULL; | ||||
| const bool use_userdata_chunk = (userdata_chunk_size != 0) && (userdata_chunk != NULL); | const bool use_userdata_chunk = (userdata_chunk_size != 0) && (userdata_chunk != NULL); | ||||
| if (use_userdata_chunk) { | if (use_userdata_chunk) { | ||||
| userdata_chunk_local = MALLOCA(userdata_chunk_size); | userdata_chunk_local = MALLOCA(userdata_chunk_size); | ||||
| memcpy(userdata_chunk_local, userdata_chunk, userdata_chunk_size); | memcpy(userdata_chunk_local, userdata_chunk, userdata_chunk_size); | ||||
| } | } | ||||
| /* Also marking it as non-threaded for the iterator callback. */ | /* Also marking it as non-threaded for the iterator callback. */ | ||||
| state->iter_shared.spin_lock = NULL; | state->iter_shared.spin_lock = NULL; | ||||
| parallel_iterator_func_do(state, userdata_chunk, 0); | parallel_iterator_func_do(state, userdata_chunk); | ||||
| if (use_userdata_chunk && settings->func_free != NULL) { | if (use_userdata_chunk && settings->func_free != NULL) { | ||||
| /* `func_free` should only free data that was created during execution of `func`. */ | /* `func_free` should only free data that was created during execution of `func`. */ | ||||
| settings->func_free(state->userdata, userdata_chunk_local); | settings->func_free(state->userdata, userdata_chunk_local); | ||||
| } | } | ||||
| } | } | ||||
| static void task_parallel_iterator_do(const TaskParallelSettings *settings, | static void task_parallel_iterator_do(const TaskParallelSettings *settings, | ||||
| TaskParallelIteratorState *state) | TaskParallelIteratorState *state) | ||||
| { | { | ||||
| TaskScheduler *task_scheduler = BLI_task_scheduler_get(); | const int num_threads = BLI_task_scheduler_num_threads(); | ||||
| const int num_threads = BLI_task_scheduler_num_threads(task_scheduler); | |||||
| task_parallel_iterator_calc_chunk_size(settings, num_threads, state); | task_parallel_calc_chunk_size( | ||||
| settings, state->tot_items, num_threads, &state->iter_shared.chunk_size); | |||||
| if (!settings->use_threading) { | if (!settings->use_threading) { | ||||
| task_parallel_iterator_no_threads(settings, state); | task_parallel_iterator_no_threads(settings, state); | ||||
| return; | return; | ||||
| } | } | ||||
| const int chunk_size = state->iter_shared.chunk_size; | const int chunk_size = state->iter_shared.chunk_size; | ||||
| const int tot_items = state->tot_items; | const int tot_items = state->tot_items; | ||||
| Show All 12 Lines | static void task_parallel_iterator_do(const TaskParallelSettings *settings, | ||||
| state->iter_shared.spin_lock = &spin_lock; | state->iter_shared.spin_lock = &spin_lock; | ||||
| void *userdata_chunk = settings->userdata_chunk; | void *userdata_chunk = settings->userdata_chunk; | ||||
| const size_t userdata_chunk_size = settings->userdata_chunk_size; | const size_t userdata_chunk_size = settings->userdata_chunk_size; | ||||
| void *userdata_chunk_local = NULL; | void *userdata_chunk_local = NULL; | ||||
| void *userdata_chunk_array = NULL; | void *userdata_chunk_array = NULL; | ||||
| const bool use_userdata_chunk = (userdata_chunk_size != 0) && (userdata_chunk != NULL); | const bool use_userdata_chunk = (userdata_chunk_size != 0) && (userdata_chunk != NULL); | ||||
| TaskPool *task_pool = BLI_task_pool_create_suspended(task_scheduler, state, TASK_PRIORITY_HIGH); | TaskPool *task_pool = BLI_task_pool_create(state, TASK_PRIORITY_HIGH); | ||||
| if (use_userdata_chunk) { | if (use_userdata_chunk) { | ||||
| userdata_chunk_array = MALLOCA(userdata_chunk_size * num_tasks); | userdata_chunk_array = MALLOCA(userdata_chunk_size * num_tasks); | ||||
| } | } | ||||
| const int thread_id = BLI_task_pool_creator_thread_id(task_pool); | |||||
| for (size_t i = 0; i < num_tasks; i++) { | for (size_t i = 0; i < num_tasks; i++) { | ||||
| if (use_userdata_chunk) { | if (use_userdata_chunk) { | ||||
| userdata_chunk_local = (char *)userdata_chunk_array + (userdata_chunk_size * i); | userdata_chunk_local = (char *)userdata_chunk_array + (userdata_chunk_size * i); | ||||
| memcpy(userdata_chunk_local, userdata_chunk, userdata_chunk_size); | memcpy(userdata_chunk_local, userdata_chunk, userdata_chunk_size); | ||||
| } | } | ||||
| /* Use this pool's pre-allocated tasks. */ | /* Use this pool's pre-allocated tasks. */ | ||||
| BLI_task_pool_push_from_thread( | BLI_task_pool_push(task_pool, parallel_iterator_func, userdata_chunk_local, false, NULL); | ||||
| task_pool, parallel_iterator_func, userdata_chunk_local, false, NULL, thread_id); | |||||
| } | } | ||||
| BLI_task_pool_work_and_wait(task_pool); | BLI_task_pool_work_and_wait(task_pool); | ||||
| BLI_task_pool_free(task_pool); | BLI_task_pool_free(task_pool); | ||||
| if (use_userdata_chunk && (settings->func_reduce != NULL || settings->func_free != NULL)) { | if (use_userdata_chunk && (settings->func_reduce != NULL || settings->func_free != NULL)) { | ||||
| for (size_t i = 0; i < num_tasks; i++) { | for (size_t i = 0; i < num_tasks; i++) { | ||||
| userdata_chunk_local = (char *)userdata_chunk_array + (userdata_chunk_size * i); | userdata_chunk_local = (char *)userdata_chunk_array + (userdata_chunk_size * i); | ||||
| ▲ Show 20 Lines • Show All 99 Lines • ▼ Show 20 Lines | |||||
| #undef MALLOCA | #undef MALLOCA | ||||
| #undef MALLOCA_FREE | #undef MALLOCA_FREE | ||||
| typedef struct ParallelMempoolState { | typedef struct ParallelMempoolState { | ||||
| void *userdata; | void *userdata; | ||||
| TaskParallelMempoolFunc func; | TaskParallelMempoolFunc func; | ||||
| } ParallelMempoolState; | } ParallelMempoolState; | ||||
| static void parallel_mempool_func(TaskPool *__restrict pool, void *taskdata, int UNUSED(threadid)) | static void parallel_mempool_func(TaskPool *__restrict pool, void *taskdata) | ||||
| { | { | ||||
| ParallelMempoolState *__restrict state = BLI_task_pool_userdata(pool); | ParallelMempoolState *__restrict state = BLI_task_pool_user_data(pool); | ||||
| BLI_mempool_iter *iter = taskdata; | BLI_mempool_iter *iter = taskdata; | ||||
| MempoolIterData *item; | MempoolIterData *item; | ||||
| while ((item = BLI_mempool_iterstep(iter)) != NULL) { | while ((item = BLI_mempool_iterstep(iter)) != NULL) { | ||||
| state->func(state->userdata, item); | state->func(state->userdata, item); | ||||
| } | } | ||||
| } | } | ||||
| Show All 9 Lines | |||||
| * | * | ||||
| * \note There is no static scheduling here. | * \note There is no static scheduling here. | ||||
| */ | */ | ||||
| void BLI_task_parallel_mempool(BLI_mempool *mempool, | void BLI_task_parallel_mempool(BLI_mempool *mempool, | ||||
| void *userdata, | void *userdata, | ||||
| TaskParallelMempoolFunc func, | TaskParallelMempoolFunc func, | ||||
| const bool use_threading) | const bool use_threading) | ||||
| { | { | ||||
| TaskScheduler *task_scheduler; | |||||
| TaskPool *task_pool; | TaskPool *task_pool; | ||||
| ParallelMempoolState state; | ParallelMempoolState state; | ||||
| int i, num_threads, num_tasks; | int i, num_threads, num_tasks; | ||||
| if (BLI_mempool_len(mempool) == 0) { | if (BLI_mempool_len(mempool) == 0) { | ||||
| return; | return; | ||||
| } | } | ||||
| if (!use_threading) { | if (!use_threading) { | ||||
| BLI_mempool_iter iter; | BLI_mempool_iter iter; | ||||
| BLI_mempool_iternew(mempool, &iter); | BLI_mempool_iternew(mempool, &iter); | ||||
| for (void *item = BLI_mempool_iterstep(&iter); item != NULL; | for (void *item = BLI_mempool_iterstep(&iter); item != NULL; | ||||
| item = BLI_mempool_iterstep(&iter)) { | item = BLI_mempool_iterstep(&iter)) { | ||||
| func(userdata, item); | func(userdata, item); | ||||
| } | } | ||||
| return; | return; | ||||
| } | } | ||||
| task_scheduler = BLI_task_scheduler_get(); | task_pool = BLI_task_pool_create(&state, TASK_PRIORITY_HIGH); | ||||
| task_pool = BLI_task_pool_create_suspended(task_scheduler, &state, TASK_PRIORITY_HIGH); | num_threads = BLI_task_scheduler_num_threads(); | ||||
| num_threads = BLI_task_scheduler_num_threads(task_scheduler); | |||||
| /* The idea here is to prevent creating task for each of the loop iterations | /* The idea here is to prevent creating task for each of the loop iterations | ||||
| * and instead have tasks which are evenly distributed across CPU cores and | * and instead have tasks which are evenly distributed across CPU cores and | ||||
| * pull next item to be crunched using the threaded-aware BLI_mempool_iter. | * pull next item to be crunched using the threaded-aware BLI_mempool_iter. | ||||
| */ | */ | ||||
| num_tasks = num_threads + 2; | num_tasks = num_threads + 2; | ||||
| state.userdata = userdata; | state.userdata = userdata; | ||||
| state.func = func; | state.func = func; | ||||
| BLI_mempool_iter *mempool_iterators = BLI_mempool_iter_threadsafe_create(mempool, | BLI_mempool_iter *mempool_iterators = BLI_mempool_iter_threadsafe_create(mempool, | ||||
| (size_t)num_tasks); | (size_t)num_tasks); | ||||
| const int thread_id = BLI_task_pool_creator_thread_id(task_pool); | |||||
| for (i = 0; i < num_tasks; i++) { | for (i = 0; i < num_tasks; i++) { | ||||
| /* Use this pool's pre-allocated tasks. */ | /* Use this pool's pre-allocated tasks. */ | ||||
| BLI_task_pool_push_from_thread( | BLI_task_pool_push(task_pool, parallel_mempool_func, &mempool_iterators[i], false, NULL); | ||||
| task_pool, parallel_mempool_func, &mempool_iterators[i], false, NULL, thread_id); | |||||
| } | } | ||||
| BLI_task_pool_work_and_wait(task_pool); | BLI_task_pool_work_and_wait(task_pool); | ||||
| BLI_task_pool_free(task_pool); | BLI_task_pool_free(task_pool); | ||||
| BLI_mempool_iter_threadsafe_free(mempool_iterators); | BLI_mempool_iter_threadsafe_free(mempool_iterators); | ||||
| } | } | ||||