Build a Mini Olric KV Store in Go: 300 Lines of Sharding, TTL, and Performance Tuning

Mini Olric Implementation

Core Data Structures

type Entry struct {
    Key       string
    Value     string
    ExpiredAt time.Time
}

type Shard struct {
    mu    sync.RWMutex
    items map[string]*Entry
}

func NewShard() *Shard {
    return &Shard{items: make(map[string]*Entry)}
}

func (s *Shard) Put(key, value string, ttl time.Duration) {
    s.mu.Lock()
    defer s.mu.Unlock()
    e := &Entry{Key: key, Value: value}
    if ttl > 0 {
        e.ExpiredAt = time.Now().Add(ttl)
    }
    s.items[key] = e
}

func (s *Shard) Get(key string) (string, bool) {
    s.mu.RLock()
    defer s.mu.RUnlock()
    e, ok := s.items[key]
    if !ok || (!e.ExpiredAt.IsZero() && time.Now().After(e.ExpiredAt)) {
        return "", false
    }
    return e.Value, true
}

func (s *Shard) Delete(key string) {
    s.mu.Lock()
    defer s.mu.Unlock()
    delete(s.items, key)
}

DMap and Shard Routing

type DMap struct {
    shards []*Shard
    n      int
}

func NewDMap(numShards int) *DMap {
    shards := make([]*Shard, numShards)
    for i := 0; i < numShards; i++ {
        shards[i] = NewShard()
    }
    return &DMap{shards: shards, n: numShards}
}

func (dm *DMap) getShard(key string) *Shard {
    idx := int(crc32.ChecksumIEEE([]byte(key))) % dm.n
    return dm.shards[idx]
}

func (dm *DMap) Put(key, value string, ttl time.Duration) {
    dm.getShard(key).Put(key, value, ttl)
}

func (dm *DMap) Get(key string) (string, bool) {
    return dm.getShard(key).Get(key)
}

func (dm *DMap) Delete(key string) {
    dm.getShard(key).Delete(key)
}

Node and Simplified RPC

type Node struct {
    Addr  string
    DMap  *DMap
    Peers []string
}

func NewNode(addr string, peers []string) *Node {
    return &Node{Addr: addr, DMap: NewDMap(16), Peers: peers}
}

func (n *Node) Start() {
    http.HandleFunc("/get", n.handleGet)
    http.HandleFunc("/put", n.handlePut)
    http.HandleFunc("/delete", n.handleDelete)
    log.Printf("Node running at %s
", n.Addr)
    log.Fatal(http.ListenAndServe(n.Addr, nil))
}

HTTP handlers expose /get, /put, and /delete endpoints that act as a lightweight RPC layer, enabling a node to read, write, and delete entries on remote peers.

Feature Optimizations

TTL / Automatic Expiration

func (s *Shard) cleanupExpired() {
    interval := 1 * time.Second // configurable
    ticker := time.NewTicker(interval)
    defer ticker.Stop()
    for range ticker.C {
        s.mu.Lock()
        for k, e := range s.items {
            if !e.ExpiredAt.IsZero() && time.Now().After(e.ExpiredAt) {
                delete(s.items, k)
            }
        }
        s.mu.Unlock()
    }
}

Node Replica Synchronization

// Put: local write + pipeline batch replica sync
func (n *Node) Put(key, value string, ttl time.Duration) {
    n.DMap.Put(key, value, ttl)
    // WAL persistence (omitted)
    var requests []*Request
    for _, peer := range n.Peers {
        if peer == n.Addr {
            continue
        }
        requests = append(requests, &Request{Peer: peer, Key: key, Value: value, TTL: ttl})
    }
    // pipeline concurrent replica sync
    sendBatch(requests)
}

Simple Rebalancing

func (n *Node) rebalance(keys []string, target string) {
    for _, key := range keys {
        val, ok := n.DMap.Get(key)
        if !ok {
            continue
        }
        go func(k, v string) {
            http.Get(fmt.Sprintf("http://%s/put?key=%s&value=%s", target, k, v))
            n.DMap.Delete(k)
        }(key, val)
    }
}

Concurrency Safety

Each Shard protects its internal map with a sync.RWMutex, allowing multiple concurrent reads while writes obtain an exclusive lock.

Performance Experiments and Optimizations

Identified Bottlenecks

Shard hot‑spot lock contention

RPC network latency

Garbage‑collection overhead and memory escape

Blocking data migration during rebalancing

Benchmark Example

func BenchmarkDMapPut(b *testing.B) {
    dm := NewDMap(16)
    b.ResetTimer()
    for i := 0; i < b.N; i++ {
        dm.Put(fmt.Sprintf("key-%d", i), "value", 0)
    }
}

Pipeline Batch RPC

func sendBatch(requests []*Request) []*Response {
    var wg sync.WaitGroup
    responses := make([]*Response, len(requests))
    for i, req := range requests {
        wg.Add(1)
        go func(i int, r *Request) {
            defer wg.Done()
            responses[i] = send(r)
        }(i, req)
    }
    wg.Wait()
    return responses
}

Memory and GC Optimizations

var entryPool = sync.Pool{
    New: func() interface{} { return &Entry{} },
}

func getEntry() *Entry { return entryPool.Get().(*Entry) }
func putEntry(e *Entry) { entryPool.Put(e) }

Optimization Results

Throughput increased 2–3×

Network latency reduced by ~20%

CPU usage lowered by ~15%

Run the benchmark with go run ./test to reproduce the performance chart.

Source Code

GitHub repository: https://github.com/louis-xie-programmer/mini-olric

Gitee repository: https://gitee.com/louis_xie/mini-olric

The implementation resides in a single Go file of roughly 300 lines and includes placeholder functions for future extensions such as distributed transactions, persistence, and advanced rebalancing.