Files
9544a8960d feat(sft): AMX MoE SFT backend with LoRA support (#1936)
* feat(sft): AMX MoE SFT backend with LoRA support

Complete SFT (Supervised Fine-Tuning) backend for MoE models using AMX SIMD:

Core C++ implementation:
- sft_moe.hpp: Forward/backward with LoRA fused operations (~5500 lines)
- moe-sft-tp.hpp: Tensor-parallel wrapper for multi-NUMA
- amx/moe-sft-tp.hpp: AMX-specific TP implementation
- avx_kernels.hpp: AVX512 SIMD kernels for LoRA GEMM
- amx_kernels.hpp: AMX tile kernels for Panel5 rank-outer optimization
- worker_pool: RDTSC profiling, Chrome trace output, SFT timer infrastructure
- ext_bindings.cpp: SFT MOE pybind bindings (BF16/INT8/INT4 + SkipLoRA variants)

Python sft/ submodule (kt_kernel.sft):
- base.py: BaseSFTMoEWrapper with buffer management (template method pattern)
- amx.py: AMXSFTMoEWrapper (weight loading, C++ task construction)
- autograd.py: KTMoEFunction (torch.autograd.Function for distributed training)
- layer.py: KTMoELayerWrapper (nn.Module replacing HF MoE layers)
- arch.py: MOEArchConfig (Qwen3/DeepSeek/Mixtral architecture detection)
- weights.py: Expert weight extraction and checkpoint loading
- lora.py: PEFT LoRA adaptation (view buffers, grad buffers, save/load adapter)
- wrapper.py: wrap_moe_layers_with_kt_wrapper, load_kt_model, build_kt_device_map
- config.py: KTConfig dataclass (DeepSpeed-style opaque config passthrough)
- dist_utils.py: Distributed gather/scatter, checkpoint-phase detection

Design decisions:
- Rank-0-only expert pattern: only rank 0 holds C++ wrapper and expert weights
- DeepSpeed-style integration: accelerate keeps only KTransformersPlugin (framework
  interaction fields), all logic in kt_kernel.sft
- Inference isolation: importing kt_kernel does not load sft/ submodule
- Old field name compatibility: _get_kt_config() converts kt_xxx→xxx automatically

Verified: Qwen3-235B-A22B 4GPU AMXBF16 training, loss converges normally.

* refactor(sft): unify KTConfig field names with kt_ prefix, add share_cache_pool, remove dead code

- KTConfig fields all use kt_ prefix matching dict keys — eliminates
  _OLD_TO_NEW mapping and prefix-stripping in wrapper.py
- Add kt_share_cache_pool field, auto-enabled when gradient_checkpointing
  is on (via training_args.py), flows through to C++ cache allocation
- Remove dead checkpoint detection code: in_ckpt_recompute,
  in_ckpt_first_forward vars (assigned but never read), fallback
  _is_in_checkpoint_first_forward() function, unused inspect import
- Remove redundant env var fallbacks in wrapper.py for share_backward_bb
  and share_cache_pool (KTConfig.__post_init__ already handles env vars)
- Simplify layer.py checkpoint logic to single _checkpoint_hook_mode() check

Verified: Qwen3-235B 3-step training on sap4, loss matches baseline
(1.2886 / 1.9824 / 1.377 vs 1.2886 / 1.9766 / 1.3809)

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

* refactor(sft): share_backward_bb default True, share_cache_pool auto-derived

- kt_share_backward_bb defaults to True (always saves memory)
- kt_share_cache_pool no longer reads from env var; defaults False,
  auto-set to True by trainer_config_process when gradient checkpointing
  is enabled

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

* fix: add missing gpu_experts_mask=None to KTMoEWrapper call in SFT wrapper

KTMoEWrapper.__new__() requires gpu_experts_mask as a positional argument,
but the SFT wrapper omitted it, causing MoE layer wrapping to fail silently
and FSDP2 to attempt broadcasting all expert weights (OOM/NCCL crash).

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

* feat(sft): support transformers v5 fused expert format

Fused experts (e.g. Qwen3MoeExperts) store weights as 3D Parameters
(gate_up_proj [E,2I,H], down_proj [E,H,I]) instead of per-expert
nn.Linear modules. PEFT cannot attach LoRA to these, so we create
KT-managed LoRA buffers with kaiming init, nn.Parameter wrappers
for the optimizer, and pre-assigned .grad for C++ backward.

- arch.py: detect_fused_experts() detection
- weights.py: fused format extraction and weight clearing
- wrapper.py: detect fused at wrap time, store _fused_experts/_lora_rank
- lora.py: _create_fused_expert_lora_buffers, save/load fused LoRA,
  get_kt_lora_params collects fused params, deduplicate wrapper finding
- layer.py: handle v5 TopKRouter tuple output, remove dead code
- autograd.py: sync_forward_sft/submit_forward_sft API rename

Verified: v5 loss/expert-LoRA values match v4 baseline, v4 backward compat.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

* feat(sft): add Qwen3.5 MoE support + fused checkpoint loading

- arch.py: add Qwen3_5Moe arch match, read config from text_config,
  _get_layers_prefix returns model.language_model.layers for Qwen3.5,
  _get_model_container_and_layers searches language_model attr
- weights.py: load_experts_from_checkpoint_files detects fused format
  (gate_up_proj in weight_map) and splits into gate/up/down
- wrapper.py: hidden_size fallback to text_config

Verified: Qwen3.5-35B-A3B (256 experts, fused format) E2E pass.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

* [fix](sft): align Python API with C++ backend after v5 refactor

- wrapper.py: pass gpu_experts_mask=None to KTMoEWrapper (required by C++ signature)
- layer.py: rename submit_forward_sft/sync_forward_sft to submit_forward/sync_forward
- autograd.py: rename sync_forward_sft to sync_forward

The sft-v5 refactor (commits 58d7eab, dd1da65) renamed Python-side method
calls but the C++ backend (AMXSFTMoEWrapper) still exposes the original
method names. This caused AttributeError on Qwen3.5-35B and other models.

* align sft branch with main: revert worker_pool, strip sft_timer, fix inference defaults

- Revert worker_pool.cpp/.h to main (remove RDTSC timer, Chrome Trace,
  sft_timer namespace, ITT API, extended do_work_stealing_job API)
- Strip all sft_timer instrumentation from sft-only files (sft_moe.hpp,
  moe-sft-tp.hpp, avx_kernels.hpp)
- Restore pin_memory=True in KExpertsCPUBuffer (inference path)
- Restore fused tensor transpose logic in convert_cpu_weights.py (main layout)

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

* revert CMakeLists.txt to main: remove debug flags and cpptrace dep

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

* clean up dev artifacts: remove SFT design docs, debug examples, bench scripts

Remove files not needed in the merge:
- docs/SFT+KTWrapper/ (6 Chinese design docs)
- docs/sft_moe_amx/ (21 dev/debug docs)
- 12 debug/test example scripts
- 6 SFT-specific bench scripts and report

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

* remove dev version stamps from ext_bindings, sft_moe, moe-sft-tp

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
Co-authored-by: JimmyPeilinLi <lipeilin@mail.nwpu.edu.cn>
2026-04-22 11:27:01 +08:00

511 lines
15 KiB
C++

#ifndef CPUINFER_OPERATOR_COMMON_HPP
#define CPUINFER_OPERATOR_COMMON_HPP
#include <map>
#include "../cpu_backend/worker_pool.h"
#include "ggml.h"
#if defined(__aarch64__) && defined(CPU_USE_KML)
#include <arm_sve.h>
#endif
#include <chrono>
#include <cmath>
#include <cstdio>
#include <cstring>
#include <stdexcept>
#include <type_traits>
// #define FORWARD_TIME_PROFILE
// #define FORWARD_TIME_REPORT
#define ASSERT_RELEASE(x, text) \
do { \
if (!(x)) { \
fprintf(stderr, "Assertion failed: %s, file %s, line %d\n", #x, __FILE__, __LINE__); \
fprintf(stderr, "Error message: %s\n", (text)); \
throw std::runtime_error((text)); \
} \
} while (0)
#define PUSH_MEM_REQ(ptr, size) mem_requests.append_pointer(&(ptr), (size))
#define PROFILE_RECORD_TIME_STAMP(name) \
do { \
auto end_time = std::chrono::high_resolution_clock::now(); \
auto duration = std::chrono::duration_cast<std::chrono::microseconds>(end_time - last).count(); \
time_map[(name)] = duration; \
last = end_time; \
} while (0)
#define DO_TPS_LOAD_WEIGHTS(pool) \
(pool)->dispense_backend()->do_numa_job([this, pool, config](int numa_id) { \
this->tps[numa_id]->config_.physical_to_logical_map = config.physical_to_logical_map; \
this->tps[numa_id]->load_weights(); \
})
#define expert_map(m, x) (m != nullptr ? m[(x)] : (x))
template <typename T, typename std::enable_if<std::is_integral<T>::value, int>::type = 0>
inline T div_up(T x, T y) {
return (x + y - 1) / y;
}
template <typename T>
T* offset_pointer(T* ptr, size_t byte_offset) {
return reinterpret_cast<T*>(reinterpret_cast<char*>(ptr) + byte_offset);
}
template <typename T>
size_t pointer_offset(T* ptr, T* b) {
return reinterpret_cast<size_t>(b) - reinterpret_cast<size_t>(ptr);
}
template <typename T>
const T* offset_pointer(const T* ptr, size_t byte_offset) {
return reinterpret_cast<const T*>(reinterpret_cast<const char*>(ptr) + byte_offset);
}
template <typename T>
T* offset_pointer_row_major(T* t, int row, int col, size_t ld) {
return offset_pointer(t, row * ld) + col;
}
template <typename T>
T* offset_pointer_col_major(T* t, int row, int col, size_t ld) {
return offset_pointer(t, col * ld) + row;
}
class TimePerf {
protected:
std::string time_perf_name;
std::map<std::string, long> time_map;
std::chrono::time_point<std::chrono::high_resolution_clock> last;
std::chrono::time_point<std::chrono::high_resolution_clock> start_time;
void forward_perf_start() {
start_time = std::chrono::high_resolution_clock::now();
last = start_time;
}
void perf_report() {
auto end_time = std::chrono::high_resolution_clock::now();
auto duration = std::chrono::duration_cast<std::chrono::microseconds>(end_time - start_time);
std::string output = time_perf_name + ", forward time: " + std::to_string(duration.count()) + " us";
// for (auto [name, t] : time_map) {
// double p = 100.0 * t / duration.count();
// // if (p < 1.0) {
// // continue; // Skip if the percentage is less than 1%
// // }
// output += ", " + name + ": " + std::to_string(t) + " us(" + std::to_string(size_t(round(p))) + "%)";
// }
// 反向遍历
for (auto it = time_map.rbegin(); it != time_map.rend(); ++it) {
const std::string& name = it->first;
long t = it->second;
double p = 100.0 * t / duration.count();
// if (p < 1.0) {
// continue; // Skip if the percentage is less than 1%
// }
output += ", " + name + ": " + std::to_string(t) + " us(" + std::to_string(size_t(round(p))) + "%)";
}
printf("%s\n", output.c_str());
}
};
struct TaskCounter {
std::vector<size_t> fold = {}, card = {};
TaskCounter(std::initializer_list<size_t> i) {
card.push_back(1);
for (auto j : i) {
push_back(j);
}
}
void push_back(size_t i) {
fold.push_back(i);
for (auto& c : card) {
c *= i;
}
card.push_back(1);
}
void push_back(std::vector<size_t> i) {
for (auto j : i) {
push_back(j);
}
}
size_t count() { return card[0]; }
size_t at(size_t id, size_t which) { return id % card.at(which) / card.at(which + 1); }
};
struct GeneralConfig {
size_t vocab_size;
size_t hidden_size;
size_t num_experts_per_tok;
size_t n_routed_experts;
size_t n_shared_experts;
size_t max_qlen = 4096;
void* lm_heads_ptr;
ggml_type lm_heads_type;
void* norm_weights_ptr;
ggml_type norm_weights_type;
void* token_embd_ptr;
ggml_type token_embd_type;
WorkerPool* pool = nullptr;
GeneralConfig() {}
};
struct GeneralMLAConfig {
size_t hidden_size;
size_t q_lora_rank;
size_t num_heads;
size_t nope_size;
size_t rope_size;
size_t kv_lora_rank;
int layer_idx = 0;
WorkerPool* pool = nullptr;
size_t token_count_in_page = 256; // token count in a page
size_t max_qlen = 1024;
size_t max_kvlen = 4096;
// rope
size_t max_position_embeddings;
double rope_scaling_factor = 1.0;
double rope_theta = 10000.0;
double rope_scaling_beta_fast;
double rope_scaling_beta_slow;
double rope_scaling_mscale;
double rope_scaling_mscale_all_dim;
double rope_scaling_original_max_position_embeddings;
void* q_a_proj;
void* q_a_norm = nullptr;
void* q_b_proj;
void* kv_a_proj_with_mqa;
void* kv_a_norm = nullptr;
void* kv_b_proj;
void* o_proj;
// for llamafile
ggml_type q_a_proj_type;
ggml_type q_a_norm_type;
ggml_type q_b_proj_type;
ggml_type kv_a_proj_with_mqa_type;
ggml_type kv_a_norm_type;
ggml_type kv_b_proj_type;
ggml_type w_o_type;
ggml_type input_type = GGML_TYPE_F32;
ggml_type output_type = GGML_TYPE_F32;
size_t m_block = 4;
size_t n_block = 4;
// for kvcache
size_t page_count = 200; // page count for kv cache
GeneralMLAConfig() {}
GeneralMLAConfig(size_t hidden_size, size_t q_lora_rank, size_t kv_lora_rank, size_t num_heads, size_t nope_size,
size_t rope_size)
: hidden_size(hidden_size),
q_lora_rank(q_lora_rank),
kv_lora_rank(kv_lora_rank),
num_heads(num_heads),
nope_size(nope_size),
rope_size(rope_size) {}
};
struct QuantConfig {
std::string quant_method = "";
int bits = 0;
int group_size = 0;
bool zero_point = false;
bool per_channel = false; // Per-channel quantization (GLM-4.7-FP8 style)
};
struct GeneralMOEConfig {
// Basic Config
int expert_num;
int num_experts_per_tok;
int hidden_size;
int intermediate_size;
int layer_idx = 0;
WorkerPool* pool = nullptr;
// SGLang offload
int num_gpu_experts = 0; // Computed from gpu_experts_mask
uint8_t* gpu_experts_mask = nullptr; // Bool mask: true = expert on GPU
void* physical_to_logical_map = nullptr;
// Compute num_gpu_experts from gpu_experts_mask
void compute_num_gpu_experts() {
num_gpu_experts = 0;
if (gpu_experts_mask) {
for (int i = 0; i < expert_num; i++) {
if (gpu_experts_mask[i]) num_gpu_experts++;
}
}
}
// Check if expert should be skipped (invalid, out of range, or on GPU)
inline bool should_skip_expert(int64_t expert_id) const {
return expert_id < 0 || expert_id >= expert_num || (gpu_experts_mask && gpu_experts_mask[expert_id]);
}
void* gate_proj = nullptr;
void* up_proj = nullptr;
void* down_proj = nullptr;
void* gate_scale = nullptr;
void* up_scale = nullptr;
void* down_scale = nullptr;
void* gate_zero = nullptr;
void* up_zero = nullptr;
void* down_zero = nullptr;
QuantConfig quant_config;
// for amx
int max_len = 0;
std::vector<std::vector<void*>> gate_projs;
std::vector<std::vector<void*>> up_projs;
std::vector<std::vector<void*>> down_projs;
std::vector<std::vector<void*>> gate_scales;
std::vector<std::vector<void*>> up_scales;
std::vector<std::vector<void*>> down_scales;
std::vector<std::vector<void*>> gate_zeros;
std::vector<std::vector<void*>> up_zeros;
std::vector<std::vector<void*>> down_zeros;
// Pre-quantized backward weights (transposed, in BufferB format) [tp_count][expert_id]
std::vector<std::vector<void*>> gate_bwd_projs;
std::vector<std::vector<void*>> up_bwd_projs;
std::vector<std::vector<void*>> down_bwd_projs;
std::vector<std::vector<void*>> gate_bwd_scales;
std::vector<std::vector<void*>> up_bwd_scales;
std::vector<std::vector<void*>> down_bwd_scales;
std::string path;
bool save = false;
bool load = false;
bool share_backward_bb = false;
bool share_cache_pool = false;
// for llamafile
int m_block = 4;
int group_min_len = 0;
int group_max_len = 0;
int gate_type;
int up_type;
int down_type;
int hidden_type;
int max_cache_depth = 1;
GeneralMOEConfig() {}
GeneralMOEConfig(int expert_num, int routed_expert_num, int hidden_size, int intermediate_size)
: expert_num(expert_num),
num_experts_per_tok(routed_expert_num),
hidden_size(hidden_size),
intermediate_size(intermediate_size) {}
int max_possible_qlen() { return std::max(max_len, group_max_len); }
};
// SFT (Supervised Fine-Tuning) configuration for MoE with LoRA
struct MOESFTConfig : public GeneralMOEConfig {
// LoRA configuration
int lora_rank = 16;
float lora_alpha = 32.0f;
float lora_scaling() const { return lora_alpha / lora_rank; }
// LoRA weight pointers (directly pointing to Python tensor memory, zero-copy)
// Layout: [expert_num, lora_rank, in_dim] for A, [expert_num, out_dim, lora_rank] for B
void* gate_lora_a = nullptr; // [expert_num, lora_rank, hidden_size]
void* gate_lora_b = nullptr; // [expert_num, intermediate_size, lora_rank]
void* up_lora_a = nullptr; // [expert_num, lora_rank, hidden_size]
void* up_lora_b = nullptr; // [expert_num, intermediate_size, lora_rank]
void* down_lora_a = nullptr; // [expert_num, lora_rank, intermediate_size]
void* down_lora_b = nullptr; // [expert_num, hidden_size, lora_rank]
MOESFTConfig() : GeneralMOEConfig() {}
MOESFTConfig(int expert_num, int routed_expert_num, int hidden_size, int intermediate_size)
: GeneralMOEConfig(expert_num, routed_expert_num, hidden_size, intermediate_size) {}
// Conversion constructor from GeneralMOEConfig (for MOE_TP_PART concept satisfaction)
explicit MOESFTConfig(const GeneralMOEConfig& base) : GeneralMOEConfig(base) {
// LoRA fields use default values (already initialized in struct definition)
}
};
struct GeneralGateConfig {
size_t hidden_size;
size_t num_experts_per_tok;
size_t n_routed_experts;
size_t n_group;
size_t topk_group;
bool norm_topk_prob = true;
float routed_scaling_factor = 2.5f;
std::string scoring_func = "sigmoid";
std::string topk_method = "noaux_tc";
int layer_idx = 0;
WorkerPool* pool = nullptr;
void* weight = nullptr;
ggml_type weight_type;
void* e_score_correction_bias = nullptr;
ggml_type e_score_correction_bias_type;
size_t max_seqlen = 25600;
GeneralGateConfig() = default;
GeneralGateConfig(int hidden_size, int num_experts_per_tok, int n_routed_experts, int n_group, int topk_group)
: hidden_size(hidden_size),
num_experts_per_tok(num_experts_per_tok),
n_routed_experts(n_routed_experts),
n_group(n_group),
topk_group(topk_group) {}
};
class MLA_Interface {
public:
virtual void forward(std::vector<int> qlens, std::vector<std::vector<int>> page_tables, std::vector<int> kv_lens,
const void* input, void* output) = 0;
};
class MoE_Interface {
public:
virtual void forward(int qlen, int k, const int64_t* expert_ids, const float* weights, const void* input,
void* output, bool incremental = false) = 0;
};
inline void init_ggml() {
static bool inited = false;
if (inited) {
return;
}
struct ggml_init_params params = {
0,
NULL,
true,
};
auto ctx_eval = ggml_init(params);
if (!ctx_eval) {
throw std::runtime_error("Failed to create ggml context");
}
inited = true;
}
template <typename A, typename B>
void convert_or_copy(A* dst, const B* src, size_t count) {
if constexpr (std::is_same_v<A, B>) {
// printf("Direct copy\n");
memcpy(dst, src, sizeof(A) * count);
} else {
if constexpr (std::is_same_v<A, float>) {
if constexpr (std::is_same_v<B, ggml_bf16_t>) {
// printf("Converting ggml_bf16_t to float\n");
ggml_bf16_to_fp32_row(src, dst, count);
} else if constexpr (std::is_same_v<B, ggml_fp16_t>) {
ggml_fp16_to_fp32_row(src, dst, count);
} else {
throw std::runtime_error("Unsupported conversion");
}
} else if constexpr (std::is_same_v<A, ggml_bf16_t>) {
if constexpr (std::is_same_v<B, float>) {
// printf("Converting float to ggml_bf16_t\n");
ggml_fp32_to_bf16_row(src, dst, count);
} else {
throw std::runtime_error("Unsupported conversion");
}
}
else {
throw std::runtime_error("Unsupported conversion");
}
}
}
template <typename A>
void convert_or_copy(A* dst, void* src, ggml_type type, size_t count) {
switch (type) {
case GGML_TYPE_BF16: {
auto src_bf16 = (ggml_bf16_t*)src;
convert_or_copy(dst, src_bf16, count);
break;
}
case GGML_TYPE_F16: {
#if defined(__aarch64__) && defined(CPU_USE_KML)
auto src_fp16 = (float16_t*)src;
convert_or_copy(dst, src_fp16, count);
#else
throw std::runtime_error("GGML_TYPE_F16 is not supported on this platform");
#endif
break;
}
case GGML_TYPE_F32: {
auto src_f32 = (float*)src;
convert_or_copy(dst, src_f32, count);
break;
}
default:
throw std::runtime_error("Unsupported type for conversion");
}
}
template <typename A>
void check_numerics(A* data, size_t count) {
for (size_t i = 0; i < count; i++) {
if (std::isnan(data[i]) || std::isinf(data[i])) {
printf("Numerics check failed at index %zu: value = %f\n", i, data[i]);
throw std::runtime_error("Numerics check failed");
}
}
printf("Numerics check passed for %zu elements.\n", count);
}
inline void debug_bf16(ggml_bf16_t* x) {
for (int i = 0; i < 10; i++) {
printf("%f ", ggml_bf16_to_fp32(x[i]));
}
printf("\n");
}
inline void debug_f32(float* x) {
for (int i = 0; i < 10; i++) {
printf("%f ", x[i]);
}
printf("\n");
}
inline void debug_f32(float* x, size_t count) {
if (count < 10) {
for (size_t i = 0; i < count; i++) {
printf("%f ", x[i]);
}
} else {
for (size_t i = 0; i < 3; i++) {
printf("%f ", x[i]);
}
printf("...");
for (size_t i = count - 3; i < count; i++) {
printf("%f ", x[i]);
}
printf("\n");
}
}
#endif