データシリアライゼーションとキャッシュ制御の基礎

分散キャッシュシステムであるgroupcacheのアーキテクチャを理解するためには、その内部で利用されている3つの主要なパッケージの挙動を把握することが不可欠です。ここでは、groupcachepb（Protocol Buffers）、lru（Least Recently Usedアルゴリズム）、そしてsingleflight（重複呼び出しの抑制）という3つの技術要素について、実装レベルで詳細を解説します。

Protocol Buffersによるデータ定義

groupcacheでは、ノード間の通信において構造化データを効率的に扱うためにProtocol Buffers（protobuf）が採用されています。これは言語やプラットフォームに依存しない拡張可能なシリアライゼーション機構を提供します。Go言語の標準パッケージであるencoding/gobと似た役割を持ちますが、protobufはスキーマファーストなアプローチをとり、定義ファイル（.proto）からコードを生成します。

以下に、groupcacheで使用されているデータ構造の定義例を示します。ここではproto2の構文を使用しており、メッセージ型やRPCサービスの定義が含まれています。

syntax = "proto2";

package demo;

// キャッシュへのリクエスト構造
message CacheRequest {
  required string target_group = 1;
  required string lookup_key = 2; 
}

// キャッシュからのレスポンス構造
message CacheResponse {
  optional bytes payload = 1;
  optional double qps_metric = 2;
}

// サービス定義
service CacheService {
  rpc Retrieve(CacheRequest) returns (CacheResponse);
}

この定義ファイルにはいくつかの重要な要素が含まれています。

• フィールドルール: requiredは必須フィールド、optionalは任意フィールドを示します。proto3ではこれらが廃止されていますが、groupcacheはproto2ベースで記述されています。
• タグ（Tag）: = 1や= 2といった数値はバイナリ形式での識別子であり、一意である必要があります。
• Service定義: RPCインターフェースを定義しており、CacheRequestを受け取りCacheResponseを返すRetrieveメソッドが宣言されています。これにより、異なるGoプロセス間でのメソッド呼び出しが可能になります。

LRUアルゴリズムの実装

メモリ管理において、キャッシュの容量が上限に達した際にどのデータを追い出すか（Evict）は重要な課題です。LRU（Least Recently Used）は、「最も最近使われていないデータ」を削除対象とするアルゴリズムです。groupcacheのlruパッケージは、このアルゴリズムを双方向リスト（container/list）とハッシュマップを組み合わせて実装しており、要素のアクセスと削除を高速に行います。

以下に、LRUキャッシュの簡易的な再実装例を示します。変数名や構造を変更しつつ、元のロジックを維持しています。

package lruimpl

import (
	"container/list"
)

// Cache はLRUアルゴリズムに基づくキャッシュ構造体です
type Cache struct {
	capacity    int
	evictFunc   func(key Key, value interface{})
	items       map[interface{}]*list.Element
	linkedList  *list.List
}

type Key interface{}

type entry struct {
	key   Key
	value interface{}
}

// New は新しいLRUキャッシュインスタンスを生成します
func New(capacity int) *Cache {
	return &Cache{
		capacity:   capacity,
		items:      make(map[interface{}]*list.Element),
		linkedList: list.New(),
	}
}

// Set はキーと値のペアをキャッシュに追加します
func (c *Cache) Set(key Key, value interface{}) {
	// 遅延初期化
	if c.items == nil {
		c.items = make(map[interface{}]*list.Element)
		c.linkedList = list.New()
	}

	// 既存キーのチェック：存在すれば先頭に移動して値を更新
	if elem, exists := c.items[key]; exists {
		c.linkedList.MoveToFront(elem)
		elem.Value.(*entry).value = value
		return
	}

	// 新規エントリの追加をリスト先頭に行う
	elem := c.linkedList.PushFront(&entry{key, value})
	c.items[key] = elem

	// 容量オーバーの場合、最も古い要素（末尾）を削除
	if c.capacity != 0 && c.linkedList.Len() > c.capacity {
		c.evictOldest()
	}
}

// Get はキーに関連付けられた値を取得します
func (c *Cache) Get(key Key) (interface{}, bool) {
	if c.items == nil {
		return nil, false
	}
	if elem, exists := c.items[key]; exists {
		c.linkedList.MoveToFront(elem)
		return elem.Value.(*entry).value, true
	}
	return nil, false
}

func (c *Cache) evictOldest() {
	elem := c.linkedList.Back()
	if elem != nil {
		c.removeElement(elem)
	}
}

func (c *Cache) removeElement(e *list.Element) {
	c.linkedList.Remove(e)
	kv := e.Value.(*entry)
	delete(c.items, kv.key)
	if c.evictFunc != nil {
		c.evictFunc(kv.key, kv.value)
	}
}

この実装のポイントは、mapでの検索結果を直接listの要素として保持することで、検索（O(1)）と順序更新（O(1)）の両立を実現している点です。データが参照されるたびにリストの先頭に移動させ、容量制限を超えた場合はリストの末尾（最も使われていない要素）を削除します。

Singleflightによる重複処理の抑制

分散システムや高負荷なキャッシュシステムでは、同一のキーに対する大量のリクエストが同時に発生する「サンダリング・ハード（thundering herd）問題」が発生する可能性があります。例えば、計算に時間がかかる関数の結果がキャッシュにない場合、多数のクライアントが同時にその計算を開始してしまうと、リソースを無駄に消費します。

singleflightパッケージは、この問題を解決するために「同じ処理中である限り、後続のリクエストを待機させ、最初の実行結果を共有する」という仕組みを提供します。

以下は、この機能を実現するための構造体とロジックの再実装例です。

package singleflight

import (
	"sync"
)

// invocation は実行中、または完了した処理を表します
type invocation struct {
	waitGroup sync.WaitGroup
	result    interface{}
	err       error
}

// Group は処理の重複実行を抑制するための名前空間です
type Group struct {
	mu    sync.Mutex
	tasks map[string]*invocation
}

// Do は指定されたキーに対して関数を実行します
// 同じキーで既に実行中の処理がある場合、その処理の完了を待って同じ結果を返します
func (g *Group) Do(key string, task func() (interface{}, error)) (interface{}, error) {
	g.mu.Lock()
	if g.tasks == nil {
		g.tasks = make(map[string]*invocation)
	}

	// 既に実行中のタスクがあれば、それを待機して結果を返す
	if running, exists := g.tasks[key]; exists {
		g.mu.Unlock()
		running.waitGroup.Wait()
		return running.result, running.err
	}

	// 新しいタスクを登録して実行を開始する
	current := &invocation{}
	current.waitGroup.Add(1)
	g.tasks[key] = current
	g.mu.Unlock()

	// 実際の処理を実行
	current.result, current.err = task()
	current.waitGroup.Done()

	// 完了したタスクをマップから削除
	g.mu.Lock()
	delete(g.tasks, key)
	g.mu.Unlock()

	return current.result, current.err
}

この実装では、sync.Mutexで保護されたmapを用いて、現在進行中の処理（invocation）を管理しています。特定のキーに対して処理が存在しない場合は新たにinvocationを作成してmapに登録し、関数を実行します。既に処理が存在する場合は、WaitGroup.Wait()によってその処理が完了するまでブロックし、完了時に同じresultとerrを全ての呼び出し元へ返します。これにより、高コストな計算の重複実行を確実に防ぐことができます。