1
0

leveldb.go 26 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034
  1. // Copyright (C) 2014 The Syncthing Authors.
  2. //
  3. // This Source Code Form is subject to the terms of the Mozilla Public
  4. // License, v. 2.0. If a copy of the MPL was not distributed with this file,
  5. // You can obtain one at http://mozilla.org/MPL/2.0/.
  6. //go:generate -command genxdr go run ../../Godeps/_workspace/src/github.com/calmh/xdr/cmd/genxdr/main.go
  7. //go:generate genxdr -o leveldb_xdr.go leveldb.go
  8. package db
  9. import (
  10. "bytes"
  11. "fmt"
  12. "runtime"
  13. "sort"
  14. "github.com/syncthing/syncthing/lib/protocol"
  15. "github.com/syncthing/syncthing/lib/sync"
  16. "github.com/syndtr/goleveldb/leveldb"
  17. "github.com/syndtr/goleveldb/leveldb/iterator"
  18. "github.com/syndtr/goleveldb/leveldb/opt"
  19. "github.com/syndtr/goleveldb/leveldb/util"
  20. )
  21. var (
  22. clockTick int64
  23. clockMut = sync.NewMutex()
  24. )
  25. func clock(v int64) int64 {
  26. clockMut.Lock()
  27. defer clockMut.Unlock()
  28. if v > clockTick {
  29. clockTick = v + 1
  30. } else {
  31. clockTick++
  32. }
  33. return clockTick
  34. }
  35. const (
  36. KeyTypeDevice = iota
  37. KeyTypeGlobal
  38. KeyTypeBlock
  39. KeyTypeDeviceStatistic
  40. KeyTypeFolderStatistic
  41. KeyTypeVirtualMtime
  42. )
  43. type fileVersion struct {
  44. version protocol.Vector
  45. device []byte
  46. }
  47. type versionList struct {
  48. versions []fileVersion
  49. }
  50. func (l versionList) String() string {
  51. var b bytes.Buffer
  52. var id protocol.DeviceID
  53. b.WriteString("{")
  54. for i, v := range l.versions {
  55. if i > 0 {
  56. b.WriteString(", ")
  57. }
  58. copy(id[:], v.device)
  59. fmt.Fprintf(&b, "{%d, %v}", v.version, id)
  60. }
  61. b.WriteString("}")
  62. return b.String()
  63. }
  64. type fileList []protocol.FileInfo
  65. func (l fileList) Len() int {
  66. return len(l)
  67. }
  68. func (l fileList) Swap(a, b int) {
  69. l[a], l[b] = l[b], l[a]
  70. }
  71. func (l fileList) Less(a, b int) bool {
  72. return l[a].Name < l[b].Name
  73. }
  74. type dbReader interface {
  75. Get([]byte, *opt.ReadOptions) ([]byte, error)
  76. }
  77. type dbWriter interface {
  78. Put([]byte, []byte)
  79. Delete([]byte)
  80. }
  81. // Flush batches to disk when they contain this many records.
  82. const batchFlushSize = 64
  83. // deviceKey returns a byte slice encoding the following information:
  84. // keyTypeDevice (1 byte)
  85. // folder (64 bytes)
  86. // device (32 bytes)
  87. // name (variable size)
  88. func deviceKey(folder, device, file []byte) []byte {
  89. return deviceKeyInto(nil, folder, device, file)
  90. }
  91. func deviceKeyInto(k []byte, folder, device, file []byte) []byte {
  92. reqLen := 1 + 64 + 32 + len(file)
  93. if len(k) < reqLen {
  94. k = make([]byte, reqLen)
  95. }
  96. k[0] = KeyTypeDevice
  97. if len(folder) > 64 {
  98. panic("folder name too long")
  99. }
  100. copy(k[1:], []byte(folder))
  101. copy(k[1+64:], device[:])
  102. copy(k[1+64+32:], []byte(file))
  103. return k[:reqLen]
  104. }
  105. func deviceKeyName(key []byte) []byte {
  106. return key[1+64+32:]
  107. }
  108. func deviceKeyFolder(key []byte) []byte {
  109. folder := key[1 : 1+64]
  110. izero := bytes.IndexByte(folder, 0)
  111. if izero < 0 {
  112. return folder
  113. }
  114. return folder[:izero]
  115. }
  116. func deviceKeyDevice(key []byte) []byte {
  117. return key[1+64 : 1+64+32]
  118. }
  119. // globalKey returns a byte slice encoding the following information:
  120. // keyTypeGlobal (1 byte)
  121. // folder (64 bytes)
  122. // name (variable size)
  123. func globalKey(folder, file []byte) []byte {
  124. k := make([]byte, 1+64+len(file))
  125. k[0] = KeyTypeGlobal
  126. if len(folder) > 64 {
  127. panic("folder name too long")
  128. }
  129. copy(k[1:], []byte(folder))
  130. copy(k[1+64:], []byte(file))
  131. return k
  132. }
  133. func globalKeyName(key []byte) []byte {
  134. return key[1+64:]
  135. }
  136. func globalKeyFolder(key []byte) []byte {
  137. folder := key[1 : 1+64]
  138. izero := bytes.IndexByte(folder, 0)
  139. if izero < 0 {
  140. return folder
  141. }
  142. return folder[:izero]
  143. }
  144. type deletionHandler func(db dbReader, batch dbWriter, folder, device, name []byte, dbi iterator.Iterator) int64
  145. func ldbGenericReplace(db *leveldb.DB, folder, device []byte, fs []protocol.FileInfo, localSize, globalSize *sizeTracker, deleteFn deletionHandler) int64 {
  146. runtime.GC()
  147. sort.Sort(fileList(fs)) // sort list on name, same as in the database
  148. start := deviceKey(folder, device, nil) // before all folder/device files
  149. limit := deviceKey(folder, device, []byte{0xff, 0xff, 0xff, 0xff}) // after all folder/device files
  150. batch := new(leveldb.Batch)
  151. l.Debugf("new batch %p", batch)
  152. snap, err := db.GetSnapshot()
  153. if err != nil {
  154. panic(err)
  155. }
  156. l.Debugf("created snapshot %p", snap)
  157. defer func() {
  158. l.Debugf("close snapshot %p", snap)
  159. snap.Release()
  160. }()
  161. dbi := snap.NewIterator(&util.Range{Start: start, Limit: limit}, nil)
  162. defer dbi.Release()
  163. moreDb := dbi.Next()
  164. fsi := 0
  165. var maxLocalVer int64
  166. isLocalDevice := bytes.Equal(device, protocol.LocalDeviceID[:])
  167. for {
  168. var newName, oldName []byte
  169. moreFs := fsi < len(fs)
  170. if !moreDb && !moreFs {
  171. break
  172. }
  173. if moreFs {
  174. newName = []byte(fs[fsi].Name)
  175. }
  176. if moreDb {
  177. oldName = deviceKeyName(dbi.Key())
  178. }
  179. cmp := bytes.Compare(newName, oldName)
  180. l.Debugf("generic replace; folder=%q device=%v moreFs=%v moreDb=%v cmp=%d newName=%q oldName=%q", folder, protocol.DeviceIDFromBytes(device), moreFs, moreDb, cmp, newName, oldName)
  181. switch {
  182. case moreFs && (!moreDb || cmp == -1):
  183. l.Debugln("generic replace; missing - insert")
  184. // Database is missing this file. Insert it.
  185. if lv := ldbInsert(batch, folder, device, fs[fsi]); lv > maxLocalVer {
  186. maxLocalVer = lv
  187. }
  188. if isLocalDevice {
  189. localSize.addFile(fs[fsi])
  190. }
  191. if fs[fsi].IsInvalid() {
  192. ldbRemoveFromGlobal(snap, batch, folder, device, newName, globalSize)
  193. } else {
  194. ldbUpdateGlobal(snap, batch, folder, device, fs[fsi], globalSize)
  195. }
  196. fsi++
  197. case moreFs && moreDb && cmp == 0:
  198. // File exists on both sides - compare versions. We might get an
  199. // update with the same version and different flags if a device has
  200. // marked a file as invalid, so handle that too.
  201. l.Debugln("generic replace; exists - compare")
  202. var ef FileInfoTruncated
  203. ef.UnmarshalXDR(dbi.Value())
  204. if !fs[fsi].Version.Equal(ef.Version) || fs[fsi].Flags != ef.Flags {
  205. l.Debugln("generic replace; differs - insert")
  206. if lv := ldbInsert(batch, folder, device, fs[fsi]); lv > maxLocalVer {
  207. maxLocalVer = lv
  208. }
  209. if isLocalDevice {
  210. localSize.removeFile(ef)
  211. localSize.addFile(fs[fsi])
  212. }
  213. if fs[fsi].IsInvalid() {
  214. ldbRemoveFromGlobal(snap, batch, folder, device, newName, globalSize)
  215. } else {
  216. ldbUpdateGlobal(snap, batch, folder, device, fs[fsi], globalSize)
  217. }
  218. } else {
  219. l.Debugln("generic replace; equal - ignore")
  220. }
  221. fsi++
  222. moreDb = dbi.Next()
  223. case moreDb && (!moreFs || cmp == 1):
  224. l.Debugln("generic replace; exists - remove")
  225. if lv := deleteFn(snap, batch, folder, device, oldName, dbi); lv > maxLocalVer {
  226. maxLocalVer = lv
  227. }
  228. moreDb = dbi.Next()
  229. }
  230. // Write out and reuse the batch every few records, to avoid the batch
  231. // growing too large and thus allocating unnecessarily much memory.
  232. if batch.Len() > batchFlushSize {
  233. l.Debugf("db.Write %p", batch)
  234. err = db.Write(batch, nil)
  235. if err != nil {
  236. panic(err)
  237. }
  238. batch.Reset()
  239. }
  240. }
  241. l.Debugf("db.Write %p", batch)
  242. err = db.Write(batch, nil)
  243. if err != nil {
  244. panic(err)
  245. }
  246. return maxLocalVer
  247. }
  248. func ldbReplace(db *leveldb.DB, folder, device []byte, fs []protocol.FileInfo, localSize, globalSize *sizeTracker) int64 {
  249. // TODO: Return the remaining maxLocalVer?
  250. return ldbGenericReplace(db, folder, device, fs, localSize, globalSize, func(db dbReader, batch dbWriter, folder, device, name []byte, dbi iterator.Iterator) int64 {
  251. // Database has a file that we are missing. Remove it.
  252. l.Debugf("delete; folder=%q device=%v name=%q", folder, protocol.DeviceIDFromBytes(device), name)
  253. ldbRemoveFromGlobal(db, batch, folder, device, name, globalSize)
  254. l.Debugf("batch.Delete %p %x", batch, dbi.Key())
  255. batch.Delete(dbi.Key())
  256. return 0
  257. })
  258. }
  259. func ldbUpdate(db *leveldb.DB, folder, device []byte, fs []protocol.FileInfo, localSize, globalSize *sizeTracker) int64 {
  260. runtime.GC()
  261. batch := new(leveldb.Batch)
  262. l.Debugf("new batch %p", batch)
  263. snap, err := db.GetSnapshot()
  264. if err != nil {
  265. panic(err)
  266. }
  267. l.Debugf("created snapshot %p", snap)
  268. defer func() {
  269. l.Debugf("close snapshot %p", snap)
  270. snap.Release()
  271. }()
  272. var maxLocalVer int64
  273. var fk []byte
  274. isLocalDevice := bytes.Equal(device, protocol.LocalDeviceID[:])
  275. for _, f := range fs {
  276. name := []byte(f.Name)
  277. fk = deviceKeyInto(fk[:cap(fk)], folder, device, name)
  278. l.Debugf("snap.Get %p %x", snap, fk)
  279. bs, err := snap.Get(fk, nil)
  280. if err == leveldb.ErrNotFound {
  281. if isLocalDevice {
  282. localSize.addFile(f)
  283. }
  284. if lv := ldbInsert(batch, folder, device, f); lv > maxLocalVer {
  285. maxLocalVer = lv
  286. }
  287. if f.IsInvalid() {
  288. ldbRemoveFromGlobal(snap, batch, folder, device, name, globalSize)
  289. } else {
  290. ldbUpdateGlobal(snap, batch, folder, device, f, globalSize)
  291. }
  292. continue
  293. }
  294. var ef FileInfoTruncated
  295. err = ef.UnmarshalXDR(bs)
  296. if err != nil {
  297. panic(err)
  298. }
  299. // Flags might change without the version being bumped when we set the
  300. // invalid flag on an existing file.
  301. if !ef.Version.Equal(f.Version) || ef.Flags != f.Flags {
  302. if isLocalDevice {
  303. localSize.removeFile(ef)
  304. localSize.addFile(f)
  305. }
  306. if lv := ldbInsert(batch, folder, device, f); lv > maxLocalVer {
  307. maxLocalVer = lv
  308. }
  309. if f.IsInvalid() {
  310. ldbRemoveFromGlobal(snap, batch, folder, device, name, globalSize)
  311. } else {
  312. ldbUpdateGlobal(snap, batch, folder, device, f, globalSize)
  313. }
  314. }
  315. // Write out and reuse the batch every few records, to avoid the batch
  316. // growing too large and thus allocating unnecessarily much memory.
  317. if batch.Len() > batchFlushSize {
  318. l.Debugf("db.Write %p", batch)
  319. err = db.Write(batch, nil)
  320. if err != nil {
  321. panic(err)
  322. }
  323. batch.Reset()
  324. }
  325. }
  326. l.Debugf("db.Write %p", batch)
  327. err = db.Write(batch, nil)
  328. if err != nil {
  329. panic(err)
  330. }
  331. return maxLocalVer
  332. }
  333. func ldbInsert(batch dbWriter, folder, device []byte, file protocol.FileInfo) int64 {
  334. l.Debugf("insert; folder=%q device=%v %v", folder, protocol.DeviceIDFromBytes(device), file)
  335. if file.LocalVersion == 0 {
  336. file.LocalVersion = clock(0)
  337. }
  338. name := []byte(file.Name)
  339. nk := deviceKey(folder, device, name)
  340. l.Debugf("batch.Put %p %x", batch, nk)
  341. batch.Put(nk, file.MustMarshalXDR())
  342. return file.LocalVersion
  343. }
  344. // ldbUpdateGlobal adds this device+version to the version list for the given
  345. // file. If the device is already present in the list, the version is updated.
  346. // If the file does not have an entry in the global list, it is created.
  347. func ldbUpdateGlobal(db dbReader, batch dbWriter, folder, device []byte, file protocol.FileInfo, globalSize *sizeTracker) bool {
  348. l.Debugf("update global; folder=%q device=%v file=%q version=%d", folder, protocol.DeviceIDFromBytes(device), file.Name, file.Version)
  349. name := []byte(file.Name)
  350. gk := globalKey(folder, name)
  351. svl, err := db.Get(gk, nil)
  352. if err != nil && err != leveldb.ErrNotFound {
  353. panic(err)
  354. }
  355. var fl versionList
  356. var oldFile protocol.FileInfo
  357. var hasOldFile bool
  358. // Remove the device from the current version list
  359. if svl != nil {
  360. err = fl.UnmarshalXDR(svl)
  361. if err != nil {
  362. panic(err)
  363. }
  364. for i := range fl.versions {
  365. if bytes.Compare(fl.versions[i].device, device) == 0 {
  366. if fl.versions[i].version.Equal(file.Version) {
  367. // No need to do anything
  368. return false
  369. }
  370. if i == 0 {
  371. // Keep the current newest file around so we can subtract it from
  372. // the globalSize if we replace it.
  373. oldFile, hasOldFile = ldbGet(db, folder, fl.versions[0].device, name)
  374. }
  375. fl.versions = append(fl.versions[:i], fl.versions[i+1:]...)
  376. break
  377. }
  378. }
  379. }
  380. nv := fileVersion{
  381. device: device,
  382. version: file.Version,
  383. }
  384. insertedAt := -1
  385. // Find a position in the list to insert this file. The file at the front
  386. // of the list is the newer, the "global".
  387. for i := range fl.versions {
  388. switch fl.versions[i].version.Compare(file.Version) {
  389. case protocol.Equal, protocol.Lesser:
  390. // The version at this point in the list is equal to or lesser
  391. // ("older") than us. We insert ourselves in front of it.
  392. fl.versions = insertVersion(fl.versions, i, nv)
  393. insertedAt = i
  394. goto done
  395. case protocol.ConcurrentLesser, protocol.ConcurrentGreater:
  396. // The version at this point is in conflict with us. We must pull
  397. // the actual file metadata to determine who wins. If we win, we
  398. // insert ourselves in front of the loser here. (The "Lesser" and
  399. // "Greater" in the condition above is just based on the device
  400. // IDs in the version vector, which is not the only thing we use
  401. // to determine the winner.)
  402. of, ok := ldbGet(db, folder, fl.versions[i].device, name)
  403. if !ok {
  404. panic("file referenced in version list does not exist")
  405. }
  406. if file.WinsConflict(of) {
  407. fl.versions = insertVersion(fl.versions, i, nv)
  408. insertedAt = i
  409. goto done
  410. }
  411. }
  412. }
  413. // We didn't find a position for an insert above, so append to the end.
  414. fl.versions = append(fl.versions, nv)
  415. insertedAt = len(fl.versions) - 1
  416. done:
  417. if insertedAt == 0 {
  418. // We just inserted a new newest version. Fixup the global size
  419. // calculation.
  420. if !file.Version.Equal(oldFile.Version) {
  421. globalSize.addFile(file)
  422. if hasOldFile {
  423. // We have the old file that was removed at the head of the list.
  424. globalSize.removeFile(oldFile)
  425. } else if len(fl.versions) > 1 {
  426. // The previous newest version is now at index 1, grab it from there.
  427. oldFile, ok := ldbGet(db, folder, fl.versions[1].device, name)
  428. if !ok {
  429. panic("file referenced in version list does not exist")
  430. }
  431. globalSize.removeFile(oldFile)
  432. }
  433. }
  434. }
  435. l.Debugf("batch.Put %p %x", batch, gk)
  436. l.Debugf("new global after update: %v", fl)
  437. batch.Put(gk, fl.MustMarshalXDR())
  438. return true
  439. }
  440. func insertVersion(vl []fileVersion, i int, v fileVersion) []fileVersion {
  441. t := append(vl, fileVersion{})
  442. copy(t[i+1:], t[i:])
  443. t[i] = v
  444. return t
  445. }
  446. // ldbRemoveFromGlobal removes the device from the global version list for the
  447. // given file. If the version list is empty after this, the file entry is
  448. // removed entirely.
  449. func ldbRemoveFromGlobal(db dbReader, batch dbWriter, folder, device, file []byte, globalSize *sizeTracker) {
  450. l.Debugf("remove from global; folder=%q device=%v file=%q", folder, protocol.DeviceIDFromBytes(device), file)
  451. gk := globalKey(folder, file)
  452. svl, err := db.Get(gk, nil)
  453. if err != nil {
  454. // We might be called to "remove" a global version that doesn't exist
  455. // if the first update for the file is already marked invalid.
  456. return
  457. }
  458. var fl versionList
  459. err = fl.UnmarshalXDR(svl)
  460. if err != nil {
  461. panic(err)
  462. }
  463. removed := false
  464. for i := range fl.versions {
  465. if bytes.Compare(fl.versions[i].device, device) == 0 {
  466. if i == 0 && globalSize != nil {
  467. f, ok := ldbGet(db, folder, device, file)
  468. if !ok {
  469. panic("removing nonexistent file")
  470. }
  471. globalSize.removeFile(f)
  472. removed = true
  473. }
  474. fl.versions = append(fl.versions[:i], fl.versions[i+1:]...)
  475. break
  476. }
  477. }
  478. if len(fl.versions) == 0 {
  479. l.Debugf("batch.Delete %p %x", batch, gk)
  480. batch.Delete(gk)
  481. } else {
  482. l.Debugf("batch.Put %p %x", batch, gk)
  483. l.Debugf("new global after remove: %v", fl)
  484. batch.Put(gk, fl.MustMarshalXDR())
  485. if removed {
  486. f, ok := ldbGet(db, folder, fl.versions[0].device, file)
  487. if !ok {
  488. panic("new global is nonexistent file")
  489. }
  490. globalSize.addFile(f)
  491. }
  492. }
  493. }
  494. func ldbWithHave(db *leveldb.DB, folder, device []byte, truncate bool, fn Iterator) {
  495. start := deviceKey(folder, device, nil) // before all folder/device files
  496. limit := deviceKey(folder, device, []byte{0xff, 0xff, 0xff, 0xff}) // after all folder/device files
  497. snap, err := db.GetSnapshot()
  498. if err != nil {
  499. panic(err)
  500. }
  501. l.Debugf("created snapshot %p", snap)
  502. defer func() {
  503. l.Debugf("close snapshot %p", snap)
  504. snap.Release()
  505. }()
  506. dbi := snap.NewIterator(&util.Range{Start: start, Limit: limit}, nil)
  507. defer dbi.Release()
  508. for dbi.Next() {
  509. f, err := unmarshalTrunc(dbi.Value(), truncate)
  510. if err != nil {
  511. panic(err)
  512. }
  513. if cont := fn(f); !cont {
  514. return
  515. }
  516. }
  517. }
  518. func ldbWithAllFolderTruncated(db *leveldb.DB, folder []byte, fn func(device []byte, f FileInfoTruncated) bool) {
  519. runtime.GC()
  520. start := deviceKey(folder, nil, nil) // before all folder/device files
  521. limit := deviceKey(folder, protocol.LocalDeviceID[:], []byte{0xff, 0xff, 0xff, 0xff}) // after all folder/device files
  522. snap, err := db.GetSnapshot()
  523. if err != nil {
  524. panic(err)
  525. }
  526. l.Debugf("created snapshot %p", snap)
  527. defer func() {
  528. l.Debugf("close snapshot %p", snap)
  529. snap.Release()
  530. }()
  531. dbi := snap.NewIterator(&util.Range{Start: start, Limit: limit}, nil)
  532. defer dbi.Release()
  533. for dbi.Next() {
  534. device := deviceKeyDevice(dbi.Key())
  535. var f FileInfoTruncated
  536. err := f.UnmarshalXDR(dbi.Value())
  537. if err != nil {
  538. panic(err)
  539. }
  540. switch f.Name {
  541. case "", ".", "..", "/": // A few obviously invalid filenames
  542. l.Infof("Dropping invalid filename %q from database", f.Name)
  543. batch := new(leveldb.Batch)
  544. ldbRemoveFromGlobal(db, batch, folder, device, nil, nil)
  545. batch.Delete(dbi.Key())
  546. db.Write(batch, nil)
  547. continue
  548. }
  549. if cont := fn(device, f); !cont {
  550. return
  551. }
  552. }
  553. }
  554. func ldbGet(db dbReader, folder, device, file []byte) (protocol.FileInfo, bool) {
  555. nk := deviceKey(folder, device, file)
  556. bs, err := db.Get(nk, nil)
  557. if err == leveldb.ErrNotFound {
  558. return protocol.FileInfo{}, false
  559. }
  560. if err != nil {
  561. panic(err)
  562. }
  563. var f protocol.FileInfo
  564. err = f.UnmarshalXDR(bs)
  565. if err != nil {
  566. panic(err)
  567. }
  568. return f, true
  569. }
  570. func ldbGetGlobal(db *leveldb.DB, folder, file []byte, truncate bool) (FileIntf, bool) {
  571. k := globalKey(folder, file)
  572. snap, err := db.GetSnapshot()
  573. if err != nil {
  574. panic(err)
  575. }
  576. l.Debugf("created snapshot %p", snap)
  577. defer func() {
  578. l.Debugf("close snapshot %p", snap)
  579. snap.Release()
  580. }()
  581. l.Debugf("snap.Get %p %x", snap, k)
  582. bs, err := snap.Get(k, nil)
  583. if err == leveldb.ErrNotFound {
  584. return nil, false
  585. }
  586. if err != nil {
  587. panic(err)
  588. }
  589. var vl versionList
  590. err = vl.UnmarshalXDR(bs)
  591. if err != nil {
  592. panic(err)
  593. }
  594. if len(vl.versions) == 0 {
  595. l.Debugln(k)
  596. panic("no versions?")
  597. }
  598. k = deviceKey(folder, vl.versions[0].device, file)
  599. l.Debugf("snap.Get %p %x", snap, k)
  600. bs, err = snap.Get(k, nil)
  601. if err != nil {
  602. panic(err)
  603. }
  604. fi, err := unmarshalTrunc(bs, truncate)
  605. if err != nil {
  606. panic(err)
  607. }
  608. return fi, true
  609. }
  610. func ldbWithGlobal(db *leveldb.DB, folder, prefix []byte, truncate bool, fn Iterator) {
  611. runtime.GC()
  612. snap, err := db.GetSnapshot()
  613. if err != nil {
  614. panic(err)
  615. }
  616. l.Debugf("created snapshot %p", snap)
  617. defer func() {
  618. l.Debugf("close snapshot %p", snap)
  619. snap.Release()
  620. }()
  621. dbi := snap.NewIterator(util.BytesPrefix(globalKey(folder, prefix)), nil)
  622. defer dbi.Release()
  623. var fk []byte
  624. for dbi.Next() {
  625. var vl versionList
  626. err := vl.UnmarshalXDR(dbi.Value())
  627. if err != nil {
  628. panic(err)
  629. }
  630. if len(vl.versions) == 0 {
  631. l.Debugln(dbi.Key())
  632. panic("no versions?")
  633. }
  634. name := globalKeyName(dbi.Key())
  635. fk = deviceKeyInto(fk[:cap(fk)], folder, vl.versions[0].device, name)
  636. l.Debugf("snap.Get %p %x", snap, fk)
  637. bs, err := snap.Get(fk, nil)
  638. if err != nil {
  639. l.Debugf("folder: %q (%x)", folder, folder)
  640. l.Debugf("key: %q (%x)", dbi.Key(), dbi.Key())
  641. l.Debugf("vl: %v", vl)
  642. l.Debugf("vl.versions[0].device: %x", vl.versions[0].device)
  643. l.Debugf("name: %q (%x)", name, name)
  644. l.Debugf("fk: %q", fk)
  645. l.Debugf("fk: %x %x %x", fk[1:1+64], fk[1+64:1+64+32], fk[1+64+32:])
  646. panic(err)
  647. }
  648. f, err := unmarshalTrunc(bs, truncate)
  649. if err != nil {
  650. panic(err)
  651. }
  652. if cont := fn(f); !cont {
  653. return
  654. }
  655. }
  656. }
  657. func ldbAvailability(db *leveldb.DB, folder, file []byte) []protocol.DeviceID {
  658. k := globalKey(folder, file)
  659. bs, err := db.Get(k, nil)
  660. if err == leveldb.ErrNotFound {
  661. return nil
  662. }
  663. if err != nil {
  664. panic(err)
  665. }
  666. var vl versionList
  667. err = vl.UnmarshalXDR(bs)
  668. if err != nil {
  669. panic(err)
  670. }
  671. var devices []protocol.DeviceID
  672. for _, v := range vl.versions {
  673. if !v.version.Equal(vl.versions[0].version) {
  674. break
  675. }
  676. n := protocol.DeviceIDFromBytes(v.device)
  677. devices = append(devices, n)
  678. }
  679. return devices
  680. }
  681. func ldbWithNeed(db *leveldb.DB, folder, device []byte, truncate bool, fn Iterator) {
  682. runtime.GC()
  683. start := globalKey(folder, nil)
  684. limit := globalKey(folder, []byte{0xff, 0xff, 0xff, 0xff})
  685. snap, err := db.GetSnapshot()
  686. if err != nil {
  687. panic(err)
  688. }
  689. l.Debugf("created snapshot %p", snap)
  690. defer func() {
  691. l.Debugf("close snapshot %p", snap)
  692. snap.Release()
  693. }()
  694. dbi := snap.NewIterator(&util.Range{Start: start, Limit: limit}, nil)
  695. defer dbi.Release()
  696. var fk []byte
  697. nextFile:
  698. for dbi.Next() {
  699. var vl versionList
  700. err := vl.UnmarshalXDR(dbi.Value())
  701. if err != nil {
  702. panic(err)
  703. }
  704. if len(vl.versions) == 0 {
  705. l.Debugln(dbi.Key())
  706. panic("no versions?")
  707. }
  708. have := false // If we have the file, any version
  709. need := false // If we have a lower version of the file
  710. var haveVersion protocol.Vector
  711. for _, v := range vl.versions {
  712. if bytes.Compare(v.device, device) == 0 {
  713. have = true
  714. haveVersion = v.version
  715. // XXX: This marks Concurrent (i.e. conflicting) changes as
  716. // needs. Maybe we should do that, but it needs special
  717. // handling in the puller.
  718. need = !v.version.GreaterEqual(vl.versions[0].version)
  719. break
  720. }
  721. }
  722. if need || !have {
  723. name := globalKeyName(dbi.Key())
  724. needVersion := vl.versions[0].version
  725. nextVersion:
  726. for i := range vl.versions {
  727. if !vl.versions[i].version.Equal(needVersion) {
  728. // We haven't found a valid copy of the file with the needed version.
  729. continue nextFile
  730. }
  731. fk = deviceKeyInto(fk[:cap(fk)], folder, vl.versions[i].device, name)
  732. l.Debugf("snap.Get %p %x", snap, fk)
  733. bs, err := snap.Get(fk, nil)
  734. if err != nil {
  735. var id protocol.DeviceID
  736. copy(id[:], device)
  737. l.Debugf("device: %v", id)
  738. l.Debugf("need: %v, have: %v", need, have)
  739. l.Debugf("key: %q (%x)", dbi.Key(), dbi.Key())
  740. l.Debugf("vl: %v", vl)
  741. l.Debugf("i: %v", i)
  742. l.Debugf("fk: %q (%x)", fk, fk)
  743. l.Debugf("name: %q (%x)", name, name)
  744. panic(err)
  745. }
  746. gf, err := unmarshalTrunc(bs, truncate)
  747. if err != nil {
  748. panic(err)
  749. }
  750. if gf.IsInvalid() {
  751. // The file is marked invalid for whatever reason, don't use it.
  752. continue nextVersion
  753. }
  754. if gf.IsDeleted() && !have {
  755. // We don't need deleted files that we don't have
  756. continue nextFile
  757. }
  758. l.Debugf("need folder=%q device=%v name=%q need=%v have=%v haveV=%d globalV=%d", folder, protocol.DeviceIDFromBytes(device), name, need, have, haveVersion, vl.versions[0].version)
  759. if cont := fn(gf); !cont {
  760. return
  761. }
  762. // This file is handled, no need to look further in the version list
  763. continue nextFile
  764. }
  765. }
  766. }
  767. }
  768. func ldbListFolders(db *leveldb.DB) []string {
  769. runtime.GC()
  770. snap, err := db.GetSnapshot()
  771. if err != nil {
  772. panic(err)
  773. }
  774. l.Debugf("created snapshot %p", snap)
  775. defer func() {
  776. l.Debugf("close snapshot %p", snap)
  777. snap.Release()
  778. }()
  779. dbi := snap.NewIterator(util.BytesPrefix([]byte{KeyTypeGlobal}), nil)
  780. defer dbi.Release()
  781. folderExists := make(map[string]bool)
  782. for dbi.Next() {
  783. folder := string(globalKeyFolder(dbi.Key()))
  784. if !folderExists[folder] {
  785. folderExists[folder] = true
  786. }
  787. }
  788. folders := make([]string, 0, len(folderExists))
  789. for k := range folderExists {
  790. folders = append(folders, k)
  791. }
  792. sort.Strings(folders)
  793. return folders
  794. }
  795. func ldbDropFolder(db *leveldb.DB, folder []byte) {
  796. runtime.GC()
  797. snap, err := db.GetSnapshot()
  798. if err != nil {
  799. panic(err)
  800. }
  801. l.Debugf("created snapshot %p", snap)
  802. defer func() {
  803. l.Debugf("close snapshot %p", snap)
  804. snap.Release()
  805. }()
  806. // Remove all items related to the given folder from the device->file bucket
  807. dbi := snap.NewIterator(util.BytesPrefix([]byte{KeyTypeDevice}), nil)
  808. for dbi.Next() {
  809. itemFolder := deviceKeyFolder(dbi.Key())
  810. if bytes.Compare(folder, itemFolder) == 0 {
  811. db.Delete(dbi.Key(), nil)
  812. }
  813. }
  814. dbi.Release()
  815. // Remove all items related to the given folder from the global bucket
  816. dbi = snap.NewIterator(util.BytesPrefix([]byte{KeyTypeGlobal}), nil)
  817. for dbi.Next() {
  818. itemFolder := globalKeyFolder(dbi.Key())
  819. if bytes.Compare(folder, itemFolder) == 0 {
  820. db.Delete(dbi.Key(), nil)
  821. }
  822. }
  823. dbi.Release()
  824. }
  825. func unmarshalTrunc(bs []byte, truncate bool) (FileIntf, error) {
  826. if truncate {
  827. var tf FileInfoTruncated
  828. err := tf.UnmarshalXDR(bs)
  829. return tf, err
  830. }
  831. var tf protocol.FileInfo
  832. err := tf.UnmarshalXDR(bs)
  833. return tf, err
  834. }
  835. func ldbCheckGlobals(db *leveldb.DB, folder []byte, globalSize *sizeTracker) {
  836. defer runtime.GC()
  837. snap, err := db.GetSnapshot()
  838. if err != nil {
  839. panic(err)
  840. }
  841. l.Debugf("created snapshot %p", snap)
  842. defer func() {
  843. l.Debugf("close snapshot %p", snap)
  844. snap.Release()
  845. }()
  846. start := globalKey(folder, nil)
  847. limit := globalKey(folder, []byte{0xff, 0xff, 0xff, 0xff})
  848. dbi := snap.NewIterator(&util.Range{Start: start, Limit: limit}, nil)
  849. defer dbi.Release()
  850. batch := new(leveldb.Batch)
  851. l.Debugf("new batch %p", batch)
  852. var fk []byte
  853. for dbi.Next() {
  854. gk := dbi.Key()
  855. var vl versionList
  856. err := vl.UnmarshalXDR(dbi.Value())
  857. if err != nil {
  858. panic(err)
  859. }
  860. // Check the global version list for consistency. An issue in previous
  861. // versions of goleveldb could result in reordered writes so that
  862. // there are global entries pointing to no longer existing files. Here
  863. // we find those and clear them out.
  864. name := globalKeyName(gk)
  865. var newVL versionList
  866. for i, version := range vl.versions {
  867. fk = deviceKeyInto(fk[:cap(fk)], folder, version.device, name)
  868. l.Debugf("snap.Get %p %x", snap, fk)
  869. _, err := snap.Get(fk, nil)
  870. if err == leveldb.ErrNotFound {
  871. continue
  872. }
  873. if err != nil {
  874. panic(err)
  875. }
  876. newVL.versions = append(newVL.versions, version)
  877. if i == 0 {
  878. fi, ok := ldbGet(snap, folder, version.device, name)
  879. if !ok {
  880. panic("nonexistent global master file")
  881. }
  882. globalSize.addFile(fi)
  883. }
  884. }
  885. if len(newVL.versions) != len(vl.versions) {
  886. l.Infof("db repair: rewriting global version list for %x %x", gk[1:1+64], gk[1+64:])
  887. batch.Put(dbi.Key(), newVL.MustMarshalXDR())
  888. }
  889. }
  890. l.Debugf("db check completed for %q", folder)
  891. db.Write(batch, nil)
  892. }