Merge pull request #112 from benbjohnson/import-cache-2

Update cache on Import (alternative)

bitmap.go

@@ -127,7 +127,7 @@ func (b *Bitmap) SetBit(i uint64) (changed bool) {
 
 // ClearBit clears the i-th bit of the bitmap.
 func (b *Bitmap) ClearBit(i uint64) (changed bool) {
-	return b.createSegmentIfNotExists(i / SliceWidth).SetBit(i)
+	return b.createSegmentIfNotExists(i / SliceWidth).ClearBit(i)
 }
 
 func (b *Bitmap) createSegmentIfNotExists(slice uint64) *BitmapSegment {

fragment.go

@@ -871,11 +871,10 @@ func (f *Fragment) Import(bitmapIDs, profileIDs []uint64) error {
 	// Process every bit.
 	// If an error occurs then reopen the storage.
 	if err := func() error {
-		lastID := uint64(0)
-		bmCounter := uint64(0)
-		var bitmap *Bitmap
+		set := make(map[uint64]struct{})
 		for i := range bitmapIDs {
 			bitmapID, profileID := bitmapIDs[i], profileIDs[i]
+
 			// Determine the position of the bit in the storage.
 			pos, err := f.pos(bitmapID, profileID)
 			if err != nil {
@@ -883,29 +882,25 @@ func (f *Fragment) Import(bitmapIDs, profileIDs []uint64) error {
 			}
 
 			// Write to storage.
-			if _, err := f.storage.Add(pos); err != nil {
+			changed, err := f.storage.Add(pos)
+			if err != nil {
 				return err
 			}
 
-			// import optimization to avoid linear foreach calls
-			// slight risk of concurrent cache counter being off but
-			// no real danger
-			if i == 0 || bitmapID != lastID {
-				bitmap = f.bitmap(bitmapID)
-				if i != 0 {
-					f.cache.Add(lastID, bmCounter)
-				}
-				bmCounter = bitmap.Count()
-				lastID = bitmapID
-			}
-			if bitmap.SetBit(profileID) {
-				bmCounter += 1
+			// Mark bitmap to be updated in cache.
+			if changed {
+				set[bitmapID] = struct{}{}
 			}
 
 			// Invalidate block checksum.
 			delete(f.checksums, int(bitmapID/HashBlockSize))
 		}
 
+		// Update cache counts for all bitmaps.
+		for bitmapID := range set {
+			f.cache.Add(bitmapID, f.bitmap(bitmapID).Count())
+		}
+
 		f.cache.Invalidate()
 		return nil
 	}(); err != nil {
@@ -1245,13 +1240,17 @@ func (s *FragmentSyncer) isClosing() bool {
 func (s *FragmentSyncer) SyncFragment() error {
 	// Determine replica set.
 	nodes := s.Cluster.FragmentNodes(s.Fragment.DB(), s.Fragment.Slice())
-
+	if len(nodes) == 1 {
+		//fmt.Println("no place to replicate", s.Fragment.DB(), s.Fragment.Frame(), s.Fragment.Slice())
+		return nil
+	}
 	// Create a set of blocks.
 	blockSets := make([][]FragmentBlock, 0, len(nodes))
 	for _, node := range nodes {
 		// Read local blocks.
 		if node.Host == s.Host {
-			blockSets = append(blockSets, s.Fragment.Blocks())
+			b := s.Fragment.Blocks()
+			blockSets = append(blockSets, b)
 			continue
 		}