forked from syncthing/syncthing
-
Notifications
You must be signed in to change notification settings - Fork 0
/
leveldb_transactions.go
260 lines (226 loc) · 6.98 KB
/
leveldb_transactions.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
// Copyright (C) 2014 The Syncthing Authors.
//
// This Source Code Form is subject to the terms of the Mozilla Public
// License, v. 2.0. If a copy of the MPL was not distributed with this file,
// You can obtain one at http://mozilla.org/MPL/2.0/.
package db
import (
"bytes"
"sync/atomic"
"github.com/syncthing/syncthing/lib/protocol"
"github.com/syndtr/goleveldb/leveldb"
)
// A readOnlyTransaction represents a database snapshot.
type readOnlyTransaction struct {
*leveldb.Snapshot
db *Instance
}
func (db *Instance) newReadOnlyTransaction() readOnlyTransaction {
snap, err := db.GetSnapshot()
if err != nil {
panic(err)
}
return readOnlyTransaction{
Snapshot: snap,
db: db,
}
}
func (t readOnlyTransaction) close() {
t.Release()
}
func (t readOnlyTransaction) getFile(folder, device, file []byte) (protocol.FileInfo, bool) {
return getFile(t, t.db.deviceKey(folder, device, file))
}
// A readWriteTransaction is a readOnlyTransaction plus a batch for writes.
// The batch will be committed on close() or by checkFlush() if it exceeds the
// batch size.
type readWriteTransaction struct {
readOnlyTransaction
*leveldb.Batch
}
func (db *Instance) newReadWriteTransaction() readWriteTransaction {
t := db.newReadOnlyTransaction()
return readWriteTransaction{
readOnlyTransaction: t,
Batch: new(leveldb.Batch),
}
}
func (t readWriteTransaction) close() {
t.flush()
t.readOnlyTransaction.close()
}
func (t readWriteTransaction) checkFlush() {
if t.Batch.Len() > batchFlushSize {
t.flush()
t.Batch.Reset()
}
}
func (t readWriteTransaction) flush() {
if err := t.db.Write(t.Batch, nil); err != nil {
panic(err)
}
atomic.AddInt64(&t.db.committed, int64(t.Batch.Len()))
}
func (t readWriteTransaction) insertFile(folder, device []byte, file protocol.FileInfo) {
l.Debugf("insert; folder=%q device=%v %v", folder, protocol.DeviceIDFromBytes(device), file)
name := []byte(file.Name)
nk := t.db.deviceKey(folder, device, name)
t.Put(nk, mustMarshal(&file))
}
// updateGlobal adds this device+version to the version list for the given
// file. If the device is already present in the list, the version is updated.
// If the file does not have an entry in the global list, it is created.
func (t readWriteTransaction) updateGlobal(folder, device []byte, file protocol.FileInfo, globalSize *sizeTracker) bool {
l.Debugf("update global; folder=%q device=%v file=%q version=%d", folder, protocol.DeviceIDFromBytes(device), file.Name, file.Version)
name := []byte(file.Name)
gk := t.db.globalKey(folder, name)
svl, err := t.Get(gk, nil)
if err != nil && err != leveldb.ErrNotFound {
panic(err)
}
var fl VersionList
var oldFile protocol.FileInfo
var hasOldFile bool
// Remove the device from the current version list
if len(svl) != 0 {
err = fl.Unmarshal(svl)
if err != nil {
panic(err)
}
for i := range fl.Versions {
if bytes.Equal(fl.Versions[i].Device, device) {
if fl.Versions[i].Version.Equal(file.Version) {
// No need to do anything
return false
}
if i == 0 {
// Keep the current newest file around so we can subtract it from
// the globalSize if we replace it.
oldFile, hasOldFile = t.getFile(folder, fl.Versions[0].Device, name)
}
fl.Versions = append(fl.Versions[:i], fl.Versions[i+1:]...)
break
}
}
}
nv := FileVersion{
Device: device,
Version: file.Version,
}
var insertedAt int
// Find a position in the list to insert this file. The file at the front
// of the list is the newer, the "global".
for i := range fl.Versions {
switch fl.Versions[i].Version.Compare(file.Version) {
case protocol.Equal, protocol.Lesser:
// The version at this point in the list is equal to or lesser
// ("older") than us. We insert ourselves in front of it.
fl.Versions = insertVersion(fl.Versions, i, nv)
insertedAt = i
goto done
case protocol.ConcurrentLesser, protocol.ConcurrentGreater:
// The version at this point is in conflict with us. We must pull
// the actual file metadata to determine who wins. If we win, we
// insert ourselves in front of the loser here. (The "Lesser" and
// "Greater" in the condition above is just based on the device
// IDs in the version vector, which is not the only thing we use
// to determine the winner.)
of, ok := t.getFile(folder, fl.Versions[i].Device, name)
if !ok {
panic("file referenced in version list does not exist")
}
if file.WinsConflict(of) {
fl.Versions = insertVersion(fl.Versions, i, nv)
insertedAt = i
goto done
}
}
}
// We didn't find a position for an insert above, so append to the end.
fl.Versions = append(fl.Versions, nv)
insertedAt = len(fl.Versions) - 1
done:
if insertedAt == 0 {
// We just inserted a new newest version. Fixup the global size
// calculation.
if !file.Version.Equal(oldFile.Version) {
globalSize.addFile(file)
if hasOldFile {
// We have the old file that was removed at the head of the list.
globalSize.removeFile(oldFile)
} else if len(fl.Versions) > 1 {
// The previous newest version is now at index 1, grab it from there.
oldFile, ok := t.getFile(folder, fl.Versions[1].Device, name)
if !ok {
panic("file referenced in version list does not exist")
}
globalSize.removeFile(oldFile)
}
}
}
l.Debugf("new global after update: %v", fl)
t.Put(gk, mustMarshal(&fl))
return true
}
// removeFromGlobal removes the device from the global version list for the
// given file. If the version list is empty after this, the file entry is
// removed entirely.
func (t readWriteTransaction) removeFromGlobal(folder, device, file []byte, globalSize *sizeTracker) {
l.Debugf("remove from global; folder=%q device=%v file=%q", folder, protocol.DeviceIDFromBytes(device), file)
gk := t.db.globalKey(folder, file)
svl, err := t.Get(gk, nil)
if err != nil {
// We might be called to "remove" a global version that doesn't exist
// if the first update for the file is already marked invalid.
return
}
var fl VersionList
err = fl.Unmarshal(svl)
if err != nil {
panic(err)
}
removed := false
for i := range fl.Versions {
if bytes.Equal(fl.Versions[i].Device, device) {
if i == 0 && globalSize != nil {
f, ok := t.getFile(folder, device, file)
if !ok {
panic("removing nonexistent file")
}
globalSize.removeFile(f)
removed = true
}
fl.Versions = append(fl.Versions[:i], fl.Versions[i+1:]...)
break
}
}
if len(fl.Versions) == 0 {
t.Delete(gk)
} else {
l.Debugf("new global after remove: %v", fl)
t.Put(gk, mustMarshal(&fl))
if removed {
f, ok := t.getFile(folder, fl.Versions[0].Device, file)
if !ok {
panic("new global is nonexistent file")
}
globalSize.addFile(f)
}
}
}
func insertVersion(vl []FileVersion, i int, v FileVersion) []FileVersion {
t := append(vl, FileVersion{})
copy(t[i+1:], t[i:])
t[i] = v
return t
}
type marshaller interface {
Marshal() ([]byte, error)
}
func mustMarshal(f marshaller) []byte {
bs, err := f.Marshal()
if err != nil {
panic(err)
}
return bs
}