🐯

Goの排他制御 API開発での注意点を考える

に公開
Go
tech

はじめに

持っていない金額以上の残高を出金できてしまったり、ステータス更新が複数ユーザーによって同時に行われた結果、誰かの処理は反映されなかったり、こんなことが起きたら非常に冷や汗ものです。排他制御を間違えると事故が簡単に起こります。その排他制御に対して、実際にどんなふうに取り組めばいいのかを自分の頭の整理を目的にまとめます。内容はもしかしたら初歩的なことかもしれませんが、何かヒントになればと思います。

TL;DR

  • グローバル変数でのsync.Mutexは同一プロセス内の排他制御しかできていない。
  • クラウド環境が主流の昨今のプロジェクトでは分散ロックにする。
    • ローカルでは起きないが、本番環境で起きるという排他制御の問題を回避できる。

背景

書籍に載っているサンプルプロジェクトや、フレームワークのクイックスタートに記載してあるソースコードを読むと、そのほとんどが排他制御について全く触れられていません。確かに、排他制御を考えると、本質的な内容がぼやけてしまうため、説明を端折るのは仕方がないことです。ですが、実際の開発では「排他制御」は事故が起こらないために特に気をつけなればならないことの一つです。システム開発において、処理が同時に実行されては困るロジックをどうするか、特に残高管理にあたっては注意を払わなければなりません。もし排他制御への注意を怠っていると、実際にある残高以上の取引ができてしまい、大事故につながります。


func main() {
  router := gin.Default()
  router.GET("/ping", func(c *gin.Context) {
    c.JSON(200, gin.H{
      "message": "pong",
    }) // <= 排他制御は当然存在しない。
  })
  router.Run()
}

「https://gin-gonic.com/en/docs/quickstart/」より引用

参照カウント付きのロックマネージャ

goはgoroutineという目玉機能がありその排他制御として標準ライブラリにsync.Mutexがあります。
Webアプリケーションの排他制御にもこれを使うのかなと考え、sync.Mapで自己流の排他制御を最初に作成しました。引数にはユニークIDをとり、同時に更新されてはいけないレコードのIDや、ログイン中のユーザーIDを取ることで、排他制御ができると思っていました。


// (!) グローバル変数にロック機構を用意している
var globalLock = &lockManager{}

type lockRef struct {
	mu    sync.Mutex
	count int32
}

func newLockRef() *lockRef {
	return &lockRef{
		count: 1,
	}
}

// LockManager 
type lockManager struct {
	locks sync.Map
}

type LockManager interface {
	Lock(uid string)
	Unlock(uid string)
}

func NewLockManager() LockManager {
	return globalLock
}

// Lock ユニークIDに基づいてロックを獲得する
func (lm *lockManager) Lock(uid string) {
    // (!) ロックが取れるまでずっと待つ。
	for {
        // LoadOrStoreは「存在しなければ作る、存在すれば既存を使う」を実現するメソッド
		actual, loaded := lm.locks.LoadOrStore(uid, newLockRef())
		lRef := actual.(*lockRef)

        // 存在する => 他の処理が走っている
		if loaded {
			// LoadInt32はアトミックにint32の値を読む。
			if atomic.LoadInt32(&lRef.count) == 0 {
                // 実質もう捨てられるはずのロックをつかんでいるのでやり直す。(※)
				continue
			}
			atomic.AddInt32(&lRef.count, 1)
		}
		lRef.mu.Lock()
		break
	}
}

// Unlock ロックを解放し、参照カウントが0になったらマップから削除
func (lm *lockManager) Unlock(uid string) {
	if value, ok := lm.locks.Load(uid); ok {
		lRef := value.(*lockRef)
		lRef.mu.Unlock()

        // 参照カウンタを-1する。0になったら削除する。
		if atomic.AddInt32(&lRef.count, -1) == 0 {
            // 上の(※)に入るのは、この「atomic.AddInt32(&lRef.count, -1) == 0」が
            // trueになり、lm.locks.Delete(uid)が実行される間。
			lm.locks.Delete(uid)
		}
	}
}

func (u *userUsecase) UpdateUserName(ctx context.Context, userID string) (*res.User, error) {

    // 「userID」を渡す => ユーザー単位でロックがかけているつもり
    u.lockManager.Lock(userID)
    // アンロックをdeferで呼び出す
    defer u.lockManager.Unlock(userID) 

    // クリティカルセクションを実行する
}

問題点

ただ、よくよく調べてみるとこの実装では、同一プロセス内でしか、排他制御ができていません。goroutineを複数立ち上げて並行処理を実行する時に使用する分には問題ないですが、WebAPIにおいて、「フロントから処理が呼ばれた際にDB更新処理を行う」のと、「ワンショットでコンテナを立ち上げて、定期実行でDB更新処理を行う」のが同時に走ると、これらは別プロセスで処理が走るので、このコードでは排他制御ができません。

補足

go run main.goで立ち上がるのがプロセスで、これでGoのランタイムが一個動きます。この上でgoroutineが動きます。APIでもリクエストを受け付けた時に一つのgoroutineが立ち上がります。sync.Mutexはこのプロセス間での排他制御であり、仮にクラウド上などで水平スケールを行い、コンテナを複数立ち上げた場合は、このプロセスやコンテナを跨いだロックは行われません。

さらに昨今はクラウド環境にアプリがデプロイされることが多いですが、APIサーバーを水平スケールをすることが多く、プロセスの数が複数になることが多々あります。
それを考えるとこのコードの排他制御をAPIで導入しても、「同一ユーザー」による「同一端末」の「同一ブラウザ」の「同一セッション」での多重実行を防いでくれるだけで、ユーザーの画面操作が別のバックグラウンド処理と重なったり、ユーザーが複数端末のログインしているときなどで簡単に破綻してしまうロジックでした。

行ロックはどうか

代替案として行ロックを使う手もあります。
gormだと、トランザクション内で以下の処理を行うと行ロックでデータを取得します。

tx.Clauses(clause.Locking{Strength: "UPDATE"}).First(model)

こうすると、他の処理がこのコードを呼び出すと、待機されます。tx.Commit()まで待ちます。
しかし、行ロックだとデッドロックの発生の可能性があるため、少し導入に慎重になる必要があります。テーブル数が多い場合だと混乱の元です。「複数のレコードを更新する場合は更新する順番を揃える」というのがありますが、複数人で開発していたり、コード量が増えてくるとこの対策が徹底されるかは不安です。自分でルールを決めてたとしても数ヶ月もすれば忘れてしまうものです。

分散ロックを導入する

WebAPI開発においては、クラウド環境が主流である以上、分散ロックが第一に取るべき方法なのかなと思います。


// redsyncパッケージを使用する。

type RedisDriver interface {
	Lock(ctx context.Context, key string) (*redsync.Mutex, error)
	UnLock(ctx context.Context, mux *redsync.Mutex) error
}

func NewRedisDriver(
	client *redis.Client,
) RedisDriver {
	pool := goredis.NewPool(client)
	rs := redsync.New(pool)
	return &redisDriver{
		rs: rs,
	}
}

type redisDriver struct {
	rs *redsync.Redsync
}

// Lock ロックを取得。複数プロセスから呼ばれても排他制御ができる。
func (d *redisDriver) Lock(ctx context.Context, key string) (*redsync.Mutex, error) {
	mutex := d.rs.NewMutex(key)
	err := mutex.LockContext(ctx)
	if err != nil {
		return nil, err
	}
	return mutex, nil
}

func (d *redisDriver) UnLock(ctx context.Context, mux *redsync.Mutex) error {
	ok, err := mux.UnlockContext(ctx)
	if err != nil {
		return err
	}
	if !ok {
		return fmt.Errorf("unlock failed: not owner of lock")
	}
	return nil
}



// LockManager 分散ロックに変更。ここではredisを使う。
type lockManager struct {
	redisDriver driver.RedisDriver
}

type LockManager interface {
    // メソッドを１個に集約
	LockWithContext(ctx context.Context, uid string) (func(), error)
}

func NewLockManager(redisDriver driver.RedisDriver) LockManager {
	return &lockManager{
		redisDriver: redisDriver,
	}
}

// LockWithContext
func (l *lockManager) LockWithContext(ctx context.Context, uid string) (func(), error) {

	lockKey := "lock:" + uid

    // ロックを取得します。
    // デフォルトは、ttlは8秒。リトライ間隔は50~250ミリ秒。
	mutex, err := l.redisDriver.Lock(ctx, lockKey)
	if err != nil {
		return nil, err
	}

	// unlock関数を返します。呼び出し元にはこれをdeferで呼んでもらうようにします。
    // 戻り値でクリーンアップ関数を返すことで、コードの意図を伝わりやすくします。
	unlockFunc := func() {
        // mutexをキャプチャします。
		if err := l.redisDriver.UnLock(context.Background(), mutex); err != nil {
            // エラーが起きたらログに残しますが、他の処理に影響がないようにしておきます。
			slog.Warn("unlock failed", "error", err, "uid", uid)
		}
	}

	return unlockFunc, nil
}

呼び出し方

LockWithContextメソッドでクリーンアップ関数が戻ることから、このコードの使用者は意図が把握しやすくなるかなと思います。

func (u *userUsecase) UpdateUserName(ctx context.Context, userID string) (*res.User, error) {

    // 「userID」を渡す => ユーザー単位でロックがかけられる
    unlock, err := u.lockManager.LockWithContext(ctx, userID)
    if err != nil {
        return nil, err
    }
    defer unlock() // deferでクリーンアップ関数を呼び出す

    // クリティカルセクションを実行する
}

きちんとテストをする

排他制御が効いているかのチェックで、APIのエンドポイントに対して、複数リクエストをするというテストもありますが、これだと水平スケールされている本番環境での担保できてません。確認すべきは複数プロセスから実行しても排他制御が効いているかどうかです。

今回のケースではダメな方法

  • go test -race
  • APIをローカルで立ち上げ、ツールで複数リクエストを実行

=> どちらの場合も同一プロセス内のメモリ競合しか検出できない

環境構築の手間がかからず、ローカル環境でサクッとテストできる方法はcmdを用意して呼び出してしまうのが一番簡単かなと思います。

go run cmd/server/main.go # APIを立ち上げる

# cmd/test/main.goにはテストすべき箇所をcontroller層を迂回して呼び出す。
go run cmd/test/main.go &
go run cmd/test/main.go &

考えなくてはならないこと

  • 競合した場合は、「待たせる」か、「409エラーか」の判断
    ロックを獲得できなかった場合は速やかにエラーを返すのか、それとも待機するのか、待つとしたら何秒まで待つのかを考えなくてはなりません。サンプルのコードではデフォルトのままにしていますが、実際は要件に合わせ調整は必要です。
    安全側に振るなら409エラーが良いですが、ユーザー目線としてはあまり印象が良くないかもしれません。この辺りを気にして、実際の開発で、ttlをデフォルトよりも少し長めに設定し、「残高不足」の場合はそれに合うエラーコード（この場合は400）を返すようにしています。

終わりに

以上で終わりです。sync.Mutexは同一プロセス内の並行処理では有効ですが、APIにおけるリクエスト単位での排他制御では悪手となります。「ローカル環境では問題ないのに本番環境では起きる」という事象を回避するためにもここは押さえておきたい大切なポイントです。

GitHubで編集を提案

Discussion