backend.go 7.8 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282
  1. // Copyright (C) 2019 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 https://mozilla.org/MPL/2.0/.
  6. package backend
  7. import (
  8. "os"
  9. "strings"
  10. "sync"
  11. "time"
  12. "github.com/syncthing/syncthing/lib/locations"
  13. )
  14. type CommitHook func(WriteTransaction) error
  15. // The Reader interface specifies the read-only operations available on the
  16. // main database and on read-only transactions (snapshots). Note that when
  17. // called directly on the database handle these operations may take implicit
  18. // transactions and performance may suffer.
  19. type Reader interface {
  20. Get(key []byte) ([]byte, error)
  21. NewPrefixIterator(prefix []byte) (Iterator, error)
  22. NewRangeIterator(first, last []byte) (Iterator, error)
  23. }
  24. // The Writer interface specifies the mutating operations available on the
  25. // main database and on writable transactions. Note that when called
  26. // directly on the database handle these operations may take implicit
  27. // transactions and performance may suffer.
  28. type Writer interface {
  29. Put(key, val []byte) error
  30. Delete(key []byte) error
  31. }
  32. // The ReadTransaction interface specifies the operations on read-only
  33. // transactions. Every ReadTransaction must be released when no longer
  34. // required.
  35. type ReadTransaction interface {
  36. Reader
  37. Release()
  38. }
  39. // The WriteTransaction interface specifies the operations on writable
  40. // transactions. Every WriteTransaction must be either committed or released
  41. // (i.e., discarded) when no longer required. No further operations must be
  42. // performed after release or commit (regardless of whether commit succeeded),
  43. // with one exception -- it's fine to release an already committed or released
  44. // transaction.
  45. //
  46. // A Checkpoint is a potential partial commit of the transaction so far, for
  47. // purposes of saving memory when transactions are in-RAM. Note that
  48. // transactions may be checkpointed *anyway* even if this is not called, due to
  49. // resource constraints, but this gives you a chance to decide when.
  50. //
  51. // Functions can be passed to Checkpoint. These are run if and only if the
  52. // checkpoint will result in a flush, and will run before the flush. The
  53. // transaction can be accessed via a closure. If an error is returned from
  54. // these functions the flush will be aborted and the error bubbled.
  55. type WriteTransaction interface {
  56. ReadTransaction
  57. Writer
  58. Checkpoint() error
  59. Commit() error
  60. }
  61. // The Iterator interface specifies the operations available on iterators
  62. // returned by NewPrefixIterator and NewRangeIterator. The iterator pattern
  63. // is to loop while Next returns true, then check Error after the loop. Next
  64. // will return false when iteration is complete (Error() == nil) or when
  65. // there is an error preventing iteration, which is then returned by
  66. // Error(). For example:
  67. //
  68. // it, err := db.NewPrefixIterator(nil)
  69. // if err != nil {
  70. // // problem preventing iteration
  71. // }
  72. // defer it.Release()
  73. // for it.Next() {
  74. // // ...
  75. // }
  76. // if err := it.Error(); err != nil {
  77. // // there was a database problem while iterating
  78. // }
  79. //
  80. // An iterator must be Released when no longer required. The Error method
  81. // can be called either before or after Release with the same results. If an
  82. // iterator was created in a transaction (whether read-only or write) it
  83. // must be released before the transaction is released (or committed).
  84. type Iterator interface {
  85. Next() bool
  86. Key() []byte
  87. Value() []byte
  88. Error() error
  89. Release()
  90. }
  91. // The Backend interface represents the main database handle. It supports
  92. // both read/write operations and opening read-only or writable
  93. // transactions. Depending on the actual implementation, individual
  94. // read/write operations may be implicitly wrapped in transactions, making
  95. // them perform quite badly when used repeatedly. For bulk operations,
  96. // consider always using a transaction of the appropriate type. The
  97. // transaction isolation level is "read committed" - there are no dirty
  98. // reads.
  99. type Backend interface {
  100. Reader
  101. Writer
  102. NewReadTransaction() (ReadTransaction, error)
  103. NewWriteTransaction(hooks ...CommitHook) (WriteTransaction, error)
  104. Close() error
  105. Compact() error
  106. }
  107. type Tuning int
  108. const (
  109. // N.b. these constants must match those in lib/config.Tuning!
  110. TuningAuto Tuning = iota
  111. TuningSmall
  112. TuningLarge
  113. )
  114. func Open(path string, tuning Tuning) (Backend, error) {
  115. if os.Getenv("USE_BADGER") != "" {
  116. l.Warnln("Using experimental badger db")
  117. if err := maybeCopyDatabase(path, strings.Replace(path, locations.BadgerDir, locations.LevelDBDir, 1), OpenBadger, OpenLevelDBRO); err != nil {
  118. return nil, err
  119. }
  120. return OpenBadger(path)
  121. }
  122. if err := maybeCopyDatabase(path, strings.Replace(path, locations.LevelDBDir, locations.BadgerDir, 1), OpenLevelDBAuto, OpenBadger); err != nil {
  123. return nil, err
  124. }
  125. return OpenLevelDB(path, tuning)
  126. }
  127. func OpenMemory() Backend {
  128. if os.Getenv("USE_BADGER") != "" {
  129. return OpenBadgerMemory()
  130. }
  131. return OpenLevelDBMemory()
  132. }
  133. type errClosed struct{}
  134. func (*errClosed) Error() string { return "database is closed" }
  135. type errNotFound struct{}
  136. func (*errNotFound) Error() string { return "key not found" }
  137. func IsClosed(err error) bool {
  138. _, ok := err.(*errClosed)
  139. return ok
  140. }
  141. func IsNotFound(err error) bool {
  142. _, ok := err.(*errNotFound)
  143. return ok
  144. }
  145. // releaser manages counting on top of a waitgroup
  146. type releaser struct {
  147. wg *closeWaitGroup
  148. once *sync.Once
  149. }
  150. func newReleaser(wg *closeWaitGroup) (*releaser, error) {
  151. if err := wg.Add(1); err != nil {
  152. return nil, err
  153. }
  154. return &releaser{
  155. wg: wg,
  156. once: new(sync.Once),
  157. }, nil
  158. }
  159. func (r releaser) Release() {
  160. // We use the Once because we may get called multiple times from
  161. // Commit() and deferred Release().
  162. r.once.Do(func() {
  163. r.wg.Done()
  164. })
  165. }
  166. // closeWaitGroup behaves just like a sync.WaitGroup, but does not require
  167. // a single routine to do the Add and Wait calls. If Add is called after
  168. // CloseWait, it will return an error, and both are safe to be used concurrently.
  169. type closeWaitGroup struct {
  170. sync.WaitGroup
  171. closed bool
  172. closeMut sync.RWMutex
  173. }
  174. func (cg *closeWaitGroup) Add(i int) error {
  175. cg.closeMut.RLock()
  176. defer cg.closeMut.RUnlock()
  177. if cg.closed {
  178. return &errClosed{}
  179. }
  180. cg.WaitGroup.Add(i)
  181. return nil
  182. }
  183. func (cg *closeWaitGroup) CloseWait() {
  184. cg.closeMut.Lock()
  185. cg.closed = true
  186. cg.closeMut.Unlock()
  187. cg.WaitGroup.Wait()
  188. }
  189. type opener func(path string) (Backend, error)
  190. // maybeCopyDatabase copies the database if the destination doesn't exist
  191. // but the source does.
  192. func maybeCopyDatabase(toPath, fromPath string, toOpen, fromOpen opener) error {
  193. if _, err := os.Lstat(toPath); !os.IsNotExist(err) {
  194. // Destination database exists (or is otherwise unavailable), do not
  195. // attempt to overwrite it.
  196. return nil
  197. }
  198. if _, err := os.Lstat(fromPath); err != nil {
  199. // Source database is not available, so nothing to copy
  200. return nil
  201. }
  202. fromDB, err := fromOpen(fromPath)
  203. if err != nil {
  204. return err
  205. }
  206. defer fromDB.Close()
  207. toDB, err := toOpen(toPath)
  208. if err != nil {
  209. // That's odd, but it will be handled & reported in the usual path
  210. // so we can ignore it here.
  211. return err
  212. }
  213. defer toDB.Close()
  214. l.Infoln("Copying database for format conversion...")
  215. if err := copyBackend(toDB, fromDB); err != nil {
  216. return err
  217. }
  218. // Move the old database out of the way to mark it as migrated.
  219. fromDB.Close()
  220. _ = os.Rename(fromPath, fromPath+".migrated."+time.Now().Format("20060102150405"))
  221. return nil
  222. }
  223. func copyBackend(to, from Backend) error {
  224. srcIt, err := from.NewPrefixIterator(nil)
  225. if err != nil {
  226. return err
  227. }
  228. defer srcIt.Release()
  229. dstTx, err := to.NewWriteTransaction()
  230. if err != nil {
  231. return err
  232. }
  233. defer dstTx.Release()
  234. for srcIt.Next() {
  235. if err := dstTx.Put(srcIt.Key(), srcIt.Value()); err != nil {
  236. return err
  237. }
  238. }
  239. if srcIt.Error() != nil {
  240. return err
  241. }
  242. srcIt.Release()
  243. return dstTx.Commit()
  244. }