|
| 1 | +/* |
| 2 | + Copyright 2025 GitHub Inc. |
| 3 | + See https://github.com/github/gh-ost/blob/master/LICENSE |
| 4 | +*/ |
| 5 | + |
| 6 | +package logic |
| 7 | + |
| 8 | +import ( |
| 9 | + "context" |
| 10 | + "sync" |
| 11 | + "sync/atomic" |
| 12 | +) |
| 13 | + |
| 14 | +const seqUninit int64 = 0 |
| 15 | + |
| 16 | +// commitBarrier implements MTS LOGICAL_CLOCK dependency tracking using a |
| 17 | +// gap-free Low Water Mark (LWM), mirroring MySQL 8.0's GAQ scheduling. |
| 18 | +// |
| 19 | +// LWM invariant: all transactions with sequence_number <= lwm have completed. |
| 20 | +// This is equivalent to MySQL's find_lwm() + move_queue_head() in rpl_rli_pdb.cc. |
| 21 | +// |
| 22 | +// Reference: MySQL 8.0 sql/rpl_mta_submode.cc |
| 23 | +// - waitForDependency corresponds to wait_for_last_committed_trx() |
| 24 | +// - commit corresponds to Worker commit + GAQ LWM advancement |
| 25 | +// - waitForAllWorkers corresponds to wait_for_workers_to_finish() |
| 26 | +type commitBarrier struct { |
| 27 | + mu sync.Mutex |
| 28 | + lwm int64 // gap-free low water mark: all seq <= lwm are complete |
| 29 | + pending map[int64]bool // sequences > lwm that committed but aren't consecutive yet |
| 30 | + delegatedJobs atomic.Int64 // jobs dispatched but not yet completed |
| 31 | + cond *sync.Cond |
| 32 | +} |
| 33 | + |
| 34 | +func newCommitBarrier() *commitBarrier { |
| 35 | + cb := &commitBarrier{ |
| 36 | + lwm: seqUninit, |
| 37 | + pending: make(map[int64]bool), |
| 38 | + } |
| 39 | + cb.cond = sync.NewCond(&cb.mu) |
| 40 | + return cb |
| 41 | +} |
| 42 | + |
| 43 | +// clockLeq implements MySQL's clock_leq: SEQ_UNINIT (0) is treated as the |
| 44 | +// minimum value in the clock domain. |
| 45 | +func clockLeq(a, b int64) bool { |
| 46 | + if a == seqUninit { |
| 47 | + return true |
| 48 | + } |
| 49 | + if b == seqUninit { |
| 50 | + return false |
| 51 | + } |
| 52 | + return a <= b |
| 53 | +} |
| 54 | + |
| 55 | +// waitForDependency blocks until lwm >= lastCommitted. |
| 56 | +// |
| 57 | +// This matches MySQL's wait_for_last_committed_trx() which waits until the |
| 58 | +// low water mark advances past the parent transaction: |
| 59 | +// |
| 60 | +// while (!clock_leq(last_committed_arg, estimate_lwm_timestamp())) |
| 61 | +// wait(logical_clock_cond) |
| 62 | +// |
| 63 | +// Cross-table dependencies (parentSeenOnStream == false) are treated as |
| 64 | +// satisfied, matching MySQL's SEQ_UNINIT handling where undefined parents |
| 65 | +// don't block scheduling. |
| 66 | +func (cb *commitBarrier) waitForDependency(ctx context.Context, lastCommitted int64, parentSeenOnStream bool) { |
| 67 | + if lastCommitted == seqUninit || !parentSeenOnStream { |
| 68 | + return |
| 69 | + } |
| 70 | + cb.mu.Lock() |
| 71 | + defer cb.mu.Unlock() |
| 72 | + for !clockLeq(lastCommitted, cb.lwm) { |
| 73 | + if ctx.Err() != nil { |
| 74 | + return |
| 75 | + } |
| 76 | + cb.cond.Wait() |
| 77 | + } |
| 78 | +} |
| 79 | + |
| 80 | +// commit records a transaction as complete and advances the LWM |
| 81 | +// as far as possible through consecutive completed sequences. |
| 82 | +// |
| 83 | +// This matches MySQL's move_queue_head() which dequeues jobs from |
| 84 | +// the GAQ head only while they are consecutively done: |
| 85 | +// |
| 86 | +// while (!empty()) { |
| 87 | +// if (ptr_g->done == 0) break; // gap — stop advancing |
| 88 | +// de_queue(&g); // remove from queue |
| 89 | +// lwm = g; // advance LWM |
| 90 | +// } |
| 91 | +func (cb *commitBarrier) commit(sequenceNumber int64) { |
| 92 | + if sequenceNumber == seqUninit { |
| 93 | + return |
| 94 | + } |
| 95 | + cb.mu.Lock() |
| 96 | + cb.pending[sequenceNumber] = true |
| 97 | + // Advance LWM through consecutive committed sequences |
| 98 | + for cb.pending[cb.lwm+1] { |
| 99 | + delete(cb.pending, cb.lwm+1) |
| 100 | + cb.lwm++ |
| 101 | + } |
| 102 | + cb.mu.Unlock() |
| 103 | + cb.cond.Broadcast() |
| 104 | +} |
| 105 | + |
| 106 | +// waitForAllWorkers blocks until all delegated jobs have completed (delegatedJobs == 0). |
| 107 | +// |
| 108 | +// Corresponds to MySQL: wait_for_workers_to_finish() |
| 109 | +func (cb *commitBarrier) waitForAllWorkers(ctx context.Context) { |
| 110 | + cb.mu.Lock() |
| 111 | + defer cb.mu.Unlock() |
| 112 | + for cb.delegatedJobs.Load() > 0 { |
| 113 | + if ctx.Err() != nil { |
| 114 | + return |
| 115 | + } |
| 116 | + cb.cond.Wait() |
| 117 | + } |
| 118 | +} |
| 119 | + |
| 120 | +// addDelegatedJob increments the delegated job counter. |
| 121 | +func (cb *commitBarrier) addDelegatedJob() { |
| 122 | + cb.delegatedJobs.Add(1) |
| 123 | +} |
| 124 | + |
| 125 | +// completeDelegatedJob decrements the delegated job counter and broadcasts wake signal. |
| 126 | +func (cb *commitBarrier) completeDelegatedJob() { |
| 127 | + cb.delegatedJobs.Add(-1) |
| 128 | + cb.cond.Broadcast() |
| 129 | +} |
| 130 | + |
| 131 | +// getLWM returns the current gap-free low water mark (thread-safe). |
| 132 | +func (cb *commitBarrier) getLWM() int64 { |
| 133 | + cb.mu.Lock() |
| 134 | + defer cb.mu.Unlock() |
| 135 | + return cb.lwm |
| 136 | +} |
0 commit comments