distributedstorage_fetch_test.go 4.2 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169
  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 TestSimpleDataReplicationFetch(t *testing.T) {
  18. // Set a low distribution range
  19. defaultDistributionRange = 5000
  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 != 1666 || 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: 1666 (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: 1666 (v:1) - lloc: 1 - "\b\f\x00\x05test2"
  76. `[1:] && res != `
  77. TestClusterMember-0 MemberStorageManager mgs1/ls_test
  78. Roots: 0=0 1=0 2=0 3=0 4=0 5=0 6=0 7=0 8=0 9=0
  79. cloc: 0 (v:1) - lloc: 1 - "\b\f\x00\x05test1"
  80. TestClusterMember-1 MemberStorageManager mgs2/ls_test
  81. Roots: 0=0 1=0 2=0 3=0 4=0 5=0 6=0 7=0 8=0 9=0
  82. cloc: 1666 (v:1) - lloc: 1 - "\b\f\x00\x05test2"
  83. cloc: 0 (v:1) - lloc: 2 - "\b\f\x00\x05test1"
  84. TestClusterMember-2 MemberStorageManager mgs3/ls_test
  85. Roots: 0=0 1=0 2=0 3=0 4=0 5=0 6=0 7=0 8=0 9=0
  86. cloc: 1666 (v:1) - lloc: 1 - "\b\f\x00\x05test2"
  87. `[1:] {
  88. t.Error("Unexpected cluster storage layout: ", res)
  89. return
  90. }
  91. // Now do a normal fetch
  92. var ret string
  93. if err := sm.Fetch(0, &ret); err != nil || ret != "test1" {
  94. t.Error("Unexpected result:", err, ret)
  95. return
  96. }
  97. if err := sm.Fetch(1666, &ret); err != nil || ret != "test2" {
  98. t.Error("Unexpected result:", err, ret)
  99. return
  100. }
  101. // Simulate an error on member 0
  102. manager.MemberErrors = make(map[string]error)
  103. defer func() { manager.MemberErrors = nil }()
  104. manager.MemberErrors[cluster3[0].MemberManager.Name()] = &testNetError{}
  105. cluster3[0].MemberManager.StopHousekeeping = true
  106. defer func() { cluster3[0].MemberManager.StopHousekeeping = false }()
  107. if err := sm.Fetch(0, &ret); err != nil || ret != "test1" {
  108. t.Error("Unexpected result:", err, ret)
  109. return
  110. }
  111. if err := sm.Fetch(1666, &ret); err != nil || ret != "test2" {
  112. t.Error("Unexpected result:", err, ret)
  113. return
  114. }
  115. sm = cluster3[2].StorageManager("test", false)
  116. if err := sm.Fetch(0, &ret); err != nil || ret != "test1" {
  117. t.Error("Unexpected result:", err, ret)
  118. return
  119. }
  120. if err := sm.Fetch(1666, &ret); err != nil || ret != "test2" {
  121. t.Error("Unexpected result:", err, ret)
  122. return
  123. }
  124. if err := sm.Fetch(1667, &ret); err.Error() !=
  125. "ClusterError: Member error (Cluster slot not found (TestClusterMember-1 - Location: 1667))" {
  126. t.Error("Unexpected result:", err)
  127. return
  128. }
  129. }