index.go 4.3 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204
  1. package storage
  2. import (
  3. "log"
  4. "sort"
  5. "sync"
  6. "github.com/coreos/etcd/Godeps/_workspace/src/github.com/google/btree"
  7. )
  8. type index interface {
  9. Get(key []byte, atRev int64) (rev, created revision, ver int64, err error)
  10. Range(key, end []byte, atRev int64) ([][]byte, []revision)
  11. Put(key []byte, rev revision)
  12. Restore(key []byte, created, modified revision, ver int64)
  13. Tombstone(key []byte, rev revision) error
  14. RangeEvents(key, end []byte, rev int64) []revision
  15. Compact(rev int64) map[revision]struct{}
  16. Equal(b index) bool
  17. }
  18. type treeIndex struct {
  19. sync.RWMutex
  20. tree *btree.BTree
  21. }
  22. func newTreeIndex() index {
  23. return &treeIndex{
  24. tree: btree.New(32),
  25. }
  26. }
  27. func (ti *treeIndex) Put(key []byte, rev revision) {
  28. keyi := &keyIndex{key: key}
  29. ti.Lock()
  30. defer ti.Unlock()
  31. item := ti.tree.Get(keyi)
  32. if item == nil {
  33. keyi.put(rev.main, rev.sub)
  34. ti.tree.ReplaceOrInsert(keyi)
  35. return
  36. }
  37. okeyi := item.(*keyIndex)
  38. okeyi.put(rev.main, rev.sub)
  39. }
  40. func (ti *treeIndex) Restore(key []byte, created, modified revision, ver int64) {
  41. keyi := &keyIndex{key: key}
  42. ti.Lock()
  43. defer ti.Unlock()
  44. item := ti.tree.Get(keyi)
  45. if item == nil {
  46. keyi.restore(created, modified, ver)
  47. ti.tree.ReplaceOrInsert(keyi)
  48. return
  49. }
  50. okeyi := item.(*keyIndex)
  51. okeyi.put(modified.main, modified.sub)
  52. }
  53. func (ti *treeIndex) Get(key []byte, atRev int64) (modified, created revision, ver int64, err error) {
  54. keyi := &keyIndex{key: key}
  55. ti.RLock()
  56. defer ti.RUnlock()
  57. item := ti.tree.Get(keyi)
  58. if item == nil {
  59. return revision{}, revision{}, 0, ErrRevisionNotFound
  60. }
  61. keyi = item.(*keyIndex)
  62. return keyi.get(atRev)
  63. }
  64. func (ti *treeIndex) Range(key, end []byte, atRev int64) (keys [][]byte, revs []revision) {
  65. if end == nil {
  66. rev, _, _, err := ti.Get(key, atRev)
  67. if err != nil {
  68. return nil, nil
  69. }
  70. return [][]byte{key}, []revision{rev}
  71. }
  72. keyi := &keyIndex{key: key}
  73. endi := &keyIndex{key: end}
  74. ti.RLock()
  75. defer ti.RUnlock()
  76. ti.tree.AscendGreaterOrEqual(keyi, func(item btree.Item) bool {
  77. if !item.Less(endi) {
  78. return false
  79. }
  80. curKeyi := item.(*keyIndex)
  81. rev, _, _, err := curKeyi.get(atRev)
  82. if err != nil {
  83. return true
  84. }
  85. revs = append(revs, rev)
  86. keys = append(keys, curKeyi.key)
  87. return true
  88. })
  89. return keys, revs
  90. }
  91. func (ti *treeIndex) Tombstone(key []byte, rev revision) error {
  92. keyi := &keyIndex{key: key}
  93. ti.Lock()
  94. defer ti.Unlock()
  95. item := ti.tree.Get(keyi)
  96. if item == nil {
  97. return ErrRevisionNotFound
  98. }
  99. ki := item.(*keyIndex)
  100. return ki.tombstone(rev.main, rev.sub)
  101. }
  102. // RangeEvents returns all revisions from key(including) to end(excluding)
  103. // at or after the given rev. The returned slice is sorted in the order
  104. // of revision.
  105. func (ti *treeIndex) RangeEvents(key, end []byte, rev int64) []revision {
  106. ti.RLock()
  107. defer ti.RUnlock()
  108. keyi := &keyIndex{key: key}
  109. if end == nil {
  110. item := ti.tree.Get(keyi)
  111. if item == nil {
  112. return nil
  113. }
  114. keyi = item.(*keyIndex)
  115. return keyi.since(rev)
  116. }
  117. endi := &keyIndex{key: end}
  118. var revs []revision
  119. ti.tree.AscendGreaterOrEqual(keyi, func(item btree.Item) bool {
  120. if !item.Less(endi) {
  121. return false
  122. }
  123. curKeyi := item.(*keyIndex)
  124. revs = append(revs, curKeyi.since(rev)...)
  125. return true
  126. })
  127. sort.Sort(revisions(revs))
  128. return revs
  129. }
  130. func (ti *treeIndex) Compact(rev int64) map[revision]struct{} {
  131. available := make(map[revision]struct{})
  132. emptyki := make([]*keyIndex, 0)
  133. log.Printf("store.index: compact %d", rev)
  134. // TODO: do not hold the lock for long time?
  135. // This is probably OK. Compacting 10M keys takes O(10ms).
  136. ti.Lock()
  137. defer ti.Unlock()
  138. ti.tree.Ascend(compactIndex(rev, available, &emptyki))
  139. for _, ki := range emptyki {
  140. item := ti.tree.Delete(ki)
  141. if item == nil {
  142. log.Panic("store.index: unexpected delete failure during compaction")
  143. }
  144. }
  145. return available
  146. }
  147. func compactIndex(rev int64, available map[revision]struct{}, emptyki *[]*keyIndex) func(i btree.Item) bool {
  148. return func(i btree.Item) bool {
  149. keyi := i.(*keyIndex)
  150. keyi.compact(rev, available)
  151. if keyi.isEmpty() {
  152. *emptyki = append(*emptyki, keyi)
  153. }
  154. return true
  155. }
  156. }
  157. func (a *treeIndex) Equal(bi index) bool {
  158. b := bi.(*treeIndex)
  159. if a.tree.Len() != b.tree.Len() {
  160. return false
  161. }
  162. equal := true
  163. a.tree.Ascend(func(item btree.Item) bool {
  164. aki := item.(*keyIndex)
  165. bki := b.tree.Get(item).(*keyIndex)
  166. if !aki.equal(bki) {
  167. equal = false
  168. return false
  169. }
  170. return true
  171. })
  172. return equal
  173. }