#include "engine/framework/core/backend.h" #include "engine/framework/modules/structural_modules.h" #include "ggml.h" #include #include #include #include #include #include #include #include #include namespace { namespace core = engine::core; namespace modules = engine::modules; struct BackendDeleter { void operator()(ggml_backend * backend) const noexcept { if (backend != nullptr) { ggml_backend_free(backend); } } }; struct GgmlContextDeleter { void operator()(ggml_context * ctx) const noexcept { if (ctx != nullptr) { ggml_free(ctx); } } }; void require(bool condition, const std::string & message) { if (!condition) { throw std::runtime_error(message); } } core::TensorValue slice_bhtd_time( core::ModuleBuildContext & ctx, const core::TensorValue & input, int64_t start, int64_t length) { return modules::SliceModule({2, start, length}).build(ctx, input); } core::TensorValue set_ring_cache_bhtd_rows( core::ModuleBuildContext & ctx, const core::TensorValue & cache, const core::TensorValue & row, const core::TensorValue & row_indexes) { require(cache.shape.rank == 4 && row.shape.rank == 4, "ring cache row update expects rank-4 tensors"); require(cache.shape.dims[0] == 1 && row.shape.dims[0] == 1, "ring cache row update expects batch size 1"); require(cache.shape.dims[1] == row.shape.dims[1], "ring cache row update head count mismatch"); require(row.shape.dims[2] == 1, "ring cache row update expects one time step"); require(cache.shape.dims[3] == row.shape.dims[3], "ring cache row update head dim mismatch"); require(row_indexes.shape.rank == 1 && row_indexes.shape.dims[0] == cache.shape.dims[1], "ring cache row update expects one index per head"); const int64_t heads = cache.shape.dims[1]; const int64_t steps = cache.shape.dims[2]; const int64_t head_dim = cache.shape.dims[3]; auto flat_cache = core::reshape_tensor(ctx, cache, core::TensorShape::from_dims({heads * steps, head_dim})); auto flat_row = core::reshape_tensor( ctx, core::ensure_backend_addressable_layout(ctx, row), core::TensorShape::from_dims({heads, head_dim})); auto * updated_raw = ggml_set_rows(ctx.ggml, flat_cache.tensor, flat_row.tensor, row_indexes.tensor); updated_raw->src[2] = cache.tensor; auto updated = core::wrap_tensor(updated_raw, flat_cache.shape, cache.type); return core::reshape_tensor(ctx, updated, cache.shape); } struct RingOracle { int64_t heads = 0; int64_t head_dim = 0; int64_t capacity = 0; int64_t end_offset = 0; std::vector key_cache; std::vector value_cache; RingOracle(int64_t input_heads, int64_t input_head_dim, int64_t input_capacity) : heads(input_heads), head_dim(input_head_dim), capacity(input_capacity), key_cache(static_cast(input_capacity * input_heads * input_head_dim), 0.0F), value_cache(static_cast(input_capacity * input_heads * input_head_dim), 0.0F) {} std::vector slots(int64_t frames) const { std::vector out(static_cast(frames)); for (int64_t frame = 0; frame < frames; ++frame) { out[static_cast(frame)] = static_cast((end_offset + frame) % capacity); } return out; } void complete(const std::vector & key_chunk, const std::vector & value_chunk, int64_t frames) { require(static_cast(key_chunk.size()) == frames * heads * head_dim, "oracle key chunk size mismatch"); require(static_cast(value_chunk.size()) == frames * heads * head_dim, "oracle value chunk size mismatch"); for (int64_t frame = 0; frame < frames; ++frame) { const int64_t slot = (end_offset + frame) % capacity; for (int64_t head = 0; head < heads; ++head) { const size_t src = static_cast((head * frames + frame) * head_dim); const size_t dst = static_cast((head * capacity + slot) * head_dim); std::copy( key_chunk.begin() + static_cast(src), key_chunk.begin() + static_cast(src + head_dim), key_cache.begin() + static_cast(dst)); std::copy( value_chunk.begin() + static_cast(src), value_chunk.begin() + static_cast(src + head_dim), value_cache.begin() + static_cast(dst)); } } end_offset += frames; } std::vector positions() const { std::vector out(static_cast(capacity), -1); const int64_t end_index = end_offset % capacity; for (int64_t index = 0; index < capacity; ++index) { const bool invalid = index >= end_offset; const int64_t delta = index - end_index; const int64_t pos = delta < 0 ? end_offset + delta : end_offset + delta - capacity; out[static_cast(index)] = invalid ? -1 : pos; } return out; } std::vector attention_mask(int64_t query_start, int64_t query_frames) const { const auto pos = positions(); std::vector out( static_cast(query_frames * capacity), -std::numeric_limits::infinity()); for (int64_t query = 0; query < query_frames; ++query) { const int64_t pos_q = query_start + query; for (int64_t slot = 0; slot < capacity; ++slot) { const int64_t pos_k = pos[static_cast(slot)]; const int64_t delta = pos_q - pos_k; if (pos_k >= 0 && delta >= 0 && delta < capacity) { out[static_cast(query * capacity + slot)] = 0.0F; } } } return out; } }; std::vector make_chunk(int64_t start, int64_t frames, int64_t heads, int64_t head_dim, float scale) { std::vector out(static_cast(frames * heads * head_dim)); for (int64_t frame = 0; frame < frames; ++frame) { const int64_t absolute_frame = start + frame; for (int64_t head = 0; head < heads; ++head) { for (int64_t dim = 0; dim < head_dim; ++dim) { const float phase = static_cast( absolute_frame * 0.013 + head * 0.071 + dim * 0.003); out[static_cast((head * frames + frame) * head_dim + dim)] = std::sin(phase) * scale + std::cos(phase * 0.37F) * (scale * 0.25F); } } } return out; } std::vector bhtd_to_bthd(const std::vector & input, int64_t frames, int64_t heads, int64_t head_dim) { std::vector output(input.size(), 0.0F); for (int64_t frame = 0; frame < frames; ++frame) { for (int64_t head = 0; head < heads; ++head) { const size_t src = static_cast((head * frames + frame) * head_dim); const size_t dst = static_cast((frame * heads + head) * head_dim); std::copy( input.begin() + static_cast(src), input.begin() + static_cast(src + head_dim), output.begin() + static_cast(dst)); } } return output; } void assert_close( const std::vector & actual, const std::vector & expected, const std::string & label, float tolerance = 1.0e-6F) { require(actual.size() == expected.size(), label + " size mismatch"); float max_diff = 0.0F; size_t max_index = 0; for (size_t index = 0; index < actual.size(); ++index) { const float diff = std::fabs(actual[index] - expected[index]); if (diff > max_diff) { max_diff = diff; max_index = index; } } if (max_diff > tolerance) { throw std::runtime_error( label + " mismatch max_abs_diff=" + std::to_string(max_diff) + " index=" + std::to_string(max_index) + " actual=" + std::to_string(actual[max_index]) + " expected=" + std::to_string(expected[max_index])); } } void assert_equal( const std::vector & actual, const std::vector & expected, const std::string & label) { require(actual.size() == expected.size(), label + " size mismatch"); for (size_t index = 0; index < actual.size(); ++index) { if (actual[index] != expected[index]) { throw std::runtime_error( label + " mismatch index=" + std::to_string(index) + " actual=" + std::to_string(actual[index]) + " expected=" + std::to_string(expected[index])); } } } class LocalRingKVCachePrimitive { public: LocalRingKVCachePrimitive( ggml_backend_t backend, core::BackendType backend_type, int threads, int64_t heads, int64_t head_dim, int64_t capacity, int64_t chunk_frames) : backend_(backend), heads_(heads), head_dim_(head_dim), capacity_(capacity), chunk_frames_(chunk_frames), key_cache_host_(static_cast(capacity * heads * head_dim), 0.0F), value_cache_host_(static_cast(capacity * heads * head_dim), 0.0F) { ctx_.reset(ggml_init({32ull * 1024ull * 1024ull, nullptr, true})); require(ctx_ != nullptr, "ring kv cache test ctx init failed"); core::ModuleBuildContext build{ctx_.get(), "personaplex.ring_kv_cache_test", backend_type}; key_cache_ = core::make_tensor(build, GGML_TYPE_F32, core::TensorShape::from_dims({1, heads, capacity, head_dim})); value_cache_ = core::make_tensor(build, GGML_TYPE_F32, key_cache_.shape); key_chunk_bthd_ = core::make_tensor(build, GGML_TYPE_F32, core::TensorShape::from_dims({1, chunk_frames, heads, head_dim})); value_chunk_bthd_ = core::make_tensor(build, GGML_TYPE_F32, key_chunk_bthd_.shape); key_chunk_ = core::ensure_backend_addressable_layout( build, modules::TransposeModule({{0, 2, 1, 3}, 4}).build(build, key_chunk_bthd_)); value_chunk_ = core::ensure_backend_addressable_layout( build, modules::TransposeModule({{0, 2, 1, 3}, 4}).build(build, value_chunk_bthd_)); for (int64_t frame = 0; frame < chunk_frames; ++frame) { slot_inputs_.push_back(core::make_tensor(build, GGML_TYPE_I32, core::TensorShape::from_dims({heads}))); } auto updated_key = key_cache_; auto updated_value = value_cache_; for (int64_t frame = 0; frame < chunk_frames; ++frame) { auto key_row = slice_bhtd_time(build, key_chunk_, frame, 1); auto value_row = slice_bhtd_time(build, value_chunk_, frame, 1); updated_key = set_ring_cache_bhtd_rows(build, updated_key, key_row, slot_inputs_[static_cast(frame)]); updated_value = set_ring_cache_bhtd_rows(build, updated_value, value_row, slot_inputs_[static_cast(frame)]); } updated_key_ = updated_key; updated_value_ = updated_value; core::set_backend_threads(backend_, threads); graph_ = ggml_new_graph_custom(ctx_.get(), 32768, false); ggml_build_forward_expand(graph_, updated_key_.tensor); ggml_build_forward_expand(graph_, updated_value_.tensor); gallocr_ = ggml_gallocr_new(ggml_backend_get_default_buffer_type(backend_)); require(gallocr_ != nullptr && ggml_gallocr_reserve(gallocr_, graph_) && ggml_gallocr_alloc_graph(gallocr_, graph_), "ring kv cache test graph allocation failed"); core::write_tensor_f32(key_cache_, key_cache_host_); core::write_tensor_f32(value_cache_, value_cache_host_); } ~LocalRingKVCachePrimitive() { if (gallocr_ != nullptr) { ggml_gallocr_free(gallocr_); } } void reset() { end_offset_ = 0; std::fill(key_cache_host_.begin(), key_cache_host_.end(), 0.0F); std::fill(value_cache_host_.begin(), value_cache_host_.end(), 0.0F); core::write_tensor_f32(key_cache_, key_cache_host_); core::write_tensor_f32(value_cache_, value_cache_host_); } void complete(const std::vector & key_chunk, const std::vector & value_chunk) { require(static_cast(key_chunk.size()) == chunk_frames_ * heads_ * head_dim_, "key chunk size mismatch"); require(static_cast(value_chunk.size()) == chunk_frames_ * heads_ * head_dim_, "value chunk size mismatch"); core::write_tensor_f32(key_chunk_bthd_, bhtd_to_bthd(key_chunk, chunk_frames_, heads_, head_dim_)); core::write_tensor_f32(value_chunk_bthd_, bhtd_to_bthd(value_chunk, chunk_frames_, heads_, head_dim_)); for (int64_t frame = 0; frame < chunk_frames_; ++frame) { const auto indexes = row_indexes(frame); core::write_tensor_i32(slot_inputs_[static_cast(frame)], indexes); } require(core::compute_backend_graph(backend_, graph_) == GGML_STATUS_SUCCESS, "ring kv cache graph failed"); key_cache_host_ = core::read_tensor_f32(updated_key_.tensor); value_cache_host_ = core::read_tensor_f32(updated_value_.tensor); ggml_backend_tensor_copy(updated_key_.tensor, key_cache_.tensor); ggml_backend_tensor_copy(updated_value_.tensor, value_cache_.tensor); end_offset_ += chunk_frames_; } std::vector positions() const { std::vector out(static_cast(capacity_), -1); const int64_t end_index = end_offset_ % capacity_; for (int64_t index = 0; index < capacity_; ++index) { const bool invalid = index >= end_offset_; const int64_t delta = index - end_index; const int64_t pos = delta < 0 ? end_offset_ + delta : end_offset_ + delta - capacity_; out[static_cast(index)] = invalid ? -1 : pos; } return out; } std::vector attention_mask(int64_t query_start) const { const auto pos = positions(); std::vector out( static_cast(chunk_frames_ * capacity_), -std::numeric_limits::infinity()); for (int64_t query = 0; query < chunk_frames_; ++query) { const int64_t pos_q = query_start + query; for (int64_t slot = 0; slot < capacity_; ++slot) { const int64_t pos_k = pos[static_cast(slot)]; const int64_t delta = pos_q - pos_k; if (pos_k >= 0 && delta >= 0 && delta < capacity_) { out[static_cast(query * capacity_ + slot)] = 0.0F; } } } return out; } const std::vector & key_cache() const noexcept { return key_cache_host_; } const std::vector & value_cache() const noexcept { return value_cache_host_; } private: std::vector row_indexes(int64_t frame) const { std::vector out(static_cast(heads_)); const int64_t slot = (end_offset_ + frame) % capacity_; for (int64_t head = 0; head < heads_; ++head) { out[static_cast(head)] = static_cast(head * capacity_ + slot); } return out; } ggml_backend_t backend_ = nullptr; int64_t heads_ = 0; int64_t head_dim_ = 0; int64_t capacity_ = 0; int64_t chunk_frames_ = 0; int64_t end_offset_ = 0; std::unique_ptr ctx_; ggml_cgraph * graph_ = nullptr; ggml_gallocr_t gallocr_ = nullptr; core::TensorValue key_cache_; core::TensorValue value_cache_; core::TensorValue key_chunk_bthd_; core::TensorValue value_chunk_bthd_; core::TensorValue key_chunk_; core::TensorValue value_chunk_; std::vector slot_inputs_; core::TensorValue updated_key_; core::TensorValue updated_value_; std::vector key_cache_host_; std::vector value_cache_host_; }; void run_case(LocalRingKVCachePrimitive & primitive, RingOracle & oracle, int64_t absolute_start, int64_t chunk_frames) { const auto key = make_chunk(absolute_start, chunk_frames, oracle.heads, oracle.head_dim, 0.73F); const auto value = make_chunk(absolute_start, chunk_frames, oracle.heads, oracle.head_dim, 0.41F); const int64_t query_start = oracle.end_offset; primitive.complete(key, value); oracle.complete(key, value, chunk_frames); assert_close(primitive.key_cache(), oracle.key_cache, "ring key cache"); assert_close(primitive.value_cache(), oracle.value_cache, "ring value cache"); assert_equal(primitive.positions(), oracle.positions(), "ring positions"); assert_close(primitive.attention_mask(query_start), oracle.attention_mask(query_start, chunk_frames), "ring attention mask"); } } // namespace int main() { try { constexpr int64_t kHeads = 8; constexpr int64_t kHeadDim = 64; constexpr int64_t kCapacity = 250; constexpr int64_t kChunkFrames = 2; constexpr int kThreads = 8; core::BackendConfig backend_config; backend_config.type = core::BackendType::Cuda; backend_config.device = 0; backend_config.threads = kThreads; std::unique_ptr backend(core::init_backend(backend_config)); require(backend != nullptr, "failed to initialize CUDA backend"); LocalRingKVCachePrimitive primitive( backend.get(), core::BackendType::Cuda, kThreads, kHeads, kHeadDim, kCapacity, kChunkFrames); RingOracle oracle(kHeads, kHeadDim, kCapacity); run_case(primitive, oracle, 0, kChunkFrames); for (int64_t step = 1; step < 124; ++step) { run_case(primitive, oracle, step * kChunkFrames, kChunkFrames); } run_case(primitive, oracle, 248, kChunkFrames); run_case(primitive, oracle, 250, kChunkFrames); run_case(primitive, oracle, 252, kChunkFrames); primitive.reset(); oracle = RingOracle(kHeads, kHeadDim, kCapacity); run_case(primitive, oracle, 1000, kChunkFrames); std::cout << "personaplex_ring_kv_cache_parity passed" << " heads=" << kHeads << " head_dim=" << kHeadDim << " capacity=" << kCapacity << " chunk_frames=" << kChunkFrames << "\n"; return 0; } catch (const std::exception & ex) { std::cerr << "personaplex_ring_kv_cache_parity failed: " << ex.what() << "\n"; return 1; } }