Merge pull request #215 from seebs/mmap-v-cache

clear container lookup cache when updating every container, handle nils with differenceInPlace, use transaction/ops log for mergeBlock.

fragment.go

@@ -1862,21 +1862,20 @@ func (f *fragment) mergeBlock(id int, data []pairSet) (sets, clears []pairSet, e
 		}
 	}
 
-	// Set local bits.
+	rowSet := make(map[uint64]struct{}, len(sets[0].columnIDs))
+	// compute positions directly, replacing columnIDs with the computed
+	// positions
 	for i := range sets[0].columnIDs {
-		if _, err := f.unprotectedSetBit(sets[0].rowIDs[i], (f.shard*ShardWidth)+sets[0].columnIDs[i]); err != nil {
-			return nil, nil, errors.Wrap(err, "setting")
-		}
+		rowSet[sets[0].rowIDs[i]] = struct{}{}
+		sets[0].columnIDs[i] += sets[0].rowIDs[i] * ShardWidth
 	}
-
-	// Clear local bits.
 	for i := range clears[0].columnIDs {
-		if _, err := f.unprotectedClearBit(clears[0].rowIDs[i], (f.shard*ShardWidth)+clears[0].columnIDs[i]); err != nil {
-			return nil, nil, errors.Wrap(err, "clearing")
-		}
+		rowSet[clears[0].rowIDs[i]] = struct{}{}
+		clears[0].columnIDs[i] += clears[0].rowIDs[i] * ShardWidth
 	}
+	err = f.importPositions(sets[0].columnIDs, clears[0].columnIDs, rowSet)
 
-	return sets[1:], clears[1:], nil
+	return sets[1:], clears[1:], err
 }
 
 // bulkImport bulk imports a set of bits and then snapshots the storage.

fragment_internal_test.go

@@ -25,6 +25,8 @@ import (
 	"math/rand"
 	"os"
 	"reflect"
+	"runtime"
+	"runtime/debug"
 	"sort"
 	"sync/atomic"
 	"testing"
@@ -3583,6 +3585,60 @@ func TestFragmentConcurrentReadWrite(t *testing.T) {
 	t.Logf("%d", acc)
 }
 
+func TestRemapCache(t *testing.T) {
+	f := mustOpenFragment("i", "f", viewStandard, 0, "")
+	// request a panic that doesn't kill the program on fault
+	wouldFault := debug.SetPanicOnFault(true)
+	defer func() {
+		debug.SetPanicOnFault(wouldFault)
+		if r := recover(); r != nil {
+			if err, ok := r.(error); ok {
+				// special case: if we caught a page fault, we diagnose that directly. sadly,
+				// we can't see the actual values that were used to generate this, probably.
+				if err.Error() == "runtime error: invalid memory address or nil pointer dereference" {
+					t.Fatalf("segfault trapped during remap test (expected failure mode)")
+				}
+			}
+			t.Fatalf("unexpected panic: %v", r)
+		}
+	}()
+	// create a container
+	_, err := f.storage.Add(65537)
+	if err != nil {
+		t.Fatalf("storage add: %v", err)
+	}
+	// cause the container to be mapped
+	err = f.Snapshot()
+	if err != nil {
+		t.Fatalf("storage snapshot: %v", err)
+	}
+	// freeze the row
+	_ = f.row(0)
+	// add a bit that isn't in that container, so that container doesn't
+	// change
+	_, err = f.storage.Add(2)
+	if err != nil {
+		t.Fatalf("storage add: %v", err)
+	}
+	// make the original container be the most recent, thus cached, container
+	_, err = f.bit(0, 65537)
+	if err != nil {
+		t.Fatalf("storage bit check: %v", err)
+	}
+	// force snapshot, remapping the containers
+	err = f.Snapshot()
+	if err != nil {
+		t.Fatalf("storage snapshot: %v", err)
+	}
+	// get rid of the old mapping
+	runtime.GC()
+	// try to read that container again
+	_, err = f.bit(0, 65537)
+	if err != nil {
+		t.Fatalf("storage bit check: %v", err)
+	}
+}
+
 func TestFragment_Bug_Q2DoubleDelete(t *testing.T) {
 	f := mustOpenFragment("i", "f", viewStandard, 0, "")
 	b := []byte{60, 48, 0, 0, 1, 0, 0, 0, 16, 0, 0, 0, 0, 0, 0, 0, 1, 0, 0, 0, 24, 0, 0, 0, 1, 0}

roaring/containers_btree.go

@@ -62,12 +62,6 @@ func (btc *bTreeContainers) Put(key uint64, c *Container) {
 	// Get can result in the tree containing a different container
 	// than we'll get on next lookup.
 	btc.lastKey, btc.lastContainer = key, c
-	// If a mapped container is added to the tree, reset the
-	// lastContainer cache so that the cache is not pointing
-	// at a read-only mmap.
-	if c.Mapped() {
-		btc.lastKey = ^uint64(0)
-	}
 	btc.tree.Set(key, c)
 }
 
@@ -119,7 +113,6 @@ func (btc *bTreeContainers) GetOrCreate(key uint64) *Container {
 		btc.lastContainer = cont
 		return cont
 	}
-
 	btc.lastContainer = v
 	return btc.lastContainer
 }
@@ -229,6 +222,9 @@ func (btc *bTreeContainers) UpdateEvery(fn func(uint64, *Container, bool) (*Cont
 	// currently not handling the error from this, but in practice it has
 	// to be io.EOF.
 	_ = e.Every(fn)
+	// invalidate cache.
+	btc.lastKey = ^uint64(0)
+	btc.lastContainer = nil
 }
 
 type btcIterator struct {

roaring/containers_slice.go

@@ -43,7 +43,8 @@ func (sc *sliceContainers) Put(key uint64, c *Container) {
 	} else {
 		sc.containers[i] = c
 	}
-
+	sc.lastKey = key
+	sc.lastContainer = c
 }
 
 func (sc *sliceContainers) PutContainerValues(key uint64, typ byte, n int, mapped bool) {
@@ -159,7 +160,7 @@ func (sc *sliceContainers) Reset() {
 	sc.keys = sc.keys[:0]
 	sc.containers = sc.containers[:0]
 	sc.lastContainer = nil
-	sc.lastKey = 0
+	sc.lastKey = ^uint64(0)
 }
 
 func (sc *sliceContainers) ResetN(n int) {
@@ -171,7 +172,7 @@ func (sc *sliceContainers) ResetN(n int) {
 		sc.containers = sc.containers[:0]
 	}
 	sc.lastContainer = nil
-	sc.lastKey = 0
+	sc.lastKey = ^uint64(0)
 }
 
 func (sc *sliceContainers) seek(key uint64) (int, bool) {
@@ -227,6 +228,9 @@ func (sc *sliceContainers) UpdateEvery(fn func(uint64, *Container, bool) (*Conta
 			sc.containers[i] = nc
 		}
 	}
+	// invalidate cache.
+	sc.lastKey = ^uint64(0)
+	sc.lastContainer = nil
 }
 
 type sliceIterator struct {

roaring/roaring.go

@@ -5540,6 +5540,10 @@ func (b *Bitmap) DifferenceInPlace(others ...*Bitmap) {
 	// Go through all the containers and remove the other bits
 	for targetItr.Next() {
 		targetKey, curContainer := targetItr.Value()
+		// no point in subtracting things from an empty container.
+		if curContainer.N() == 0 {
+			removeContainerKeys = append(removeContainerKeys, targetKey)
+		}
 		// Loop until every iters current value has been handled.
 		for _, iIter := range bitmapIters {
 			if !iIter.hasNext {
@@ -5554,16 +5558,18 @@ func (b *Bitmap) DifferenceInPlace(others ...*Bitmap) {
 					break
 				}
 			}
-
 			if targetKey == iKey {
-				if curContainer.frozen() {
-					curContainer = curContainer.Clone()
-					b.Containers.Put(targetKey, curContainer)
-				}
-				curContainer.differenceInPlace(iContainer)
-				if curContainer.N() == 0 {
-					removeContainerKeys = append(removeContainerKeys, iKey)
-					break
+				// note: a nil container is valid, and has N == 0.
+				if iContainer.N() != 0 {
+					if curContainer.frozen() {
+						curContainer = curContainer.Clone()
+						b.Containers.Put(targetKey, curContainer)
+					}
+					curContainer.differenceInPlace(iContainer)
+					if curContainer.N() == 0 {
+						removeContainerKeys = append(removeContainerKeys, targetKey)
+						break
+					}
 				}
 				iIter.hasNext = iIter.iter.Next()
 			}
@@ -5578,6 +5584,9 @@ func (b *Bitmap) DifferenceInPlace(others ...*Bitmap) {
 }
 
 func (c *Container) differenceInPlace(other *Container) {
+	if other == nil {
+		return
+	}
 	if other.isArray() {
 		if c.isArray() {
 			differenceArrayArrayInPlace(c, other)