kvstore_test.go 22 KB

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