1// SPDX-License-Identifier: MIT
2
3package kv
4
5import (
6 "fmt"
7 "io"
8 "time"
9)
10
11// maxGroupBytes caps the encoded payload of one commit group. A single batch
12// larger than this forms a group of one.
13const maxGroupBytes = 4 << 20
14
15// request is a batch waiting for the writer. The writer answers on reply
16// exactly once. Requests are allocated per Apply and never pooled: a pooled
17// channel would be shared across testing/synctest bubbles, and receiving on a
18// bubbled channel from outside its bubble is fatal.
19type request struct {
20 b *Batch
21 reply chan result // capacity 1
22}
23
24type result struct {
25 seq uint64
26 err error
27}
28
29// writer is the single goroutine that owns the log file and builds the new
30// state. It exits on Close or when the database fails.
31func (db *DB) writer() {
32 defer close(db.done)
33 var pending *request // dequeued but did not fit in the previous group
34 defer func() {
35 // A request taken but never grouped must still be answered.
36 if pending != nil {
37 err := db.stopped()
38 if err == nil {
39 err = ErrClosed // not reachable: the writer only exits on close or failure
40 }
41 pending.reply <- result{err: err}
42 }
43 }()
44 for {
45 // Closing takes priority over queued requests.
46 select {
47 case <-db.closing:
48 return
49 default:
50 }
51 first := pending
52 pending = nil
53 if first == nil {
54 select {
55 case <-db.closing:
56 return
57 case first = <-db.reqCh:
58 }
59 }
60 pending = db.collect(first)
61 if !db.commit() {
62 return
63 }
64 if !db.maybeCompact() {
65 return
66 }
67 }
68}
69
70// collect fills db.group starting with first: it drains the queued requests
71// and, with GroupWindow > 0, waits for more until the window expires or the
72// database is closing. It returns a request that was taken but did not fit
73// under maxGroupBytes.
74func (db *DB) collect(first *request) (overflow *request) {
75 db.group = append(db.group[:0], first)
76 size := recordHead + first.b.encodedSize()
77 var expired <-chan time.Time
78 if w := db.opts.GroupWindow; w > 0 {
79 t := time.NewTimer(w)
80 defer t.Stop()
81 expired = t.C
82 }
83 for {
84 var r *request
85 if expired == nil {
86 select {
87 case r = <-db.reqCh:
88 default:
89 return nil
90 }
91 } else {
92 select {
93 case r = <-db.reqCh:
94 case <-expired:
95 return nil
96 case <-db.closing:
97 return nil
98 }
99 }
100 if p := r.b.encodedSize(); size+p <= maxGroupBytes {
101 db.group = append(db.group, r)
102 size += p
103 } else {
104 return r
105 }
106 }
107}
108
109// commit applies db.group to a fresh builder, writes and syncs the records,
110// publishes the new state and answers every request. It reports false if the
111// database failed.
112func (db *DB) commit() bool {
113 defer clear(db.group)
114 cur := db.snap.Load()
115 bld := cur.t.builder()
116 seq := cur.seq
117 buf := db.buf[:0]
118 results := make([]result, len(db.group))
119 var batches int
120 var firstSeq uint64
121
122 for i, r := range db.group {
123 b := r.b
124 if !checkPreconditions(bld, b.pre) {
125 results[i] = result{err: ErrConflict}
126 continue
127 }
128 if len(b.ops) == 0 {
129 results[i] = result{seq: seq}
130 continue
131 }
132 seq++
133 for _, o := range b.ops {
134 if o.kind == opPut {
135 bld.put(o.key, o.value, seq)
136 } else {
137 bld.delete(o.key)
138 }
139 }
140 if batches == 0 {
141 firstSeq = seq
142 buf = beginRecord(buf, seq)
143 }
144 buf = appendBatch(buf, b.ops)
145 batches++
146 results[i] = result{seq: seq}
147 }
148 db.buf = buf[:0]
149 if cap(db.buf) > 2*maxGroupBytes {
150 db.buf = nil // do not pin the memory of one huge batch
151 }
152
153 if batches > 0 {
154 finishRecord(buf, batches)
155 syncTime, err := db.persist(buf)
156 if err != nil {
157 err = db.fail(err)
158 for _, r := range db.group {
159 r.reply <- result{err: err}
160 }
161 return false
162 }
163 db.logSize += int64(len(buf))
164 db.snap.Store(newSnapshot(bld.freeze(), seq))
165 db.log.Debug().Uint64("seq", firstSeq).Int("batches", batches).Int("bytes", len(buf)).
166 Dur("sync", syncTime).Msg("group committed")
167 }
168 for i, r := range db.group {
169 r.reply <- results[i]
170 }
171 return true
172}
173
174// persist appends buf to the log and syncs it (unless NoSync). It returns
175// the time spent in the sync.
176func (db *DB) persist(buf []byte) (time.Duration, error) {
177 n, err := db.file.Write(buf)
178 if err == nil && n < len(buf) {
179 err = io.ErrShortWrite
180 }
181 if err != nil {
182 return 0, fmt.Errorf("write log: %w", err)
183 }
184 if db.opts.NoSync {
185 return 0, nil
186 }
187 start := time.Now()
188 if err := db.file.Sync(); err != nil {
189 return 0, fmt.Errorf("sync log: %w", err)
190 }
191 return time.Since(start), nil
192}
193
194// fail puts the database in the failed state, keeping the first cause, and
195// returns the error that requests receive.
196func (db *DB) fail(cause error) error {
197 db.failOnce.Do(func() {
198 db.failErr = fmt.Errorf("%w: %w", ErrFailed, cause)
199 db.log.Error().Err(cause).Msg("database failed")
200 close(db.failed)
201 })
202 return db.failErr
203}
204
205// checkPreconditions reports whether all preconditions hold on bld.
206func checkPreconditions(bld *builder, pre []precond) bool {
207 for _, p := range pre {
208 item, ok := bld.get(p.key)
209 if p.absent && ok || !p.absent && (!ok || item.Version != p.version) {
210 return false
211 }
212 }
213 return true
214}