distributedstorage_update_test.go 4.6 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174
  1. /*
  2. * EliasDB
  3. *
  4. * Copyright 2016 Matthias Ladkau. All rights reserved.
  5. *
  6. * This Source Code Form is subject to the terms of the Mozilla Public
  7. * License, v. 2.0. If a copy of the MPL was not distributed with this
  8. * file, You can obtain one at http://mozilla.org/MPL/2.0/.
  9. */
  10. package cluster
  11. import (
  12. "math"
  13. "testing"
  14. "time"
  15. "devt.de/krotik/eliasdb/cluster/manager"
  16. )
  17. func TestSimpleDataReplicationUpdate(t *testing.T) {
  18. // Set a low distribution range
  19. defaultDistributionRange = 10
  20. defer func() { defaultDistributionRange = math.MaxUint64 }()
  21. // Setup a cluster
  22. manager.FreqHousekeeping = 5
  23. defer func() { manager.FreqHousekeeping = 1000 }()
  24. // Log transfer worker runs
  25. logTransferWorker = true
  26. defer func() { logTransferWorker = false }()
  27. // Create a cluster with 3 members and a replication factor of 2
  28. cluster3, ms := createCluster(3, 2)
  29. // Debug output
  30. // manager.LogDebug = manager.LogInfo
  31. // log.SetOutput(os.Stderr)
  32. // defer func() { log.SetOutput(ioutil.Discard) }()
  33. for i, dd := range cluster3 {
  34. dd.Start()
  35. defer dd.Close()
  36. if i > 0 {
  37. err := dd.MemberManager.JoinCluster(cluster3[0].MemberManager.Name(), cluster3[0].MemberManager.NetAddr())
  38. if err != nil {
  39. t.Error(err)
  40. return
  41. }
  42. }
  43. }
  44. sm := cluster3[1].StorageManager("test", true)
  45. // Insert two strings into the store
  46. if loc, err := sm.Insert("test1"); loc != 0 || err != nil {
  47. t.Error("Unexpected result:", loc, err)
  48. return
  49. }
  50. sm.Flush()
  51. time.Sleep(10 * time.Millisecond)
  52. if loc, err := sm.Insert("test2"); loc != 3 || err != nil {
  53. t.Error("Unexpected result:", loc, err)
  54. return
  55. }
  56. sm.Flush()
  57. // Ensure the transfer worker is running on all members
  58. for _, m := range ms {
  59. m.transferWorker()
  60. for m.transferRunning {
  61. time.Sleep(time.Millisecond)
  62. }
  63. }
  64. // Check that we have a certain storage layout in the cluster
  65. if res := clusterLayout(ms, "test"); res != `
  66. TestClusterMember-0 MemberStorageManager mgs1/ls_test
  67. Roots: 0=0 1=0 2=0 3=0 4=0 5=0 6=0 7=0 8=0 9=0
  68. cloc: 0 (v:1) - lloc: 1 - "\b\f\x00\x05test1"
  69. TestClusterMember-1 MemberStorageManager mgs2/ls_test
  70. Roots: 0=0 1=0 2=0 3=0 4=0 5=0 6=0 7=0 8=0 9=0
  71. cloc: 0 (v:1) - lloc: 1 - "\b\f\x00\x05test1"
  72. cloc: 3 (v:1) - lloc: 2 - "\b\f\x00\x05test2"
  73. TestClusterMember-2 MemberStorageManager mgs3/ls_test
  74. Roots: 0=0 1=0 2=0 3=0 4=0 5=0 6=0 7=0 8=0 9=0
  75. cloc: 3 (v:1) - lloc: 1 - "\b\f\x00\x05test2"
  76. `[1:] {
  77. t.Error("Unexpected cluster storage layout: ", res)
  78. return
  79. }
  80. // Simulate a failure on member 0
  81. manager.MemberErrors = make(map[string]error)
  82. defer func() { manager.MemberErrors = nil }()
  83. manager.MemberErrors[cluster3[0].MemberManager.Name()] = &testNetError{}
  84. cluster3[0].MemberManager.StopHousekeeping = true
  85. if err := sm.Update(0, "test1updated"); err != nil {
  86. t.Error("Unexpected result:", err)
  87. return
  88. }
  89. // Check the expected cluster layout. The record should be updated
  90. // on member 1 and a transfer record for member 0
  91. if res := clusterLayout(ms, "test"); res != `
  92. TestClusterMember-0 MemberStorageManager mgs1/ls_test
  93. Roots: 0=0 1=0 2=0 3=0 4=0 5=0 6=0 7=0 8=0 9=0
  94. cloc: 0 (v:1) - lloc: 1 - "\b\f\x00\x05test1"
  95. TestClusterMember-1 MemberStorageManager mgs2/ls_test
  96. Roots: 0=0 1=0 2=0 3=0 4=0 5=0 6=0 7=0 8=0 9=0
  97. cloc: 0 (v:2) - lloc: 1 - "\x0f\f\x00\ftest1updated"
  98. cloc: 3 (v:1) - lloc: 2 - "\b\f\x00\x05test2"
  99. transfer: [TestClusterMember-0] - Update {"Loc":0,"StoreName":"test","Ver":2} "\x0f\f\x00\ftest1updated"
  100. TestClusterMember-2 MemberStorageManager mgs3/ls_test
  101. Roots: 0=0 1=0 2=0 3=0 4=0 5=0 6=0 7=0 8=0 9=0
  102. cloc: 3 (v:1) - lloc: 1 - "\b\f\x00\x05test2"
  103. `[1:] {
  104. t.Error("Unexpected cluster storage layout: ", res)
  105. return
  106. }
  107. // Simulate that member 0 comes back
  108. delete(manager.MemberErrors, cluster3[0].MemberManager.Name())
  109. cluster3[0].MemberManager.StopHousekeeping = false
  110. // Make sure the transfer worker runs
  111. // Ensure the transfer worker is running on all members
  112. for _, m := range ms {
  113. m.transferWorker()
  114. for m.transferRunning {
  115. time.Sleep(time.Millisecond)
  116. }
  117. }
  118. // Check the expected cluster layout. All should be updated now.
  119. if res := clusterLayout(ms, "test"); res != `
  120. TestClusterMember-0 MemberStorageManager mgs1/ls_test
  121. Roots: 0=0 1=0 2=0 3=0 4=0 5=0 6=0 7=0 8=0 9=0
  122. cloc: 0 (v:2) - lloc: 1 - "\x0f\f\x00\ftest1updated"
  123. TestClusterMember-1 MemberStorageManager mgs2/ls_test
  124. Roots: 0=0 1=0 2=0 3=0 4=0 5=0 6=0 7=0 8=0 9=0
  125. cloc: 0 (v:2) - lloc: 1 - "\x0f\f\x00\ftest1updated"
  126. cloc: 3 (v:1) - lloc: 2 - "\b\f\x00\x05test2"
  127. TestClusterMember-2 MemberStorageManager mgs3/ls_test
  128. Roots: 0=0 1=0 2=0 3=0 4=0 5=0 6=0 7=0 8=0 9=0
  129. cloc: 3 (v:1) - lloc: 1 - "\b\f\x00\x05test2"
  130. `[1:] {
  131. t.Error("Unexpected cluster storage layout: ", res)
  132. return
  133. }
  134. }