昨年のAdvent Calendarに引き続きChannelネタです。

前置き

データベースアクセスを削減する為に取得したデータをローカルにキャッシュする、みたいなことをよくやります。

シンプルに書くと↓みたいなカンジでしょうか。

var cache sync.Map

func get(key string) (v interface{}, err error) {
	v, ok := cache.Load(key)
	if !ok {
		v, err = getFromDB(key)
		if err != nil {
			return nil, err
		}
		cache.Store(key, v)
	}
	return v, nil
}

キャッシュが存在しなければDBに取得しに行きます。

さて、上記コード、一つ問題があります。
キャッシュされていない状態で同じkeyでcache.Load→cache.Storeの間にアクセスが集中すると、それらは全てデータベースに同じデータを取得しに行ってしまい、無駄が多いです。

¹

そこで、誰かがデータを取得しに行ってる間、同じデータが必要な後続リクエストはそれを待機する様にしたいと思います。

¹

作戦

先発リクエストデータ取得中の後続リクエストの待機にChannelを利用します。
Channelはキャッシュのエントリ毎に保持します。

先発リクエストはまず空のキャッシュエントリを作成（予約）しChannelを保存する
先発リクエストがデータ取得開始
後続リクエストは1で作成されたChannelをread。先発リクエストのデータ取得を待機
先発リクエストはデータを取得したらエントリに格納し、Channelをclose
待機していた後続リクエストが動き出しデータを参照

Channelは結構サイズ（メモリ使用量）が大きいので節約する為にキャッシュ完了したらnilをセットしておきます。

実装

キャッシュエントリ


type entry struct {
	lock  chan struct{} // lock for fetch
	value interface{}
	err   error
}

func (ce *entry) getWithTimeout(dst interface{}, timeout time.Duration) (interface{}, error) {
	if lock := ce.lock; lock != nil { // nil lock means cache is ready
		if timeout < 0 { // no timeout
			<-ce.lock
		} else {
			select {
			case <-lock:
			case <-time.After(timeout):
				return nil, ErrGetCacheTimeout
			}
		}
	}

	if ce.err != nil {
		return nil, ce.err
	}

	return ce.value, nil
}

エントリのデータを参照する前にかならずChannelをreadします。
データが格納されるまではブロックされます。
データが格納されたらChannelがcloseされるので、それからデータを参照します。

引数でタイムアウトを指定できる様にしました。-1ならば無制限、0ならばTry-Lock（ロックされていたら待機せず処理続行）の意味合いとなります。

キャッシュ本体


type Cache struct {
	cache sync.Map
}

func (c *Cache) Get(key interface{}) (value interface{}, err error) {
	return c.GetWithTimeout(key, -1)
}

func (c *Cache) GetWithTimeout(key interface{}, timeout time.Duration) (value interface{}, err error) {
	e, ok := c.cache.Load(key)
	if !ok || e == nil {
		return nil, ErrEntryNotFound
	}
	return e.(*entry).getWithTimeout(key, timeout)
}

type ResolveFunc func(entity interface{}, err error)

func (c *Cache) Reserve(key interface{}) ResolveFunc {
	entry := &entry{lock: make(chan struct{})}

	resolve := func(entity interface{}, err error) {
		entry.value, entry.err = entity, err
		close(entry.lock)
		entry.lock = nil // set nil to save memory
	}

	c.cache.Store(key, entry)

	return resolve
}

利用方法

cache := new(Cache) // 生成

value, err := cache.Get(key)
if err == ErrEntryNotFound {
	resolve := cache.Reserve(key) // エントリ予約

	value, err = getFromDB(key) // DB参照

	resolve(value, err) // キャッシュ保存

	if err != nil {
		return err
	}
}

厳密には cache.Get(key) から cache.Reserve(key) の間にアクセスがあると重複してDBにフェッチが走ってしまいますが、この間隔は極小の為それほど問題にはならないかと思います。気になる場合はMutexで排他してもよいでしょう。

弱点（デメリット）

**とにかくChannelのサイズが大きい！！**😣

メモリアロケーションを計測しましたが、Channel1個辺り約1KiBくらいは消費します。
キャッシュしたらnilを設定して解放してるのでそれほど問題にはならないかと思いますが、同時に大量キーにアクセスされると瞬間的なメモリ使用量は大きくなる可能性があります。

Channelの代わりにsync.Mutex(or sync.RWMutex）を利用することも可能です。サイズは半分以下になります。
ただしMutexはタイムアウトやTry-Lockが実装出来ない（よね？）という制限があります。それら機能が不要ならばMutexを使うのもアリかと思います。

Channel, Mutex以外にもよい方法をご存知の方はコメントで教えて頂けるととても嬉しいです🙇‍♂️

作った

本記事内容のライブラリを作成しました^^

有効期限指定や、統計機能なども追加してあります。
内部のキャッシュ実装はHashiCorpさんのgolang-lruをそのまま利用させてもらったのでサイズ上限も指定可能です。

使い方はREADMEやGodocを参照ください。

本記事内で使用しているGopher画像はRenée Frenchさんのデザインを加工したものです ↩ ↩²