diff --git a/bitmap.go b/bitmap.go index b9a56a802..083ed1529 100644 --- a/bitmap.go +++ b/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 { diff --git a/fragment.go b/fragment.go index 90deb153f..cf55f600d 100644 --- a/fragment.go +++ b/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 }