Merge pull request #95 from benbjohnson/sync-closing

Check for early close during AAE

fragment.go

@@ -1170,6 +1170,18 @@ type FragmentSyncer struct {
 
 	Host    string
 	Cluster *Cluster
+
+	Closing <-chan struct{}
+}
+
+// isClosing returns true if the closing channel is closed.
+func (s *FragmentSyncer) isClosing() bool {
+	select {
+	case <-s.Closing:
+		return true
+	default:
+		return false
+	}
 }
 
 // SyncFragment compares checksums for the local and remote fragments and
@@ -1197,6 +1209,11 @@ func (s *FragmentSyncer) SyncFragment() error {
 			return err
 		}
 		blockSets = append(blockSets, blocks)
+
+		// Verify sync is not prematurely closing.
+		if s.isClosing() {
+			return nil
+		}
 	}
 
 	// Iterate over all blocks and find differences.
@@ -1257,6 +1274,11 @@ func (s *FragmentSyncer) syncBlock(id int) error {
 			continue
 		}
 
+		// Verify sync is not prematurely closing.
+		if s.isClosing() {
+			return nil
+		}
+
 		client, err := NewClient(node.Host)
 		if err != nil {
 			return err
@@ -1274,6 +1296,11 @@ func (s *FragmentSyncer) syncBlock(id int) error {
 		})
 	}
 
+	// Verify sync is not prematurely closing.
+	if s.isClosing() {
+		return nil
+	}
+
 	// Merge blocks together.
 	sets, clears, err := f.MergeBlock(id, pairSets)
 	if err != nil {
@@ -1298,6 +1325,11 @@ func (s *FragmentSyncer) syncBlock(id int) error {
 			fmt.Fprintf(&buf, "ClearBit(frame=%q, id=%d, profileID=%d)\n", f.Frame(), clear.BitmapIDs[j], (f.Slice()*SliceWidth)+clear.ProfileIDs[j])
 		}
 
+		// Verify sync is not prematurely closing.
+		if s.isClosing() {
+			return nil
+		}
+
 		// Execute query.
 		_, err := clients[i].ExecuteQuery(f.DB(), buf.String(), false)
 		if err != nil {

index.go

@@ -203,6 +203,9 @@ type IndexSyncer struct {
 
 	Host    string
 	Cluster *Cluster
+
+	// Signals that the sync should stop.
+	Closing <-chan struct{}
 }
 
 // SyncIndex compares the index on host with the local index and resolves differences.
@@ -218,6 +221,13 @@ func (s *IndexSyncer) SyncIndex() error {
 					continue
 				}
 
+				// Verify syncer has not closed.
+				select {
+				case <-s.Closing:
+					return nil
+				default:
+				}
+
 				// Sync fragment if own it.
 				if err := s.syncFragment(di.Name, fi.Name, slice); err != nil {
 					return fmt.Errorf("sync error: db=%s, frame=%s, slice=%d, err=%s", di.Name, fi.Name, slice, err)
@@ -242,9 +252,11 @@ func (s *IndexSyncer) syncFragment(db, frame string, slice uint64) error {
 		Fragment: f,
 		Host:     s.Host,
 		Cluster:  s.Cluster,
+		Closing:  s.Closing,
 	}
 	if err := fs.SyncFragment(); err != nil {
 		return err
 	}
+
 	return nil
 }

server.go

@@ -167,6 +167,7 @@ func (s *Server) monitorAntiEntropy() {
 		syncer.Index = s.Index
 		syncer.Host = s.Host
 		syncer.Cluster = s.Cluster
+		syncer.Closing = s.closing
 
 		// Sync indexes.
 		if err := syncer.SyncIndex(); err != nil {