第 13 章
Day 13：Key 到期機制
目前使用 map[string]string 儲存資料：
type mapStorage struct {
storage map[string]string
mu sync.RWMutex
}這個簡單結構無法直接處理資料到期。
如何實作？
根據 Redis 的到期機制說明，key 會以被動與主動兩種方式到期：被動方式在 client 存取已到期的 key 時刪除；但有些 key 到期後不再被存取，因此還需要定期抽查帶有到期時間的 keys，刪除其中已到期的資料。
因此 value 除了字串，還需要記錄 expiration。
Object
定義 Object 型別，取代原本儲存的 string，把資料與到期資訊放在一起：
package object
import "time"
type Object interface {
Key() string
ExpiresAt() *time.Time
}
type base struct {
key string
expiresAt *time.Time
}
func (obj base) Key() string {
return obj.key
}
func (obj base) ExpiresAt() *time.Time {
return obj.expiresAt
}String 是可儲存的 Object 型別之一：
package object
import "time"
type String struct {
base
val string
}
func NewString(k, v string, expiresAt *time.Time) *String {
return &String{
val: v,
base: base{key: k, expiresAt: expiresAt},
}
}
func (str *String) String() string {
return str.val
}更新 mapStorage：
// file: internal/repo/storage.go
func NewStorage() Storage {
s := mapStorage{}
- s.storage = make(map[string]string)
+ s.storage = make(map[string]object.Object)
return &s
}
type mapStorage struct {
- storage map[string]string
+ storage map[string]object.Object
mu sync.Mutex
}func (s *mapStorage) Set(_ context.Context, obj object.Object) error {
s.mu.Lock()
defer s.mu.Unlock()
s.storage[obj.Key()] = obj
return nil
}
func (s *mapStorage) Get(_ context.Context, k string) (v object.Object, err error) {
s.mu.RLock()
defer s.mu.RUnlock()
var ok bool
if v, ok = s.storage[k]; !ok {
return nil, fmt.Errorf("%w: miss", ErrNotFound)
}
if v.ExpiresAt() != nil && time.Now().After(*v.ExpiresAt()) {
delete(s.storage, k) // remove key from storage when expired
return nil, fmt.Errorf("%w: expired", ErrNotFound)
}
return
}讀取時刪除已到期的 key，就是被動刪除。
讓 SET 支援到期時間
SET 可以使用以下選項：
EX {sec}：在指定秒數後到期。PX {msec}：在指定毫秒數後到期。EXAT {usec}：在指定 Unix timestamp 到期，單位為秒。PXAT {umsec}：在指定 Unix timestamp 到期，單位為毫秒。
完整語法如下：
SET key value [NX | XX | IFEQ ifeq-value | IFNE ifne-value |
IFDEQ ifdeq-digest | IFDNE ifdne-digest] [GET] [EX seconds |
PX milliseconds | EXAT unix-time-seconds |
PXAT unix-time-milliseconds | KEEPTTL]為 Set 加入 parse()，回傳 (parsedSet, error)：
func (c *Set) parse(cmd *resp.Command) (p parsedSet, err error) {
const (
awaitingKey = iota
awaitingValue
awaitingOption
awaitingConditionValue
awaitingPostConditionOption
awaitingPostGetOption
awaitingExpirationValue
done
invalid
)
state := awaitingKey
var conditionType int
var expirationOption string
var expirationOptionIndex int
for i, arg := range cmd.Args() {
switch state {
case awaitingKey:
p.K = arg.String()
state = awaitingValue
case awaitingValue:
p.V = arg.String()
state = awaitingOption
case awaitingOption:
switch strings.ToUpper(arg.String()) {
case "NX":
p.Cond.Typ = nx
state = awaitingPostConditionOption
case "XX":
p.Cond.Typ = xx
state = awaitingPostConditionOption
case "IFEQ":
conditionType = ifeq
state = awaitingConditionValue
case "IFNE":
conditionType = ifne
state = awaitingConditionValue
case "IFDEQ":
conditionType = ifdeq
state = awaitingConditionValue
case "IFDNE":
conditionType = ifdne
state = awaitingConditionValue
case "GET":
p.Get = true
state = awaitingPostGetOption
case "EX", "PX", "EXAT", "PXAT":
expirationOption = strings.ToUpper(arg.String())
expirationOptionIndex = i
state = awaitingExpirationValue
case "KEEPTTL":
p.Exp = new(time.Time)
state = done
default:
state = invalid
}
case awaitingConditionValue:
p.Cond = parseSetCond{Typ: conditionType, Val: arg.String()}
state = awaitingPostConditionOption
case awaitingPostConditionOption:
switch strings.ToUpper(arg.String()) {
case "GET":
p.Get = true
state = awaitingPostGetOption
case "EX", "PX", "EXAT", "PXAT":
expirationOption = strings.ToUpper(arg.String())
expirationOptionIndex = i
state = awaitingExpirationValue
case "KEEPTTL":
p.Exp = new(time.Time)
state = done
default:
state = invalid
}
case awaitingPostGetOption:
switch strings.ToUpper(arg.String()) {
case "EX", "PX", "EXAT", "PXAT":
expirationOption = strings.ToUpper(arg.String())
expirationOptionIndex = i
state = awaitingExpirationValue
case "KEEPTTL":
p.Exp = new(time.Time)
state = done
default:
state = invalid
}
case awaitingExpirationValue:
p.Exp, err = parseExpiration(expirationOption, arg.String())
if err != nil {
return parsedSet{}, fmt.Errorf("%w: %w", ErrSyntax, err)
}
state = done
case done:
state = invalid
}
}
if state != awaitingOption && state != awaitingPostConditionOption && state != awaitingPostGetOption && state != done {
return parsedSet{}, fmt.Errorf("%w: invalid SET syntax", ErrSyntax)
}
if p.Exp != nil && !p.Exp.IsZero() {
cmd.UpdateAOF(expirationOptionIndex+1, resp.NewBulkString("PXAT"))
cmd.UpdateAOF(expirationOptionIndex+2, resp.NewBulkString(strconv.FormatInt(p.Exp.UnixMilli(), 10)))
}
return p, nil
}
func parseExpiration(option, value string) (*time.Time, error) {
switch option {
case "EX", "PX":
unit := "s"
if option == "PX" {
unit = "ms"
}
duration, err := parseDuration(value, unit)
if err != nil {
return nil, err
}
return new(time.Now().Add(duration)), nil
case "EXAT", "PXAT":
timestamp, err := strconv.Atoi(value)
if err != nil {
return nil, err
}
if timestamp <= 0 {
return nil, errors.New("invalid expire time")
}
if option == "EXAT" {
return new(time.Unix(int64(timestamp), 0)), nil
}
return new(time.UnixMilli(int64(timestamp))), nil
}
return nil, fmt.Errorf("unknown expiration option %q", option)
}
func parseDuration(s, unit string) (time.Duration, error) {
if _, err := strconv.Atoi(s); err != nil {
return 0, fmt.Errorf("invalid expire time: %w", err)
}
var sb strings.Builder
sb.WriteString(s)
sb.WriteString(unit)
if d, err := time.ParseDuration(sb.String()); err != nil {
return d, err
} else if d <= 0 {
return d, errors.New("invalid expire time")
} else {
return d, nil
}
}
const (
nx = iota + 1 // set value only not exists
xx // set value only exists
ifeq // set only value == cond.Val
ifne // set only value != cond.Val
ifdeq // set only XXH3(value) == cond.Val
ifdne // set only XXH3(value) != cond.Val
)
type parseSetCond struct {
Typ int
Val string
}
type parsedSet struct {
K string
V string
Cond parseSetCond
Get bool
Exp *time.Time
}注意：這裡以 finite state machine 建立 parser。程式較長，但狀態轉移清楚。
SetParam interface
Set.parse() 回傳 parsedSet，而不是直接將 object.Object 傳給 repo.Storage.Set()。因此讓 storage 接受 repo.SetParam interface，再由 parsedSet 實作它：
type SetParam interface {
Obj() object.Object
}type parsedSet struct {
K string
V string
Cond parseSetCond
Get bool
Exp *time.Time
}
var _ repo.SetParam = parsedSet{}
func (parsed parsedSet) Obj() object.Object {
return object.NewString(parsed.K, parsed.V, parsed.Exp)
}var _ repo.SetParam = parsedSet{} 能在編譯時確認型別符合介面。
更新 repo.Storage.Set()：
func (s *mapStorage) Set(_ context.Context, param SetParam) error {
s.mu.Lock()
defer s.mu.Unlock()
s.storage[param.Obj().Key()] = param.Obj()
return nil
}這讓未來的 storage 除了 object.String，也能處理 HSET 所需的 object.Hash 等型別。
主動到期
前面完成的是存取時刪除已到期資料。主動方式則定期抽查並清除到期 keys。Redis 每 100 毫秒執行這類工作。
使用 Worker 重構
我們已經用 goroutine 實作過背景工作：依設定每秒同步 AOF。
建立 service.Worker 統一管理這些 goroutines，讓 Application.Run() 流程更清楚：
package service
import (
"context"
"errors"
"time"
"golang.org/x/sync/errgroup"
"olivine/internal/data"
"olivine/internal/repo"
)
type Worker interface {
Start(context.Context) error
}
func NewWorker(cfg *data.Config, aof AOF, storage repo.Storage) Worker {
return &worker{
cfg: cfg,
aof: aof,
storage: storage,
}
}
type worker struct {
cfg *data.Config
storage repo.Storage
aof AOF
}
func (w *worker) Start(ctx context.Context) error {
g, ctx := errgroup.WithContext(ctx)
if w.cfg.AOFEnabled && w.cfg.AOFFsync == data.AOFFsyncEverySec {
g.Go(w.aofSyncer(ctx))
}
g.Go(w.storagePruner(ctx))
return g.Wait()
}
func (w *worker) aofSyncer(ctx context.Context) func() error {
return func() error {
ticker := time.NewTicker(time.Second)
defer ticker.Stop()
for {
select {
case <-ctx.Done():
return nil
case <-ticker.C:
if err := w.aof.Sync(); err != nil {
return err
}
}
}
}
}再於應用程式的 errgroup goroutine 呼叫 Worker.Start()。Worker 的實作如下：
package service
import (
"context"
"time"
"golang.org/x/sync/errgroup"
"olivine/internal/data"
)
type Worker interface {
Start(context.Context) error
}
func NewWorker(cfg *data.Config, aof AOF) Worker {
return &worker{
cfg: cfg,
aof: aof,
}
}
type worker struct {
cfg *data.Config
aof AOF
}
func (w *worker) Start(ctx context.Context) error {
g, ctx := errgroup.WithContext(ctx)
if w.cfg.AOFEnabled && w.cfg.AOFFsync == data.AOFFsyncAlways {
g.Go(w.aofSyncer(ctx))
}
return g.Wait()
}
func (w *worker) aofSyncer(ctx context.Context) func() error {
return func() error {
ticker := time.NewTicker(time.Second)
defer ticker.Stop()
for {
select {
case <-ctx.Done():
return nil
case <-ticker.C:
if err := w.aof.Sync(); err != nil {
return err
}
}
}
}
}Storage.Prune() 與 Worker 註冊
func (s *mapStorage) Prune(ctx context.Context) error {
for {
select {
case <-ctx.Done():
return ctx.Err()
default:
if stop := s.tryPrune(); stop {
return nil
}
}
}
}
func (s *mapStorage) tryPrune() bool {
s.mu.Lock()
defer s.mu.Unlock()
if len(s.storage) == 0 {
return true
}
const sampleSize = 10
sampled := 0
expired := 0
now := time.Now()
for k, v := range s.storage {
if sampled == sampleSize {
break
}
sampled++
if v.ExpiresAt() != nil && now.After(*v.ExpiresAt()) {
delete(s.storage, k)
expired++
}
}
return expired*4 < sampled
}// file: internal/service/worker.go
func (w *worker) Start(ctx context.Context) error {
g, ctx := errgroup.WithContext(ctx)
if w.cfg.AOFEnabled && w.cfg.AOFFsync == data.AOFFsyncEverySec {
g.Go(w.aofSyncer(ctx))
}
+ g.Go(w.storagePruner(ctx))
return g.Wait()
}
+func (w *worker) storagePruner(ctx context.Context) func() error {
+ return func() error {
+ ticker := time.NewTicker(time.Millisecond * 100)
+ defer ticker.Stop()
+
+ for {
+ select {
+ case <-ctx.Done():
+ return nil
+ case <-ticker.C:
+ if err := w.storage.Prune(ctx); err != nil {
+ if ctx.Err() != nil && errors.Is(err, ctx.Err()) {
+ return nil
+ }
+ return err
+ }
+ }
+ }
+ }
+}到期資料與 AOF
直接把原始 SET 寫入 AOF，會在還原時產生意料外的結果：
- 00:00:00：執行
SET foo bar EX 10，同時寫入 storage 與 AOF。 - 00:00:05：
GET foo回傳bar。 - 00:00:11：key 到期，
GET foo回傳(nil)。 - 00:01:00：關閉服務。
- 01:00:00：啟動服務，重播 AOF。
- 01:00:05：
GET foo又回傳bar，不符合預期。
解法是將 AOF 中的相對到期時間改為絕對 Unix timestamp。
在 resp.Command 加入 aof
帶有到期選項的 SET 需要寫入不同版本的指令，所以在 resp.Command 加入供 AOF writer 使用的 aof：
type Command struct {
raw Value
+ aof Array
cmd BulkString
args []BulkString
}aof 是 raw 的 deep clone；resp.ReadCommand() 已確認 raw 為 resp.Array。
func ReadCommand(rd *Reader) (*Command, error) {
v, err := rd.Read()
if err != nil {
return nil, err
}
arr, ok := v.(Array)
if !ok {
return nil, fmt.Errorf("%w: expected array, got %T(%+v)", ErrProtocol, v, v)
}
if arr.null || len(arr.data) == 0 {
return nil, fmt.Errorf("%w: empty array", ErrProtocol)
}
cmd, ok := arr.data[0].(BulkString)
if !ok {
return nil, fmt.Errorf("%w: command expected bulk string, got %T(%+v)", ErrProtocol, arr.data[0], arr.data[0])
}
args := make([]BulkString, 0, len(arr.data)-1)
for i := range arr.data[1:] {
arg, ok := arr.data[i+1].(BulkString)
if !ok {
return nil, fmt.Errorf("%w: argument [%d] expected bulk string, got %T(%+v)", ErrProtocol, i, arr.data[i+1], arr.data[i+1])
}
args = append(args, arg)
}
return &Command{
raw: v,
aof: arr.Clone(),
cmd: cmd,
args: args,
}, nil
}加入 UpdateAOF()：
func (cmd *Command) UpdateAOF(i int, v Value) {
cmd.aof.data[i] = v
}在 SET 呼叫 UpdateAOF()
if p.Exp != nil && !p.Exp.IsZero() {
cmd.UpdateAOF(expirationOptionIndex+1, resp.NewBulkString("PXAT"))
cmd.UpdateAOF(expirationOptionIndex+2, resp.NewBulkString(strconv.FormatInt(p.Exp.UnixMilli(), 10)))
}將 EX、PX 改寫為 PXAT 後再寫入 AOF。
TTL 指令
實作並註冊 TTL：
package cmd
import (
"context"
"errors"
"fmt"
"time"
"olivine/internal/repo"
"olivine/pkg/resp"
)
type TTL struct {
storage repo.Storage
}
func NewTTL(storage repo.Storage) *TTL {
return &TTL{
storage: storage,
}
}
func (c *TTL) Command() string {
return "TTL"
}
func (c *TTL) Exec(ctx context.Context, cmd *resp.Command) (resp.Value, error) {
if err := c.parse(cmd); err != nil {
return nil, err
}
args := cmd.Args()
k := args[0]
v, err := c.storage.Get(ctx, k.String())
if err != nil {
if errors.Is(err, repo.ErrNotFound) {
return resp.Integer(-2), nil
}
return nil, fmt.Errorf("%w: %w", ErrStorage, err)
}
if v.ExpiresAt() == nil {
return resp.Integer(-1), nil
}
return resp.Integer(time.Until(*v.ExpiresAt()) / time.Second), nil
}
func (c *TTL) parse(cmd *resp.Command) error {
const (
awaitingKey = iota
keyReceived
tooManyKeys
)
state := awaitingKey
for range cmd.Args() {
switch state {
case awaitingKey:
state = keyReceived
case keyReceived:
state = tooManyKeys
}
}
if state != keyReceived {
return fmt.Errorf("%w: argument count mismatch: expect %d got %d", ErrSyntax, 1, len(cmd.Args()))
}
return nil
}// file: internal/service/cmd/cmd.go
func NewCommands(storage repo.Storage) []Command {
return []Command{
&Ping{},
NewSet(storage),
NewGet(storage),
+ NewTTL(storage),
}
}補充
這個主題很難寫。到期機制涉及許多電腦科學知識，但全部展開會讓文章過長。
我想解釋為什麼 Redis 同時採用主動與被動到期，而不使用以 expiration *time.Time 排序的 min-heap（priority queue）。
也想說明 Redis 如何在 redisObject 使用 bit fields 儲存到期資訊，以及這裡為什麼自行建立 finite state machine，而放棄 regular expression。
這一階段包含不少其他修改，但無法在本章逐一說明。