From 16eff6de8c2b6ff867492cfe29dd1744b5a40cf6 Mon Sep 17 00:00:00 2001 From: Matt Jaffee Date: Wed, 8 Aug 2018 14:41:54 -0500 Subject: [PATCH 1/6] prevent anti entropy and cluster resize from running simultaneously --- cluster.go | 33 +++++++++++++++++++++++++++++++++ holder.go | 5 ++++- server.go | 22 +++++++++++++++++++++- 3 files changed, 58 insertions(+), 2 deletions(-) diff --git a/cluster.go b/cluster.go index 309846e0b..689706d75 100644 --- a/cluster.go +++ b/cluster.go @@ -204,6 +204,8 @@ type cluster struct { // nolint: maligned joining chan struct{} joined bool + abortAntiEntropyCh chan struct{} + mu sync.RWMutex jobs map[int64]*resizeJob currentJob *resizeJob @@ -235,6 +237,33 @@ func newCluster() *cluster { } } +// initializeAntiEntropy is called by the anti entropy routine when it starts. +// If the AE channel is created without a routine reading from it, cluster will +// block indefinitely when calling abortAntiEntropy(). +func (c *cluster) initializeAntiEntropy() { + c.mu.Lock() + c.abortAntiEntropyCh = make(chan struct{}) + c.mu.Unlock() +} + +// abortAntiEntropyQ checks whether the cluster wants to abort the anti entropy +// process (so that it can resize). It does not block. +func (c *cluster) abortAntiEntropyQ() bool { + select { + case <-c.abortAntiEntropyCh: + return true + default: + return false + } +} + +// abortAntiEntropy blocks until the anti-entropy routine calls abortAntiEntropyQ +func (c *cluster) abortAntiEntropy() { + if c.abortAntiEntropyCh != nil { + c.abortAntiEntropyCh <- struct{}{} + } +} + func (c *cluster) coordinatorNode() *Node { c.mu.RLock() defer c.mu.RUnlock() @@ -405,6 +434,10 @@ func (c *cluster) unprotectedSetState(state string) { c.state = state + if state == ClusterStateResizing { + c.abortAntiEntropy() + } + // TODO: consider NOT running cleanup on an active node that has // been removed. // It's safe to do a cleanup after state changes back to normal. diff --git a/holder.go b/holder.go index c2e8f5a6e..cf895e802 100644 --- a/holder.go +++ b/holder.go @@ -577,8 +577,11 @@ type holderSyncer struct { Closing <-chan struct{} } -// IsClosing returns true if the syncer has been marked to close. +// IsClosing returns true if the syncer has been asked to close. func (s *holderSyncer) IsClosing() bool { + if s.Cluster.abortAntiEntropyQ() { + return true + } select { case <-s.Closing: return true diff --git a/server.go b/server.go index 2cc4accdc..d120fcc09 100644 --- a/server.go +++ b/server.go @@ -417,6 +417,8 @@ func (s *Server) monitorAntiEntropy() { if s.antiEntropyInterval == 0 { return // anti entropy disabled } + s.cluster.initializeAntiEntropy() + ticker := time.NewTicker(s.antiEntropyInterval) defer ticker.Stop() @@ -428,11 +430,17 @@ func (s *Server) monitorAntiEntropy() { select { case <-s.closing: return + case <-s.cluster.abortAntiEntropyCh: // receive here so we don't block resizing + continue case <-ticker.C: s.holder.Stats.Count("AntiEntropy", 1, 1.0) } t := time.Now() - + if s.cluster.State() == ClusterStateResizing { + continue // don't launch anti-entropy during resize. + // the cluster sets its state to resizing and *then* sends to + // abortAntiEntropyCh before starting to resize + } // Sync holders. s.logger.Printf("holder sync beginning") if err := s.syncer.SyncHolder(); err != nil { @@ -444,6 +452,18 @@ func (s *Server) monitorAntiEntropy() { s.logger.Printf("holder sync complete") dif := time.Since(t) s.holder.Stats.Histogram("AntiEntropyDuration", float64(dif), 1.0) + + // Drain tick channel since we just finished anti-entropy. If the AE + // process took a long time, we don't want them to pile up on each + // other. + for { + select { + case <-ticker.C: + continue + default: + } + break + } } } From 0e467e54927b8c810475cf8e2de327fa42c27163 Mon Sep 17 00:00:00 2001 From: Matt Jaffee Date: Wed, 8 Aug 2018 15:11:39 -0500 Subject: [PATCH 2/6] rename cluster.nodes and fix race in API --- api.go | 2 +- cluster.go | 62 +++++++++++++++++++++++----------------- cluster_internal_test.go | 13 +++++---- executor.go | 8 +++--- holder.go | 4 +-- holder_internal_test.go | 6 ++-- server.go | 4 +-- utils_internal_test.go | 6 ++-- 8 files changed, 58 insertions(+), 47 deletions(-) diff --git a/api.go b/api.go index 847cc7ebd..bc71cc5a2 100644 --- a/api.go +++ b/api.go @@ -428,7 +428,7 @@ func (api *API) FragmentBlocks(_ context.Context, indexName string, fieldName st // Hosts returns a list of the hosts in the cluster including their ID, // URL, and which is the coordinator. func (api *API) Hosts(_ context.Context) []*Node { - return api.cluster.Nodes + return api.cluster.Nodes() } // Node gets the ID, URI and coordinator status for this particular node. diff --git a/cluster.go b/cluster.go index 689706d75..84f648d91 100644 --- a/cluster.go +++ b/cluster.go @@ -169,7 +169,7 @@ type nodeAction struct { type cluster struct { // nolint: maligned id string Node *Node - Nodes []*Node // TODO phase this out? + nodes []*Node // TODO phase this out? // Hashing algorithm used to assign partitions to nodes. Hasher Hasher @@ -331,7 +331,7 @@ func (c *cluster) unprotectedUpdateCoordinator(n *Node) bool { c.Coordinator = n.ID changed = true } - for _, node := range c.Nodes { + for _, node := range c.nodes { if node.ID == n.ID { node.IsCoordinator = true } else { @@ -388,7 +388,7 @@ func (c *cluster) removeNode(nodeID string) error { // nodeIDs returns the list of IDs in the cluster. func (c *cluster) nodeIDs() []string { - return Nodes(c.Nodes).IDs() + return Nodes(c.nodes).IDs() } func (c *cluster) unprotectedSetID(id string) { @@ -522,7 +522,7 @@ func (c *cluster) unprotectedStatus() *ClusterStatus { return &ClusterStatus{ ClusterID: c.id, State: c.state, - Nodes: c.Nodes, + Nodes: c.nodes, } } @@ -534,7 +534,7 @@ func (c *cluster) nodeByID(id string) *Node { // unprotectedNodeByID returns a node reference by ID. func (c *cluster) unprotectedNodeByID(id string) *Node { - for _, n := range c.Nodes { + for _, n := range c.nodes { if n.ID == id { return n } @@ -555,7 +555,7 @@ func (c *cluster) topologyContainsNode(id string) bool { // nodePositionByID returns the position of the node in slice c.Nodes. func (c *cluster) nodePositionByID(nodeID string) int { - for i, n := range c.Nodes { + for i, n := range c.nodes { if n.ID == nodeID { return i } @@ -571,14 +571,24 @@ func (c *cluster) addNodeBasicSorted(node *Node) bool { return false } - c.Nodes = append(c.Nodes, node) + c.nodes = append(c.nodes, node) // All hosts must be merged in the same order on all nodes in the cluster. - sort.Sort(byID(c.Nodes)) + sort.Sort(byID(c.nodes)) return true } +// Nodes returns a copy of the slice of nodes in the cluster. Safe for +// concurrent use, result may be modified. +func (c *cluster) Nodes() []*Node { + c.mu.Lock() + defer c.mu.Unlock() + ret := make([]*Node, len(c.nodes)) + copy(ret, c.nodes) + return ret +} + // removeNodeBasicSorted removes a node from the cluster, maintaining the sort // order. Returns true if the node was removed. unprotected. func (c *cluster) removeNodeBasicSorted(nodeID string) bool { @@ -587,9 +597,9 @@ func (c *cluster) removeNodeBasicSorted(nodeID string) bool { return false } - copy(c.Nodes[i:], c.Nodes[i+1:]) - c.Nodes[len(c.Nodes)-1] = nil - c.Nodes = c.Nodes[:len(c.Nodes)-1] + copy(c.nodes[i:], c.nodes[i+1:]) + c.nodes[len(c.nodes)-1] = nil + c.nodes = c.nodes[:len(c.nodes)-1] return true } @@ -663,8 +673,8 @@ func (c *cluster) fragCombos(idx string, maxShard uint64, fieldViews viewsByFiel // added or removed. An error is returned for any case other than where // exactly one node is added or removed. unprotected. func (c *cluster) diff(other *cluster) (action string, nodeID string, err error) { - lenFrom := len(c.Nodes) - lenTo := len(other.Nodes) + lenFrom := len(c.nodes) + lenTo := len(other.nodes) // Determine if a node is being added or removed. if lenFrom == lenTo { return "", "", errors.New("clusters are the same size") @@ -676,7 +686,7 @@ func (c *cluster) diff(other *cluster) (action string, nodeID string, err error) } action = resizeJobActionAdd // Determine the node ID that is being added. - for _, n := range other.Nodes { + for _, n := range other.nodes { if c.unprotectedNodeByID(n.ID) == nil { nodeID = n.ID break @@ -689,7 +699,7 @@ func (c *cluster) diff(other *cluster) (action string, nodeID string, err error) } action = resizeJobActionRemove // Determine the node ID that is being removed. - for _, n := range c.Nodes { + for _, n := range c.nodes { if other.unprotectedNodeByID(n.ID) == nil { nodeID = n.ID break @@ -711,7 +721,7 @@ func (c *cluster) fragSources(to *cluster, idx *Index) (map[string][]*ResizeSour } // Initialize the map with all the nodes in `to`. - for _, n := range to.Nodes { + for _, n := range to.nodes { m[n.ID] = nil } @@ -724,7 +734,7 @@ func (c *cluster) fragSources(to *cluster, idx *Index) (map[string][]*ResizeSour srcCluster := c if action == resizeJobActionAdd && c.ReplicaN > 1 { srcCluster = newCluster() - srcCluster.Nodes = Nodes(c.Nodes).Clone() + srcCluster.nodes = Nodes(c.nodes).Clone() srcCluster.Hasher = c.Hasher srcCluster.partitionN = c.partitionN srcCluster.ReplicaN = 1 @@ -814,19 +824,19 @@ func (c *cluster) partitionNodes(partitionID int) []*Node { // Default replica count to between one and the number of nodes. // The replica count can be zero if there are no nodes. replicaN := c.ReplicaN - if replicaN > len(c.Nodes) { - replicaN = len(c.Nodes) + if replicaN > len(c.nodes) { + replicaN = len(c.nodes) } else if replicaN == 0 { replicaN = 1 } // Determine primary owner node. - nodeIndex := c.Hasher.Hash(uint64(partitionID), len(c.Nodes)) + nodeIndex := c.Hasher.Hash(uint64(partitionID), len(c.nodes)) // Collect nodes around the ring. nodes := make([]*Node, replicaN) for i := 0; i < replicaN; i++ { - nodes[i] = c.Nodes[(nodeIndex+i)%len(c.Nodes)] + nodes[i] = c.nodes[(nodeIndex+i)%len(c.nodes)] } return nodes @@ -1132,12 +1142,12 @@ func (c *cluster) unprotectedGenerateResizeJob(nodeAction nodeAction) (*resizeJo // Broadcaster is associated to the resizeJob here for use in broadcasting // the resize instructions to other nodes in the cluster. func (c *cluster) unprotectedGenerateResizeJobByAction(nodeAction nodeAction) (*resizeJob, error) { - j := newResizeJob(c.Nodes, nodeAction.node, nodeAction.action) + j := newResizeJob(c.nodes, nodeAction.node, nodeAction.action) j.Broadcaster = c.broadcaster // toCluster is a clone of Cluster with the new node added/removed for comparison. toCluster := newCluster() - toCluster.Nodes = Nodes(c.Nodes).Clone() + toCluster.nodes = Nodes(c.nodes).Clone() toCluster.Hasher = c.Hasher toCluster.partitionN = c.partitionN toCluster.ReplicaN = c.ReplicaN @@ -1150,7 +1160,7 @@ func (c *cluster) unprotectedGenerateResizeJobByAction(nodeAction nodeAction) (* // multiIndex is a map of sources initialized with all the nodes in toCluster. multiIndex := make(map[string][]*ResizeSource) - for _, n := range toCluster.Nodes { + for _, n := range toCluster.nodes { multiIndex[n.ID] = nil } @@ -1805,7 +1815,7 @@ func (c *cluster) mergeClusterStatus(cs *ClusterStatus) error { // except for self. Generate a list to remove first // so that nodes aren't removed mid-loop. nodeIDsToRemove := []string{} - for _, node := range c.Nodes { + for _, node := range c.nodes { // Don't remove this node. if node.ID == c.Node.ID { continue @@ -1839,7 +1849,7 @@ func (c *cluster) setStatic(hosts []string) error { if err != nil { return errors.Wrap(err, "getting URI") } - c.Nodes = append(c.Nodes, &Node{URI: *uri}) + c.nodes = append(c.nodes, &Node{URI: *uri}) } return nil } diff --git a/cluster_internal_test.go b/cluster_internal_test.go index 07a6521d4..0d9603031 100644 --- a/cluster_internal_test.go +++ b/cluster_internal_test.go @@ -22,6 +22,7 @@ import ( "strings" "testing" "testing/quick" + "time" "github.com/davecgh/go-spew/spew" "github.com/pkg/errors" @@ -316,7 +317,7 @@ func TestResizeJob(t *testing.T) { // Ensure the cluster can fairly distribute partitions across the nodes. func TestCluster_Owners(t *testing.T) { c := cluster{ - Nodes: []*Node{ + nodes: []*Node{ {URI: NewTestURIFromHostPort("serverA", 1000)}, {URI: NewTestURIFromHostPort("serverB", 1000)}, {URI: NewTestURIFromHostPort("serverC", 1000)}, @@ -326,12 +327,12 @@ func TestCluster_Owners(t *testing.T) { } // Verify nodes are distributed. - if a := c.partitionNodes(0); !reflect.DeepEqual(a, []*Node{c.Nodes[0], c.Nodes[1]}) { + if a := c.partitionNodes(0); !reflect.DeepEqual(a, []*Node{c.nodes[0], c.nodes[1]}) { t.Fatalf("unexpected owners: %s", spew.Sdump(a)) } // Verify nodes go around the ring. - if a := c.partitionNodes(2); !reflect.DeepEqual(a, []*Node{c.Nodes[2], c.Nodes[0]}) { + if a := c.partitionNodes(2); !reflect.DeepEqual(a, []*Node{c.nodes[2], c.nodes[0]}) { t.Fatalf("unexpected owners: %s", spew.Sdump(a)) } } @@ -384,7 +385,7 @@ func TestHasher(t *testing.T) { func TestCluster_ContainsShards(t *testing.T) { c := NewTestCluster(5) c.ReplicaN = 3 - shards := c.containsShards("test", 10, c.Nodes[2]) + shards := c.containsShards("test", 10, c.nodes[2]) if !reflect.DeepEqual(shards, []uint64{0, 2, 3, 5, 6, 9, 10}) { t.Fatalf("unexpected shars for node's index: %v", shards) @@ -755,8 +756,8 @@ func TestCluster_UpdateCoordinator(t *testing.T) { t.Run("UpdateCoordinator", func(t *testing.T) { c := NewTestCluster(2) - oldNode := c.Nodes[0] - newNode := c.Nodes[1] + oldNode := c.nodes[0] + newNode := c.nodes[1] // Update coordinator to the same value. if c.updateCoordinator(oldNode) { diff --git a/executor.go b/executor.go index 10026ae2c..d7e70ce81 100644 --- a/executor.go +++ b/executor.go @@ -1225,7 +1225,7 @@ func (e *executor) executeSetRowAttrs(ctx context.Context, index string, c *pql. } // Execute on remote nodes in parallel. - nodes := Nodes(e.Cluster.Nodes).FilterID(e.Node.ID) + nodes := Nodes(e.Cluster.nodes).FilterID(e.Node.ID) resp := make(chan error, len(nodes)) for _, node := range nodes { go func(node *Node) { @@ -1311,7 +1311,7 @@ func (e *executor) executeBulkSetRowAttrs(ctx context.Context, index string, cal } // Execute on remote nodes in parallel. - nodes := Nodes(e.Cluster.Nodes).FilterID(e.Node.ID) + nodes := Nodes(e.Cluster.nodes).FilterID(e.Node.ID) resp := make(chan error, len(nodes)) for _, node := range nodes { go func(node *Node) { @@ -1360,7 +1360,7 @@ func (e *executor) executeSetColumnAttrs(ctx context.Context, index string, c *p } // Execute on remote nodes in parallel. - nodes := Nodes(e.Cluster.Nodes).FilterID(e.Node.ID) + nodes := Nodes(e.Cluster.nodes).FilterID(e.Node.ID) resp := make(chan error, len(nodes)) for _, node := range nodes { go func(node *Node) { @@ -1431,7 +1431,7 @@ func (e *executor) mapReduce(ctx context.Context, index string, shards []uint64, // processing should be done locally so we start with just the local node. var nodes []*Node if !opt.Remote { - nodes = Nodes(e.Cluster.Nodes).Clone() + nodes = Nodes(e.Cluster.nodes).Clone() } else { nodes = []*Node{e.Cluster.unprotectedNodeByID(e.Node.ID)} } diff --git a/holder.go b/holder.go index cf895e802..3242a714d 100644 --- a/holder.go +++ b/holder.go @@ -669,7 +669,7 @@ func (s *holderSyncer) syncIndex(index string) error { s.Stats.CountWithCustomTags("ColumnAttrStoreBlocks", int64(len(blks)), 1.0, []string{indexTag}) // Sync with every other host. - for _, node := range Nodes(s.Cluster.Nodes).FilterID(s.Node.ID) { + for _, node := range Nodes(s.Cluster.nodes).FilterID(s.Node.ID) { // Retrieve attributes from differing blocks. // Skip update and recomputation if no attributes have changed. m, err := s.Cluster.InternalClient.ColumnAttrDiff(context.Background(), &node.URI, index, blks) @@ -713,7 +713,7 @@ func (s *holderSyncer) syncField(index, name string) error { s.Stats.CountWithCustomTags("RowAttrStoreBlocks", int64(len(blks)), 1.0, []string{indexTag, fieldTag}) // Sync with every other host. - for _, node := range Nodes(s.Cluster.Nodes).FilterID(s.Node.ID) { + for _, node := range Nodes(s.Cluster.nodes).FilterID(s.Node.ID) { // Retrieve attributes from differing blocks. // Skip update and recomputation if no attributes have changed. m, err := s.Cluster.InternalClient.RowAttrDiff(context.Background(), &node.URI, index, name, blks) diff --git a/holder_internal_test.go b/holder_internal_test.go index 566bc71f4..e9ab1977e 100644 --- a/holder_internal_test.go +++ b/holder_internal_test.go @@ -183,7 +183,7 @@ func TestHolderCleaner_CleanHolder(t *testing.T) { // Mock 2-node, fully replicated cluster. cluster.ReplicaN = 2 - cluster.Nodes[0].URI = NewTestURIFromHostPort("localhost", 0) + cluster.nodes[0].URI = NewTestURIFromHostPort("localhost", 0) // Create fields on nodes. for _, hldr := range []*tHolder{hldr0} { @@ -215,7 +215,7 @@ func TestHolderCleaner_CleanHolder(t *testing.T) { // Set up cleaner for replication 2. cleaner2 := holderCleaner{ - Node: cluster.Nodes[0], + Node: cluster.nodes[0], Holder: hldr0.Holder, Cluster: cluster, } @@ -252,7 +252,7 @@ func TestHolderCleaner_CleanHolder(t *testing.T) { // Set up cleaner for replication 1. cleaner1 := holderCleaner{ - Node: cluster.Nodes[0], + Node: cluster.nodes[0], Holder: hldr0.Holder, Cluster: cluster, } diff --git a/server.go b/server.go index d120fcc09..c556f77f1 100644 --- a/server.go +++ b/server.go @@ -562,7 +562,7 @@ func (s *Server) SendSync(m Message) error { return fmt.Errorf("marshaling message: %v", err) } msg = append([]byte{getMessageType(m)}, msg...) - for _, node := range s.cluster.Nodes { + for _, node := range s.cluster.nodes { node := node s.logger.Printf("SendSync to: %s", node.URI) // Don't forward the message to ourselves. @@ -662,7 +662,7 @@ func (s *Server) monitorDiagnostics() { s.diagnostics.SetVersion(Version) s.diagnostics.Set("Host", s.uri.Host) s.diagnostics.Set("Cluster", strings.Join(s.cluster.nodeIDs(), ",")) - s.diagnostics.Set("NumNodes", len(s.cluster.Nodes)) + s.diagnostics.Set("NumNodes", len(s.cluster.nodes)) s.diagnostics.Set("NumCPU", runtime.NumCPU()) s.diagnostics.Set("NodeID", s.nodeID) s.diagnostics.Set("ClusterID", s.cluster.id) diff --git a/utils_internal_test.go b/utils_internal_test.go index 85a225c34..ff149e8cf 100644 --- a/utils_internal_test.go +++ b/utils_internal_test.go @@ -40,14 +40,14 @@ func NewTestCluster(n int) *cluster { c.Topology = newTopology() for i := 0; i < n; i++ { - c.Nodes = append(c.Nodes, &Node{ + c.nodes = append(c.nodes, &Node{ ID: fmt.Sprintf("node%d", i), URI: NewTestURI("http", fmt.Sprintf("host%d", i), uint16(0)), }) } - c.Node = c.Nodes[0] - c.Coordinator = c.Nodes[0].ID + c.Node = c.nodes[0] + c.Coordinator = c.nodes[0].ID c.SetState(ClusterStateNormal) return c From 287ea370fde33d95fb52507db75eb4a421a81600 Mon Sep 17 00:00:00 2001 From: Matt Jaffee Date: Wed, 8 Aug 2018 15:11:52 -0500 Subject: [PATCH 3/6] add cluster tests for anti entropy abort --- cluster_internal_test.go | 58 ++++++++++++++++++++++++++++++++++++++++ 1 file changed, 58 insertions(+) diff --git a/cluster_internal_test.go b/cluster_internal_test.go index 0d9603031..8bfd364bc 100644 --- a/cluster_internal_test.go +++ b/cluster_internal_test.go @@ -751,6 +751,64 @@ func TestCluster_ResizeStates(t *testing.T) { }) } +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) + }() + 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) + }() + select { + case <-ch: + t.Fatalf("aborting anti entropy on an initialized 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") + } + }) + +} + // Ensures that coordinator can be changed. func TestCluster_UpdateCoordinator(t *testing.T) { t.Run("UpdateCoordinator", func(t *testing.T) { From b761121f47d435ae6083cbf6962d53d01cbe2eab Mon Sep 17 00:00:00 2001 From: Matt Jaffee Date: Thu, 9 Aug 2018 15:54:31 -0500 Subject: [PATCH 4/6] test antiEntropy set to 0 works as expected --- server_internal_test.go | 27 +++++++++++++++++++++++++++ 1 file changed, 27 insertions(+) diff --git a/server_internal_test.go b/server_internal_test.go index e64f80a0f..e22681764 100644 --- a/server_internal_test.go +++ b/server_internal_test.go @@ -15,8 +15,10 @@ package pilosa import ( + "io/ioutil" "runtime" "testing" + "time" ) // Ensure the file handle count is working @@ -33,3 +35,28 @@ func TestCountOpenFiles(t *testing.T) { t.Error("countOpenFiles returned invalid value 0.") } } + +func TestMonitorAntiEntropyZero(t *testing.T) { + + td, err := ioutil.TempDir("", "") + if err != nil { + t.Fatalf("getting temp dir: %v", err) + } + s, err := NewServer(OptServerDataDir(td), + OptServerAntiEntropyInterval(0)) + if err != nil { + t.Fatalf("making new server: %v", err) + } + + ch := make(chan struct{}) + go func() { + s.monitorAntiEntropy() + close(ch) + }() + + select { + case <-ch: + case <-time.After(time.Second): + t.Fatalf("monitorAntiEntropy should have returned immediately with duration 0") + } +} From ab63c8e7b6894b34121eaa274041f5fcfdc5eaae Mon Sep 17 00:00:00 2001 From: Matt Jaffee Date: Thu, 9 Aug 2018 16:07:05 -0500 Subject: [PATCH 5/6] add forgotten server changes which support AE test --- server.go | 14 +++++++++----- 1 file changed, 9 insertions(+), 5 deletions(-) diff --git a/server.go b/server.go index c556f77f1..819d1d80d 100644 --- a/server.go +++ b/server.go @@ -232,11 +232,12 @@ func OptServerClusterHasher(h Hasher) ServerOption { // NewServer returns a new instance of Server. func NewServer(opts ...ServerOption) (*Server, error) { s := &Server{ - closing: make(chan struct{}), - cluster: newCluster(), - holder: NewHolder(), - diagnostics: newDiagnosticsCollector(defaultDiagnosticServer), - systemInfo: newNopSystemInfo(), + closing: make(chan struct{}), + cluster: newCluster(), + holder: NewHolder(), + diagnostics: newDiagnosticsCollector(defaultDiagnosticServer), + systemInfo: newNopSystemInfo(), + defaultClient: nopInternalClient{}, gcNotifier: NopGCNotifier, @@ -246,6 +247,9 @@ func NewServer(opts ...ServerOption) (*Server, error) { logger: NopLogger, } + s.executor = newExecutor(optExecutorInternalQueryClient(s.defaultClient)) + s.cluster.InternalClient = s.defaultClient + s.diagnostics.server = s for _, opt := range opts { From c133c53d0d448ccc4cd313f8cca7541689ac3d5f Mon Sep 17 00:00:00 2001 From: Matt Jaffee Date: Mon, 20 Aug 2018 10:39:41 -0500 Subject: [PATCH 6/6] add missing word in test fail message --- cluster_internal_test.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/cluster_internal_test.go b/cluster_internal_test.go index 757e6c9b4..879ad2f35 100644 --- a/cluster_internal_test.go +++ b/cluster_internal_test.go @@ -831,7 +831,7 @@ func TestAE(t *testing.T) { }() select { case <-ch: - t.Fatalf("aborting anti entropy on an initialized didn't block") + t.Fatalf("aborting anti entropy on an initialized cluster didn't block") case <-time.After(time.Microsecond * 100): } })