mirror of
https://github.com/featurebasedb/featurebase.git
synced 2026-10-07 19:37:51 +00:00
32fec70816
This is a lot more complex than it sounds like it will be. We shut down the cache flush when a holder is closed, but if you're deleting an index, we don't check for that, and can have a cache flush still creating cache files in an index which could conceivably result in os.RemoteAll() failing. This shouldn't happen often, but it's happened at least once. To address this, first, we make sure that every tier of this operation bails as quickly as it can after the thing it's working on closes. Second, we retry RemoveAll. Unfortunately, some things get reopened, so we have to handle that, have mutexes covering the access to the channel, and so on. Also, some things were getting double-closed, which was previously harmless but could now fail. So, first, catch all the existing double-closes and remove them, second, make the double-close fail with an error. Note that virtually none of the tests check for errors on close. This passes tests and should be unable to hit the original problem. Unfortunately, it's unreasonably hard to check that, because it requires an incredible coincidence of timing on the delete aligning with a cache flush.
625 lines
16 KiB
Go
625 lines
16 KiB
Go
// Copyright 2021 Molecula Corp. All rights reserved.
|
|
package pilosa
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"math/rand"
|
|
"reflect"
|
|
"strings"
|
|
"testing"
|
|
"testing/quick"
|
|
"time"
|
|
|
|
"github.com/davecgh/go-spew/spew"
|
|
pnet "github.com/molecula/featurebase/v3/net"
|
|
"github.com/molecula/featurebase/v3/roaring"
|
|
"github.com/molecula/featurebase/v3/testhook"
|
|
"github.com/molecula/featurebase/v3/topology"
|
|
. "github.com/molecula/featurebase/v3/vprint" // nolint:staticcheck
|
|
)
|
|
|
|
// Ensure that fragCombos creates the correct fragment mapping.
|
|
func TestFragCombos(t *testing.T) {
|
|
uri0, err := pnet.NewURIFromAddress("host0")
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
uri1, err := pnet.NewURIFromAddress("host1")
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
|
|
node0 := &topology.Node{ID: "node0", URI: *uri0}
|
|
node1 := &topology.Node{ID: "node1", URI: *uri1}
|
|
|
|
c := newCluster()
|
|
c.addNodeBasicSorted(node0)
|
|
c.addNodeBasicSorted(node1)
|
|
|
|
tests := []struct {
|
|
idx string
|
|
availableShards *roaring.Bitmap
|
|
fieldViews viewsByField
|
|
expected fragsByHost
|
|
}{
|
|
{
|
|
idx: "i",
|
|
availableShards: roaring.NewBitmap(0, 1, 2),
|
|
fieldViews: viewsByField{"f": []string{"v1", "v2"}},
|
|
expected: fragsByHost{
|
|
"node0": []frag{{"f", "v1", uint64(0)}, {"f", "v2", uint64(0)}},
|
|
"node1": []frag{{"f", "v1", uint64(1)}, {"f", "v2", uint64(1)}, {"f", "v1", uint64(2)}, {"f", "v2", uint64(2)}},
|
|
},
|
|
},
|
|
{
|
|
idx: "foo",
|
|
availableShards: roaring.NewBitmap(0, 1, 2, 3),
|
|
fieldViews: viewsByField{"f": []string{"v0"}},
|
|
expected: fragsByHost{
|
|
"node0": []frag{{"f", "v0", uint64(1)}, {"f", "v0", uint64(2)}},
|
|
"node1": []frag{{"f", "v0", uint64(0)}, {"f", "v0", uint64(3)}},
|
|
},
|
|
},
|
|
}
|
|
for _, test := range tests {
|
|
actual := c.fragCombos(test.idx, test.availableShards, test.fieldViews)
|
|
if !reflect.DeepEqual(actual, test.expected) {
|
|
t.Errorf("expected: %v, but got: %v", test.expected, actual)
|
|
}
|
|
|
|
}
|
|
}
|
|
|
|
// newHolderWithTempPath returns a new instance of Holder.
|
|
func newHolderWithTempPath(tb testing.TB, backend string) *Holder {
|
|
path, err := testhook.TempDirInDir(tb, *TempDir, "pilosa-holder-")
|
|
if err != nil {
|
|
panic(err)
|
|
}
|
|
cfg := mustHolderConfig()
|
|
cfg.StorageConfig.Backend = backend
|
|
h := NewHolder(path, cfg)
|
|
PanicOn(h.Open())
|
|
testhook.Cleanup(tb, func() {
|
|
h.Close()
|
|
})
|
|
return h
|
|
}
|
|
|
|
// newIndexWithTempPath returns a new instance of Index.
|
|
func newIndexWithTempPath(tb testing.TB, name string) *Index {
|
|
path, err := testhook.TempDirInDir(tb, *TempDir, "pilosa-index-")
|
|
if err != nil {
|
|
panic(err)
|
|
}
|
|
cfg := DefaultHolderConfig()
|
|
cfg.StorageConfig.FsyncEnabled = false
|
|
cfg.RBFConfig.FsyncEnabled = false
|
|
h := NewHolder(path, cfg)
|
|
PanicOn(h.Open())
|
|
index, err := h.CreateIndex(name, IndexOptions{})
|
|
testhook.Cleanup(tb, func() {
|
|
h.Close()
|
|
})
|
|
if err != nil {
|
|
panic(err)
|
|
}
|
|
return index
|
|
}
|
|
|
|
// Ensure that fragSources creates the correct fragment mapping.
|
|
func TestFragSources(t *testing.T) {
|
|
uri0, err := pnet.NewURIFromAddress("host0")
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
uri1, err := pnet.NewURIFromAddress("host1")
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
uri2, err := pnet.NewURIFromAddress("host2")
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
uri3, err := pnet.NewURIFromAddress("host3")
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
|
|
node0 := &topology.Node{ID: "node0", URI: *uri0}
|
|
node1 := &topology.Node{ID: "node1", URI: *uri1}
|
|
node2 := &topology.Node{ID: "node2", URI: *uri2}
|
|
node3 := &topology.Node{ID: "node3", URI: *uri3}
|
|
|
|
c1 := newCluster()
|
|
c1.ReplicaN = 1
|
|
c1.addNodeBasicSorted(node0)
|
|
c1.addNodeBasicSorted(node1)
|
|
|
|
c2 := newCluster()
|
|
c2.ReplicaN = 1
|
|
c2.addNodeBasicSorted(node0)
|
|
c2.addNodeBasicSorted(node1)
|
|
c2.addNodeBasicSorted(node2)
|
|
|
|
c3 := newCluster()
|
|
c3.ReplicaN = 2
|
|
c3.addNodeBasicSorted(node0)
|
|
c3.addNodeBasicSorted(node1)
|
|
|
|
c4 := newCluster()
|
|
c4.ReplicaN = 2
|
|
c4.addNodeBasicSorted(node0)
|
|
c4.addNodeBasicSorted(node1)
|
|
c4.addNodeBasicSorted(node2)
|
|
|
|
c5 := newCluster()
|
|
c5.ReplicaN = 2
|
|
c5.addNodeBasicSorted(node0)
|
|
c5.addNodeBasicSorted(node1)
|
|
c5.addNodeBasicSorted(node2)
|
|
c5.addNodeBasicSorted(node3)
|
|
|
|
idx := newIndexWithTempPath(t, "i")
|
|
|
|
field, err := idx.CreateFieldIfNotExists("f", OptFieldTypeDefault())
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
|
|
// Obtain transaction.
|
|
var shard uint64
|
|
tx := idx.holder.txf.NewTx(Txo{Write: writable, Index: idx, Shard: shard})
|
|
defer tx.Rollback()
|
|
|
|
_, err = field.SetBit(tx, 1, 101, nil)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
PanicOn(tx.Commit())
|
|
|
|
shard = 1
|
|
tx = idx.holder.txf.NewTx(Txo{Write: writable, Index: idx, Shard: shard})
|
|
defer tx.Rollback()
|
|
_, err = field.SetBit(tx, 1, ShardWidth*shard+1, nil)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
PanicOn(tx.Commit())
|
|
|
|
shard = 2
|
|
tx = idx.holder.txf.NewTx(Txo{Write: writable, Index: idx, Shard: shard})
|
|
defer tx.Rollback()
|
|
|
|
_, err = field.SetBit(tx, 1, ShardWidth*shard+1, nil)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
PanicOn(tx.Commit())
|
|
|
|
shard = 3
|
|
tx = idx.holder.txf.NewTx(Txo{Write: writable, Index: idx, Shard: shard})
|
|
defer tx.Rollback()
|
|
|
|
_, err = field.SetBit(tx, 1, ShardWidth*shard+1, nil)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
PanicOn(tx.Commit())
|
|
|
|
tests := []struct {
|
|
from *cluster
|
|
to *cluster
|
|
idx *Index
|
|
expected map[string][]*ResizeSource
|
|
err string
|
|
}{
|
|
{
|
|
from: c1,
|
|
to: c2,
|
|
idx: idx,
|
|
expected: map[string][]*ResizeSource{
|
|
"node0": {},
|
|
"node1": {},
|
|
"node2": {
|
|
{&topology.Node{ID: "node0", URI: pnet.URI{Scheme: "http", Host: "host0", Port: 10101}, IsPrimary: false}, "i", "f", "standard", uint64(0)},
|
|
{&topology.Node{ID: "node1", URI: pnet.URI{Scheme: "http", Host: "host1", Port: 10101}, IsPrimary: false}, "i", "f", "standard", uint64(2)},
|
|
},
|
|
},
|
|
err: "",
|
|
},
|
|
{
|
|
from: c4,
|
|
to: c3,
|
|
idx: idx,
|
|
expected: map[string][]*ResizeSource{
|
|
"node0": {
|
|
{&topology.Node{ID: "node1", URI: pnet.URI{Scheme: "http", Host: "host1", Port: 10101}, IsPrimary: false}, "i", "f", "standard", uint64(1)},
|
|
},
|
|
"node1": {
|
|
{&topology.Node{ID: "node0", URI: pnet.URI{Scheme: "http", Host: "host0", Port: 10101}, IsPrimary: false}, "i", "f", "standard", uint64(0)},
|
|
{&topology.Node{ID: "node0", URI: pnet.URI{Scheme: "http", Host: "host0", Port: 10101}, IsPrimary: false}, "i", "f", "standard", uint64(2)},
|
|
},
|
|
},
|
|
err: "",
|
|
},
|
|
{
|
|
from: c5,
|
|
to: c4,
|
|
idx: idx,
|
|
expected: map[string][]*ResizeSource{
|
|
"node0": {
|
|
{&topology.Node{ID: "node2", URI: pnet.URI{Scheme: "http", Host: "host2", Port: 10101}, IsPrimary: false}, "i", "f", "standard", uint64(0)},
|
|
{&topology.Node{ID: "node2", URI: pnet.URI{Scheme: "http", Host: "host2", Port: 10101}, IsPrimary: false}, "i", "f", "standard", uint64(2)},
|
|
},
|
|
"node1": {
|
|
{&topology.Node{ID: "node0", URI: pnet.URI{Scheme: "http", Host: "host0", Port: 10101}, IsPrimary: false}, "i", "f", "standard", uint64(3)},
|
|
},
|
|
"node2": {},
|
|
},
|
|
err: "",
|
|
},
|
|
{
|
|
from: c2,
|
|
to: c4,
|
|
idx: idx,
|
|
expected: nil,
|
|
err: "clusters are the same size",
|
|
},
|
|
{
|
|
from: c1,
|
|
to: c5,
|
|
idx: idx,
|
|
expected: nil,
|
|
err: "adding more than one node at a time is not supported",
|
|
},
|
|
{
|
|
from: c5,
|
|
to: c1,
|
|
idx: idx,
|
|
expected: nil,
|
|
err: "removing more than one node at a time is not supported",
|
|
},
|
|
}
|
|
for _, test := range tests {
|
|
|
|
actual, err := (test.from).fragSources(test.to, test.idx)
|
|
if test.err != "" {
|
|
if !strings.Contains(err.Error(), test.err) {
|
|
t.Fatalf("expected error: %s, got: %s", test.err, err.Error())
|
|
}
|
|
} else {
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if !reflect.DeepEqual(actual, test.expected) {
|
|
t.Errorf("expected: %v, but got: %v", test.expected, actual)
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
// Ensure that fragSources creates the correct fragment mapping.
|
|
func TestResizeJob(t *testing.T) {
|
|
|
|
uri0, err := pnet.NewURIFromAddress("host0")
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
uri1, err := pnet.NewURIFromAddress("host1")
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
uri2, err := pnet.NewURIFromAddress("host2")
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
|
|
node0 := &topology.Node{ID: "node0", URI: *uri0}
|
|
node1 := &topology.Node{ID: "node1", URI: *uri1}
|
|
node2 := &topology.Node{ID: "node2", URI: *uri2}
|
|
|
|
tests := []struct {
|
|
existingNodes []*topology.Node
|
|
node *topology.Node
|
|
action string
|
|
expectedIDs map[string]bool
|
|
}{
|
|
{
|
|
existingNodes: []*topology.Node{node0, node1},
|
|
node: node2,
|
|
action: resizeJobActionAdd,
|
|
expectedIDs: map[string]bool{node0.ID: false, node1.ID: false, node2.ID: false},
|
|
},
|
|
{
|
|
existingNodes: []*topology.Node{node0, node1, node2},
|
|
node: node2,
|
|
action: resizeJobActionRemove,
|
|
expectedIDs: map[string]bool{node0.ID: false, node1.ID: false},
|
|
},
|
|
}
|
|
for _, test := range tests {
|
|
|
|
actual := newResizeJob(test.existingNodes, test.node, test.action)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if !reflect.DeepEqual(actual.IDs, test.expectedIDs) {
|
|
t.Errorf("expected: %v, but got: %v", test.expectedIDs, actual.IDs)
|
|
}
|
|
}
|
|
}
|
|
|
|
// Ensure the cluster can fairly distribute partitions across the nodes.
|
|
func TestCluster_Owners(t *testing.T) {
|
|
c := cluster{
|
|
noder: topology.NewLocalNoder([]*topology.Node{
|
|
{URI: NewTestURIFromHostPort("serverA", 1000)},
|
|
{URI: NewTestURIFromHostPort("serverB", 1000)},
|
|
{URI: NewTestURIFromHostPort("serverC", 1000)},
|
|
}),
|
|
Hasher: NewTestModHasher(),
|
|
ReplicaN: 2,
|
|
}
|
|
|
|
cNodes := c.noder.Nodes()
|
|
|
|
// Create a snapshot of the cluster to use for node/partition calculations.
|
|
snap := topology.NewClusterSnapshot(c.noder, c.Hasher, c.ReplicaN)
|
|
|
|
// Verify nodes are distributed.
|
|
if a := snap.PartitionNodes(0); !reflect.DeepEqual(a, []*topology.Node{cNodes[0], cNodes[1]}) {
|
|
t.Fatalf("unexpected owners: %s", spew.Sdump(a))
|
|
}
|
|
|
|
// Verify nodes go around the ring.
|
|
if a := snap.PartitionNodes(2); !reflect.DeepEqual(a, []*topology.Node{cNodes[2], cNodes[0]}) {
|
|
t.Fatalf("unexpected owners: %s", spew.Sdump(a))
|
|
}
|
|
}
|
|
|
|
// Ensure the partitioner can assign a fragment to a partition.
|
|
func TestCluster_Partition(t *testing.T) {
|
|
if err := quick.Check(func(index string, shard uint64, partitionN int) bool {
|
|
c := newCluster()
|
|
c.partitionN = partitionN
|
|
|
|
partitionID := topology.ShardToShardPartition(index, shard, partitionN)
|
|
if partitionID < 0 || partitionID >= partitionN {
|
|
t.Errorf("partition out of range: shard=%d, p=%d, n=%d", shard, partitionID, partitionN)
|
|
}
|
|
|
|
return true
|
|
}, &quick.Config{
|
|
Values: func(values []reflect.Value, rand *rand.Rand) {
|
|
values[0], _ = quick.Value(reflect.TypeOf(""), rand)
|
|
values[1] = reflect.ValueOf(uint64(rand.Uint32()))
|
|
values[2] = reflect.ValueOf(rand.Intn(1000) + 1)
|
|
},
|
|
}); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
}
|
|
|
|
// Ensure the hasher can hash correctly.
|
|
func TestHasher(t *testing.T) {
|
|
for _, tt := range []struct {
|
|
key uint64
|
|
bucket []int
|
|
}{
|
|
// Generated from the reference C++ code
|
|
{0, []int{0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0}},
|
|
{1, []int{0, 0, 0, 0, 0, 0, 6, 6, 6, 6, 6, 6, 6, 6, 6, 6, 6, 17, 17}},
|
|
{0xdeadbeef, []int{0, 1, 2, 3, 3, 5, 5, 5, 5, 5, 5, 5, 5, 5, 5, 5, 16, 16, 16}},
|
|
{0x0ddc0ffeebadf00d, []int{0, 1, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 15, 15, 15, 15}},
|
|
} {
|
|
for i, v := range tt.bucket {
|
|
hasher := &topology.Jmphasher{}
|
|
if got := hasher.Hash(tt.key, i+1); got != v {
|
|
t.Errorf("hash(%v,%v)=%v, want %v", tt.key, i+1, got, v)
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
// Ensure ContainsShards can find the actual shard list for node and index.
|
|
func TestCluster_ContainsShards(t *testing.T) {
|
|
c := NewTestCluster(t, 5)
|
|
c.ReplicaN = 3
|
|
cNodes := c.noder.Nodes()
|
|
|
|
// Create a snapshot of the cluster to use for node/partition calculations.
|
|
snap := topology.NewClusterSnapshot(c.noder, c.Hasher, c.ReplicaN)
|
|
|
|
shards := snap.ContainsShards("test", roaring.NewBitmap(0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10), cNodes[2])
|
|
|
|
if !reflect.DeepEqual(shards, []uint64{0, 2, 3, 5, 6, 9, 10}) {
|
|
t.Fatalf("unexpected shars for node's index: %v", shards)
|
|
}
|
|
}
|
|
|
|
func TestCluster_Nodes(t *testing.T) {
|
|
const urisCount = 4
|
|
var uris []pnet.URI
|
|
arbitraryPorts := []int{17384, 17385, 17386, 17387}
|
|
for i := 0; i < urisCount; i++ {
|
|
uris = append(uris, NewTestURIFromHostPort(fmt.Sprintf("node%d", i), uint16(arbitraryPorts[i])))
|
|
}
|
|
|
|
node0 := &topology.Node{ID: "node0", URI: uris[0]}
|
|
node1 := &topology.Node{ID: "node1", URI: uris[1]}
|
|
node2 := &topology.Node{ID: "node2", URI: uris[2]}
|
|
node3 := &topology.Node{ID: "node3", URI: uris[3]}
|
|
|
|
nodes := []*topology.Node{node0, node1, node2}
|
|
|
|
t.Run("NodeIDs", func(t *testing.T) {
|
|
actual := topology.Nodes(nodes).IDs()
|
|
expected := []string{node0.ID, node1.ID, node2.ID}
|
|
if !reflect.DeepEqual(actual, expected) {
|
|
t.Errorf("expected: %v, but got: %v", expected, actual)
|
|
}
|
|
})
|
|
|
|
t.Run("Filter", func(t *testing.T) {
|
|
actual := topology.Nodes(topology.Nodes(nodes).Filter(nodes[1])).URIs()
|
|
expected := []pnet.URI{uris[0], uris[2]}
|
|
if !reflect.DeepEqual(actual, expected) {
|
|
t.Errorf("expected: %v, but got: %v", expected, actual)
|
|
}
|
|
})
|
|
|
|
t.Run("FilterURI", func(t *testing.T) {
|
|
actual := topology.Nodes(topology.Nodes(nodes).FilterURI(uris[1])).URIs()
|
|
expected := []pnet.URI{uris[0], uris[2]}
|
|
if !reflect.DeepEqual(actual, expected) {
|
|
t.Errorf("expected: %v, but got: %v", expected, actual)
|
|
}
|
|
})
|
|
|
|
t.Run("Contains", func(t *testing.T) {
|
|
actualTrue := topology.Nodes(nodes).Contains(node1)
|
|
actualFalse := topology.Nodes(nodes).Contains(node3)
|
|
if !reflect.DeepEqual(actualTrue, true) {
|
|
t.Errorf("expected: %v, but got: %v", true, actualTrue)
|
|
}
|
|
if !reflect.DeepEqual(actualFalse, false) {
|
|
t.Errorf("expected: %v, but got: %v", false, actualTrue)
|
|
}
|
|
})
|
|
|
|
t.Run("Clone", func(t *testing.T) {
|
|
clone := topology.Nodes(nodes).Clone()
|
|
actual := topology.Nodes(clone).URIs()
|
|
expected := []pnet.URI{uris[0], uris[1], uris[2]}
|
|
if !reflect.DeepEqual(actual, expected) {
|
|
t.Errorf("expected: %v, but got: %v", expected, actual)
|
|
}
|
|
})
|
|
}
|
|
|
|
func TestCluster_PreviousNode(t *testing.T) {
|
|
node0 := &topology.Node{ID: "node0"}
|
|
node1 := &topology.Node{ID: "node1"}
|
|
node2 := &topology.Node{ID: "node2"}
|
|
|
|
t.Run("OneNode", func(t *testing.T) {
|
|
c := newCluster()
|
|
c.addNodeBasicSorted(node0)
|
|
|
|
c.Node = node0
|
|
if prev := c.unprotectedPreviousNode(); prev != nil {
|
|
t.Errorf("expected: nil, but got: %v", prev)
|
|
}
|
|
})
|
|
|
|
t.Run("TwoNode", func(t *testing.T) {
|
|
c := newCluster()
|
|
c.addNodeBasicSorted(node0)
|
|
c.addNodeBasicSorted(node1)
|
|
|
|
c.Node = node0
|
|
if prev := c.unprotectedPreviousNode(); prev != node1 {
|
|
t.Errorf("expected: node1, but got: %v", prev)
|
|
}
|
|
|
|
c.Node = node1
|
|
if prev := c.unprotectedPreviousNode(); prev != node0 {
|
|
t.Errorf("expected: node0, but got: %v", prev)
|
|
}
|
|
})
|
|
|
|
t.Run("ThreeNode", func(t *testing.T) {
|
|
c := newCluster()
|
|
c.addNodeBasicSorted(node0)
|
|
c.addNodeBasicSorted(node1)
|
|
c.addNodeBasicSorted(node2)
|
|
|
|
c.Node = node0
|
|
if prev := c.unprotectedPreviousNode(); prev != node2 {
|
|
t.Errorf("expected: node2, but got: %v", prev)
|
|
}
|
|
|
|
c.Node = node1
|
|
if prev := c.unprotectedPreviousNode(); prev != node0 {
|
|
t.Errorf("expected: node0, but got: %v", prev)
|
|
}
|
|
|
|
c.Node = node2
|
|
if prev := c.unprotectedPreviousNode(); prev != node1 {
|
|
t.Errorf("expected: node1, but got: %v", prev)
|
|
}
|
|
})
|
|
}
|
|
|
|
func TestAE(t *testing.T) {
|
|
t.Run("AbortDoesn'tBlockUninitialized", func(t *testing.T) {
|
|
c := newCluster()
|
|
ch := make(chan struct{})
|
|
go func() {
|
|
c.abortAntiEntropy()
|
|
close(ch)
|
|
}()
|
|
defer c.abortAntiEntropyQ() // avoid leaking a goroutine.
|
|
select {
|
|
case <-ch:
|
|
return
|
|
case <-time.After(time.Second):
|
|
t.Fatalf("aborting anti entropy on a new cluster blocked")
|
|
}
|
|
})
|
|
|
|
t.Run("AbortBlocksInitialized", func(t *testing.T) {
|
|
c := newCluster()
|
|
c.initializeAntiEntropy()
|
|
|
|
ch := make(chan struct{})
|
|
go func() {
|
|
c.abortAntiEntropy()
|
|
close(ch)
|
|
}()
|
|
defer c.abortAntiEntropyQ() // avoid leak of goroutine.
|
|
select {
|
|
case <-ch:
|
|
t.Fatalf("aborting anti entropy on an initialized cluster didn't block")
|
|
case <-time.After(time.Microsecond * 100):
|
|
}
|
|
})
|
|
|
|
t.Run("AbortAntiEntropyQ", func(t *testing.T) {
|
|
c := newCluster()
|
|
c.initializeAntiEntropy()
|
|
if c.abortAntiEntropyQ() {
|
|
t.Fatalf("abortAntiEntropyQ should report false when abort not called")
|
|
}
|
|
go func() {
|
|
for {
|
|
if c.abortAntiEntropyQ() {
|
|
break
|
|
}
|
|
}
|
|
}()
|
|
ch := make(chan struct{})
|
|
go func() {
|
|
c.abortAntiEntropy()
|
|
close(ch)
|
|
}()
|
|
select {
|
|
case <-ch:
|
|
case <-time.After(time.Second):
|
|
t.Fatalf("abort should not have blocked this long")
|
|
}
|
|
})
|
|
}
|
|
|
|
func TestTranslateIndexKey(t *testing.T) {
|
|
c := newCluster()
|
|
node0 := &topology.Node{ID: "node0"}
|
|
c.addNodeBasicSorted(node0)
|
|
c.holder = newHolderWithTempPath(t, "rbf")
|
|
_, e := c.translateIndexKey(context.Background(), "i", "a", false)
|
|
if e == nil {
|
|
t.Fatal("expecting error")
|
|
}
|
|
}
|