[PR #1251] [CLOSED] [Store][Feature] Update KV event publish system #2107

New issue

Closed

opened 2026-10-29 21:32:50 +00:00 by bsakaguchi · 0 comments

bsakaguchi commented

2026-10-29 21:32:50 +00:00

Owner

📋 Pull Request Information

Original PR: https://github.com/kvcache-ai/Mooncake/pull/1251
Author: @Asher-XunZhang
Created: 12/22/2025
Status: ❌ Closed

Base: main ← Head: zmq_event_publisher

📝 Commits (7)

b0289e0 Update KV event publisher
0303c1a Fix some CI review suggestions
101c2af Merge remote-tracking branch 'upstream/main' into zmq_event_publisher
281ac83 Merge remote-tracking branch 'upstream/main' into zmq_event_publisher
600a29f Merge remote-tracking branch 'upstream/main' into zmq_event_publisher
cff97a9 Merge remote-tracking branch 'upstream/main' into zmq_event_publisher
dc597bd Merge remote-tracking branch 'upstream/main' into zmq_event_publisher

📊 Changes

31 files changed (+5004 additions, -46 deletions)

View changed files

📝 dependencies.sh (+140 -0)
📝 mooncake-integration/store/store_py.cpp (+29 -2)
📝 mooncake-store/include/client_service.h (+12 -3)
📝 mooncake-store/include/dummy_client.h (+2 -1)
➕ mooncake-store/include/kv_event/kv_event.hpp (+214 -0)
➕ mooncake-store/include/kv_event/kv_event_consumer.h (+220 -0)
➕ mooncake-store/include/kv_event/kv_event_producer.h (+163 -0)
➕ mooncake-store/include/kv_event/kv_event_publisher_config.h (+47 -0)
➕ mooncake-store/include/kv_event/kv_event_system.h (+161 -0)
➕ mooncake-store/include/kv_event/kv_event_types.h (+50 -0)
➕ mooncake-store/include/kv_event_publisher.h (+13 -0)
📝 mooncake-store/include/master_client.h (+7 -4)
📝 mooncake-store/include/master_config.h (+77 -0)
📝 mooncake-store/include/master_service.h (+39 -1)
📝 mooncake-store/include/pyclient.h (+3 -1)
📝 mooncake-store/include/real_client.h (+4 -2)
📝 mooncake-store/include/rpc_service.h (+6 -4)
📝 mooncake-store/include/thread_pool.h (+7 -1)
➕ mooncake-store/include/thread_safe_queue.h (+564 -0)
📝 mooncake-store/src/CMakeLists.txt (+7 -1)

...and 11 more files

📄 Description

Description

motivation: [RFC]: Mooncake-Conductor: Design and Implementation of a Global Scheduler Module for KV-Cache-Centric Disaggregated Architecture #977

Add a kv event publish system: it consists of events and an asynchronous sender, a kv event publisher based on ZMQ. Co-work with @Liziqi-77 @yejj710

PR Overview

Summary of the KV Event System File Structure

classDiagram
    %% Event Class Hierarchy
    class KVCacheEvent {
        <<interface>>
        +pack(msgpack::packer~msgpack::sbuffer~) void
        +type_tag() string_view
    }
    
    KVCacheEvent <|-- BlockStoreEvent
    KVCacheEvent <|-- BlockUpdateEvent
    KVCacheEvent <|-- RemoveAllEvent
    
    %% Core Components
    class KVEventProducer {
        -shared_ptr~KVEventQueue~ event_queue_
        -unique_ptr~ThreadPool~ enqueue_pool_ : enqueue worker(1)
        
        +publish~Event, Args...~(args) future~bool~
        +publish_event_async(KVEventPtr) future~bool~
        +shutdown() void
    }
    
    class KVEventConsumer {
        -zmq::context_t context_
        -shared_ptr~KVEventQueue~ event_queue_
        -jthread publisher_thread_
        
        +KVEventConsumer(shared_ptr~KVEventQueue~, Config)
        +shutdown() void
        +is_running() bool
    }
    
    class ThreadSafeQueue~T~ {
        <<alias KVEventQueue>>
    }
    
    %% Facade Class
    class KVEventSystem {
        -shared_ptr~KVEventQueue~ event_queue_
        
        +KVEventSystem(const KVEventPublisherConfig&)
        +~KVEventSystem()
        +shutdown() void
        +is_running() bool
        +publish~Event, Args...~(args) future~bool~
    }
    
    %% Configuration and Event Classes
    class KVEventPublisherConfig {
    }
    class StoreEventInfo {
    }
    class EventBatch {
        +serialize() msgpack::sbuffer
    }
    
    %% Relationships
    KVEventSystem --> KVEventPublisherConfig : reads
    KVEventSystem --> KVEventProducer : contains
    KVEventSystem --> KVEventConsumer : contains
    KVEventSystem --> ThreadSafeQueue~KVEventPtr~ : contains
    
    KVEventProducer --> ThreadSafeQueue~KVEventPtr~ : enqueues
    KVEventConsumer --> ThreadSafeQueue~KVEventPtr~ : dequeues
    
    KVEventProducer --> KVCacheEvent : creates
    BlockStoreEvent --> StoreEventInfo : contains
    KVCacheEvent <.. EventBatch : aggregates
    EventBatch <.. KVEventConsumer : serialize & publish

mooncake-store/
- include/
  - kv_event/
    - kv_event.hpp
      - Purpose：Defines the base class and concrete implementations for all KV cache events
      - Classes and Structs：
        
        KVCacheEvent：Abstract base class for events, defines serialization interface
        
        BlockStoreEvent：Event for the first time KV cache is stored
        
        BlockUpdateEvent：Event for any replica update operations
        
        RemoveAllEvent：Event for clearing all cached data
        
        EventBatch：Event batch class for improving transmission efficiency
      - Details：Uses msgpack for serialization, supports event type identification, and includes StoreEventInfo metadata
    - kv_event_types.h
      - Purpose：Provides unified type definitions and thread-safe queue implementation
      - Classes and Structs：
        
        QueuedItem：Queue item struct, combines an event and a promise.
      - Details：Uses concepts for type checking, defines type aliases KVEventPtr and KVEventQueue, implements thread-safe queue based on ThreadSafeQueue
    - kv_event_producer.h
      - Purpose：Responsible for creating and publishing events
      - Classes and Structs：
        
        KVEventProducer：Event producer class, manages event creation and queueing
      - Details：Maintains a thread pool for asynchronous event queueing, implements a retry mechanism, tracks event types and statistics
    - kv_event_consumer.h
      - Purpose：Consumes events from the queue and publishes them via ZeroMQ
      - Classes and Structs：
        
        KVEventConsumer：Event consumer class, processes queued events and publishes them over the network
      - Details：Uses ZeroMQ for network communication, supports event batching, maintains a replay buffer, implements automatic port selection
    - kv_event_system.h
      - Purpose：Provides a simplified system interface, coordinating all components
      - Classes and Structs：
        
        KVEventSystem：Event system facade class, hides internal complexity
      - Details：Creates and configures all components, provides a unified publishing interface, aggregates statistics, supports graceful shutdown
    - kv_event_publisher_config.h
      - Purpose：Defines all configuration parameters for the system
      - Classes and Structs：
        
        KVEventPublisherConfig：System configuration struct
      - Details：Contains configuration items for networking, performance, batching, thread pool, etc., and offers a configuration validation method
  - kv_event_publisher.h
    - Purpose：Serves as the public API entry point for the KV event system, allowing external modules to include all necessary components at once
    - Details：Aggregates and includes the 6 core header files from the kv_event/ subdirectory
  - thread_safe_queue.h
    - Purpose：Provides a thread-safe queue template class, implements the producer-consumer pattern for multi-threaded environments, supports bounded capacity and graceful shutdown mechanisms, optimized for MPSC (Multiple Producers, Single Consumer) scenarios
    - Details：
      - ThreadSafeQueue<T>：Core thread-safe queue template class, provides blocking/timeout push/pop operations, batch processing, queue state query, and graceful shutdown functionality, supporting flexible type for template parameter T
- src/
  - kv_event_publisher.cpp
    - Purpose：Core implementation file for the KV event publisher, responsible for the concrete implementation of configuration validation, event queue management, asynchronous event production, ZeroMQ event publishing, and replay services for the KV event system

System Architecture Relationships

sequenceDiagram
    participant App as Application
    participant ES as KVEventSystem
    participant EP as KVEventProducer
    participant TSQ as ThreadSafeQueue
    participant ZMQ as KVEventConsumer
    participant Network as ZeroMQNetwork
    
    App->>ES: 1. publish<KVEvent>(...)
    ES->>EP: 2. publish<KVEvent>(...)
    EP->>TSQ: 3. push(event)<br>(create KVCacheEvent object)
    Note over TSQ: KVEvent Buffer
    ZMQ->>TSQ: 4. peek_batch()
    ZMQ->>ZMQ: 5. batch processing
    ZMQ->>Network: 6. ZeroMQ Publish
    Network-->>ZMQ: 7. Publish confirmation
    ZMQ-->>TSQ: 8. pop_batch()
    ES-->>App: 9. return future
    
    Note over App,Network: Asynchronous processing flow

Data Flow：
- Events are published via KVEventSystem::publish
- KVEventSystemdelegates to KVEventProducer to create and enqueue events
- Events are placed into the KVEventQueue
- KVEventConsumer consumes events from the queue, batches them, and publishes via ZeroMQ
Component Interaction：
- KVEventSystem acts as a facade class, coordinating all components
- Producers and consumers communicate via the KVEventQueue
- All components depend on the event types defined in kv_event.hpp
Implementation Features：
- Uses the same ZeroMQ implementation as the inference engine for high-performance network communication
- Uses the same cross-language msgpack as the inference engine for serializing kv-event information

Type of Change

Types
- Bug fix
- New feature
  - Transfer Engine
  - Mooncake Store
  - Mooncake EP
  - Integration
  - P2P Store
  - Python Wheel
- Breaking change
- CI/CD
- Documentation update
- Other

How Has This Been Tested?

I've added 2 new UTs: kv_event_publisher_test and thread_safe_queue_test

Checklist

I have performed a self-review of my own code.
I have updated the documentation.
I have added tests to prove my changes are effective.

_{🔄 This issue represents a GitHub Pull Request. It cannot be merged through Gitea due to API limitations.}

## 📋 Pull Request Information **Original PR:** https://github.com/kvcache-ai/Mooncake/pull/1251 **Author:** [@Asher-XunZhang](https://github.com/Asher-XunZhang) **Created:** 12/22/2025 **Status:** ❌ Closed **Base:** `main` ← **Head:** `zmq_event_publisher` --- ### 📝 Commits (7) - [`b0289e0`](https://github.com/kvcache-ai/Mooncake/commit/b0289e06e91f33b9cd5745873159fde0a533c77c) Update KV event publisher - [`0303c1a`](https://github.com/kvcache-ai/Mooncake/commit/0303c1abaf9edb74fbca940a02c6292066ae7fad) Fix some CI review suggestions - [`101c2af`](https://github.com/kvcache-ai/Mooncake/commit/101c2af83c1acc3126e30f9f03e97e04317d1aea) Merge remote-tracking branch 'upstream/main' into zmq_event_publisher - [`281ac83`](https://github.com/kvcache-ai/Mooncake/commit/281ac833f593e63d7b56a354c9cf4c85fd2e4c04) Merge remote-tracking branch 'upstream/main' into zmq_event_publisher - [`600a29f`](https://github.com/kvcache-ai/Mooncake/commit/600a29f7f94770f627b226e6cf2b93bd4563838a) Merge remote-tracking branch 'upstream/main' into zmq_event_publisher - [`cff97a9`](https://github.com/kvcache-ai/Mooncake/commit/cff97a9492b19b72bc3eb956b5be2dcb8b429c93) Merge remote-tracking branch 'upstream/main' into zmq_event_publisher - [`dc597bd`](https://github.com/kvcache-ai/Mooncake/commit/dc597bd56c8d32476c5d66fa6558d91efc13d7c5) Merge remote-tracking branch 'upstream/main' into zmq_event_publisher ### 📊 Changes **31 files changed** (+5004 additions, -46 deletions) <details> <summary>View changed files</summary> 📝 `dependencies.sh` (+140 -0) 📝 `mooncake-integration/store/store_py.cpp` (+29 -2) 📝 `mooncake-store/include/client_service.h` (+12 -3) 📝 `mooncake-store/include/dummy_client.h` (+2 -1) ➕ `mooncake-store/include/kv_event/kv_event.hpp` (+214 -0) ➕ `mooncake-store/include/kv_event/kv_event_consumer.h` (+220 -0) ➕ `mooncake-store/include/kv_event/kv_event_producer.h` (+163 -0) ➕ `mooncake-store/include/kv_event/kv_event_publisher_config.h` (+47 -0) ➕ `mooncake-store/include/kv_event/kv_event_system.h` (+161 -0) ➕ `mooncake-store/include/kv_event/kv_event_types.h` (+50 -0) ➕ `mooncake-store/include/kv_event_publisher.h` (+13 -0) 📝 `mooncake-store/include/master_client.h` (+7 -4) 📝 `mooncake-store/include/master_config.h` (+77 -0) 📝 `mooncake-store/include/master_service.h` (+39 -1) 📝 `mooncake-store/include/pyclient.h` (+3 -1) 📝 `mooncake-store/include/real_client.h` (+4 -2) 📝 `mooncake-store/include/rpc_service.h` (+6 -4) 📝 `mooncake-store/include/thread_pool.h` (+7 -1) ➕ `mooncake-store/include/thread_safe_queue.h` (+564 -0) 📝 `mooncake-store/src/CMakeLists.txt` (+7 -1) _...and 11 more files_ </details> ### 📄 Description ## Description motivation: [[RFC]: Mooncake-Conductor: Design and Implementation of a Global Scheduler Module for KV-Cache-Centric Disaggregated Architecture #977](https://github.com/kvcache-ai/Mooncake/issues/977) Add a kv event publish system: it consists of events and an asynchronous sender, a kv event publisher based on ZMQ. Co-work with @Liziqi-77 @yejj710 ### PR Overview #### Summary of the KV Event System File Structure ```mermaid classDiagram %% Event Class Hierarchy class KVCacheEvent { <<interface>> +pack(msgpack::packer~msgpack::sbuffer~) void +type_tag() string_view } KVCacheEvent <|-- BlockStoreEvent KVCacheEvent <|-- BlockUpdateEvent KVCacheEvent <|-- RemoveAllEvent %% Core Components class KVEventProducer { -shared_ptr~KVEventQueue~ event_queue_ -unique_ptr~ThreadPool~ enqueue_pool_ : enqueue worker(1) +publish~Event, Args...~(args) future~bool~ +publish_event_async(KVEventPtr) future~bool~ +shutdown() void } class KVEventConsumer { -zmq::context_t context_ -shared_ptr~KVEventQueue~ event_queue_ -jthread publisher_thread_ +KVEventConsumer(shared_ptr~KVEventQueue~, Config) +shutdown() void +is_running() bool } class ThreadSafeQueue~T~ { <<alias KVEventQueue>> } %% Facade Class class KVEventSystem { -shared_ptr~KVEventQueue~ event_queue_ +KVEventSystem(const KVEventPublisherConfig&) +~KVEventSystem() +shutdown() void +is_running() bool +publish~Event, Args...~(args) future~bool~ } %% Configuration and Event Classes class KVEventPublisherConfig { } class StoreEventInfo { } class EventBatch { +serialize() msgpack::sbuffer } %% Relationships KVEventSystem --> KVEventPublisherConfig : reads KVEventSystem --> KVEventProducer : contains KVEventSystem --> KVEventConsumer : contains KVEventSystem --> ThreadSafeQueue~KVEventPtr~ : contains KVEventProducer --> ThreadSafeQueue~KVEventPtr~ : enqueues KVEventConsumer --> ThreadSafeQueue~KVEventPtr~ : dequeues KVEventProducer --> KVCacheEvent : creates BlockStoreEvent --> StoreEventInfo : contains KVCacheEvent <.. EventBatch : aggregates EventBatch <.. KVEventConsumer : serialize & publish ``` - mooncake-store/ - `include/` - `kv_event/` - `kv_event.hpp` - **Purpose**：Defines the base class and concrete implementations for all KV cache events - **Classes and Structs**： - `KVCacheEvent`：Abstract base class for events, defines serialization interface - `BlockStoreEvent`：Event for the first time KV cache is stored - `BlockUpdateEvent`：Event for any replica update operations - `RemoveAllEvent`：Event for clearing all cached data - `EventBatch`：Event batch class for improving transmission efficiency - **Details**：Uses msgpack for serialization, supports event type identification, and includes `StoreEventInfo` metadata - `kv_event_types.h` - **Purpose**：Provides unified type definitions and thread-safe queue implementation - **Classes and Structs**： - `QueuedItem`：Queue item struct, combines an event and a promise. - **Details**：Uses concepts for type checking, defines type aliases `KVEventPtr` and `KVEventQueue`, implements thread-safe queue based on `ThreadSafeQueue` - `kv_event_producer.h` - **Purpose**：Responsible for creating and publishing events - **Classes and Structs**： - `KVEventProducer`：Event producer class, manages event creation and queueing - **Details**：Maintains a thread pool for asynchronous event queueing, implements a retry mechanism, tracks event types and statistics - `kv_event_consumer.h` - **Purpose**：Consumes events from the queue and publishes them via ZeroMQ - **Classes and Structs**： - `KVEventConsumer`：Event consumer class, processes queued events and publishes them over the network - **Details**：Uses ZeroMQ for network communication, supports event batching, maintains a replay buffer, implements automatic port selection - `kv_event_system.h` - **Purpose**：Provides a simplified system interface, coordinating all components - **Classes and Structs**： - `KVEventSystem`：Event system facade class, hides internal complexity - **Details**：Creates and configures all components, provides a unified publishing interface, aggregates statistics, supports graceful shutdown - `kv_event_publisher_config.h` - **Purpose**：Defines all configuration parameters for the system - **Classes and Structs**： - `KVEventPublisherConfig`：System configuration struct - **Details**：Contains configuration items for networking, performance, batching, thread pool, etc., and offers a configuration validation method - `kv_event_publisher.h` - **Purpose**：Serves as the public API entry point for the KV event system, allowing external modules to include all necessary components at once - **Details**：Aggregates and includes the 6 core header files from the `kv_event/` subdirectory - `thread_safe_queue.h` - **Purpose**：Provides a thread-safe queue template class, implements the producer-consumer pattern for multi-threaded environments, supports bounded capacity and graceful shutdown mechanisms, optimized for MPSC (Multiple Producers, Single Consumer) scenarios - **Details**： - `ThreadSafeQueue<T>`：Core thread-safe queue template class, provides blocking/timeout push/pop operations, batch processing, queue state query, and graceful shutdown functionality, supporting flexible type for template parameter T - `src/` - `kv_event_publisher.cpp` - **Purpose**：Core implementation file for the KV event publisher, responsible for the concrete implementation of configuration validation, event queue management, asynchronous event production, ZeroMQ event publishing, and replay services for the KV event system #### System Architecture Relationships ```mermaid sequenceDiagram participant App as Application participant ES as KVEventSystem participant EP as KVEventProducer participant TSQ as ThreadSafeQueue participant ZMQ as KVEventConsumer participant Network as ZeroMQNetwork App->>ES: 1. publish<KVEvent>(...) ES->>EP: 2. publish<KVEvent>(...) EP->>TSQ: 3. push(event)<br>(create KVCacheEvent object) Note over TSQ: KVEvent Buffer ZMQ->>TSQ: 4. peek_batch() ZMQ->>ZMQ: 5. batch processing ZMQ->>Network: 6. ZeroMQ Publish Network-->>ZMQ: 7. Publish confirmation ZMQ-->>TSQ: 8. pop_batch() ES-->>App: 9. return future Note over App,Network: Asynchronous processing flow ``` - Data Flow： - Events are published via `KVEventSystem::publish` - `KVEventSystem`delegates to `KVEventProducer` to create and enqueue events - Events are placed into the `KVEventQueue` - `KVEventConsumer` consumes events from the queue, batches them, and publishes via `ZeroMQ` - Component Interaction： - `KVEventSystem` acts as a facade class, coordinating all components - Producers and consumers communicate via the `KVEventQueue` - All components depend on the event types defined in `kv_event.hpp` - Implementation Features： - Uses the same `ZeroMQ` implementation as the inference engine for high-performance network communication - Uses the same cross-language `msgpack` as the inference engine for serializing `kv-event` information ## Type of Change * Types - [ ] Bug fix - [ ] New feature - [ ] Transfer Engine - [x] Mooncake Store - [ ] Mooncake EP - [x] Integration - [ ] P2P Store - [x] Python Wheel - [ ] Breaking change - [x] CI/CD - [ ] Documentation update - [ ] Other ## How Has This Been Tested? I've added 2 new UTs: kv_event_publisher_test and thread_safe_queue_test ## Checklist - [x] I have performed a self-review of my own code. - [ ] I have updated the documentation. - [x] I have added tests to prove my changes are effective. --- <sub>🔄 This issue represents a GitHub Pull Request. It cannot be merged through Gitea due to API limitations.</sub>

bsakaguchi

2026-10-29 21:32:50 +00:00