kvstore_test.go 22 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833
  1. // Copyright 2015 The etcd Authors
  2. //
  3. // Licensed under the Apache License, Version 2.0 (the "License");
  4. // you may not use this file except in compliance with the License.
  5. // You may obtain a copy of the License at
  6. //
  7. // http://www.apache.org/licenses/LICENSE-2.0
  8. //
  9. // Unless required by applicable law or agreed to in writing, software
  10. // distributed under the License is distributed on an "AS IS" BASIS,
  11. // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  12. // See the License for the specific language governing permissions and
  13. // limitations under the License.
  14. package mvcc
  15. import (
  16. "crypto/rand"
  17. "encoding/binary"
  18. "fmt"
  19. "math"
  20. mrand "math/rand"
  21. "os"
  22. "reflect"
  23. "sync"
  24. "testing"
  25. "time"
  26. "go.etcd.io/etcd/lease"
  27. "go.etcd.io/etcd/mvcc/backend"
  28. "go.etcd.io/etcd/mvcc/mvccpb"
  29. "go.etcd.io/etcd/pkg/schedule"
  30. "go.etcd.io/etcd/pkg/testutil"
  31. "go.uber.org/zap"
  32. )
  33. func TestStoreRev(t *testing.T) {
  34. b, tmpPath := backend.NewDefaultTmpBackend()
  35. s := NewStore(zap.NewExample(), b, &lease.FakeLessor{}, nil)
  36. defer s.Close()
  37. defer os.Remove(tmpPath)
  38. for i := 1; i <= 3; i++ {
  39. s.Put([]byte("foo"), []byte("bar"), lease.NoLease)
  40. if r := s.Rev(); r != int64(i+1) {
  41. t.Errorf("#%d: rev = %d, want %d", i, r, i+1)
  42. }
  43. }
  44. }
  45. func TestStorePut(t *testing.T) {
  46. kv := mvccpb.KeyValue{
  47. Key: []byte("foo"),
  48. Value: []byte("bar"),
  49. CreateRevision: 1,
  50. ModRevision: 2,
  51. Version: 1,
  52. }
  53. kvb, err := kv.Marshal()
  54. if err != nil {
  55. t.Fatal(err)
  56. }
  57. tests := []struct {
  58. rev revision
  59. r indexGetResp
  60. rr *rangeResp
  61. wrev revision
  62. wkey []byte
  63. wkv mvccpb.KeyValue
  64. wputrev revision
  65. }{
  66. {
  67. revision{1, 0},
  68. indexGetResp{revision{}, revision{}, 0, ErrRevisionNotFound},
  69. nil,
  70. revision{2, 0},
  71. newTestKeyBytes(revision{2, 0}, false),
  72. mvccpb.KeyValue{
  73. Key: []byte("foo"),
  74. Value: []byte("bar"),
  75. CreateRevision: 2,
  76. ModRevision: 2,
  77. Version: 1,
  78. Lease: 1,
  79. },
  80. revision{2, 0},
  81. },
  82. {
  83. revision{1, 1},
  84. indexGetResp{revision{2, 0}, revision{2, 0}, 1, nil},
  85. &rangeResp{[][]byte{newTestKeyBytes(revision{2, 1}, false)}, [][]byte{kvb}},
  86. revision{2, 0},
  87. newTestKeyBytes(revision{2, 0}, false),
  88. mvccpb.KeyValue{
  89. Key: []byte("foo"),
  90. Value: []byte("bar"),
  91. CreateRevision: 2,
  92. ModRevision: 2,
  93. Version: 2,
  94. Lease: 2,
  95. },
  96. revision{2, 0},
  97. },
  98. {
  99. revision{2, 0},
  100. indexGetResp{revision{2, 1}, revision{2, 0}, 2, nil},
  101. &rangeResp{[][]byte{newTestKeyBytes(revision{2, 1}, false)}, [][]byte{kvb}},
  102. revision{3, 0},
  103. newTestKeyBytes(revision{3, 0}, false),
  104. mvccpb.KeyValue{
  105. Key: []byte("foo"),
  106. Value: []byte("bar"),
  107. CreateRevision: 2,
  108. ModRevision: 3,
  109. Version: 3,
  110. Lease: 3,
  111. },
  112. revision{3, 0},
  113. },
  114. }
  115. for i, tt := range tests {
  116. s := newFakeStore()
  117. b := s.b.(*fakeBackend)
  118. fi := s.kvindex.(*fakeIndex)
  119. s.currentRev = tt.rev.main
  120. fi.indexGetRespc <- tt.r
  121. if tt.rr != nil {
  122. b.tx.rangeRespc <- *tt.rr
  123. }
  124. s.Put([]byte("foo"), []byte("bar"), lease.LeaseID(i+1))
  125. data, err := tt.wkv.Marshal()
  126. if err != nil {
  127. t.Errorf("#%d: marshal err = %v, want nil", i, err)
  128. }
  129. wact := []testutil.Action{
  130. {Name: "seqput", Params: []interface{}{keyBucketName, tt.wkey, data}},
  131. }
  132. if tt.rr != nil {
  133. wact = []testutil.Action{
  134. {Name: "seqput", Params: []interface{}{keyBucketName, tt.wkey, data}},
  135. }
  136. }
  137. if g := b.tx.Action(); !reflect.DeepEqual(g, wact) {
  138. t.Errorf("#%d: tx action = %+v, want %+v", i, g, wact)
  139. }
  140. wact = []testutil.Action{
  141. {Name: "get", Params: []interface{}{[]byte("foo"), tt.wputrev.main}},
  142. {Name: "put", Params: []interface{}{[]byte("foo"), tt.wputrev}},
  143. }
  144. if g := fi.Action(); !reflect.DeepEqual(g, wact) {
  145. t.Errorf("#%d: index action = %+v, want %+v", i, g, wact)
  146. }
  147. if s.currentRev != tt.wrev.main {
  148. t.Errorf("#%d: rev = %+v, want %+v", i, s.currentRev, tt.wrev)
  149. }
  150. s.Close()
  151. }
  152. }
  153. func TestStoreRange(t *testing.T) {
  154. key := newTestKeyBytes(revision{2, 0}, false)
  155. kv := mvccpb.KeyValue{
  156. Key: []byte("foo"),
  157. Value: []byte("bar"),
  158. CreateRevision: 1,
  159. ModRevision: 2,
  160. Version: 1,
  161. }
  162. kvb, err := kv.Marshal()
  163. if err != nil {
  164. t.Fatal(err)
  165. }
  166. wrev := int64(2)
  167. tests := []struct {
  168. idxr indexRangeResp
  169. r rangeResp
  170. }{
  171. {
  172. indexRangeResp{[][]byte{[]byte("foo")}, []revision{{2, 0}}},
  173. rangeResp{[][]byte{key}, [][]byte{kvb}},
  174. },
  175. {
  176. indexRangeResp{[][]byte{[]byte("foo"), []byte("foo1")}, []revision{{2, 0}, {3, 0}}},
  177. rangeResp{[][]byte{key}, [][]byte{kvb}},
  178. },
  179. }
  180. ro := RangeOptions{Limit: 1, Rev: 0, Count: false}
  181. for i, tt := range tests {
  182. s := newFakeStore()
  183. b := s.b.(*fakeBackend)
  184. fi := s.kvindex.(*fakeIndex)
  185. s.currentRev = 2
  186. b.tx.rangeRespc <- tt.r
  187. fi.indexRangeRespc <- tt.idxr
  188. ret, err := s.Range([]byte("foo"), []byte("goo"), ro)
  189. if err != nil {
  190. t.Errorf("#%d: err = %v, want nil", i, err)
  191. }
  192. if w := []mvccpb.KeyValue{kv}; !reflect.DeepEqual(ret.KVs, w) {
  193. t.Errorf("#%d: kvs = %+v, want %+v", i, ret.KVs, w)
  194. }
  195. if ret.Rev != wrev {
  196. t.Errorf("#%d: rev = %d, want %d", i, ret.Rev, wrev)
  197. }
  198. wstart := newRevBytes()
  199. revToBytes(tt.idxr.revs[0], wstart)
  200. wact := []testutil.Action{
  201. {Name: "range", Params: []interface{}{keyBucketName, wstart, []byte(nil), int64(0)}},
  202. }
  203. if g := b.tx.Action(); !reflect.DeepEqual(g, wact) {
  204. t.Errorf("#%d: tx action = %+v, want %+v", i, g, wact)
  205. }
  206. wact = []testutil.Action{
  207. {Name: "range", Params: []interface{}{[]byte("foo"), []byte("goo"), wrev}},
  208. }
  209. if g := fi.Action(); !reflect.DeepEqual(g, wact) {
  210. t.Errorf("#%d: index action = %+v, want %+v", i, g, wact)
  211. }
  212. if s.currentRev != 2 {
  213. t.Errorf("#%d: current rev = %+v, want %+v", i, s.currentRev, 2)
  214. }
  215. s.Close()
  216. }
  217. }
  218. func TestStoreDeleteRange(t *testing.T) {
  219. key := newTestKeyBytes(revision{2, 0}, false)
  220. kv := mvccpb.KeyValue{
  221. Key: []byte("foo"),
  222. Value: []byte("bar"),
  223. CreateRevision: 1,
  224. ModRevision: 2,
  225. Version: 1,
  226. }
  227. kvb, err := kv.Marshal()
  228. if err != nil {
  229. t.Fatal(err)
  230. }
  231. tests := []struct {
  232. rev revision
  233. r indexRangeResp
  234. rr rangeResp
  235. wkey []byte
  236. wrev revision
  237. wrrev int64
  238. wdelrev revision
  239. }{
  240. {
  241. revision{2, 0},
  242. indexRangeResp{[][]byte{[]byte("foo")}, []revision{{2, 0}}},
  243. rangeResp{[][]byte{key}, [][]byte{kvb}},
  244. newTestKeyBytes(revision{3, 0}, true),
  245. revision{3, 0},
  246. 2,
  247. revision{3, 0},
  248. },
  249. }
  250. for i, tt := range tests {
  251. s := newFakeStore()
  252. b := s.b.(*fakeBackend)
  253. fi := s.kvindex.(*fakeIndex)
  254. s.currentRev = tt.rev.main
  255. fi.indexRangeRespc <- tt.r
  256. b.tx.rangeRespc <- tt.rr
  257. n, _ := s.DeleteRange([]byte("foo"), []byte("goo"))
  258. if n != 1 {
  259. t.Errorf("#%d: n = %d, want 1", i, n)
  260. }
  261. data, err := (&mvccpb.KeyValue{
  262. Key: []byte("foo"),
  263. }).Marshal()
  264. if err != nil {
  265. t.Errorf("#%d: marshal err = %v, want nil", i, err)
  266. }
  267. wact := []testutil.Action{
  268. {Name: "seqput", Params: []interface{}{keyBucketName, tt.wkey, data}},
  269. }
  270. if g := b.tx.Action(); !reflect.DeepEqual(g, wact) {
  271. t.Errorf("#%d: tx action = %+v, want %+v", i, g, wact)
  272. }
  273. wact = []testutil.Action{
  274. {Name: "range", Params: []interface{}{[]byte("foo"), []byte("goo"), tt.wrrev}},
  275. {Name: "tombstone", Params: []interface{}{[]byte("foo"), tt.wdelrev}},
  276. }
  277. if g := fi.Action(); !reflect.DeepEqual(g, wact) {
  278. t.Errorf("#%d: index action = %+v, want %+v", i, g, wact)
  279. }
  280. if s.currentRev != tt.wrev.main {
  281. t.Errorf("#%d: rev = %+v, want %+v", i, s.currentRev, tt.wrev)
  282. }
  283. }
  284. }
  285. func TestStoreCompact(t *testing.T) {
  286. s := newFakeStore()
  287. defer s.Close()
  288. b := s.b.(*fakeBackend)
  289. fi := s.kvindex.(*fakeIndex)
  290. s.currentRev = 3
  291. fi.indexCompactRespc <- map[revision]struct{}{{1, 0}: {}}
  292. key1 := newTestKeyBytes(revision{1, 0}, false)
  293. key2 := newTestKeyBytes(revision{2, 0}, false)
  294. b.tx.rangeRespc <- rangeResp{[][]byte{key1, key2}, nil}
  295. s.Compact(3)
  296. s.fifoSched.WaitFinish(1)
  297. if s.compactMainRev != 3 {
  298. t.Errorf("compact main rev = %d, want 3", s.compactMainRev)
  299. }
  300. end := make([]byte, 8)
  301. binary.BigEndian.PutUint64(end, uint64(4))
  302. wact := []testutil.Action{
  303. {Name: "put", Params: []interface{}{metaBucketName, scheduledCompactKeyName, newTestRevBytes(revision{3, 0})}},
  304. {Name: "range", Params: []interface{}{keyBucketName, make([]byte, 17), end, int64(10000)}},
  305. {Name: "delete", Params: []interface{}{keyBucketName, key2}},
  306. {Name: "put", Params: []interface{}{metaBucketName, finishedCompactKeyName, newTestRevBytes(revision{3, 0})}},
  307. }
  308. if g := b.tx.Action(); !reflect.DeepEqual(g, wact) {
  309. t.Errorf("tx actions = %+v, want %+v", g, wact)
  310. }
  311. wact = []testutil.Action{
  312. {Name: "compact", Params: []interface{}{int64(3)}},
  313. }
  314. if g := fi.Action(); !reflect.DeepEqual(g, wact) {
  315. t.Errorf("index action = %+v, want %+v", g, wact)
  316. }
  317. }
  318. func TestStoreRestore(t *testing.T) {
  319. s := newFakeStore()
  320. b := s.b.(*fakeBackend)
  321. fi := s.kvindex.(*fakeIndex)
  322. putkey := newTestKeyBytes(revision{3, 0}, false)
  323. putkv := mvccpb.KeyValue{
  324. Key: []byte("foo"),
  325. Value: []byte("bar"),
  326. CreateRevision: 4,
  327. ModRevision: 4,
  328. Version: 1,
  329. }
  330. putkvb, err := putkv.Marshal()
  331. if err != nil {
  332. t.Fatal(err)
  333. }
  334. delkey := newTestKeyBytes(revision{5, 0}, true)
  335. delkv := mvccpb.KeyValue{
  336. Key: []byte("foo"),
  337. }
  338. delkvb, err := delkv.Marshal()
  339. if err != nil {
  340. t.Fatal(err)
  341. }
  342. b.tx.rangeRespc <- rangeResp{[][]byte{finishedCompactKeyName}, [][]byte{newTestRevBytes(revision{3, 0})}}
  343. b.tx.rangeRespc <- rangeResp{[][]byte{scheduledCompactKeyName}, [][]byte{newTestRevBytes(revision{3, 0})}}
  344. b.tx.rangeRespc <- rangeResp{[][]byte{putkey, delkey}, [][]byte{putkvb, delkvb}}
  345. b.tx.rangeRespc <- rangeResp{nil, nil}
  346. s.restore()
  347. if s.compactMainRev != 3 {
  348. t.Errorf("compact rev = %d, want 5", s.compactMainRev)
  349. }
  350. if s.currentRev != 5 {
  351. t.Errorf("current rev = %v, want 5", s.currentRev)
  352. }
  353. wact := []testutil.Action{
  354. {Name: "range", Params: []interface{}{metaBucketName, finishedCompactKeyName, []byte(nil), int64(0)}},
  355. {Name: "range", Params: []interface{}{metaBucketName, scheduledCompactKeyName, []byte(nil), int64(0)}},
  356. {Name: "range", Params: []interface{}{keyBucketName, newTestRevBytes(revision{1, 0}), newTestRevBytes(revision{math.MaxInt64, math.MaxInt64}), int64(restoreChunkKeys)}},
  357. }
  358. if g := b.tx.Action(); !reflect.DeepEqual(g, wact) {
  359. t.Errorf("tx actions = %+v, want %+v", g, wact)
  360. }
  361. gens := []generation{
  362. {created: revision{4, 0}, ver: 2, revs: []revision{{3, 0}, {5, 0}}},
  363. {created: revision{0, 0}, ver: 0, revs: nil},
  364. }
  365. ki := &keyIndex{key: []byte("foo"), modified: revision{5, 0}, generations: gens}
  366. wact = []testutil.Action{
  367. {Name: "keyIndex", Params: []interface{}{ki}},
  368. {Name: "insert", Params: []interface{}{ki}},
  369. }
  370. if g := fi.Action(); !reflect.DeepEqual(g, wact) {
  371. t.Errorf("index action = %+v, want %+v", g, wact)
  372. }
  373. }
  374. func TestRestoreDelete(t *testing.T) {
  375. oldChunk := restoreChunkKeys
  376. restoreChunkKeys = mrand.Intn(3) + 2
  377. defer func() { restoreChunkKeys = oldChunk }()
  378. b, tmpPath := backend.NewDefaultTmpBackend()
  379. s := NewStore(zap.NewExample(), b, &lease.FakeLessor{}, nil)
  380. defer os.Remove(tmpPath)
  381. keys := make(map[string]struct{})
  382. for i := 0; i < 20; i++ {
  383. ks := fmt.Sprintf("foo-%d", i)
  384. k := []byte(ks)
  385. s.Put(k, []byte("bar"), lease.NoLease)
  386. keys[ks] = struct{}{}
  387. switch mrand.Intn(3) {
  388. case 0:
  389. // put random key from past via random range on map
  390. ks = fmt.Sprintf("foo-%d", mrand.Intn(i+1))
  391. s.Put([]byte(ks), []byte("baz"), lease.NoLease)
  392. keys[ks] = struct{}{}
  393. case 1:
  394. // delete random key via random range on map
  395. for k := range keys {
  396. s.DeleteRange([]byte(k), nil)
  397. delete(keys, k)
  398. break
  399. }
  400. }
  401. }
  402. s.Close()
  403. s = NewStore(zap.NewExample(), b, &lease.FakeLessor{}, nil)
  404. defer s.Close()
  405. for i := 0; i < 20; i++ {
  406. ks := fmt.Sprintf("foo-%d", i)
  407. r, err := s.Range([]byte(ks), nil, RangeOptions{})
  408. if err != nil {
  409. t.Fatal(err)
  410. }
  411. if _, ok := keys[ks]; ok {
  412. if len(r.KVs) == 0 {
  413. t.Errorf("#%d: expected %q, got deleted", i, ks)
  414. }
  415. } else if len(r.KVs) != 0 {
  416. t.Errorf("#%d: expected deleted, got %q", i, ks)
  417. }
  418. }
  419. }
  420. func TestRestoreContinueUnfinishedCompaction(t *testing.T) {
  421. b, tmpPath := backend.NewDefaultTmpBackend()
  422. s0 := NewStore(zap.NewExample(), b, &lease.FakeLessor{}, nil)
  423. defer os.Remove(tmpPath)
  424. s0.Put([]byte("foo"), []byte("bar"), lease.NoLease)
  425. s0.Put([]byte("foo"), []byte("bar1"), lease.NoLease)
  426. s0.Put([]byte("foo"), []byte("bar2"), lease.NoLease)
  427. // write scheduled compaction, but not do compaction
  428. rbytes := newRevBytes()
  429. revToBytes(revision{main: 2}, rbytes)
  430. tx := s0.b.BatchTx()
  431. tx.Lock()
  432. tx.UnsafePut(metaBucketName, scheduledCompactKeyName, rbytes)
  433. tx.Unlock()
  434. s0.Close()
  435. s1 := NewStore(zap.NewExample(), b, &lease.FakeLessor{}, nil)
  436. // wait for scheduled compaction to be finished
  437. time.Sleep(100 * time.Millisecond)
  438. if _, err := s1.Range([]byte("foo"), nil, RangeOptions{Rev: 1}); err != ErrCompacted {
  439. t.Errorf("range on compacted rev error = %v, want %v", err, ErrCompacted)
  440. }
  441. // check the key in backend is deleted
  442. revbytes := newRevBytes()
  443. revToBytes(revision{main: 1}, revbytes)
  444. // The disk compaction is done asynchronously and requires more time on slow disk.
  445. // try 5 times for CI with slow IO.
  446. for i := 0; i < 5; i++ {
  447. tx = s1.b.BatchTx()
  448. tx.Lock()
  449. ks, _ := tx.UnsafeRange(keyBucketName, revbytes, nil, 0)
  450. tx.Unlock()
  451. if len(ks) != 0 {
  452. time.Sleep(100 * time.Millisecond)
  453. continue
  454. }
  455. return
  456. }
  457. t.Errorf("key for rev %+v still exists, want deleted", bytesToRev(revbytes))
  458. }
  459. type hashKVResult struct {
  460. hash uint32
  461. compactRev int64
  462. }
  463. // TestHashKVWhenCompacting ensures that HashKV returns correct hash when compacting.
  464. func TestHashKVWhenCompacting(t *testing.T) {
  465. b, tmpPath := backend.NewDefaultTmpBackend()
  466. s := NewStore(zap.NewExample(), b, &lease.FakeLessor{}, nil)
  467. defer os.Remove(tmpPath)
  468. rev := 10000
  469. for i := 2; i <= rev; i++ {
  470. s.Put([]byte("foo"), []byte(fmt.Sprintf("bar%d", i)), lease.NoLease)
  471. }
  472. hashCompactc := make(chan hashKVResult, 1)
  473. donec := make(chan struct{})
  474. var wg sync.WaitGroup
  475. for i := 0; i < 10; i++ {
  476. wg.Add(1)
  477. go func() {
  478. defer wg.Done()
  479. for {
  480. hash, _, compactRev, err := s.HashByRev(int64(rev))
  481. if err != nil {
  482. t.Fatal(err)
  483. }
  484. select {
  485. case <-donec:
  486. return
  487. case hashCompactc <- hashKVResult{hash, compactRev}:
  488. }
  489. }
  490. }()
  491. }
  492. go func() {
  493. defer close(donec)
  494. revHash := make(map[int64]uint32)
  495. for round := 0; round < 1000; round++ {
  496. r := <-hashCompactc
  497. if revHash[r.compactRev] == 0 {
  498. revHash[r.compactRev] = r.hash
  499. }
  500. if r.hash != revHash[r.compactRev] {
  501. t.Fatalf("Hashes differ (current %v) != (saved %v)", r.hash, revHash[r.compactRev])
  502. }
  503. }
  504. }()
  505. wg.Add(1)
  506. go func() {
  507. defer wg.Done()
  508. for i := 100; i >= 0; i-- {
  509. _, err := s.Compact(int64(rev - 1 - i))
  510. if err != nil {
  511. t.Fatal(err)
  512. }
  513. time.Sleep(10 * time.Millisecond)
  514. }
  515. }()
  516. select {
  517. case <-donec:
  518. wg.Wait()
  519. case <-time.After(10 * time.Second):
  520. testutil.FatalStack(t, "timeout")
  521. }
  522. }
  523. // TestHashKVZeroRevision ensures that "HashByRev(0)" computes
  524. // correct hash value with latest revision.
  525. func TestHashKVZeroRevision(t *testing.T) {
  526. b, tmpPath := backend.NewDefaultTmpBackend()
  527. s := NewStore(zap.NewExample(), b, &lease.FakeLessor{}, nil)
  528. defer os.Remove(tmpPath)
  529. rev := 1000
  530. for i := 2; i <= rev; i++ {
  531. s.Put([]byte("foo"), []byte(fmt.Sprintf("bar%d", i)), lease.NoLease)
  532. }
  533. if _, err := s.Compact(int64(rev / 2)); err != nil {
  534. t.Fatal(err)
  535. }
  536. hash1, _, _, err := s.HashByRev(int64(rev))
  537. if err != nil {
  538. t.Fatal(err)
  539. }
  540. var hash2 uint32
  541. hash2, _, _, err = s.HashByRev(0)
  542. if err != nil {
  543. t.Fatal(err)
  544. }
  545. if hash1 != hash2 {
  546. t.Errorf("hash %d (rev %d) != hash %d (rev 0)", hash1, rev, hash2)
  547. }
  548. }
  549. func TestTxnPut(t *testing.T) {
  550. // assign arbitrary size
  551. bytesN := 30
  552. sliceN := 100
  553. keys := createBytesSlice(bytesN, sliceN)
  554. vals := createBytesSlice(bytesN, sliceN)
  555. b, tmpPath := backend.NewDefaultTmpBackend()
  556. s := NewStore(zap.NewExample(), b, &lease.FakeLessor{}, nil)
  557. defer cleanup(s, b, tmpPath)
  558. for i := 0; i < sliceN; i++ {
  559. txn := s.Write()
  560. base := int64(i + 2)
  561. if rev := txn.Put(keys[i], vals[i], lease.NoLease); rev != base {
  562. t.Errorf("#%d: rev = %d, want %d", i, rev, base)
  563. }
  564. txn.End()
  565. }
  566. }
  567. func TestTxnBlockBackendForceCommit(t *testing.T) {
  568. b, tmpPath := backend.NewDefaultTmpBackend()
  569. s := NewStore(zap.NewExample(), b, &lease.FakeLessor{}, nil)
  570. defer os.Remove(tmpPath)
  571. txn := s.Read()
  572. done := make(chan struct{})
  573. go func() {
  574. s.b.ForceCommit()
  575. done <- struct{}{}
  576. }()
  577. select {
  578. case <-done:
  579. t.Fatalf("failed to block ForceCommit")
  580. case <-time.After(100 * time.Millisecond):
  581. }
  582. txn.End()
  583. select {
  584. case <-done:
  585. case <-time.After(5 * time.Second): // wait 5 seconds for CI with slow IO
  586. testutil.FatalStack(t, "failed to execute ForceCommit")
  587. }
  588. }
  589. // TODO: test attach key to lessor
  590. func newTestRevBytes(rev revision) []byte {
  591. bytes := newRevBytes()
  592. revToBytes(rev, bytes)
  593. return bytes
  594. }
  595. func newTestKeyBytes(rev revision, tombstone bool) []byte {
  596. bytes := newRevBytes()
  597. revToBytes(rev, bytes)
  598. if tombstone {
  599. bytes = appendMarkTombstone(zap.NewExample(), bytes)
  600. }
  601. return bytes
  602. }
  603. func newFakeStore() *store {
  604. b := &fakeBackend{&fakeBatchTx{
  605. Recorder: &testutil.RecorderBuffered{},
  606. rangeRespc: make(chan rangeResp, 5)}}
  607. fi := &fakeIndex{
  608. Recorder: &testutil.RecorderBuffered{},
  609. indexGetRespc: make(chan indexGetResp, 1),
  610. indexRangeRespc: make(chan indexRangeResp, 1),
  611. indexRangeEventsRespc: make(chan indexRangeEventsResp, 1),
  612. indexCompactRespc: make(chan map[revision]struct{}, 1),
  613. }
  614. s := &store{
  615. b: b,
  616. le: &lease.FakeLessor{},
  617. kvindex: fi,
  618. currentRev: 0,
  619. compactMainRev: -1,
  620. fifoSched: schedule.NewFIFOScheduler(),
  621. stopc: make(chan struct{}),
  622. lg: zap.NewExample(),
  623. }
  624. s.ReadView, s.WriteView = &readView{s}, &writeView{s}
  625. return s
  626. }
  627. type rangeResp struct {
  628. keys [][]byte
  629. vals [][]byte
  630. }
  631. type fakeBatchTx struct {
  632. testutil.Recorder
  633. rangeRespc chan rangeResp
  634. }
  635. func (b *fakeBatchTx) Lock() {}
  636. func (b *fakeBatchTx) Unlock() {}
  637. func (b *fakeBatchTx) UnsafeCreateBucket(name []byte) {}
  638. func (b *fakeBatchTx) UnsafePut(bucketName []byte, key []byte, value []byte) {
  639. b.Recorder.Record(testutil.Action{Name: "put", Params: []interface{}{bucketName, key, value}})
  640. }
  641. func (b *fakeBatchTx) UnsafeSeqPut(bucketName []byte, key []byte, value []byte) {
  642. b.Recorder.Record(testutil.Action{Name: "seqput", Params: []interface{}{bucketName, key, value}})
  643. }
  644. func (b *fakeBatchTx) UnsafeRange(bucketName []byte, key, endKey []byte, limit int64) (keys [][]byte, vals [][]byte) {
  645. b.Recorder.Record(testutil.Action{Name: "range", Params: []interface{}{bucketName, key, endKey, limit}})
  646. r := <-b.rangeRespc
  647. return r.keys, r.vals
  648. }
  649. func (b *fakeBatchTx) UnsafeDelete(bucketName []byte, key []byte) {
  650. b.Recorder.Record(testutil.Action{Name: "delete", Params: []interface{}{bucketName, key}})
  651. }
  652. func (b *fakeBatchTx) UnsafeForEach(bucketName []byte, visitor func(k, v []byte) error) error {
  653. return nil
  654. }
  655. func (b *fakeBatchTx) Commit() {}
  656. func (b *fakeBatchTx) CommitAndStop() {}
  657. type fakeBackend struct {
  658. tx *fakeBatchTx
  659. }
  660. func (b *fakeBackend) BatchTx() backend.BatchTx { return b.tx }
  661. func (b *fakeBackend) ReadTx() backend.ReadTx { return b.tx }
  662. func (b *fakeBackend) Hash(ignores map[backend.IgnoreKey]struct{}) (uint32, error) { return 0, nil }
  663. func (b *fakeBackend) Size() int64 { return 0 }
  664. func (b *fakeBackend) SizeInUse() int64 { return 0 }
  665. func (b *fakeBackend) Snapshot() backend.Snapshot { return nil }
  666. func (b *fakeBackend) ForceCommit() {}
  667. func (b *fakeBackend) Defrag() error { return nil }
  668. func (b *fakeBackend) Close() error { return nil }
  669. type indexGetResp struct {
  670. rev revision
  671. created revision
  672. ver int64
  673. err error
  674. }
  675. type indexRangeResp struct {
  676. keys [][]byte
  677. revs []revision
  678. }
  679. type indexRangeEventsResp struct {
  680. revs []revision
  681. }
  682. type fakeIndex struct {
  683. testutil.Recorder
  684. indexGetRespc chan indexGetResp
  685. indexRangeRespc chan indexRangeResp
  686. indexRangeEventsRespc chan indexRangeEventsResp
  687. indexCompactRespc chan map[revision]struct{}
  688. }
  689. func (i *fakeIndex) Revisions(key, end []byte, atRev int64) []revision {
  690. _, rev := i.Range(key, end, atRev)
  691. return rev
  692. }
  693. func (i *fakeIndex) Get(key []byte, atRev int64) (rev, created revision, ver int64, err error) {
  694. i.Recorder.Record(testutil.Action{Name: "get", Params: []interface{}{key, atRev}})
  695. r := <-i.indexGetRespc
  696. return r.rev, r.created, r.ver, r.err
  697. }
  698. func (i *fakeIndex) Range(key, end []byte, atRev int64) ([][]byte, []revision) {
  699. i.Recorder.Record(testutil.Action{Name: "range", Params: []interface{}{key, end, atRev}})
  700. r := <-i.indexRangeRespc
  701. return r.keys, r.revs
  702. }
  703. func (i *fakeIndex) Put(key []byte, rev revision) {
  704. i.Recorder.Record(testutil.Action{Name: "put", Params: []interface{}{key, rev}})
  705. }
  706. func (i *fakeIndex) Tombstone(key []byte, rev revision) error {
  707. i.Recorder.Record(testutil.Action{Name: "tombstone", Params: []interface{}{key, rev}})
  708. return nil
  709. }
  710. func (i *fakeIndex) RangeSince(key, end []byte, rev int64) []revision {
  711. i.Recorder.Record(testutil.Action{Name: "rangeEvents", Params: []interface{}{key, end, rev}})
  712. r := <-i.indexRangeEventsRespc
  713. return r.revs
  714. }
  715. func (i *fakeIndex) Compact(rev int64) map[revision]struct{} {
  716. i.Recorder.Record(testutil.Action{Name: "compact", Params: []interface{}{rev}})
  717. return <-i.indexCompactRespc
  718. }
  719. func (i *fakeIndex) Keep(rev int64) map[revision]struct{} {
  720. i.Recorder.Record(testutil.Action{Name: "keep", Params: []interface{}{rev}})
  721. return <-i.indexCompactRespc
  722. }
  723. func (i *fakeIndex) Equal(b index) bool { return false }
  724. func (i *fakeIndex) Insert(ki *keyIndex) {
  725. i.Recorder.Record(testutil.Action{Name: "insert", Params: []interface{}{ki}})
  726. }
  727. func (i *fakeIndex) KeyIndex(ki *keyIndex) *keyIndex {
  728. i.Recorder.Record(testutil.Action{Name: "keyIndex", Params: []interface{}{ki}})
  729. return nil
  730. }
  731. func createBytesSlice(bytesN, sliceN int) [][]byte {
  732. rs := [][]byte{}
  733. for len(rs) != sliceN {
  734. v := make([]byte, bytesN)
  735. if _, err := rand.Read(v); err != nil {
  736. panic(err)
  737. }
  738. rs = append(rs, v)
  739. }
  740. return rs
  741. }