RDBMSを自作して学ぶ：第11章 ロックと同時実行制御
前回は、トランザクションの開始・コミット・ロールバックを管理するTransactionManagerを実装しました。しかし、テストで確かめたとおり、まだ別のトランザクションが未コミットのデータを読めてしまいます。今回は、複数のトランザクションが同時に動いても互いに干渉しないようにするロックマネージャを実装し、ACIDの「I（分離性）」を実現します。2相ロック（2PL）とデッドロックの検出が中心的なテーマです。
並行実行で起きる異常
ロックなしで複数のトランザクションを同時に動かすと、次のような異常が起きます。
① ダーティリード（Dirty Read）
コミットされていない変更を、別のトランザクションが読んでしまう現象です。
TX1: BEGIN
TX1: UPDATE accounts SET balance = 0 WHERE id = 1 （まだコミットしていない）
TX2: SELECT balance FROM accounts WHERE id = 1 → 0 を読む
TX1: ROLLBACK （TX2が読んだ「0」は存在しなかったことになる）前章のテストで確かめたのは、これのINSERT版です。
② ノンリピータブルリード（Non-Repeatable Read）
同じトランザクションの中で同じ行を2回読んだとき、その間に別のトランザクションがコミットした変更によって、値が変わってしまう現象です。
TX1: SELECT balance FROM accounts WHERE id = 1 → 1000
TX2: UPDATE accounts SET balance = 2000 WHERE id = 1
TX2: COMMIT
TX1: SELECT balance FROM accounts WHERE id = 1 → 2000（さっきと違う）③ ファントムリード（Phantom Read）
同じ条件で2回検索したとき、その間に別のトランザクションが行を挿入（または削除）してコミットしたことで、結果の行数が変わってしまう現象です。新しく現れた行を「幻（ファントム）」と呼びます。
TX1: SELECT COUNT(*) FROM users WHERE age > 20 → 100件
TX2: INSERT INTO users VALUES (999, 'new', 25)
TX2: COMMIT
TX1: SELECT COUNT(*) FROM users WHERE age > 20 → 101件前章で見たとおり、SQL標準の分離レベルは、これらの現象をどこまで許すかで定義されています。SERIALIZABLEでは、いずれも起こりません。
ロックの種類
ロックには、読み取り用と書き込み用の2種類を用意します。
| ロック | 略称 | 用途 |
|---|---|---|
| 共有ロック（Shared Lock） | Sロック | 読み取り |
| 排他ロック（Exclusive Lock） | Xロック | 書き込み |
同じ資源に対して、あるトランザクションがロックを持っているときに、別のトランザクションがロックを取れるかどうかは、次の**両立性（互換性）**の表で決まります。
| 相手がSを保持 | 相手がXを保持 | |
|---|---|---|
| Sを要求 | 取れる | 待つ |
| Xを要求 | 待つ | 待つ |
読み取りどうしは互いに影響しないので、Sロックは何人でも同時に持てます。書き込みは他のすべての操作と衝突するので、Xロックは1人しか持てず、Sロックとも両立しません。
ロックの粒度 ── 行か、テーブルか
ロックをどの単位でかけるかをロックの粒度と呼びます。
- 行ロック: 行（RID）ごとにロックする。関係のない行を操作するトランザクションどうしは待たずに済むので、並行性が高い
- テーブルロック: テーブル全体を1つの単位としてロックする。並行性は低いが、仕組みは単純
行ロックだけでは、ファントムリードを防げないという問題があります。行ロックでロックできるのは「すでに存在する行」だけで、これから挿入される行はロックのしようがないからです。先ほどの例で、TX1が age > 20 の行すべてにSロックをかけても、TX2が新しい行を挿入することは止められません。これを防ぐには、「age > 20 という条件そのもの」をロックする述語ロックや、インデックス上の「行と行の間」をロックするギャップロック（MySQL InnoDBのネクストキーロック）といった仕組みが必要になります。PostgreSQLは、ロックの代わりにトランザクション間の読み書きの依存関係を追跡する方式（SSI: Serializable Snapshot Isolation）でSERIALIZABLEを実現しています。
本シリーズでは実装を単純に保つため、テーブル単位でロックをかけます。
- SELECTは、読むテーブル（JOINするテーブルも含む）にSロックをかける
- INSERTは、書き込むテーブルにXロックをかける
テーブル全体をロックしているので、TX1がusersテーブルを読んでいる間は、TX2はusersに行を挿入できません。ファントムも起こりえません。その代わり、同じテーブルへのINSERTは同時に1つしか実行できず、読み取りと書き込みも互いを待ちます。
実際のデータベースは、テーブルロックと行ロックを組み合わせた多粒度ロックを使います。行をロックする前にテーブルに「この中のどこかの行をロックしている」という意図ロック（ISロック・IXロック）をかけておくことで、テーブル全体のロックと行ロックが衝突するかをすばやく判定できるようにしています。
2相ロック（2PL）
ロックを取ってすぐ放すだけでは足りない
ロックがあっても、使い方を誤ると分離性は守れません。例えば「読む直前にSロックを取り、読み終わったらすぐ放す」という使い方では、2回の読み取りの間に他のトランザクションが書き込めてしまい、ノンリピータブルリードが起きます。
2相ロックのルール
**2相ロック（Two-Phase Locking, 2PL）**は、ロックの取り方と放し方について、次のルールを定めたプロトコルです。
- 成長相（Growing Phase）: ロックを取得していく。この間はロックを放さない
- 縮退相（Shrinking Phase）: ロックを放していく。一度でもロックを放したら、それ以降は新しいロックを取得しない
保持しているロックの数
│ ┌────────┐
│ ┌─┘ └─┐
│ ┌─┘ └─┐
│ ┌─┘ └─┐
└──┴────────────────────┴──▶ 時間
└── 成長相 ──┘└─ 縮退相 ─┘すべてのトランザクションが2PLに従えば、同時に実行した結果は、トランザクションをある順番で1つずつ実行した結果と必ず同じになる（直列化可能になる）ことが知られています。直感的には、各トランザクションが「必要なロックをすべて取り終えた瞬間」の順番に並べると、その順に1つずつ実行したのと同じ結果になる、ということです。
厳格2相ロック（Strict 2PL）
ただし素朴な2PLには、縮退相でXロックを放した直後に、そのトランザクションがロールバックするかもしれない、という問題があります。放されたXロックを取って未コミットのデータを読んだ別のトランザクションも、道連れでロールバックしなければなりません（カスケードアボート）。
そこで実際には、Xロックをコミットまたはロールバックの時点まで放さない「厳格2相ロック（Strict 2PL）」が使われます。本シリーズではさらに単純に、SロックもXロックも、すべてトランザクションの終了時にまとめて放します（これを Strong Strict 2PL や Rigorous 2PL と呼び分けることもあります）。
本シリーズのロックはテーブル単位で、トランザクションが終わるまで放さないので、SERIALIZABLEが実現できます。
デッドロック
デッドロックとは
ロックを使うと、トランザクションどうしが互いに相手のロックの解放を待ち合って、永遠に進めなくなることがあります。これがデッドロックです。
TX1: テーブルAのXロックを持っている → テーブルBのロックを待つ
TX2: テーブルBのXロックを持っている → テーブルAのロックを待つ
→ どちらも永遠に進めない待機グラフによる検出
デッドロックを見つけるには**待機グラフ（Wait-for Graph）を使います。トランザクションをノードにし、「TX1がTX2のロック解放を待っている」ことを TX1 → TX2 という有向辺で表します。このグラフに閉路（サイクル）**があれば、デッドロックです。
TX1 ──待つ──▶ TX2
▲ │
└────待つ──────┘ 閉路 = デッドロック本シリーズでは、あるトランザクションがロックを待つことになったとき、その時点のロックの状態から待機グラフを作り、自分自身から出発して自分に戻ってくる閉路がないかを深さ優先探索で調べます。閉路が見つかったら、ロックを要求したそのトランザクションを犠牲者（ビクティム）に選び、ロックの要求を諦めさせて ErrDeadlock を返します。犠牲者はロールバックしてロックを放し、他のトランザクションを先に進ませます。
犠牲者の選び方には、ほかにも「最も新しく始まったトランザクション（TxIDが最大のもの）」「取り消す作業が最も少ないトランザクション」などがあります。例えばMySQL InnoDBは、挿入・更新・削除した行が少ない、つまりロールバックのコストが小さいトランザクションを犠牲者に選びます。本シリーズでは「閉路を完成させた要求を諦めさせる」のが最も実装しやすいので、この方法を取ります。
なお、デッドロックへの対策には、起きてから検出する方法のほかに、一定時間待ったら諦める（タイムアウト）方法や、TxIDの大小で「待つか・諦めるか」を決めてそもそも閉路ができないようにする方法（wait-die、wound-wait）もあります。
ロックのアップグレード
同じトランザクションが、SロックからXロックへ強めることもあります。例えば、あるテーブルをSELECTした後、同じテーブルにINSERTする場合です。自分しかSロックを持っていなければ、その場でXロックに強められます。
注意が必要なのは、2つのトランザクションが同じテーブルのSロックを持ったまま、両方がXロックに強めようとする場合です。お互いが相手のSロックの解放を待つので、これもデッドロックになります。待機グラフでは TX1 → TX2 → TX1 という閉路として検出されます。
実装：LockManager
// transaction/lock/lock_manager.go
package lock
import (
"errors"
"sync"
"github.com/yourname/go-rdbms/transaction/txid"
)
// Mode はロックの種類
type Mode int
const (
Shared Mode = iota // Sロック（読み取り用）
Exclusive // Xロック（書き込み用）
)
func (m Mode) String() string {
if m == Shared {
return "S"
}
return "X"
}
// ErrDeadlock はデッドロックを検出し、このロック要求を諦めたことを表す
// 受け取った側はトランザクションをロールバックしなければならない
var ErrDeadlock = errors.New("lock: デッドロックを検出しました")
// compatible は2つのロックが同時に成立するかを返す（S同士だけが両立する）
func compatible(a, b Mode) bool {
return a == Shared && b == Shared
}
// resource は1つの資源（本シリーズではテーブル）のロック状態
type resource struct {
holders map[txid.TxID]Mode // ロックを持っているトランザクションとそのモード
}
// waitInfo はロック待ちをしているトランザクションが、何を待っているか
type waitInfo struct {
resource string
mode Mode
}
// LockManager はロックの付与・解放と、デッドロックの検出を行う
type LockManager struct {
mu sync.Mutex
cond *sync.Cond // ロックが解放されたことを待機中のゴルーチンに知らせる
resources map[string]*resource
waiting map[txid.TxID]waitInfo
}
func NewLockManager() *LockManager {
lm := &LockManager{
resources: make(map[string]*resource),
waiting: make(map[txid.TxID]waitInfo),
}
lm.cond = sync.NewCond(&lm.mu)
return lm
}
// Lock はtxに資源nameのロックを与える。すぐに与えられなければ、与えられるまで待つ
// 待っている間にデッドロックになった場合は ErrDeadlock を返す
func (lm *LockManager) Lock(tx txid.TxID, name string, mode Mode) error {
lm.mu.Lock()
defer lm.mu.Unlock()
for {
// 眠っている間にReleaseAllで資源が片付けられていることがあるので、毎回引き直す
r, ok := lm.resources[name]
if !ok {
r = &resource{holders: make(map[txid.TxID]Mode)}
lm.resources[name] = r
}
// 既に同じか強いロックを持っていれば何もしない
if held, ok := r.holders[tx]; ok && (held == Exclusive || mode == Shared) {
delete(lm.waiting, tx)
return nil
}
if lm.grantable(r, tx, mode) {
delete(lm.waiting, tx)
r.holders[tx] = mode // SからXへのアップグレードの場合は上書きになる
return nil
}
lm.waiting[tx] = waitInfo{resource: name, mode: mode}
if lm.hasCycle(tx) {
delete(lm.waiting, tx)
return ErrDeadlock
}
lm.cond.Wait() // 誰かがロックを解放するまで眠る（待機中はlm.muを手放す）
}
}
// grantable はtxにmodeのロックを今すぐ与えられるかを返す
func (lm *LockManager) grantable(r *resource, tx txid.TxID, mode Mode) bool {
for holder, held := range r.holders {
if holder == tx {
continue // 自分が持っているSロックは、Xへのアップグレードを妨げない
}
if !compatible(held, mode) {
return false
}
}
return true
}
// ReleaseAll はtxが持つロックをすべて解放する（コミット・ロールバック時に呼ぶ）
func (lm *LockManager) ReleaseAll(tx txid.TxID) {
lm.mu.Lock()
defer lm.mu.Unlock()
for name, r := range lm.resources {
delete(r.holders, tx)
if len(r.holders) == 0 {
delete(lm.resources, name)
}
}
delete(lm.waiting, tx)
lm.cond.Broadcast() // 待っているゴルーチンをすべて起こし、ロックを取れるか確かめさせる
}
// waitsFor はtxがロック解放を待っている相手（待機グラフの辺の行き先）を返す
func (lm *LockManager) waitsFor(tx txid.TxID) []txid.TxID {
w, ok := lm.waiting[tx]
if !ok {
return nil
}
r, ok := lm.resources[w.resource]
if !ok {
return nil // 資源が片付けられた = 待っている相手はもういない
}
var result []txid.TxID
for holder, held := range r.holders {
if holder != tx && !compatible(held, w.mode) {
result = append(result, holder)
}
}
return result
}
// hasCycle は待機グラフをstartからたどり、startに戻ってくる閉路があるかを調べる
// グラフは現在のロック状態から毎回作り直すので、古い辺が残ることはない
func (lm *LockManager) hasCycle(start txid.TxID) bool {
visited := map[txid.TxID]bool{}
var visit func(tx txid.TxID) bool
visit = func(tx txid.TxID) bool {
for _, next := range lm.waitsFor(tx) {
if next == start {
return true
}
if !visited[next] {
visited[next] = true
if visit(next) {
return true
}
}
}
return false
}
return visit(start)
}実装のポイントを見ていきます。
1つのミューテックスと条件変数で守る
ロックの状態（resources と waiting）はすべて lm.mu で守ります。ロックを取れなかったゴルーチンは sync.Cond の Wait で眠り、誰かが ReleaseAll でロックを放すと Broadcast で全員が起こされて、ロックを取れるかを確かめ直します。Wait は眠っている間 lm.mu を手放すので、他のゴルーチンがロックを放すのを妨げません。
起きたら資源の状態を引き直す
ReleaseAll は、ロックの保持者がいなくなった資源を resources から取り除きます。そのため、眠る前に取得した resource へのポインタを起きた後も使い続けると、すでに取り除かれた古いエントリにロックを登録してしまいます。その後に同じ資源を要求した別のトランザクションは新しいエントリを作るので、両立しないはずのロックが2つ同時に成立してしまいます。これを避けるため、ループのたびに lm.resources[name] から引き直しています。
待機グラフはその都度作る
待機グラフを別のデータとして持ち続けると、ロックが取れて待ちが終わったときや、トランザクションが終わったときに、辺を消し忘れる危険があります。古い辺が残っていると、実際には存在しない閉路を見つけて、デッドロックではないのにトランザクションを犠牲にしてしまいます。本実装では、waiting（誰が何を待っているか）と resources（誰が何を持っているか）から、waitsFor でその場で辺を求めます。
ロックを取れるまでの判定を、起きるたびにやり直す
眠っている間に状況は変わります。ほかのトランザクションがロックを取った結果、新しい閉路ができることもあるので、デッドロックの確認も起きるたびに行います。
書き込み側が待たされ続けることがある
この実装は「今すぐ両立するならロックを与える」だけで、待っている順番を考慮しません。Sロックを持つトランザクションが次々に入れ替わり続けると、Xロックを待つトランザクションがいつまでも待たされる可能性があります（スタベーション）。実際のロックマネージャは、要求を待ち行列に並べて順番を守るなどの工夫をしています。
TransactionManagerとExecutorへの組み込み
TransactionManager
TransactionManagerにロックマネージャを持たせ、コミットとロールバックの最後にロックをまとめて解放します。ロールバックでは、取り消し操作が終わるまでXロックを持ち続ける必要があります。
@@ -8,4 +8,5 @@ import (
"github.com/yourname/go-rdbms/storage/btree"
+ "github.com/yourname/go-rdbms/transaction/lock"
"github.com/yourname/go-rdbms/transaction/txid"
)
@@ -13,13 +14,15 @@ import (
// TransactionManager はトランザクションの開始・コミット・ロールバックを管理する
type TransactionManager struct {
- mu sync.Mutex
- nextID txid.TxID
- active map[txid.TxID]*Transaction
+ mu sync.Mutex
+ nextID txid.TxID
+ active map[txid.TxID]*Transaction
+ lockMgr *lock.LockManager
}
-func NewTransactionManager() *TransactionManager {
+func NewTransactionManager(lockMgr *lock.LockManager) *TransactionManager {
return &TransactionManager{
- nextID: 1,
- active: make(map[txid.TxID]*Transaction),
+ nextID: 1,
+ active: make(map[txid.TxID]*Transaction),
+ lockMgr: lockMgr,
}
}
@@ -42,4 +45,7 @@ func (tm *TransactionManager) Commit(tx *Transaction) error {
}
// 第12章でここにCOMMITログの書き込みとfsyncが入る
+
+ // 厳格2PL: ロックはトランザクションの終了時にまとめて解放する
+ tm.lockMgr.ReleaseAll(tx.id)
return nil
}
@@ -64,5 +70,10 @@ func (tm *TransactionManager) Rollback(tx *Transaction) error {
}
}
- return tm.finish(tx, StatusAborted)
+ if err := tm.finish(tx, StatusAborted); err != nil {
+ return err
+ }
+ // 取り消しが終わるまではXロックを持ち続け、最後に解放する
+ tm.lockMgr.ReleaseAll(tx.id)
+ return nil
}
Executor
Executorは、テーブルを読む前にSロックを、書き込む前にXロックを取ります。ロックを待つ間は、ページのピンやヒープファイルのミューテックスを一切持っていない状態にしておくことが大切です。何かを持ったまま眠ると、そのせいで他のトランザクションが進めず、待機グラフに現れないデッドロックが起きてしまいます。
@@ -9,4 +9,5 @@ import (
"github.com/yourname/go-rdbms/sql/planner"
"github.com/yourname/go-rdbms/transaction"
+ "github.com/yourname/go-rdbms/transaction/lock"
)
@@ -23,8 +24,9 @@ type Executor struct {
catalog *catalog.Catalog
planner *planner.Planner
+ lockMgr *lock.LockManager
}
-func New(cat *catalog.Catalog) *Executor {
- return &Executor{catalog: cat, planner: planner.New(cat)}
+func New(cat *catalog.Catalog, lockMgr *lock.LockManager) *Executor {
+ return &Executor{catalog: cat, planner: planner.New(cat), lockMgr: lockMgr}
}
@@ -33,5 +35,5 @@ func (e *Executor) Execute(tx *transaction.Transaction, stmt parser.Statement) (
switch s := stmt.(type) {
case *parser.SelectStatement:
- return e.executeSelect(s)
+ return e.executeSelect(tx, s)
case *parser.InsertStatement:
return e.executeInsert(tx, s)
@@ -45,5 +47,16 @@ func (e *Executor) Execute(tx *transaction.Transaction, stmt parser.Statement) (
// ---- SELECT ----
-func (e *Executor) executeSelect(stmt *parser.SelectStatement) (*Result, error) {
+func (e *Executor) executeSelect(tx *transaction.Transaction, stmt *parser.SelectStatement) (*Result, error) {
+ // 読むテーブルすべてにSロックをかける（コミットまで保持する）
+ tables := []string{stmt.From}
+ for _, j := range stmt.Joins {
+ tables = append(tables, j.Table)
+ }
+ for _, name := range tables {
+ if err := e.lockMgr.Lock(tx.ID(), name, lock.Shared); err != nil {
+ return nil, err
+ }
+ }
+
plan, err := e.planner.PlanSelect(stmt)
if err != nil {
@@ -121,4 +134,8 @@ func (e *Executor) executeInsert(tx *transaction.Transaction, stmt *parser.Inser
return nil, err
}
+ // 書き込むテーブルにXロックをかける（コミットまで保持する）
+ if err := e.lockMgr.Lock(tx.ID(), table.Name, lock.Exclusive); err != nil {
+ return nil, err
+ }
values, err := insertValues(stmt, table.Schema)
if err != nil {ErrDeadlock を受け取った呼び出し側は、そのトランザクションをロールバックしなければなりません。この処理は第13章のセッションで実装します。
テストを書く
ロックマネージャ単体のテストでは、別のゴルーチンでロックを要求し、「一定時間待っても返ってこない（待たされている）」「解放したら返ってくる」ことを確かめます。
// transaction/lock/lock_manager_test.go
package lock_test
import (
"errors"
"sync"
"testing"
"time"
"github.com/yourname/go-rdbms/transaction/lock"
"github.com/yourname/go-rdbms/transaction/txid"
)
// lockAsync は別のゴルーチンでロックを要求し、結果を返すチャネルを返す
func lockAsync(lm *lock.LockManager, tx txid.TxID, name string, mode lock.Mode) <-chan error {
done := make(chan error, 1)
go func() { done <- lm.Lock(tx, name, mode) }()
return done
}
// assertBlocked はロック要求がまだ待たされていることを確かめる
func assertBlocked(t *testing.T, done <-chan error) {
t.Helper()
select {
case err := <-done:
t.Fatalf("待たされるはずのロックが返ってきた: err=%v", err)
case <-time.After(100 * time.Millisecond):
}
}
// assertGranted はロック要求が成功して返ってくることを確かめる
func assertGranted(t *testing.T, done <-chan error) {
t.Helper()
select {
case err := <-done:
if err != nil {
t.Fatalf("ロックが取れるはず: %v", err)
}
case <-time.After(time.Second):
t.Fatal("ロックが取れないまま待たされている")
}
}
func TestLockManager_SharedLocksCoexist(t *testing.T) {
lm := lock.NewLockManager()
var wg sync.WaitGroup
for i := 1; i <= 5; i++ {
wg.Add(1)
go func(tx txid.TxID) {
defer wg.Done()
if err := lm.Lock(tx, "USERS", lock.Shared); err != nil {
t.Errorf("TX %d: Sロックの取得に失敗: %v", tx, err)
}
}(txid.TxID(i))
}
wg.Wait() // 誰も待たされずに全員がSロックを取れる
}
func TestLockManager_ExclusiveBlocks(t *testing.T) {
lm := lock.NewLockManager()
if err := lm.Lock(1, "USERS", lock.Exclusive); err != nil {
t.Fatal(err)
}
done := lockAsync(lm, 2, "USERS", lock.Shared)
assertBlocked(t, done) // TX1のXロックがあるのでTX2は待たされる
lm.ReleaseAll(1)
assertGranted(t, done) // TX1が解放するとTX2が進める
}
func TestLockManager_Upgrade(t *testing.T) {
lm := lock.NewLockManager()
// 自分しかSロックを持っていなければ、そのままXにアップグレードできる
lm.Lock(1, "USERS", lock.Shared)
if err := lm.Lock(1, "USERS", lock.Exclusive); err != nil {
t.Fatalf("アップグレードに失敗: %v", err)
}
// Xを持っている間は他のTXのSロックは待たされる
done := lockAsync(lm, 2, "USERS", lock.Shared)
assertBlocked(t, done)
lm.ReleaseAll(1)
assertGranted(t, done)
}
func TestLockManager_Deadlock(t *testing.T) {
lm := lock.NewLockManager()
lm.Lock(1, "A", lock.Exclusive)
lm.Lock(2, "B", lock.Exclusive)
// TX1 は B を待つ（TX1 → TX2）
done1 := lockAsync(lm, 1, "B", lock.Exclusive)
assertBlocked(t, done1)
// TX2 が A を要求すると TX2 → TX1 の辺ができて閉路になる
if err := lm.Lock(2, "A", lock.Exclusive); !errors.Is(err, lock.ErrDeadlock) {
t.Fatalf("デッドロックが検出されるはず: got %v", err)
}
// TX2 がロールバックしてロックを解放すると、TX1 が進める
lm.ReleaseAll(2)
assertGranted(t, done1)
}
func TestLockManager_UpgradeDeadlock(t *testing.T) {
lm := lock.NewLockManager()
// 2つのTXが同じ資源のSロックを持ったまま、両方ともXにアップグレードしようとする
lm.Lock(1, "USERS", lock.Shared)
lm.Lock(2, "USERS", lock.Shared)
done1 := lockAsync(lm, 1, "USERS", lock.Exclusive)
assertBlocked(t, done1) // TX2のSロックがあるので待つ
if err := lm.Lock(2, "USERS", lock.Exclusive); !errors.Is(err, lock.ErrDeadlock) {
t.Fatalf("デッドロックが検出されるはず: got %v", err)
}
lm.ReleaseAll(2)
assertGranted(t, done1)
}
func TestLockManager_WaiterKeepsExclusion(t *testing.T) {
lm := lock.NewLockManager()
// TX2 が待っている間に TX1 が解放し、TX2 がSロックを得る
lm.Lock(1, "USERS", lock.Exclusive)
done2 := lockAsync(lm, 2, "USERS", lock.Shared)
assertBlocked(t, done2)
lm.ReleaseAll(1)
assertGranted(t, done2)
// TX2 がSロックを持っているので、TX3 のXロックは待たされなければならない
done3 := lockAsync(lm, 3, "USERS", lock.Exclusive)
assertBlocked(t, done3)
lm.ReleaseAll(2)
assertGranted(t, done3)
}トランザクションのテストでは、前章のダーティリードのテストを、「未コミットのデータを読もうとすると待たされ、ロールバック後には0件が返る」テストに置き換えます。SQLレベルでのデッドロックもテストします。
// transaction/transaction_manager_test.go（追加・変更部分）
func TestTransaction_NoDirtyRead(t *testing.T) {
e := setup(t)
tx := e.tm.Begin()
e.exec(tx, "CREATE TABLE accounts (id INT PRIMARY KEY, owner VARCHAR(10))")
e.tm.Commit(tx)
// TX1がINSERTしてまだコミットしていない（ITEMSのXロックを持っている）
tx1 := e.tm.Begin()
e.exec(tx1, "INSERT INTO items VALUES (1, 'apple')")
// 別のトランザクションがITEMSを読もうとすると、TX1が終わるまで待たされる
done := make(chan int, 1)
go func() { done <- e.count() }()
select {
case n := <-done:
t.Fatalf("TX1の終了前に読めてしまった（%d件）", n)
case <-time.After(100 * time.Millisecond):
}
// 関係ないテーブルは待たずに読める
tx3 := e.tm.Begin()
if _, err := e.exec(tx3, "SELECT * FROM accounts"); err != nil {
t.Fatal(err)
}
e.tm.Commit(tx3)
// TX1がロールバックすると、待っていた読み取りは「0件」を返す
e.tm.Rollback(tx1)
select {
case n := <-done:
if n != 0 {
t.Errorf("ロールバックされた行が見えている: %d件", n)
}
case <-time.After(time.Second):
t.Fatal("TX1の終了後も待たされている")
}
}
func TestTransaction_DeadlockVictimRollsBack(t *testing.T) {
e := setup(t)
tx := e.tm.Begin()
e.exec(tx, "CREATE TABLE accounts (id INT PRIMARY KEY, owner VARCHAR(10))")
e.tm.Commit(tx)
tx1, tx2 := e.tm.Begin(), e.tm.Begin()
e.exec(tx1, "SELECT * FROM items") // TX1: ITEMS の S
e.exec(tx2, "SELECT * FROM accounts") // TX2: ACCOUNTS の S
// TX1 は ACCOUNTS に書こうとして TX2 を待つ
done := make(chan error, 1)
go func() {
_, err := e.exec(tx1, "INSERT INTO accounts VALUES (1, 'alice')")
done <- err
}()
time.Sleep(100 * time.Millisecond)
// TX2 が ITEMS に書こうとすると閉路ができ、TX2 の要求がデッドロックとして失敗する
_, err := e.exec(tx2, "INSERT INTO items VALUES (1, 'apple')")
if !errors.Is(err, lock.ErrDeadlock) {
t.Fatalf("デッドロックになるはず: got %v", err)
}
// デッドロックの犠牲になったTX2をロールバックすると、TX1が先へ進める
e.tm.Rollback(tx2)
if err := <-done; err != nil {
t.Fatalf("TX1のINSERTが失敗した: %v", err)
}
if err := e.tm.Commit(tx1); err != nil {
t.Fatal(err)
}
}go test ./transaction/... -v期待される出力（=== RUN の行は省略）：
--- PASS: TestTransaction_Commit (0.02s)
--- PASS: TestTransaction_Rollback (0.01s)
--- PASS: TestTransaction_NoDirtyRead (0.12s)
--- PASS: TestTransaction_DeadlockVictimRollsBack (0.12s)
PASS
ok github.com/yourname/go-rdbms/transaction 0.667s
--- PASS: TestLockManager_SharedLocksCoexist (0.00s)
--- PASS: TestLockManager_ExclusiveBlocks (0.10s)
--- PASS: TestLockManager_Upgrade (0.10s)
--- PASS: TestLockManager_Deadlock (0.10s)
--- PASS: TestLockManager_UpgradeDeadlock (0.10s)
--- PASS: TestLockManager_WaiterKeepsExclusion (0.20s)
PASS
ok github.com/yourname/go-rdbms/transaction/lock 0.818s並行処理のテストはタイミングに依存しやすいので、go test -race -count=10 ./transaction/... のように、データ競合の検出を有効にして何度か繰り返し実行しておくと安心です。
ここまでのコード全体像
go-rdbms/
├── storage/
├── sql/
│ └── executor/
│ └── executor.go ← テーブルにS/Xロックをかける（本章）
└── transaction/
├── txid/
├── lock/
│ └── lock_manager.go ← ロックマネージャ・デッドロック検出（本章）
├── transaction.go
└── transaction_manager.go ← 終了時にロックを解放（本章）ロックマネージャの設計で意識したポイント
① テーブル単位のロックで、確実にSERIALIZABLEにする
行ロックでは防げないファントムも、テーブル全体をロックすれば起こりません。並行性より正しさと単純さを優先しました。
② ロックはトランザクションの終了時にまとめて放す
厳格2PLにすることで、直列化可能性に加えて、未コミットのデータを他のトランザクションが読むこと（とカスケードアボート）を防いでいます。
③ 待機グラフを持ち続けず、その都度作る
グラフを現在のロック状態から毎回作るので、古い辺による誤検出が起きません。
④ 眠った後は、すべてを確かめ直す
条件変数で起こされた後は、資源の引き直し、ロックの可否、デッドロックの有無を毎回判定し直します。「眠る前の前提は、起きた後には崩れているかもしれない」というのが、並行処理の基本です。
まとめ
- ロックなしの並行実行では、ダーティリード・ノンリピータブルリード・ファントムリードが起きる
- Sロックどうしは両立し、Xロックは他のどのロックとも両立しない
- 行ロックだけではファントムを防げない。本シリーズはテーブル単位でロックをかけてSERIALIZABLEを実現した
- 2相ロックに従えば直列化可能になる。本シリーズでは、すべてのロックを終了時にまとめて放す厳格2PLを採用した
- デッドロックは待機グラフの閉路として検出し、閉路を完成させた要求を犠牲にして
ErrDeadlockを返す
次章では、クラッシュや電源断からデータを守るWAL（Write-Ahead Logging）と、再起動時のリカバリを実装します。ACIDの残り、原子性（A）と永続性（D）を、クラッシュに対しても保証できるようにします。