// Note: once we move threading into a separate C++ file // will use std::hardware_destructive_interference_size instead of hardcoding it here // and we'll use C++ attribute syntax. #define GGML_CACHE_LINE 64
// synchronization primitives
atomic_int n_graph; // incremented when there is work to be done (i.e each graph)
atomic_int GGML_CACHE_ALIGN n_barrier;
atomic_int GGML_CACHE_ALIGN n_barrier_passed;
atomic_int GGML_CACHE_ALIGN current_chunk; // currently processing chunk during Mat_Mul, shared between all the threads.
// these are atomic as an annotation for thread-sanitizer
atomic_bool stop; // Used for stopping the threadpool altogether
atomic_bool pause; // Used for pausing the threadpool or individual threads
atomic_int abort; // Used for aborting processing of a graph
struct ggml_compute_state * workers; // per thread state int n_threads_max; // number of threads in the pool
atomic_int n_threads_cur; // number of threads used in the current graph
struct ggml_numa_node {
uint32_t cpus[GGML_NUMA_MAX_CPUS]; // hardware threads on this node
uint32_t n_cpus;
};
struct ggml_numa_nodes { enum ggml_numa_strategy numa_strategy; struct ggml_numa_node nodes[GGML_NUMA_MAX_NODES];
uint32_t n_nodes;
uint32_t total_cpus; // hardware threads on system
uint32_t current_node; // node on which main process is execting #ifdefined(__gnu_linux__)
cpu_set_t cpuset; // cpuset from numactl #else
uint32_t cpuset; // no NUMA support outside of Linux at this time. Use a portable datatype #endif
};
GGML_PRINT_DEBUG("found %u numa nodes, %u CPUs\n", g_state.numa.n_nodes, g_state.numa.total_cpus);
// figure out which node we're on
uint current_cpu; int getcpu_ret = 0; #if __GLIBC__ > 2 || (__GLIBC__ == 2 && __GLIBC_MINOR__ > 33) || defined(__COSMOPOLITAN__)
getcpu_ret = getcpu(¤t_cpu, &g_state.numa.current_node); #else // old glibc doesn't have a wrapper for this call. Fall back on direct syscall # if !defined(SYS_getcpu) && defined(SYS_get_cpu) # define SYS_getcpu SYS_get_cpu // some older glibc versions use this name # endif
getcpu_ret = syscall(SYS_getcpu, ¤t_cpu, &g_state.numa.current_node); #endif
// desc: when src1 is not a contiguous memory block we have to calculate the offset using the strides // if it is, then we have either copied the data to params->wdata and made it contiguous or we are using // the original src1 data pointer, so we should index using the indices directly // TODO: this is a bit of a hack, we should probably have a better way to handle this constchar * src1_col = (constchar*)wdata +
(src1_cont || src1->type != vec_dot_type
? (i11 + i12 * ne11 + i13 * ne12 * ne11) * row_size
: (i11 * nb11 + i12 * nb12 + i13 * nb13)); float * dst_col = (float*)((char*)dst->data + (i1 * nb1 + i2 * nb2 + i3 * nb3));
if (ith == 0) { // Every thread starts at ith, so the first unprocessed chunk is nth. This save a bit of coordination right at the start.
atomic_store_explicit(¶ms->threadpool->current_chunk, nth, memory_order_relaxed);
}
// This is the size of the first dimension of the result, so we can iterate that way. (see the ASSERT above, these are the same numbers) const int64_t nr0 = ne0;
// This is the size of the rest of the dimensions of the result const int64_t nr1 = ne1 * ne2 * ne3;
// Now select a reasonable chunk size. int chunk_size = 16;
// We need to step up the size if it's small if (nr0 == 1 || nr1 == 1) {
chunk_size = 64;
}
// distribute the work across the inner or outer loop based on which one is larger // The number of chunks in the 0/1 dim. // CEIL(nr0/chunk_size)
int64_t nchunk0 = (nr0 + chunk_size - 1) / chunk_size;
int64_t nchunk1 = (nr1 + chunk_size - 1) / chunk_size;
// If the chunking is poor for the number of threads on this setup, scrap the whole plan. Re-chunk it by thread. // Also, chunking by thread was measured to have perform better on NUMA systems. See https://github.com/ggml-org/llama.cpp/pull/6915 // In theory, chunking should be just as useful on NUMA and non NUMA systems, but testing disagreed with that. if (nchunk0 * nchunk1 < nth * 4 || ggml_is_numa()) { // distribute the thread work across the inner or outer loop based on which one is larger
nchunk0 = nr0 > nr1 ? nth : 1; // parallelize by src0 rows
nchunk1 = nr0 > nr1 ? 1 : nth; // parallelize by src1 rows
}
// The number of elements in each chunk const int64_t dr0 = (nr0 + nchunk0 - 1) / nchunk0; const int64_t dr1 = (nr1 + nchunk1 - 1) / nchunk1;
// The first chunk comes from our thread_id, the rest will get auto-assigned. int current_chunk = ith;
// dot kernels can handle 1 row and col at a time, but mmla kernels can process 2 rows and cols
int64_t num_rows_per_vec_dot = vec_dot_num_rows;
// these checks are needed to avoid crossing dim1 boundaries // can be optimized, but the logic would become more complicated, so keeping it like this for simplicity if ((nr0 % 2 != 0) || (ne11 % 2 != 0) || ((ir0_end - ir0_start) % 2 != 0) || ((ir1_end - ir1_start) % 2 != 0)) {
num_rows_per_vec_dot = 1;
}
ggml_compute_forward_mul_mat_one_chunk(params, dst, src0->type, num_rows_per_vec_dot, ir0_start, ir0_end, ir1_start, ir1_end);
// desc: when src1 is not a contiguous memory block we have to calculate the offset using the strides // if it is, then we have either copied the data to params->wdata and made it contiguous or we are using // the original src1 data pointer, so we should index using the indices directly // TODO: this is a bit of a hack, we should probably have a better way to handle this constchar * src1_col = (constchar *) wdata +
(src1_cont || src1->type != vec_dot_type
? (i11 + i12*ne11)*row_size
: (i11*nb11 + i12*nb12));
int chunk_size = 16; if (nr0 == 1 || nr1 == 1) {
chunk_size = 64;
}
#ifdefined(__aarch64__) // disable for ARM constbool disable_chunking = true; #else // disable for NUMA constbool disable_chunking = ggml_is_numa(); #endif// defined(__aarch64__)
if (tensor->op == GGML_OP_NONE || ggml_is_empty(tensor)) { return;
}
// extra_buffer op? if (ggml_cpu_extra_compute_forward(params, tensor)) { return;
}
switch (tensor->op) { case GGML_OP_DUP:
{
ggml_compute_forward_dup(params, tensor);
} break; case GGML_OP_ADD:
{
ggml_compute_forward_add(params, tensor);
} break; case GGML_OP_ADD_ID:
{
ggml_compute_forward_add_id(params, tensor);
} break; case GGML_OP_ADD1:
{
ggml_compute_forward_add1(params, tensor);
} break; case GGML_OP_ACC:
{
ggml_compute_forward_acc(params, tensor);
} break; case GGML_OP_SUB:
{
ggml_compute_forward_sub(params, tensor);
} break; case GGML_OP_MUL:
{
ggml_compute_forward_mul(params, tensor);
} break; case GGML_OP_DIV:
{
ggml_compute_forward_div(params, tensor);
} break; case GGML_OP_SQR:
{
ggml_compute_forward_sqr(params, tensor);
} break; case GGML_OP_SQRT:
{
ggml_compute_forward_sqrt(params, tensor);
} break; case GGML_OP_LOG:
{
ggml_compute_forward_log(params, tensor);
} break; case GGML_OP_SIN:
{
ggml_compute_forward_sin(params, tensor);
} break; case GGML_OP_COS:
{
ggml_compute_forward_cos(params, tensor);
} break; case GGML_OP_SUM:
{
ggml_compute_forward_sum(params, tensor);
} break; case GGML_OP_SUM_ROWS:
{
ggml_compute_forward_sum_rows(params, tensor);
} break; case GGML_OP_MEAN:
{
ggml_compute_forward_mean(params, tensor);
} break; case GGML_OP_ARGMAX:
{
ggml_compute_forward_argmax(params, tensor);
} break; case GGML_OP_COUNT_EQUAL:
{
ggml_compute_forward_count_equal(params, tensor);
} break; case GGML_OP_REPEAT:
{
ggml_compute_forward_repeat(params, tensor);
} break; case GGML_OP_REPEAT_BACK:
{
ggml_compute_forward_repeat_back(params, tensor);
} break; case GGML_OP_CONCAT:
{
ggml_compute_forward_concat(params, tensor);
} break; case GGML_OP_SILU_BACK:
{
ggml_compute_forward_silu_back(params, tensor);
} break; case GGML_OP_NORM:
{
ggml_compute_forward_norm(params, tensor);
} break; case GGML_OP_RMS_NORM:
{
ggml_compute_forward_rms_norm(params, tensor);
} break; case GGML_OP_RMS_NORM_BACK:
{
ggml_compute_forward_rms_norm_back(params, tensor);
} break; case GGML_OP_GROUP_NORM:
{
ggml_compute_forward_group_norm(params, tensor);
} break; case GGML_OP_L2_NORM:
{
ggml_compute_forward_l2_norm(params, tensor);
} break; case GGML_OP_MUL_MAT:
{
ggml_compute_forward_mul_mat(params, tensor);
} break; case GGML_OP_MUL_MAT_ID:
{
ggml_compute_forward_mul_mat_id(params, tensor);
} break; case GGML_OP_OUT_PROD:
{
ggml_compute_forward_out_prod(params, tensor);
} break; case GGML_OP_SCALE:
{
ggml_compute_forward_scale(params, tensor);
} break; case GGML_OP_SET:
{
ggml_compute_forward_set(params, tensor);
} break; case GGML_OP_CPY:
{
ggml_compute_forward_cpy(params, tensor);
} break; case GGML_OP_CONT:
{
ggml_compute_forward_cont(params, tensor);
} break; case GGML_OP_RESHAPE:
{
ggml_compute_forward_reshape(params, tensor);
} break; case GGML_OP_VIEW:
{
ggml_compute_forward_view(params, tensor);
} break; case GGML_OP_PERMUTE:
{
ggml_compute_forward_permute(params, tensor);
} break; case GGML_OP_TRANSPOSE:
{
ggml_compute_forward_transpose(params, tensor);
} break; case GGML_OP_GET_ROWS:
{
ggml_compute_forward_get_rows(params, tensor);
} break; case GGML_OP_GET_ROWS_BACK:
{
ggml_compute_forward_get_rows_back(params, tensor);
} break; case GGML_OP_SET_ROWS:
{
ggml_compute_forward_set_rows(params, tensor);
} break; case GGML_OP_DIAG:
{
ggml_compute_forward_diag(params, tensor);
} break; case GGML_OP_DIAG_MASK_INF:
{
ggml_compute_forward_diag_mask_inf(params, tensor);
} break; case GGML_OP_DIAG_MASK_ZERO:
{
ggml_compute_forward_diag_mask_zero(params, tensor);
} break; case GGML_OP_SOFT_MAX:
{
ggml_compute_forward_soft_max(params, tensor);
} break; case GGML_OP_SOFT_MAX_BACK:
{
ggml_compute_forward_soft_max_ext_back(params, tensor);
} break; case GGML_OP_ROPE:
{
ggml_compute_forward_rope(params, tensor);
} break; case GGML_OP_ROPE_BACK:
{
ggml_compute_forward_rope_back(params, tensor);
} break; case GGML_OP_CLAMP:
{
ggml_compute_forward_clamp(params, tensor);
} break; case GGML_OP_CONV_TRANSPOSE_1D:
{
ggml_compute_forward_conv_transpose_1d(params, tensor);
} break; case GGML_OP_IM2COL:
{
ggml_compute_forward_im2col(params, tensor);
} break; case GGML_OP_IM2COL_BACK:
{
ggml_compute_forward_im2col_back_f32(params, tensor);
} break; case GGML_OP_CONV_2D:
{
ggml_compute_forward_conv_2d(params, tensor);
} break; case GGML_OP_CONV_2D_DW:
{
ggml_compute_forward_conv_2d_dw(params, tensor);
} break; case GGML_OP_CONV_TRANSPOSE_2D:
{
ggml_compute_forward_conv_transpose_2d(params, tensor);
} break; case GGML_OP_POOL_1D:
{
ggml_compute_forward_pool_1d(params, tensor);
} break; case GGML_OP_POOL_2D:
{
ggml_compute_forward_pool_2d(params, tensor);
} break; case GGML_OP_POOL_2D_BACK:
{
ggml_compute_forward_pool_2d_back(params, tensor);
} break; case GGML_OP_UPSCALE:
{
ggml_compute_forward_upscale(params, tensor);
} break; case GGML_OP_PAD:
{
ggml_compute_forward_pad(params, tensor);
} break; case GGML_OP_PAD_REFLECT_1D:
{
ggml_compute_forward_pad_reflect_1d(params, tensor);
} break; case GGML_OP_ROLL:
{
ggml_compute_forward_roll(params, tensor);
} break; case GGML_OP_ARANGE:
{
ggml_compute_forward_arange(params, tensor);
} break; case GGML_OP_TIMESTEP_EMBEDDING:
{
ggml_compute_forward_timestep_embedding(params, tensor);
} break; case GGML_OP_ARGSORT:
{
ggml_compute_forward_argsort(params, tensor);
} break; case GGML_OP_LEAKY_RELU:
{
ggml_compute_forward_leaky_relu(params, tensor);
} break; case GGML_OP_FLASH_ATTN_EXT:
{
ggml_compute_forward_flash_attn_ext(params, tensor);
} break; case GGML_OP_FLASH_ATTN_BACK:
{
int32_t t = ggml_get_op_params_i32(tensor, 0);
GGML_ASSERT(t == 0 || t == 1); bool masked = t != 0;
ggml_compute_forward_flash_attn_back(params, masked, tensor);
} break; case GGML_OP_SSM_CONV:
{
ggml_compute_forward_ssm_conv(params, tensor);
} break; case GGML_OP_SSM_SCAN:
{
ggml_compute_forward_ssm_scan(params, tensor);
} break; case GGML_OP_WIN_PART:
{
ggml_compute_forward_win_part(params, tensor);
} break; case GGML_OP_WIN_UNPART:
{
ggml_compute_forward_win_unpart(params, tensor);
} break; case GGML_OP_UNARY:
{
ggml_compute_forward_unary(params, tensor);
} break; case GGML_OP_GLU:
{
ggml_compute_forward_glu(params, tensor);
} break; case GGML_OP_GET_REL_POS:
{
ggml_compute_forward_get_rel_pos(params, tensor);
} break; case GGML_OP_ADD_REL_POS:
{
ggml_compute_forward_add_rel_pos(params, tensor);
} break; case GGML_OP_RWKV_WKV6:
{
ggml_compute_forward_rwkv_wkv6(params, tensor);
} break; case GGML_OP_GATED_LINEAR_ATTN:
{
ggml_compute_forward_gla(params, tensor);
} break; case GGML_OP_RWKV_WKV7:
{
ggml_compute_forward_rwkv_wkv7(params, tensor);
} break; case GGML_OP_MAP_CUSTOM1:
{
ggml_compute_forward_map_custom1(params, tensor);
} break; case GGML_OP_MAP_CUSTOM2:
{
ggml_compute_forward_map_custom2(params, tensor);
} break; case GGML_OP_MAP_CUSTOM3:
{
ggml_compute_forward_map_custom3(params, tensor);
} break; case GGML_OP_CUSTOM:
{
ggml_compute_forward_custom(params, tensor);
} break; case GGML_OP_CROSS_ENTROPY_LOSS:
{
ggml_compute_forward_cross_entropy_loss(params, tensor);
} break; case GGML_OP_CROSS_ENTROPY_LOSS_BACK:
{
ggml_compute_forward_cross_entropy_loss_back(params, tensor);
} break; case GGML_OP_OPT_STEP_ADAMW:
{
ggml_compute_forward_opt_step_adamw(params, tensor);
} break; case GGML_OP_OPT_STEP_SGD:
{
ggml_compute_forward_opt_step_sgd(params, tensor);
} break; case GGML_OP_NONE:
{ // nop
} break; case GGML_OP_COUNT:
{
GGML_ABORT("fatal error");
}
}
}
// Android's libc implementation "bionic" does not support setting affinity #ifdefined(__gnu_linux__) staticvoid set_numa_thread_affinity(int thread_n) { if (!ggml_is_numa()) { return;
}
int node_num; int rv;
size_t setsize = CPU_ALLOC_SIZE(g_state.numa.total_cpus);
switch(g_state.numa.numa_strategy) { case GGML_NUMA_STRATEGY_DISTRIBUTE: // run thread on node_num thread_n / (threads per node)
node_num = thread_n % g_state.numa.n_nodes; break; case GGML_NUMA_STRATEGY_ISOLATE: // run thread on current_node
node_num = g_state.numa.current_node; break; case GGML_NUMA_STRATEGY_NUMACTL: // use the cpuset that numactl gave us
rv = pthread_setaffinity_np(pthread_self(), setsize, &g_state.numa.cpuset); if (rv) {
fprintf(stderr, "warning: pthread_setaffinity_np() failed: %s\n",strerror(rv));
} return; default: return;
}
cpu_set_t * cpus = CPU_ALLOC(g_state.numa.total_cpus);
CPU_ZERO_S(setsize, cpus); for (unsigned i = 0; i < g_state.numa.total_cpus; ++i) {
CPU_SET_S(i, setsize, cpus);
}
int rv = pthread_setaffinity_np(pthread_self(), setsize, cpus); if (rv) {
fprintf(stderr, "warning: pthread_setaffinity_np() failed: %s\n", strerror(rv));
}
CPU_FREE(cpus);
} #else // TODO: Windows etc. // (the linux implementation may also work on BSD, someone should test) staticvoid set_numa_thread_affinity(int thread_n) { UNUSED(thread_n); } staticvoid clear_numa_thread_affinity(void) {} #endif
staticint ggml_get_n_tasks(struct ggml_tensor * node, int n_threads) { int n_tasks = 0;
if (ggml_is_empty(node)) { // no need to multi-thread a no-op
n_tasks = 1; return n_tasks;
}
switch (node->op) { case GGML_OP_CPY: case GGML_OP_DUP: case GGML_OP_CONT: case GGML_OP_ADD: case GGML_OP_ADD_ID: case GGML_OP_ADD1: case GGML_OP_ACC:
{
n_tasks = n_threads;
} break; case GGML_OP_SUB: case GGML_OP_SQR: case GGML_OP_SQRT: case GGML_OP_LOG: case GGML_OP_SIN: case GGML_OP_COS: case GGML_OP_SUM: case GGML_OP_SUM_ROWS: case GGML_OP_MEAN: case GGML_OP_ARGMAX:
{
n_tasks = 1;
} break; case GGML_OP_COUNT_EQUAL:
{
n_tasks = n_threads;
} break; case GGML_OP_REPEAT: case GGML_OP_REPEAT_BACK: case GGML_OP_LEAKY_RELU:
{
n_tasks = 1;
} break; case GGML_OP_UNARY: switch (ggml_get_unary_op(node)) { case GGML_UNARY_OP_ABS: case GGML_UNARY_OP_SGN: case GGML_UNARY_OP_NEG: case GGML_UNARY_OP_STEP: case GGML_UNARY_OP_TANH: case GGML_UNARY_OP_ELU: case GGML_UNARY_OP_RELU: case GGML_UNARY_OP_SIGMOID: case GGML_UNARY_OP_HARDSWISH: case GGML_UNARY_OP_HARDSIGMOID: case GGML_UNARY_OP_EXP:
{
n_tasks = 1;
} break;
case GGML_UNARY_OP_GELU: case GGML_UNARY_OP_GELU_ERF: case GGML_UNARY_OP_GELU_QUICK: case GGML_UNARY_OP_SILU:
{
n_tasks = n_threads;
} break; default:
GGML_ABORT("fatal error");
} break; case GGML_OP_GLU: switch (ggml_get_glu_op(node)) { case GGML_GLU_OP_REGLU: case GGML_GLU_OP_GEGLU: case GGML_GLU_OP_SWIGLU: case GGML_GLU_OP_SWIGLU_OAI: case GGML_GLU_OP_GEGLU_ERF: case GGML_GLU_OP_GEGLU_QUICK:
{
n_tasks = n_threads;
} break; default:
GGML_ABORT("fatal error");
} break; case GGML_OP_SILU_BACK: case GGML_OP_MUL: case GGML_OP_DIV: case GGML_OP_NORM: case GGML_OP_RMS_NORM: case GGML_OP_RMS_NORM_BACK: case GGML_OP_L2_NORM: case GGML_OP_GROUP_NORM: case GGML_OP_CONCAT: case GGML_OP_MUL_MAT: case GGML_OP_MUL_MAT_ID: case GGML_OP_OUT_PROD:
{
n_tasks = n_threads;
} break; case GGML_OP_GET_ROWS: case GGML_OP_SET_ROWS:
{ // FIXME: get_rows can use additional threads, but the cost of launching additional threads // decreases performance with GPU offloading //n_tasks = n_threads;
n_tasks = 1;
} break; case GGML_OP_SCALE: case GGML_OP_SET: case GGML_OP_RESHAPE: case GGML_OP_VIEW: case GGML_OP_PERMUTE: case GGML_OP_TRANSPOSE: case GGML_OP_GET_ROWS_BACK: case GGML_OP_DIAG:
{
n_tasks = 1;
} break; case GGML_OP_DIAG_MASK_ZERO: case GGML_OP_DIAG_MASK_INF: case GGML_OP_SOFT_MAX_BACK: case GGML_OP_ROPE: case GGML_OP_ROPE_BACK: case GGML_OP_ADD_REL_POS:
{
n_tasks = n_threads;
} break; case GGML_OP_CLAMP:
{
n_tasks = 1; //TODO
} break; case GGML_OP_SOFT_MAX:
{
n_tasks = MIN(n_threads, ggml_nrows(node->src[0]));
} break; case GGML_OP_IM2COL: case GGML_OP_IM2COL_BACK: case GGML_OP_CONV_2D: case GGML_OP_CONV_2D_DW: case GGML_OP_CONV_TRANSPOSE_1D: case GGML_OP_CONV_TRANSPOSE_2D:
{
n_tasks = n_threads;
} break; case GGML_OP_POOL_1D: case GGML_OP_POOL_2D: case GGML_OP_POOL_2D_BACK:
{
n_tasks = 1;
} break; case GGML_OP_UPSCALE: case GGML_OP_PAD: case GGML_OP_PAD_REFLECT_1D: case GGML_OP_ROLL: case GGML_OP_ARANGE: case GGML_OP_TIMESTEP_EMBEDDING: case GGML_OP_ARGSORT: case GGML_OP_FLASH_ATTN_EXT: case GGML_OP_FLASH_ATTN_BACK: case GGML_OP_SSM_CONV: case GGML_OP_SSM_SCAN: case GGML_OP_RWKV_WKV6: case GGML_OP_GATED_LINEAR_ATTN: case GGML_OP_RWKV_WKV7:
{
n_tasks = n_threads;
} break; case GGML_OP_WIN_PART: case GGML_OP_WIN_UNPART: case GGML_OP_GET_REL_POS:
{
n_tasks = 1;
} break; case GGML_OP_MAP_CUSTOM1:
{ struct ggml_map_custom1_op_params p;
memcpy(&p, node->op_params, sizeof(p)); if (p.n_tasks == GGML_N_TASKS_MAX) {
n_tasks = n_threads;
} else {
n_tasks = MIN(p.n_tasks, n_threads);
}
} break; case GGML_OP_MAP_CUSTOM2:
{ struct ggml_map_custom2_op_params p;
memcpy(&p, node->op_params, sizeof(p)); if (p.n_tasks == GGML_N_TASKS_MAX) {
n_tasks = n_threads;
} else {
n_tasks = MIN(p.n_tasks, n_threads);
}
} break; case GGML_OP_MAP_CUSTOM3:
{ struct ggml_map_custom3_op_params p;
memcpy(&p, node->op_params, sizeof(p)); if (p.n_tasks == GGML_N_TASKS_MAX) {
n_tasks = n_threads;
} else {
n_tasks = MIN(p.n_tasks, n_threads);
}
} break; case GGML_OP_CUSTOM:
{ struct ggml_custom_op_params p;
memcpy(&p, node->op_params, sizeof(p)); if (p.n_tasks == GGML_N_TASKS_MAX) {
n_tasks = n_threads;
} else {
n_tasks = MIN(p.n_tasks, n_threads);
}
} break; case GGML_OP_CROSS_ENTROPY_LOSS: case GGML_OP_CROSS_ENTROPY_LOSS_BACK: case GGML_OP_OPT_STEP_ADAMW: case GGML_OP_OPT_STEP_SGD:
{
n_tasks = n_threads;
} break; case GGML_OP_NONE:
{
n_tasks = 1;
} break; case GGML_OP_COUNT:
{
GGML_ABORT("fatal error");
} default:
{
fprintf(stderr, "%s: op not implemented: ", __func__); if (node->op < GGML_OP_COUNT) {
fprintf(stderr, "%s\n", ggml_op_name(node->op));
} else {
fprintf(stderr, "%d\n", node->op);
}
GGML_ABORT("fatal error");
}
}
// TODO: support > 64 CPUs staticbool ggml_thread_apply_affinity(bool * mask) {
HANDLE h = GetCurrentThread();
uint64_t bitmask = 0ULL;
assert(GGML_MAX_N_THREADS >= 64);
for (int32_t i = 0; i < 8; i++) {
int32_t idx = i * 8;
uint8_t val = 0;
val |= mask[idx + 0] << 0;
val |= mask[idx + 1] << 1;
val |= mask[idx + 2] << 2;
val |= mask[idx + 3] << 3;
val |= mask[idx + 4] << 4;
val |= mask[idx + 5] << 5;
val |= mask[idx + 6] << 6;
val |= mask[idx + 7] << 7;
bitmask |= (uint64_t)val << idx;
}
for (int32_t i = 64; i < GGML_MAX_N_THREADS; i++) { if (mask[i]) {
fprintf(stderr, "warn: setting thread-affinity for > 64 CPUs isn't supported on windows!\n"); break;
}
}
DWORD_PTR m = (DWORD_PTR)bitmask;
m = SetThreadAffinityMask(h, m);
return m != 0;
}
staticbool ggml_thread_apply_priority(int32_t prio) { // Note that on Windows the Process Priority Class must be updated in order to set Thread priority. // This is up to the applications.
DWORD p = THREAD_PRIORITY_NORMAL; switch (prio) { case GGML_SCHED_PRIO_LOW: p = THREAD_PRIORITY_BELOW_NORMAL; break; case GGML_SCHED_PRIO_NORMAL: p = THREAD_PRIORITY_NORMAL; break; case GGML_SCHED_PRIO_MEDIUM: p = THREAD_PRIORITY_ABOVE_NORMAL; break; case GGML_SCHED_PRIO_HIGH: p = THREAD_PRIORITY_HIGHEST; break; case GGML_SCHED_PRIO_REALTIME: p = THREAD_PRIORITY_TIME_CRITICAL; break;
}
if (prio != GGML_SCHED_PRIO_LOW) { // Tell Windows that this thread should not be throttled (needs its own CPU core). // Newer Windows 11 versions aggresively park (offline) CPU cores and often place // all our threads onto the first 4 cores which results in terrible performance with // n_threads > 4 #if _WIN32_WINNT >= 0x0602
THREAD_POWER_THROTTLING_STATE t;
ZeroMemory(&t, sizeof(t));
t.Version = THREAD_POWER_THROTTLING_CURRENT_VERSION;
t.ControlMask = THREAD_POWER_THROTTLING_EXECUTION_SPEED;
t.StateMask = 0;
if (!SetThreadInformation(GetCurrentThread(), ThreadPowerThrottling, &t, sizeof(t))) {
GGML_LOG_DEBUG("failed to disable thread power throttling %d : (%d)\n", prio, (int) GetLastError()); returnfalse;
} #endif
}
int32_t err = pthread_setschedparam(pthread_self(), policy, &p); if (err != 0) {
fprintf(stderr, "warn: failed to set thread priority %d : %s (%d)\n", prio, strerror(err), err); returnfalse;
}
return true;
}
#elifdefined(__gnu_linux__) // TODO: this may not work on BSD, to be verified
staticbool ggml_thread_apply_affinity(constbool * mask) {
cpu_set_t cpuset; int err;
CPU_ZERO(&cpuset);
for (uint32_t i = 0; i < GGML_MAX_N_THREADS; i++) { if (mask[i]) {
GGML_PRINT_DEBUG("Thread %lx: adding %d to cpuset\n", pthread_self(), i);
CPU_SET(i, &cpuset);
}
}
// thread scheduling for the different operations + work buffer size estimation for (int i = 0; i < cgraph->n_nodes; i++) { struct ggml_tensor * node = cgraph->nodes[i];
if ((node->src[0]->type == GGML_TYPE_F16 ||
node->src[0]->type == GGML_TYPE_BF16) &&
node->src[1]->type == GGML_TYPE_F32) {
cur += sizeof(ggml_fp16_t)*ne00*ne01*ne02;
cur += sizeof(ggml_fp16_t)*ne10*ne11;
} elseif (node->src[0]->type == GGML_TYPE_F32 &&
node->src[1]->type == GGML_TYPE_F32) {
cur += sizeof(float)*ne00*ne01*ne02;
cur += sizeof(float)*ne10*ne11;
} else {
GGML_ABORT("fatal error");
}
} break; case GGML_OP_CONV_2D:
{
cur = GGML_IM2COL_WORK_SIZE;
} break; case GGML_OP_CONV_TRANSPOSE_2D:
{ const int64_t ne00 = node->src[0]->ne[0]; // W const int64_t ne01 = node->src[0]->ne[1]; // H const int64_t ne02 = node->src[0]->ne[2]; // Channels Out const int64_t ne03 = node->src[0]->ne[3]; // Channels In
const int64_t ne10 = node->src[1]->ne[0]; // W const int64_t ne11 = node->src[1]->ne[1]; // H const int64_t ne12 = node->src[1]->ne[2]; // Channels In
cur += sizeof(ggml_fp16_t)*ne00*ne01*ne02*ne03;
cur += sizeof(ggml_fp16_t)*ne10*ne11*ne12;
} break; case GGML_OP_FLASH_ATTN_EXT:
{ const int64_t ne10 = node->src[1]->ne[0]; // DK const int64_t ne20 = node->src[2]->ne[0]; // DV
cur = sizeof(float)*(1*ne10 + 2*ne20)*n_tasks; // 1x head size K + 2x head size V (per thread)
} break; case GGML_OP_FLASH_ATTN_BACK:
{ const int64_t D = node->src[0]->ne[0]; const int64_t ne11 = ggml_up(node->src[1]->ne[1], GGML_SOFT_MAX_UNROLL); const int64_t mxDn = MAX(D, ne11) * 2; // *2 because of S and SM in ggml_compute_forward_flash_attn_back if (node->src[1]->type == GGML_TYPE_F32) {
cur = sizeof(float)*mxDn*n_tasks; // TODO: this can become (n_tasks-1)
cur += sizeof(float)*mxDn*n_tasks; // this is overestimated by x2
} elseif (node->src[1]->type == GGML_TYPE_F16) {
cur = sizeof(float)*mxDn*n_tasks; // TODO: this can become (n_tasks-1)
cur += sizeof(float)*mxDn*n_tasks; // this is overestimated by x2
} elseif (node->src[1]->type == GGML_TYPE_BF16) {
cur = sizeof(float)*mxDn*n_tasks; // TODO: this can become (n_tasks-1)
cur += sizeof(float)*mxDn*n_tasks; // this is overestimated by x2
}
} break;
case GGML_OP_CROSS_ENTROPY_LOSS:
{
cur = ggml_type_size(node->type)*(n_tasks + node->src[0]->ne[0]*n_tasks);
} break; case GGML_OP_COUNT:
{
GGML_ABORT("fatal error");
} default: break;
}
}
work_size = MAX(work_size, cur);
}
if (work_size > 0) {
work_size += CACHE_LINE_SIZE*(n_threads);
}
if (node_n + 1 < cgraph->n_nodes) {
ggml_barrier(state->threadpool);
}
}
ggml_barrier(state->threadpool);
return0;
}
#ifndef GGML_USE_OPENMP
// check if thread is active staticinlinebool ggml_graph_compute_thread_active(struct ggml_compute_state * state) { struct ggml_threadpool * threadpool = state->threadpool; int n_threads = atomic_load_explicit(&threadpool->n_threads_cur, memory_order_relaxed); return (state->ith < n_threads);
}
// check if thread is ready to proceed (exit from polling or sleeping) staticinlinebool ggml_graph_compute_thread_ready(struct ggml_compute_state * state) { struct ggml_threadpool * threadpool = state->threadpool;
if (state->pending || threadpool->stop || threadpool->pause) { return true; }
// check for new graph/work int new_graph = atomic_load_explicit(&threadpool->n_graph, memory_order_relaxed); if (new_graph != state->last_graph) {
state->pending = ggml_graph_compute_thread_active(state);
state->last_graph = new_graph;
}
return state->pending;
}
// sync thread state after polling staticinlinevoid ggml_graph_compute_thread_sync(struct ggml_compute_state * state) { // TSAN doesn't support standalone fence yet, we use a dummy read-modify-write instead #ifdef GGML_TSAN_ENABLED
atomic_fetch_add_explicit(&state->threadpool->n_graph, 0, memory_order_seq_cst); #else
atomic_thread_fence(memory_order_seq_cst); #endif
UNUSED(state);
}
// Skip polling for unused threads if (!ggml_graph_compute_thread_active(state)) { return state->pending;
}
// This seems to make 0 ... 100 a decent range for polling level across modern processors. // Perhaps, we can adjust it dynamically based on load and things. const uint64_t n_rounds = 1024UL * 128 * threadpool->poll;
for (uint64_t i=0; !ggml_graph_compute_thread_ready(state) && i < n_rounds; i++) { // No new work. Keep polling.
ggml_thread_cpu_relax();
}
if (ggml_graph_compute_poll_for_work(state)) {
ggml_graph_compute_thread_sync(state); return state->pending;
}
ggml_mutex_lock_shared(&threadpool->mutex); while (!ggml_graph_compute_thread_ready(state)) { // No new work. Wait for the signal.
GGML_PRINT_DEBUG("thread #%d waiting for work (sleeping)\n", state->ith);
ggml_cond_wait(&threadpool->cond, &threadpool->mutex);
}
ggml_mutex_unlock_shared(&threadpool->mutex);
if (threadpool->thread_create_callback) {
threadpool->thread_create_callback();
}
ggml_thread_apply_priority(threadpool->prio); if (ggml_thread_cpumask_is_valid(state->cpumask)) {
ggml_thread_apply_affinity(state->cpumask);
}
while (true) { // Check if we need to sleep while (threadpool->pause) {
GGML_PRINT_DEBUG("thread #%d inside pause loop\n", state->ith);
ggml_mutex_lock_shared(&threadpool->mutex); if (threadpool->pause) {
ggml_cond_wait(&threadpool->cond, &threadpool->mutex);
}
GGML_PRINT_DEBUG("thread #%d resuming after wait\n", state->ith);
ggml_mutex_unlock_shared(&threadpool->mutex);
}
// This needs to be checked for after the cond_wait if (threadpool->stop) break;
// Check if there is new work // The main thread is the only one that can dispatch new work
ggml_graph_compute_check_for_work(state); if (state->pending) {
state->pending = false;
ggml_graph_compute_thread(state);
}
}
if (threadpool->thread_destroy_callback) {
threadpool->thread_destroy_callback();
}
return (thread_ret_t) 0;
}
// Start processing new graph staticvoid ggml_graph_compute_kickoff(struct ggml_threadpool * threadpool, int n_threads)
{ // Always take the mutex here because the worker threads are doing hybrid poll/wait
// Update the number of active threads
atomic_store_explicit(&threadpool->n_threads_cur, n_threads, memory_order_relaxed);
// Indicate the graph is ready to be processed // We need the full seq-cst fence here because of the polling threads (used in thread_sync)
atomic_fetch_add_explicit(&threadpool->n_graph, 1, memory_order_seq_cst);
if (threadpool->pause) { // Update main thread prio and affinity to match the threadpool settings
ggml_thread_apply_priority(threadpool->prio); if (ggml_thread_cpumask_is_valid(threadpool->workers[0].cpumask)) {
ggml_thread_apply_affinity(threadpool->workers[0].cpumask);
}
if (!threadpool->pause) { // Update main thread prio and affinity at the start, otherwise we'll do it in resume
ggml_thread_apply_priority(threadpool->prio); if (ggml_thread_cpumask_is_valid(threadpool->workers[0].cpumask)) {
ggml_thread_apply_affinity(threadpool->workers[0].cpumask);
}
} #endif// GGML_USE_OPENMP
int n_threads = cplan->n_threads; struct ggml_threadpool * threadpool = cplan->threadpool;
bool disposable_threadpool = false;
if (threadpool == NULL) { //GGML_PRINT_DEBUG("Threadpool is not specified. Will create a disposable threadpool : n_threads %d\n", n_threads);
disposable_threadpool = true;
struct ggml_threadpool_params ttp = ggml_threadpool_params_default(n_threads);
threadpool = ggml_threadpool_new_impl(&ttp, cgraph, cplan);
} else { // Reset some of the parameters that need resetting // No worker threads should be accessing the parameters below at this stage
threadpool->cgraph = cgraph;
threadpool->cplan = cplan;
threadpool->current_chunk = 0;
threadpool->abort = -1;
threadpool->ec = GGML_STATUS_SUCCESS;
}
#ifdef GGML_USE_OPENMP if (n_threads > 1) { #pragma omp parallel num_threads(n_threads)
{ #pragma omp single
{ // update the number of threads from the actual number of threads that we got from OpenMP
n_threads = omp_get_num_threads();
atomic_store_explicit(&threadpool->n_threads_cur, n_threads, memory_order_relaxed);
}
ggml_graph_compute_thread(&threadpool->workers[omp_get_thread_num()]);
}
} else {
atomic_store_explicit(&threadpool->n_threads_cur, 1, memory_order_relaxed);
ggml_graph_compute_thread(&threadpool->workers[0]);
} #else if (n_threads > threadpool->n_threads_max) {
GGML_LOG_WARN("cplan requested more threads (%d) than available (%d)\n", n_threads, threadpool->n_threads_max);
n_threads = threadpool->n_threads_max;
}
// Kick all threads to start the new graph
ggml_graph_compute_kickoff(threadpool, n_threads);
// This is a work thread too
ggml_graph_compute_thread(&threadpool->workers[0]); #endif
// don't leave affinity set on the main thread
clear_numa_thread_affinity();
enum ggml_status ret = threadpool->ec;
if (disposable_threadpool) {
ggml_threadpool_free(threadpool);
}
GGML_PRINT_DEBUG("%s: GELU, Quick GELU, SILU and EXP tables initialized in %f ms\n", __func__, (t_end - t_start)/1000.0);
#ifdef GGML_USE_OPENMP //if (!getenv("OMP_WAIT_POLICY")) { // // set the wait policy to active, so that OpenMP threads don't sleep // putenv("OMP_WAIT_POLICY=active"); //}
if (!getenv("KMP_BLOCKTIME")) { // set the time to wait before sleeping a thread // this is less aggressive than setting the wait policy to active, but should achieve similar results in most cases
putenv("KMP_BLOCKTIME=200"); // 200ms
} #endif
}
Die Informationen auf dieser Webseite wurden
nach bestem Wissen sorgfältig zusammengestellt. Es wird jedoch weder Vollständigkeit, noch Richtigkeit,
noch Qualität der bereit gestellten Informationen zugesichert.
Bemerkung:
Die farbliche Syntaxdarstellung und die Messung sind noch experimentell.